From adcd1184985b193c11521f08cac36e6a8ad39543 Mon Sep 17 00:00:00 2001 From: wecom-fullstack Date: Wed, 21 Oct 2026 18:47:19 +0800 Subject: [PATCH] =?UTF-8?q?feat(maps):=20P1=20=E5=9B=9B=E5=B1=82=E5=9B=A0?= =?UTF-8?q?=E6=9E=9C=E7=BB=93=E6=9E=84=E5=8C=96=20map=5Fconnections=20?= =?UTF-8?q?=E8=A1=A8=20+=20=E8=BF=81=E7=A7=BB=E8=84=9A=E6=9C=AC=20+=20anal?= =?UTF-8?q?ysis=20=E5=8F=AA=E8=AF=BB=E7=AB=AF=E7=82=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新表 map_connections（entity_id/map_id/from_node/to_node/from_layer/to_layer/ from_objective_id/to_objective_id/effect/label/sort_order），唯一键 (map_id,from,to) 保幂等 - 节点双格式兼容：层:中文目标名（生产存量 31 条）与 层-序号（前端现用） - canvas_data 保留不动（双写期）；POST/DELETE 写表 + 同步 JSON，失败仅告警不阻断 - GET /connections 表优先、表空回落 canvas_data；新增 GET /connections/analysis （层统计/跨层分布/断链/反向告警不拦截） - scripts/migrate_map_connections.py 四态（check/apply/idem/rollback）+ canvas 快照回滚点 - tests: 9 条（双格式解析、双写、回落、删除同步、分析统计与告警） --- backend/app/api/maps.py | 294 +++++++++++- backend/app/models/__init__.py | 31 ++ backend/scripts/migrate_map_connections.py | 524 +++++++++++++++++++++ backend/tests/test_map_connections_p1.py | 166 +++++++ 4 files changed, 1005 insertions(+), 10 deletions(-) create mode 100644 backend/scripts/migrate_map_connections.py create mode 100644 backend/tests/test_map_connections_p1.py diff --git a/backend/app/api/maps.py b/backend/app/api/maps.py index a8ec85e2..ee426c9e 100644 --- a/backend/app/api/maps.py +++ b/backend/app/api/maps.py @@ -7,12 +7,13 @@ from datetime import date, datetime, timedelta from app.database import get_db from app.deps import get_entity_id from app.auth_middleware import require_auth, require_role -from app.models import StrategicMap, OperationLog, MapObjective, Objective, KR, KPIDefinition, ReviewRecord, User +from app.models import StrategicMap, OperationLog, MapObjective, Objective, KR, KPIDefinition, ReviewRecord, User, MapConnection from app.utils.kr_name import sanitize_kr_name from app.utils.kpi_value_scope import actual_only_clause import copy import json import logging +import re logger = logging.getLogger(__name__) @@ -507,7 +508,165 @@ def batch_delete_maps(data: dict, db: Session = Depends(get_db), entity_id: int return {"message": f"已删除 {deleted} 个地图", "deleted": deleted} -# ── 连线管理 ───────────────────────────────── +# ── 连线管理（P1 四层因果结构化：map_connections 表 ⇄ canvas_data 双写）── +# +# 2026-10-21 cma-map-connections-table-p1： +# · 生产存量节点标识是 `层:中文目标名`（如 `learning:贡献率思维全员培训`）， +# 当前前端 MapCanvas 生成的是 `层-序号`（如 `learning-0`）⇒ 两种格式都必须能解析； +# · 表是新增读模型（归因/校验用），canvas_data 保持原样供前端渲染 ⇒ 双写期； +# · 表缺失时（迁移前/回滚后）所有写路径降级为只写 canvas_data，不阻断前端。 + +LAYER_ORDER = ("learning", "process", "customer", "finance") +_LAYER_INDEX = {k: i for i, k in enumerate(LAYER_ORDER)} +_NODE_COLON = re.compile(r"^([a-z_]+):(.+)$") +_NODE_INDEX = re.compile(r"^([a-z_]+)-(\d+)$") + + +def _norm_obj_name(s) -> str: + return re.sub(r"\s+", " ", str(s or "").strip()) + + +def parse_node_ref(node): + """节点标识解析（双格式兼容）→ dict(layer, kind, value, error) + + `learning:贡献率思维全员培训` → layer=learning, kind=name, value=中文名 + `learning-0` → layer=learning, kind=index, value=0 + """ + if not isinstance(node, str) or not node.strip(): + return {"layer": None, "kind": None, "value": None, "error": "node_empty"} + s = node.strip() + m = _NODE_COLON.match(s) + if m: + layer, name = m.group(1), m.group(2).strip() + if layer not in _LAYER_INDEX: + return {"layer": None, "kind": None, "value": None, "error": f"unknown_layer:{layer}"} + if not name: + return {"layer": layer, "kind": None, "value": None, "error": "node_name_empty"} + return {"layer": layer, "kind": "name", "value": name, "error": None} + m = _NODE_INDEX.match(s) + if m: + layer, idx = m.group(1), int(m.group(2)) + if layer not in _LAYER_INDEX: + return {"layer": None, "kind": None, "value": None, "error": f"unknown_layer:{layer}"} + return {"layer": layer, "kind": "index", "value": idx, "error": None} + return {"layer": None, "kind": None, "value": None, "error": "node_format_unrecognized"} + + +def _objective_index(db: Session, m: StrategicMap): + """(名字索引, 层内顺序索引) —— 名字索引键 = (dimension_key, 归一化名称)。""" + rows = db.query(MapObjective).filter(MapObjective.map_id == m.id).order_by( + MapObjective.sort_order, MapObjective.id).all() + by_name, by_layer = {}, {} + for o in rows: + by_name[(o.dimension_key, _norm_obj_name(o.name))] = o.id + by_layer.setdefault(o.dimension_key, []).append(o) + return by_name, by_layer + + +def resolve_node(db: Session, m: StrategicMap, node): + """→ (layer, objective_id|None, note|None) + + `层-序号` 的序号语义 = 前端 MapCanvas 使用的 `dimensions[层].objectives` 数组下标； + dimensions 缺失时退化到 map_objectives.sort_order / 层内顺序。 + """ + ref = parse_node_ref(node) + layer = ref["layer"] + if ref["error"]: + return layer, None, ref["error"] + by_name, by_layer = _objective_index(db, m) + if ref["kind"] == "name": + want = _norm_obj_name(ref["value"]) + hit = by_name.get((layer, want)) + if hit: + return layer, hit, None + return layer, None, "objective_unresolved" + idx = ref["value"] + dims = _dims_list(m.dimensions) + for d in dims: + if d.get("key") == layer: + objs = d.get("objectives") or [] + if 0 <= idx < len(objs): + name = _norm_obj_name((objs[idx] or {}).get("name")) + hit = by_name.get((layer, name)) + return layer, hit, (None if hit else "objective_unresolved") + return layer, None, "index_out_of_range" + cand = by_layer.get(layer) or [] + for o in cand: + if (o.sort_order or 0) == idx: + return layer, o.id, "resolved_by_sort_order" + if 0 <= idx < len(cand): + return layer, cand[idx].id, "resolved_by_position" + return layer, None, "index_out_of_range" + + +def _row_payload_from_canvas(db: Session, m: StrategicMap, c: dict, sort_order: int) -> dict: + frm, to = c.get("from", ""), c.get("to", "") + fl, f_obj, f_note = resolve_node(db, m, frm) + tl, t_obj, t_note = resolve_node(db, m, to) + return { + "entity_id": m.entity_id if m.entity_id is not None else 1, + "map_id": m.id, "from_node": frm, "to_node": to, + "from_layer": fl, "to_layer": tl, + "from_objective_id": f_obj, "to_objective_id": t_obj, + "effect": c.get("effect") or "positive", + "label": (c.get("label") or None), + "sort_order": sort_order, + } + + +def _table_ready(db: Session) -> bool: + """新表是否可用（迁移前/回滚后为 False）——用 SAVEPOINT 隔离探测，避免污染事务。""" + try: + with db.begin_nested(): + db.query(MapConnection).limit(1).all() + return True + except Exception: + return False + + +def _table_rows(db: Session, m: StrategicMap): + # 账套隔离已在 map 查询处完成；这里只按 map_id 取行，避免历史 map.entity_id 为空时 + # 冗余列与 map 不一致导致「表里有行却读不到」的假回落。 + rows = db.query(MapConnection).filter( + MapConnection.map_id == m.id, + ).order_by(MapConnection.sort_order, MapConnection.id).all() + return rows + + +def _table_upsert(db: Session, m: StrategicMap, payload: dict) -> bool: + """双写：失败只告警、不阻断 canvas_data 主路径（兼容期容错）。""" + try: + with db.begin_nested(): + existing = db.query(MapConnection).filter( + MapConnection.map_id == payload["map_id"], + MapConnection.from_node == payload["from_node"], + MapConnection.to_node == payload["to_node"], + ).first() + if existing: + for k, v in payload.items(): + setattr(existing, k, v) + else: + db.add(MapConnection(**payload)) + db.flush() + return True + except Exception as exc: # 表缺失/授权不足/约束冲突 → 主路径继续 + logger.warning("map_connections 双写失败（canvas_data 已保留）: %s: %s", type(exc).__name__, exc) + return False + + +def _table_delete(db: Session, m: StrategicMap, from_node: str, to_node: str) -> bool: + try: + with db.begin_nested(): + db.query(MapConnection).filter( + MapConnection.map_id == m.id, + MapConnection.from_node == from_node, + MapConnection.to_node == to_node, + ).delete(synchronize_session=False) + return True + except Exception as exc: + logger.warning("map_connections 删除同步失败: %s: %s", type(exc).__name__, exc) + return False + def _get_connections(m: StrategicMap) -> list: if not m.canvas_data: @@ -522,9 +681,122 @@ def _get_connections(m: StrategicMap) -> list: return m.canvas_data["connections"] +@router.get("/{map_id}/connections") +def list_connections(map_id: int, db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)): + """读连线：表优先，表空回落 canvas_data（兼容期）。""" + m = db.query(StrategicMap).filter(StrategicMap.id == map_id, StrategicMap.entity_id == entity_id).first() + if not m: + raise HTTPException(404, "战略地图不存在") + + if _table_ready(db): + rows = _table_rows(db, m) + if rows: + return { + "source": "table", "total": len(rows), + "connections": [{ + "from": r.from_node, "to": r.to_node, "style": "solid", + "effect": r.effect or "positive", "label": r.label or "", + "from_layer": r.from_layer, "to_layer": r.to_layer, + "from_objective_id": r.from_objective_id, "to_objective_id": r.to_objective_id, + } for r in rows], + } + + conns = _get_connections(m) + return { + "source": "canvas_data", "total": len(conns), + "connections": [{ + "from": c.get("from"), "to": c.get("to"), + "style": c.get("style") or "solid", + "effect": c.get("effect") or "positive", + "label": c.get("label") or "", + } for c in conns if isinstance(c, dict)], + } + + +@router.get("/{map_id}/connections/analysis") +def analyze_connections(map_id: int, db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)): + """连线只读分析（归因/复盘用）：层统计 + 跨层分布 + 断链 + 反向告警（不拦截）。 + + 读路径：表内有行 → 读表；否则回落 canvas_data（并对回落数据现算解析结果）。 + """ + m = db.query(StrategicMap).filter(StrategicMap.id == map_id, StrategicMap.entity_id == entity_id).first() + if not m: + raise HTTPException(404, "战略地图不存在") + + source = "canvas_data" + items = [] + table_rows = _table_rows(db, m) if _table_ready(db) else [] + if table_rows: + source = "table" + for r in table_rows: + items.append({ + "from": r.from_node, "to": r.to_node, + "from_layer": r.from_layer, "to_layer": r.to_layer, + "from_objective_id": r.from_objective_id, "to_objective_id": r.to_objective_id, + "effect": r.effect or "positive", "label": r.label, + }) + else: + for i, c in enumerate(_get_connections(m)): + if not isinstance(c, dict): + continue + row = _row_payload_from_canvas(db, m, c, i) + items.append({ + "from": row["from_node"], "to": row["to_node"], + "from_layer": row["from_layer"], "to_layer": row["to_layer"], + "from_objective_id": row["from_objective_id"], + "to_objective_id": row["to_objective_id"], + "effect": row["effect"], "label": row["label"], + }) + + layer_stats = {k: {"as_from": 0, "as_to": 0} for k in LAYER_ORDER} + cross_layer, reversed_edges, broken, unknown = {}, [], [], [] + same_layer = 0 + for it in items: + fl, tl = it.get("from_layer"), it.get("to_layer") + if fl in layer_stats: + layer_stats[fl]["as_from"] += 1 + if tl in layer_stats: + layer_stats[tl]["as_to"] += 1 + if fl not in _LAYER_INDEX or tl not in _LAYER_INDEX: + unknown.append({"from": it["from"], "to": it["to"], + "from_layer": fl, "to_layer": tl}) + else: + pair = f"{fl}->{tl}" + cross_layer[pair] = cross_layer.get(pair, 0) + 1 + if fl == tl: + same_layer += 1 + elif _LAYER_INDEX[tl] < _LAYER_INDEX[fl]: + reversed_edges.append({"from": it["from"], "to": it["to"], + "from_layer": fl, "to_layer": tl}) + if not it.get("from_objective_id"): + broken.append({"from": it["from"], "to": it["to"], "side": "from", + "node": it["from"], "reason": parse_node_ref(it["from"]).get("error") or "objective_unresolved"}) + if not it.get("to_objective_id"): + broken.append({"from": it["from"], "to": it["to"], "side": "to", + "node": it["to"], "reason": parse_node_ref(it["to"]).get("error") or "objective_unresolved"}) + + warnings = [] + if reversed_edges: + warnings.append(f"反向连线 {len(reversed_edges)} 条（与 learning→process→customer→finance 相反，" + f"仅告警不拦截）") + if broken: + warnings.append(f"断链 {len(broken)} 处（节点未能解析到 map_objectives）") + if unknown: + warnings.append(f"层前缀无法识别 {len(unknown)} 条") + + return { + "map_id": m.id, "entity_id": m.entity_id, "source": source, + "layer_order": list(LAYER_ORDER), "total": len(items), + "layer_stats": layer_stats, "cross_layer": cross_layer, + "same_layer_count": same_layer, + "reversed": reversed_edges, "broken": broken, + "unknown_layer": unknown, "warnings": warnings, + } + + @router.post("/{map_id}/connections") def add_connection(map_id: int, data: dict, db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)): - """新增因果连线: {"from": "learning-0", "to": "process-0"}""" + """新增因果连线: {"from": "learning-0", "to": "process-0"}（表 + canvas_data 双写）""" m = db.query(StrategicMap).filter(StrategicMap.id == map_id, StrategicMap.entity_id == entity_id).first() if not m: raise HTTPException(404, "战略地图不存在") @@ -539,10 +811,6 @@ def add_connection(map_id: int, data: dict, db: Session = Depends(get_db), entit if from_id == to_id: raise HTTPException(400, "不能自身连线") - # 校验: 维度不能相同 (但放开允许同层连线, 仅禁止自连) - from_dim = from_id.rsplit("-", 1)[0] - to_dim = to_id.rsplit("-", 1)[0] - conns = _get_connections(m) # 校验: 不能重复 @@ -560,14 +828,19 @@ def add_connection(map_id: int, data: dict, db: Session = Depends(get_db), entit m.canvas_data["connections"] = conns from sqlalchemy.orm.attributes import flag_modified flag_modified(m, "canvas_data") # 同上：JSON 列需显式标脏才能落库 + + # 双写（表缺失/失败不阻断前端主路径） + payload = _row_payload_from_canvas(db, m, conns[-1], len(conns) - 1) + db_synced = _table_upsert(db, m, payload) + db.commit() # updated_at: 供前端采纳为新基准（并发保护 base_updated_at），避免自伤 409 - return {"connections": conns, "updated_at": _stamp(m.updated_at)} + return {"connections": conns, "updated_at": _stamp(m.updated_at), "db_synced": db_synced} @router.delete("/{map_id}/connections") def delete_connection_by_key(map_id: int, data: dict, db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)): - """根据 from/to 删除连线""" + """根据 from/to 删除连线（表 + canvas_data 同步删）""" m = db.query(StrategicMap).filter(StrategicMap.id == map_id, StrategicMap.entity_id == entity_id).first() if not m: raise HTTPException(404, "战略地图不存在") @@ -586,9 +859,10 @@ def delete_connection_by_key(map_id: int, data: dict, db: Session = Depends(get_ # 否则接口返回 200 且响应列表已减 1，但 commit 不 flush，重读仍是旧值（删连线变哑弹）。 from sqlalchemy.orm.attributes import flag_modified flag_modified(m, "canvas_data") + db_synced = _table_delete(db, m, from_id, to_id) db.commit() return {"connections": new_conns, "removed": {"from": from_id, "to": to_id}, - "updated_at": _stamp(m.updated_at)} + "updated_at": _stamp(m.updated_at), "db_synced": db_synced} # ── 版本管理 ───────────────────────────────── diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index d3e19c1e..216c393b 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -364,6 +364,37 @@ class MapObjective(Base): created_at = Column(DateTime, server_default=func.now()) updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now()) +class MapConnection(Base): + """战略地图四层因果连线（2026-10-21 P1：canvas_data.connections → 独立表） + + 迁移来源 = `strategic_maps.canvas_data->connections`，双写期表与 JSON 并存 + （JSON 保留供前端渲染；表供归因/校验/分析）。 + + 节点标识双兼容（生产实况两种格式并存）： + - `dim:中文目标名`（生产存量实际格式，如 `learning:贡献率思维全员培训`） + - `dim-序号`（当前前端 MapCanvas 生成的格式，如 `learning-0`） + 原始串原样保存在 from_node/to_node，解析结果落 from_layer/to_layer/from_objective_id。 + """ + __tablename__ = "map_connections" + __table_args__ = ( + UniqueConstraint("map_id", "from_node", "to_node", name="uk_map_conn_edge"), + ) + id = Column(Integer, primary_key=True, index=True) + entity_id = Column(Integer, default=1, index=True, comment="企业ID（冗余，便于按账套直查）") + map_id = Column(Integer, ForeignKey("strategic_maps.id", ondelete="CASCADE"), nullable=False, index=True, comment="关联战略地图") + from_node = Column(String(255), nullable=False, comment="源节点原始标识（原样保留）") + to_node = Column(String(255), nullable=False, comment="目标节点原始标识（原样保留）") + from_layer = Column(String(20), nullable=True, comment="解析层: learning/process/customer/finance") + to_layer = Column(String(20), nullable=True, comment="解析层: learning/process/customer/finance") + from_objective_id = Column(Integer, ForeignKey("map_objectives.id", ondelete="SET NULL"), nullable=True, comment="源目标ID（可解析时回填）") + to_objective_id = Column(Integer, ForeignKey("map_objectives.id", ondelete="SET NULL"), nullable=True, comment="目标目标ID（可解析时回填）") + effect = Column(String(20), default="positive", comment="positive/negative（存量缺省 positive）") + label = Column(String(255), nullable=True, comment="连线标签") + sort_order = Column(Integer, default=0, comment="画布内原始顺序（保持渲染一致）") + created_at = Column(DateTime, server_default=func.now()) + updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now()) + + class KPICausality(Base): """KPI因果链 — 记录KPI间的因果关系""" __tablename__ = "kpi_causality" diff --git a/backend/scripts/migrate_map_connections.py b/backend/scripts/migrate_map_connections.py new file mode 100644 index 00000000..117f1c5b --- /dev/null +++ b/backend/scripts/migrate_map_connections.py @@ -0,0 +1,524 @@ +#!/usr/bin/env python3 +"""canvas_data.connections → map_connections 迁移（P1 四层因果结构化） + +任务：cma-map-connections-table-p1 / project-20261021-cma-map-connections-table-p1-fullstack + +设计要点 +-------- +1. **只搬不删**：canvas_data 原样保留（双写期，前端渲染仍读 JSON）。本脚本对 + strategic_maps 只做 SELECT，任何情况下不 UPDATE。 +2. **幂等**：目标表唯一键 (map_id, from_node, to_node)。重跑 `--apply` 只补缺， + 已存在的行 SKIP；`--idem` 为只读等价复跑（断言全部已在位），零写。 +3. **双格式兼容**（生产实况）： + `dim:中文目标名` → 按 (dimension_key, 目标名) 解析 map_objectives.id + `dim-序号` → 按 dimensions[dim].objectives 顺序解析（缺 dimensions 时退 sort_order） + 原始串原样落 from_node/to_node，解析结果落 *_layer / *_objective_id。 +4. **四态**：--check（默认拒绝无标志调用）/ --apply / --idem / --rollback + 写库通道：生产 apply/rollback = root over unix socket（cma_api 无新表 DML 授权）； + --check 走环境变量只读账号（cma_reader/cma_api 的 SELECT）。 + +用法 +---- + python3 migrate_map_connections.py --check [--json-out PATH] + python3 migrate_map_connections.py --show-create + python3 migrate_map_connections.py --apply [--backup-dir DIR] [--evidence PATH] + python3 migrate_map_connections.py --idem + python3 migrate_map_connections.py --rollback + +退出码：0 成功 ｜ 2 用法/缺标志 ｜ 3 表不存在（先走 db_migrate.py --apply）｜ 4 画布 JSON 不可解析（fail-closed）｜ 5 自检不一致 +""" +import argparse +import hashlib +import json +import os +import re +import sys +from datetime import datetime + +HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, os.path.dirname(HERE)) + +LAYERS = ("learning", "process", "customer", "finance") +LAYER_ORDER = {k: i for i, k in enumerate(LAYERS)} +NODE_COLON = re.compile(r"^([a-z_]+):(.+)$") +NODE_INDEX = re.compile(r"^([a-z_]+)-(\d+)$") + + +# ── 连接（凭据只从 env 读，不回显）──────────────────────────────── +def _mask(url: str) -> str: + return re.sub(r"://([^:/@]+):[^@]*@", r"://\1:***@", url) + + +def build_engine(db_url: str | None, mode: str): + from sqlalchemy import create_engine + + if db_url: + url = db_url + elif mode == "check" or mode == "show-create": + user = os.getenv("CMA_DB_USER", "cma_reader") + pwd = os.getenv("CMA_DB_PASS", "") + host = os.getenv("CMA_DB_HOST", "127.0.0.1") + port = os.getenv("CMA_DB_PORT", "3306") + name = os.getenv("CMA_DB_NAME", "cma") + if not pwd: + print("[fail] 缺少 CMA_DB_PASS（只读预检需要 SELECT 账号凭据）", file=sys.stderr) + sys.exit(2) + url = f"mysql+pymysql://{user}:{pwd}@{host}:{port}/{name}?charset=utf8mb4" + else: + name = os.getenv("CMA_DB_NAME", "cma") + url = f"mysql+pymysql://root@localhost/{name}?unix_socket=/var/run/mysqld/mysqld.sock&charset=utf8mb4" + return create_engine(url, future=True), url + + +def load_dotenv(): + try: + from dotenv import load_dotenv as _ld + _ld(os.path.join(os.path.dirname(HERE), ".env")) + except Exception: + pass + + +# ── JSON 规范化 ──────────────────────────────────────────────── +def as_json(v): + """DB JSON 列可能是 dict/list 或 str（SQLite/部分驱动）。""" + if isinstance(v, (bytes, bytearray)): + v = v.decode("utf-8") + if isinstance(v, str): + return json.loads(v) + return v + + +def canvas_md5(canvas) -> str: + raw = json.dumps(canvas, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + return hashlib.md5(raw.encode("utf-8")).hexdigest() + + +def norm_name(s: str) -> str: + return re.sub(r"\s+", " ", (s or "").strip()) + + +# ── 节点解析 ─────────────────────────────────────────────────── +def parse_node(node): + """→ (layer|None, ref|None, err|None) + + ref = {"kind": "name", "name": ...} | {"kind": "index", "index": n} + """ + if not isinstance(node, str) or not node.strip(): + return None, None, "node_empty" + s = node.strip() + m = NODE_COLON.match(s) + if m: + layer = m.group(1) + name = m.group(2).strip() + if layer not in LAYERS: + return None, None, f"unknown_layer:{layer}" + if not name: + return None, None, "node_name_empty" + return layer, {"kind": "name", "name": name}, None + m = NODE_INDEX.match(s) + if m: + layer = m.group(1) + if layer not in LAYERS: + return None, None, f"unknown_layer:{layer}" + return layer, {"kind": "index", "index": int(m.group(2))}, None + return None, None, "node_format_unrecognized" + + +class ObjectiveIndex: + """按 map 建索引：名字 → objective_id；层 + 序号 → objective_id。""" + + def __init__(self, dims_by_map, objs_by_map): + self.by_name = {} + self.by_order = {} + self.dims = dims_by_map + for map_id, objs in objs_by_map.items(): + name_idx, order_idx = {}, {} + for o in objs: + name_idx[(o["dimension_key"], norm_name(o["name"]))] = o["id"] + order_idx.setdefault(o["dimension_key"], []).append(o) + for k in order_idx: + order_idx[k].sort(key=lambda o: (o.get("sort_order") or 0, o["id"])) + self.by_name[map_id] = name_idx + self.by_order[map_id] = order_idx + + def resolve(self, map_id, layer, ref): + """→ (objective_id|None, note|None)""" + if ref["kind"] == "name": + hit = self.by_name.get(map_id, {}).get((layer, norm_name(ref["name"]))) + if hit: + return hit, None + # 仅大小写/空白差异的近似命中：报告但不静默 + want = norm_name(ref["name"]).replace(" ", "") + for (lk, ln), oid in self.by_name.get(map_id, {}).items(): + if lk == layer and ln.replace(" ", "") == want: + return oid, "normalized_match" + return None, "objective_unresolved" + # index：优先按 dimensions[dim].objectives 顺序（前端 key 就是该数组下标） + idx = ref["index"] + dims = self.dims.get(map_id) or [] + for d in dims if isinstance(dims, list) else []: + if d.get("key") == layer: + objs = d.get("objectives") or [] + if 0 <= idx < len(objs): + name = norm_name(objs[idx].get("name")) + hit = self.by_name.get(map_id, {}).get((layer, name)) + if hit: + return hit, None + return None, "objective_unresolved" + return None, "index_out_of_range" + # dimensions 缺失 → 退 sort_order + cand = self.by_order.get(map_id, {}).get(layer) or [] + for o in cand: + if (o.get("sort_order") or 0) == idx: + return o["id"], "resolved_by_sort_order" + if 0 <= idx < len(cand): + return cand[idx]["id"], "resolved_by_position" + return None, "index_out_of_range" + + +# ── 计划构建（纯读）──────────────────────────────────────────── +def load_state(session): + from sqlalchemy import inspect + from app.models import StrategicMap, MapObjective, MapConnection + + insp = inspect(session.get_bind()) + has_table = insp.has_table(MapConnection.__tablename__) + + maps = [] + for m in session.query(StrategicMap).order_by(StrategicMap.id).all(): + maps.append({ + "id": m.id, "entity_id": m.entity_id, "title": m.title, "status": m.status, + "canvas": as_json(m.canvas_data), "dimensions": as_json(m.dimensions), + }) + + objs = {} + for o in session.query(MapObjective).order_by(MapObjective.map_id, MapObjective.sort_order, MapObjective.id).all(): + objs.setdefault(o.map_id, []).append({ + "id": o.id, "dimension_key": o.dimension_key, "name": o.name, "sort_order": o.sort_order, + }) + + existing = {} + if has_table: + for c in session.query(MapConnection).all(): + existing[(c.map_id, c.from_node, c.to_node)] = c.id + return maps, objs, existing, has_table + + +def build_plan(maps, objs): + oi = ObjectiveIndex({m["id"]: m["dimensions"] for m in maps}, objs) + plan, per_map, anomalies = [], [], [] + + def note(map_id, idx, kind, node, detail=""): + anomalies.append({"map_id": map_id, "canvas_index": idx, "kind": kind, + "node": node, "detail": detail}) + + for m in maps: + canvas = m["canvas"] + conns = [] + if canvas is None: + canvas = {} + if not isinstance(canvas, dict): + note(m["id"], None, "canvas_not_object", None, f"type={type(canvas).__name__}") + per_map.append({"map_id": m["id"], "entity_id": m["entity_id"], "title": m["title"], + "canvas_connections": None, "plan_rows": 0, "anomalies": 1}) + continue + conns = canvas.get("connections") or [] + if not isinstance(conns, list): + note(m["id"], None, "connections_not_list", None, f"type={type(conns).__name__}") + conns = [] + + rows, seen = [], {} + for i, c in enumerate(conns): + if not isinstance(c, dict): + note(m["id"], i, "connection_not_object", None, f"type={type(c).__name__}") + continue + frm, to = c.get("from"), c.get("to") + if not isinstance(frm, str) or not isinstance(to, str) or not frm or not to: + note(m["id"], i, "endpoint_missing", f"{frm!r}->{to!r}", "from/to 缺失或非字符串") + continue + key = (frm, to) + if key in seen: + note(m["id"], i, "duplicate_edge", f"{frm} -> {to}", f"与 index={seen[key]} 重复（唯一键会合并）") + continue + seen[key] = i + + fl, fref, ferr = parse_node(frm) + tl, tref, terr = parse_node(to) + if ferr: + note(m["id"], i, ferr, frm, "from 解析失败") + if terr: + note(m["id"], i, terr, to, "to 解析失败") + if frm == to: + note(m["id"], i, "self_loop", frm, "自连") + if fl and tl and LAYER_ORDER[tl] < LAYER_ORDER[fl]: + note(m["id"], i, "reverse_direction", f"{frm} -> {to}", + f"{fl}→{tl} 与 learning→process→customer→finance 相反（告警不拦截）") + + f_obj = f_note = t_obj = t_note = None + if fref: + f_obj, f_note = oi.resolve(m["id"], fl, fref) + if f_note and f_note != "normalized_match": + note(m["id"], i, f_note, frm, "from 目标未解析") + if tref: + t_obj, t_note = oi.resolve(m["id"], tl, tref) + if t_note and t_note != "normalized_match": + note(m["id"], i, t_note, to, "to 目标未解析") + + rows.append({ + "entity_id": m["entity_id"] if m["entity_id"] is not None else 1, + "map_id": m["id"], "from_node": frm, "to_node": to, + "from_layer": fl, "to_layer": tl, + "from_objective_id": f_obj, "to_objective_id": t_obj, + "effect": c.get("effect") or "positive", + "label": c.get("label") or None, + "sort_order": i, + }) + + plan.extend(rows) + per_map.append({ + "map_id": m["id"], "entity_id": m["entity_id"], "title": m["title"], + "status": m["status"], "canvas_connections": len(conns), "plan_rows": len(rows), + "anomalies": sum(1 for a in anomalies if a["map_id"] == m["id"]), + }) + return plan, per_map, anomalies + + +def summarize(plan, per_map, anomalies, existing=None): + by_pair, by_layer = {}, {} + for r in plan: + pair = f"{r['from_layer'] or '?'}->{r['to_layer'] or '?'}" + by_pair[pair] = by_pair.get(pair, 0) + 1 + for lay in (r["from_layer"], r["to_layer"]): + if lay: + by_layer[lay] = by_layer.get(lay, 0) + 1 + existing = existing or {} + to_insert = sum(1 for r in plan if (r["map_id"], r["from_node"], r["to_node"]) not in existing) + return { + "canvas_connections": sum(p["canvas_connections"] or 0 for p in per_map), + "plan_rows": len(plan), + "already_in_table": len([r for r in plan if (r["map_id"], r["from_node"], r["to_node"]) in existing]), + "to_insert": to_insert, + "unresolved_layer": sum(1 for r in plan if not r["from_layer"] or not r["to_layer"]), + "unresolved_objective": sum(1 for r in plan if not r["from_objective_id"] or not r["to_objective_id"]), + "anomalies": len(anomalies), + "layer_pair_distribution": by_pair, + "layer_node_refs": by_layer, + } + + +# ── 三种模式的实现 ───────────────────────────────────────────── +def cmd_check(engine, args): + from sqlalchemy.orm import Session + with Session(engine) as s: + maps, objs, existing, has_table = load_state(s) + plan, per_map, anomalies = build_plan(maps, objs) + report = { + "generated_at": datetime.now().isoformat(timespec="seconds"), + "mode": "check", "read_only": True, + "db": _mask(engine.url.render_as_string(hide_password=True).replace("***", "***")), + "engine_dialect": engine.dialect.name, + "table_exists": has_table, + "totals": summarize(plan, per_map, anomalies, existing), + "maps": per_map, + "anomalies": anomalies, + "sample_rows": plan[:5], + } + if args.json_out: + os.makedirs(os.path.dirname(os.path.abspath(args.json_out)), exist_ok=True) + with open(args.json_out, "w", encoding="utf-8") as f: + json.dump(report, f, ensure_ascii=False, indent=2) + print(f"[check] 库={report['db']} 方言={report['engine_dialect']} 目标表存在={has_table}") + t = report["totals"] + print(f"[check] 画布连线={t['canvas_connections']} 计划行={t['plan_rows']} " + f"已在表={t['already_in_table']} 待插入={t['to_insert']}") + print(f"[check] 层未解析={t['unresolved_layer']} 目标未解析={t['unresolved_objective']} 异常条目={t['anomalies']}") + print(f"[check] 层对分布={t['layer_pair_distribution']}") + for p in per_map: + print(f"[check] map {p['map_id']} (entity={p['entity_id']}, {p['status']}): " + f"canvas={p['canvas_connections']} → plan={p['plan_rows']} 异常={p['anomalies']}") + for a in anomalies[:50]: + print(f"[check] ✗ map={a['map_id']} idx={a['canvas_index']} {a['kind']} {a['node']} {a['detail']}") + if not has_table: + print("[check] ⚠ 目标表不存在：建表须先走 db_migrate.py --apply（受批面）") + return 0 + + +def cmd_show_create(): + from sqlalchemy.schema import CreateTable + from sqlalchemy.dialects import mysql + from app.models import MapConnection + print(CreateTable(MapConnection.__table__).compile(dialect=mysql.dialect())) + return 0 + + +def cmd_apply(engine, args): + from sqlalchemy.orm import Session + from sqlalchemy.exc import IntegrityError + from app.models import MapConnection + + with Session(engine) as s: + maps, objs, existing, has_table = load_state(s) + if not has_table: + print("[fail] map_connections 表不存在 → 先执行（受批）db_migrate.py --apply", file=sys.stderr) + return 3 + plan, per_map, anomalies = build_plan(maps, objs) + hard = [a for a in anomalies if a["kind"] in ("canvas_not_object", "connections_not_list", "connection_not_object")] + if hard: + print(f"[fail] 画布结构不可解析 {len(hard)} 处 → fail-closed，未写任何行：" + f"{hard[:3]}", file=sys.stderr) + return 4 + + # 回滚点：canvas_data 全量快照 + md5（只读 SELECT 得到） + ts = datetime.now().strftime("%Y%m%d-%H%M%S") + backup_dir = os.path.abspath(args.backup_dir or f"/srv/hermes/backups/map_connections_p1/{ts}") + os.makedirs(backup_dir, exist_ok=True) + backup_path = os.path.join(backup_dir, "canvas_data_snapshot.json") + with open(backup_path, "w", encoding="utf-8") as f: + json.dump([{"map_id": m["id"], "entity_id": m["entity_id"], "title": m["title"], + "canvas_data": m["canvas"], "md5": canvas_md5(m["canvas"])} for m in maps], + f, ensure_ascii=False, indent=2) + pre_md5 = {m["id"]: canvas_md5(m["canvas"]) for m in maps} + + inserted, skipped, errors = [], 0, [] + with Session(engine) as s: + try: + for r in plan: + key = (r["map_id"], r["from_node"], r["to_node"]) + if key in existing: + skipped += 1 + continue + obj = MapConnection(**r) + s.add(obj) + try: + s.flush() + except IntegrityError: + s.rollback() + skipped += 1 + continue + inserted.append({"id": obj.id, "map_id": r["map_id"], + "from_node": r["from_node"], "to_node": r["to_node"]}) + s.commit() + except Exception as exc: + s.rollback() + print(f"[fail] 写入失败已回滚：{type(exc).__name__}: {exc}", file=sys.stderr) + return 1 + + # 写后自检：行数 + canvas 未变 + with Session(engine) as s: + maps2, _, _, _ = load_state(s) + table_rows = s.query(MapConnection).filter( + MapConnection.map_id.in_([m["id"] for m in maps])).count() if maps else 0 + post_md5 = {m["id"]: canvas_md5(m["canvas"]) for m in maps2} + canvas_unchanged = pre_md5 == post_md5 + expected = len({(r["map_id"], r["from_node"], r["to_node"]) for r in plan}) + + evidence = { + "action": "apply", "status": "ok", + "started_at": ts, "finished_at": datetime.now().isoformat(timespec="seconds"), + "db_engine": engine.dialect.name, + "backup_path": backup_path, "backup_pre_md5": pre_md5, "post_md5": post_md5, + "canvas_unchanged": canvas_unchanged, + "plan_rows": len(plan), "inserted": len(inserted), "skipped_existing": skipped, + "table_rows_for_scope": table_rows, "expected_distinct_edges": expected, + "anomalies": anomalies, "maps": per_map, "inserted_ids": inserted, + } + ok = canvas_unchanged and table_rows == expected and not errors + evidence["verdict"] = "PASS" if ok else "FAIL" + ev_path = os.path.abspath(args.evidence or os.path.join(backup_dir, "apply_evidence.json")) + os.makedirs(os.path.dirname(ev_path), exist_ok=True) + with open(ev_path, "w", encoding="utf-8") as f: + json.dump(evidence, f, ensure_ascii=False, indent=2) + + print(f"[apply] 计划 {len(plan)} 行；新插入 {len(inserted)}，已存在跳过 {skipped}") + print(f"[apply] 表内本范围行数 {table_rows} / 期望 {expected}；canvas 未改={canvas_unchanged}") + print(f"[apply] 备份 {backup_path}") + print(f"[apply] 证据 {ev_path}") + print(f"[apply] verdict={evidence['verdict']}") + return 0 if ok else 5 + + +def cmd_idem(engine, args): + from sqlalchemy.orm import Session + with Session(engine) as s: + maps, objs, existing, has_table = load_state(s) + if not has_table: + print("[fail] 表不存在", file=sys.stderr) + return 3 + plan, per_map, anomalies = build_plan(maps, objs) + mismatches = [] + for r in plan: + key = (r["map_id"], r["from_node"], r["to_node"]) + cid = existing.get(key) + if not cid: + mismatches.append({"kind": "missing", "key": list(key)}) + continue + row = s.get(__import__("app.models", fromlist=["MapConnection"]).MapConnection, cid) + for f in ("from_layer", "to_layer", "from_objective_id", "to_objective_id"): + if (getattr(row, f) or None) != (r[f] or None): + mismatches.append({"kind": "value_diff", "key": list(key), "field": f, + "db": getattr(row, f), "plan": r[f]}) + print(f"[idem] 计划 {len(plan)} 行，已在位 {len(plan) - len([m for m in mismatches if m['kind'] == 'missing'])}；" + f"差异 {len(mismatches)}") + for m in mismatches[:20]: + print(f"[idem] ✗ {m}") + print(f"[idem] verdict={'PASS' if not mismatches else 'FAIL'}") + return 0 if not mismatches else 5 + + +def cmd_rollback(engine, args): + from sqlalchemy.orm import Session + from app.models import MapConnection + path = args.rollback + if not path or not os.path.exists(path): + print(f"[fail] 回滚证据不存在：{path}", file=sys.stderr) + return 2 + with open(path, encoding="utf-8") as f: + ev = json.load(f) + ids = [x["id"] for x in ev.get("inserted_ids", [])] + if not ids: + print("[rollback] 证据里没有 inserted_ids，无需回滚") + return 0 + with Session(engine) as s: + n = s.query(MapConnection).filter(MapConnection.id.in_(ids)).delete(synchronize_session=False) + s.commit() + left = s.query(MapConnection).filter(MapConnection.id.in_(ids)).count() + print(f"[rollback] 删除 {n} 行，残留 {left}（证据 {path}）；canvas_data 本脚本从未写入，无需还原") + return 0 if left == 0 else 5 + + +def main(): + ap = argparse.ArgumentParser(description="canvas_data.connections → map_connections（四态幂等迁移）") + g = ap.add_mutually_exclusive_group() + g.add_argument("--check", action="store_true", help="只读预检（dry-run）：行数/异常/待插入") + g.add_argument("--apply", action="store_true", help="唯一写：幂等插入（需表已存在）") + g.add_argument("--idem", action="store_true", help="只读幂等复跑：断言全部已在位") + g.add_argument("--rollback", metavar="EVIDENCE_JSON", help="按 apply 证据删除本批插入行") + g.add_argument("--show-create", action="store_true", help="打印模型编译出的 CREATE TABLE") + ap.add_argument("--db-url", help="覆盖连接串（演练副本/SQLite 用）") + ap.add_argument("--json-out", help="--check 的机器可读报告路径") + ap.add_argument("--backup-dir", help="--apply 回滚点目录") + ap.add_argument("--evidence", help="--apply 证据 JSON 路径") + args = ap.parse_args() + + load_dotenv() + + if args.show_create: + return cmd_show_create() + if not any([args.check, args.apply, args.idem, args.rollback]): + print("拒绝无标志调用：请显式给 --check / --apply / --idem / --rollback", file=sys.stderr) + return 2 + + mode = "check" if args.check else "apply" if args.apply else "idem" if args.idem else "rollback" + engine, url = build_engine(args.db_url, mode) + print(f"[{mode}] 通道 {_mask(url)}") + if args.check: + return cmd_check(engine, args) + if args.apply: + return cmd_apply(engine, args) + if args.idem: + return cmd_idem(engine, args) + return cmd_rollback(engine, args) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/backend/tests/test_map_connections_p1.py b/backend/tests/test_map_connections_p1.py new file mode 100644 index 00000000..eaf5b680 --- /dev/null +++ b/backend/tests/test_map_connections_p1.py @@ -0,0 +1,166 @@ +"""P1 四层因果结构化：map_connections 表 + /connections 读/写/analysis 回归 + +任务 project-20261021-cma-map-connections-table-p1-fullstack +覆盖：双格式解析（层:中文名 / 层-序号）、双写、读表回落 canvas、分析与告警不拦截。 +""" +import pytest +from fastapi.testclient import TestClient +from sqlalchemy.orm import Session + +from app.models import StrategicMap, MapObjective, MapConnection +from app.api.maps import parse_node_ref, resolve_node +from tests.conftest import create_test_user, get_token_for_user, auth_header + +DIMS = [ + {"key": "finance", "name": "财务层", "icon": "💰", "color": "#F56C6C", + "objectives": [{"name": "营业收入"}, {"name": "营业利润"}]}, + {"key": "customer", "name": "客户层", "icon": "👥", "color": "#409EFF", + "objectives": [{"name": "客户留存"}]}, + {"key": "process", "name": "流程层", "icon": "⚙️", "color": "#67C23A", + "objectives": [{"name": "交付效率"}, {"name": "质量闭环"}]}, + {"key": "learning", "name": "学习层", "icon": "📚", "color": "#E6A23C", + "objectives": [{"name": "全员培训"}, {"name": "能力认证"}]}, +] + + +def _seed_map(db: Session, title="P1连线测试", canvas=None) -> StrategicMap: + m = StrategicMap(entity_id=1, title=title, status="draft", dimensions=DIMS, + canvas_data=canvas or {"connections": []}) + db.add(m) + db.commit() + db.refresh(m) + for d in DIMS: + for i, o in enumerate(d["objectives"]): + db.add(MapObjective(entity_id=1, map_id=m.id, dimension_key=d["key"], + name=o["name"], sort_order=i)) + db.commit() + return m + + +@pytest.fixture(autouse=True) +def _clean(db: Session): + db.query(MapConnection).delete() + db.query(MapObjective).filter(MapObjective.map_id >= 900000).delete() + db.query(StrategicMap).filter(StrategicMap.id >= 900000).delete() + db.commit() + yield + + +class TestNodeParse: + def test_colon_format(self): + r = parse_node_ref("learning:全员培训") + assert (r["layer"], r["kind"], r["value"], r["error"]) == ("learning", "name", "全员培训", None) + + def test_index_format(self): + r = parse_node_ref("process-1") + assert (r["layer"], r["kind"], r["value"], r["error"]) == ("process", "index", 1, None) + + def test_unknown_formats(self): + assert parse_node_ref("nonsense")["error"] == "node_format_unrecognized" + assert parse_node_ref("hr:某目标")["error"] == "unknown_layer:hr" + assert parse_node_ref("")["error"] == "node_empty" + + def test_resolve_both_formats(self, db: Session): + m = _seed_map(db) + # 冒号中文名 + layer, oid, note = resolve_node(db, m, "learning:能力认证") + assert (layer, note) == ("learning", None) and oid is not None + # 序号（dimensions 数组下标） + layer, oid2, note = resolve_node(db, m, "process-1") + assert (layer, note) == ("process", None) + obj = db.query(MapObjective).filter(MapObjective.id == oid2).first() + assert obj.name == "质量闭环" + + +class TestConnectionApi: + def test_post_dual_write_and_get_table(self, client: TestClient, db: Session): + create_test_user(db) + token = get_token_for_user(client) + m = _seed_map(db, title="P1双写") + mid = m.id + + resp = client.post(f"/api/cma/maps/{mid}/connections", headers=auth_header(token), + json={"from": "learning:全员培训", "to": "process:交付效率"}) + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["db_synced"] is True + assert body["connections"][0]["from"] == "learning:全员培训" + + # canvas_data 保留（前端渲染主路径） + db.expire_all() + m2 = db.query(StrategicMap).filter(StrategicMap.id == mid).first() + assert m2.canvas_data["connections"][0]["to"] == "process:交付效率" + # 表内落行且层/目标解析正确 + row = db.query(MapConnection).filter(MapConnection.map_id == mid).one() + assert (row.from_layer, row.to_layer) == ("learning", "process") + assert row.from_objective_id and row.to_objective_id + + got = client.get(f"/api/cma/maps/{mid}/connections", headers=auth_header(token)) + assert got.status_code == 200 + assert got.json()["source"] == "table" + assert got.json()["total"] == 1 + + def test_index_format_persisted(self, client: TestClient, db: Session): + create_test_user(db) + token = get_token_for_user(client) + m = _seed_map(db, title="P1序号格式") + resp = client.post(f"/api/cma/maps/{m.id}/connections", headers=auth_header(token), + json={"from": "learning-0", "to": "process-1"}) + assert resp.status_code == 200, resp.text + row = db.query(MapConnection).filter(MapConnection.map_id == m.id).one() + assert (row.from_node, row.from_layer) == ("learning-0", "learning") + assert row.from_objective_id and row.to_objective_id + + def test_get_falls_back_to_canvas(self, client: TestClient, db: Session): + create_test_user(db) + token = get_token_for_user(client) + m = _seed_map(db, title="P1回落", + canvas={"connections": [{"from": "learning:全员培训", "to": "process:质量闭环"}]}) + got = client.get(f"/api/cma/maps/{m.id}/connections", headers=auth_header(token)) + assert got.status_code == 200 + assert got.json()["source"] == "canvas_data" + assert got.json()["total"] == 1 + + def test_delete_removes_both(self, client: TestClient, db: Session): + create_test_user(db) + token = get_token_for_user(client) + m = _seed_map(db, title="P1删除") + client.post(f"/api/cma/maps/{m.id}/connections", headers=auth_header(token), + json={"from": "learning:全员培训", "to": "process:交付效率"}) + resp = client.request("DELETE", f"/api/cma/maps/{m.id}/connections", headers=auth_header(token), + json={"from": "learning:全员培训", "to": "process:交付效率"}) + assert resp.status_code == 200, resp.text + assert resp.json()["db_synced"] is True + assert db.query(MapConnection).filter(MapConnection.map_id == m.id).count() == 0 + db.expire_all() + m2 = db.query(StrategicMap).filter(StrategicMap.id == m.id).first() + assert m2.canvas_data["connections"] == [] + + def test_analysis_stats_reverse_and_broken(self, client: TestClient, db: Session): + create_test_user(db) + token = get_token_for_user(client) + m = _seed_map(db, title="P1分析") + pairs = [ + ("learning:全员培训", "process:交付效率"), # 正常 + ("process:质量闭环", "customer:客户留存"), # 正常 + ("customer:客户留存", "finance:营业收入"), # 正常 + ("finance:营业利润", "customer:客户留存"), # 反向（告警不拦截） + ("finance:营业收入", "finance:营业利润"), # 同层 + ("learning:不存在目标", "process:交付效率"), # 断链（照样可写） + ] + for f, t in pairs: + r = client.post(f"/api/cma/maps/{m.id}/connections", headers=auth_header(token), + json={"from": f, "to": t}) + assert r.status_code == 200, r.text + + a = client.get(f"/api/cma/maps/{m.id}/connections/analysis", headers=auth_header(token)) + assert a.status_code == 200, a.text + d = a.json() + assert d["source"] == "table" and d["total"] == 6 + assert d["layer_stats"]["learning"]["as_from"] == 2 + assert d["layer_stats"]["finance"]["as_to"] == 2 # customer→finance 1 + finance→finance 1 + assert d["cross_layer"]["learning->process"] == 2 # 正常 1 + 断链 1（层可解析、目标不可解析） + assert d["same_layer_count"] == 1 + assert len(d["reversed"]) == 1 and d["reversed"][0]["from_layer"] == "finance" + assert any(b["side"] == "from" for b in d["broken"]) + assert any("反向连线" in w for w in d["warnings"])