sxbh_admin/cma-management
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(maps): 战略地图版本管理完整实现（方案B）— versions API列表/详情/回滚带自动备份/备注 + MapVersions独立版本页 + 路由注册 + 版本弹窗接真实数据 (project-20260908-map-versions-page)

Browse Source
This commit is contained in:
Hermes CI Fix
2026-09-08 20:11:59 +08:00
parent fcc511d3b9
commit 397c365263
6 changed files with 627 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/versions.py
+191 -44
View File
@@ -1,4 +1,19 @@
"""战略地图版本管理 API"""
"""战略地图版本管理 API（2026-09-08 完整实现）
功能闭环（任总报告\"版本连接点击打不开\"→ 方案B）:
- GET /{map_id}/versions 版本列表（含维度/目标/KPI/连线摘要 + is_current）
- GET /{map_id}/versions/{vid} 单版本详情（dimensions/canvas_data 全量，供\"查看快照画布\"）
- POST /{map_id}/versions/snapshot 手动创建快照（保留，历史兼容）
- POST /{map_id}/versions/{vid}/rollback 回滚到指定版本 —— 回滚前自动备份当前内容为新版本快照，防覆盖丢数据
- PUT /{map_id}/versions/{vid} 更新版本备注 comment
安全:
- 全部端点按 map_id + entity_id 隔离（地图归属 404；版本行再校验 entity_id，防跨账套回滚）
- 回滚 = 先备份后覆盖；备份版本号由当前地图 version 派生（vX.Y → vX.(Y+1)），撞号自动 +1
- 不改 _auto_snapshot / 发布逻辑（maps.py 自动快照保持原样）
"""
import json
import re
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app.database import get_db
@@ -11,43 +26,138 @@ router = APIRouter(prefix="/api/cma/maps", tags=["战略地图版本"],
)
def _get_map(db: Session, map_id: int, entity_id: int) -> StrategicMap:
m = db.query(StrategicMap).filter(
StrategicMap.id == map_id, StrategicMap.entity_id == entity_id
).first()
if not m:
raise HTTPException(404, "战略地图不存在")
return m
def _get_version(db: Session, map_id: int, vid: int, entity_id: int) -> StrategicMapVersion:
v = db.query(StrategicMapVersion).filter(
StrategicMapVersion.id == vid,
StrategicMapVersion.map_id == map_id,
StrategicMapVersion.entity_id == entity_id,
).first()
if not v:
raise HTTPException(404, "版本不存在")
return v
def _parse_json(raw, default):
if raw is None:
return default
if isinstance(raw, (dict, list)):
return raw
try:
return json.loads(raw)
except Exception:
return default
def _next_minor_version(s: str) -> str:
"""把 'v1.0' → 'v1.1'；解析失败返回 None"""
m = re.search(r"v(\d+)\.(\d+)", s or "")
if m:
return f"v{m.group(1)}.{int(m.group(2)) + 1}"
return None
def _summary_dict(v: StrategicMapVersion, m: StrategicMap = None) -> dict:
dims = _parse_json(v.dimensions, [])
canvas = _parse_json(v.canvas_data, {"connections": []})
codes = set()
objective_count = 0
for dim in dims:
for obj in dim.get("objectives", []):
objective_count += 1
for c in (obj.get("kpis") or []):
if c:
codes.add(str(c))
for kr in (obj.get("krs") or []):
if isinstance(kr, dict) and kr.get("kpi_code"):
codes.add(str(kr["kpi_code"]))
conns = canvas.get("connections", []) if isinstance(canvas, dict) else []
return {
"id": v.id,
"map_id": v.map_id,
"entity_id": v.entity_id,
"version": v.version,
"comment": v.comment,
"created_by": v.created_by,
"created_at": v.created_at.isoformat() if v.created_at else None,
"dimension_count": len(dims),
"objective_count": objective_count,
"kpi_count": len(codes),
"connection_count": len(conns),
"is_current": bool(m and m.version == v.version),
}
def _full_dict(v: StrategicMapVersion, m: StrategicMap = None) -> dict:
d = {c.name: getattr(v, c.name) for c in v.__table__.columns}
if d.get("created_at") is not None:
d["created_at"] = d["created_at"].isoformat()
d.update(_summary_dict(v, m))
return d
@router.get("/{map_id}/versions")
def list_versions(map_id: int, db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)):
"""查看版本历史（账套隔离, OpenCode审查#9）"""
"""查看版本历史（账套隔离；地图归属校验 404）"""
m = _get_map(db, map_id, entity_id)
versions = db.query(StrategicMapVersion).filter(
StrategicMapVersion.map_id == map_id,
StrategicMapVersion.entity_id == entity_id,
).order_by(StrategicMapVersion.id.desc()).all()
return {"data": [v_to_dict(v) for v in versions]}
).order_by(StrategicMapVersion.created_at.desc(), StrategicMapVersion.id.desc()).all()
return {"data": [_summary_dict(v, m) for v in versions]}
@router.get("/{map_id}/versions/{vid}")
def get_version(map_id: int, vid: int, db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)):
"""单版本详情：dimensions/canvas_data 全量（供\"查看该版本画布\"）"""
m = _get_map(db, map_id, entity_id)
v = _get_version(db, map_id, vid, entity_id)
return _full_dict(v, m)
@router.put("/{map_id}/versions/{vid}")
def update_version_comment(map_id: int, vid: int, data: dict, db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)):
"""更新版本备注 comment"""
_get_map(db, map_id, entity_id)
if "comment" not in data:
raise HTTPException(400, "缺少 comment 字段")
v = _get_version(db, map_id, vid, entity_id)
v.comment = data.get("comment")
db.commit()
db.refresh(v)
return _summary_dict(v, _get_map(db, map_id, entity_id))
def _next_snapshot_version(db: Session, map_id: int) -> str:
"""该地图下一个快照版本号：最新快照行 minor+1；无快照 → v1.0（与 _auto_snapshot 语义一致）"""
latest = db.query(StrategicMapVersion).filter(
StrategicMapVersion.map_id == map_id
).order_by(StrategicMapVersion.id.desc()).first()
if latest:
nxt = _next_minor_version(latest.version)
if nxt:
return nxt
return "v1.0"
@router.post("/{map_id}/versions/snapshot")
def create_snapshot(map_id: int, data: dict, db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)):
"""手动创建快照（账套隔离）"""
m = db.query(StrategicMap).filter(StrategicMap.id == map_id, StrategicMap.entity_id == entity_id).first()
if not m:
raise HTTPException(404, "战略地图不存在")
m = _get_map(db, map_id, entity_id)
import json
dims = m.dimensions
canvas = m.canvas_data
if isinstance(dims, str):
dims = json.loads(dims)
if isinstance(canvas, str):
canvas = json.loads(canvas)
dims = _parse_json(m.dimensions, [])
canvas = _parse_json(m.canvas_data, {"connections": []})
# 自动版本号
existing = db.query(StrategicMapVersion).filter(
StrategicMapVersion.map_id == map_id
).order_by(StrategicMapVersion.id.desc()).first()
if existing:
import re
match = re.search(r"v(\d+)\.(\d+)", existing.version)
major = int(match.group(1)) if match else 1
minor = int(match.group(2)) + 1 if match else 0
new_ver = f"v{major}.{minor}"
else:
new_ver = "v1.0"
# 自动版本号: 与 _auto_snapshot 一致 —— 最新快照行 minor+1；无快照 → v1.0
new_ver = _next_snapshot_version(db, map_id)
snapshot = StrategicMapVersion(
map_id=map_id,
@@ -60,30 +170,67 @@ def create_snapshot(map_id: int, data: dict, db: Session = Depends(get_db), enti
db.add(snapshot)
db.commit()
db.refresh(snapshot)
return v_to_dict(snapshot)
return _full_dict(snapshot, m)
@router.post("/{map_id}/versions/{ver_id}/rollback")
def rollback_version(map_id: int, ver_id: int, db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)):
"""回滚到指定版本"""
m = db.query(StrategicMap).filter(StrategicMap.id == map_id, StrategicMap.entity_id == entity_id).first()
if not m:
raise HTTPException(404, "战略地图不存在")
@router.post("/{map_id}/versions/{vid}/rollback")
def rollback_version(map_id: int, vid: int, db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)):
"""回滚到指定版本。
v = db.query(StrategicMapVersion).filter(
StrategicMapVersion.id == ver_id,
安全设计（2026-09-08 完整实现，修复旧版无备份覆盖事故）:
1. 回滚前把地图【当前内容】备份为新版本快照（版本号由当前 map.version 派生 vX.(Y+1)，撞号自动 +1）
—— 任何情况下不因回滚丢失当前画布内容
2. 用目标版本 dimensions/canvas_data 覆盖 strategic_maps 当前行，version 设为该版本号
3. 地图与版本行双重 entity 校验（跨账套一律 404）
"""
m = _get_map(db, map_id, entity_id)
v = _get_version(db, map_id, vid, entity_id)
# 1) 先备份当前内容 → 新版本快照
cur_dims = _parse_json(m.dimensions, [])
cur_canvas = _parse_json(m.canvas_data, {"connections": []})
backup_ver = _next_snapshot_version(db, map_id)
# 防撞号: 极端场景最新快照版本号不可解析回落 v1.0 时循环 +1（正常递增天然唯一）
exists = db.query(StrategicMapVersion).filter(
StrategicMapVersion.map_id == map_id,
StrategicMapVersion.version == backup_ver,
).first()
if not v:
raise HTTPException(404, "版本不存在")
while exists:
nxt = _next_minor_version(backup_ver)
backup_ver = nxt if nxt else backup_ver + ".1"
exists = db.query(StrategicMapVersion).filter(
StrategicMapVersion.map_id == map_id,
StrategicMapVersion.version == backup_ver,
).first()
m.dimensions = v.dimensions
m.canvas_data = v.canvas_data
m.version = f"rollback-{v.version}"
m.status = "draft"
backup = StrategicMapVersion(
map_id=map_id,
entity_id=entity_id,
version=backup_ver,
dimensions=cur_dims,
canvas_data=cur_canvas,
comment="回滚前自动备份",
)
db.add(backup)
# 2) 覆盖当前行内容 + 版本号
m.dimensions = _parse_json(v.dimensions, [])
m.canvas_data = _parse_json(v.canvas_data, {"connections": []})
m.version = v.version
db.commit()
return {"message": f"已回滚到 {v.version}", "version": m.version}
# 3) 同步目标表（与 update_map 保存路径一致，保证回顾会/KR 取数一致）
try:
from app.api.maps import _sync_map_objectives, _sync_objectives_krs
_sync_map_objectives(m, db)
_sync_objectives_krs(m, db)
except Exception as e:
# 同步失败不回滚已完成的版本覆盖（主数据已安全）
import logging
logging.getLogger("maps_versions").warning("rollback 后目标表同步失败: %s", e)
def v_to_dict(v):
return {c.name: getattr(v, c.name) for c in v.__table__.columns}
return {
"message": f"已回滚到 {v.version}（当前内容已自动备份为 {backup_ver}）",
"version": m.version,
"backup_version": backup_ver,
}
backend/tests/test_map_versions.py
+218
View File
@@ -0,0 +1,218 @@
"""
战略地图版本管理完整实现测试（project-20260908-map-versions-page）
覆盖: 列表摘要/详情全量/回滚自动备份/备注编辑/跨entity 404/不存在404
"""
import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from app.models import Entity, User, UserEntity, StrategicMap, StrategicMapVersion
from tests.conftest import create_test_user, get_token_for_user, auth_header
def _create_second_entity(db: Session, user: User) -> int:
db.add(Entity(id=2, name="第二企业", short_name="企业2", status="active"))
db.add(UserEntity(user_id=user.id, entity_id=2, granted_by=1))
db.commit()
return 2
def _token_for_entity(client: TestClient, entity_id: int) -> str:
resp = client.post("/api/cma/auth/login", json={
"username": "testadmin", "password": "admin123", "entity_id": entity_id,
})
assert resp.status_code == 200
data = resp.json()
return data.get("token") or data.get("access_token")
def _create_published_map(client: TestClient, token: str, title: str) -> int:
"""建模板地图 → 发布 → 返回 map_id（自动产生 v1.0 快照）"""
create_resp = client.post(
"/api/cma/maps/create-with-template",
headers=auth_header(token),
json={"title": title},
)
assert create_resp.status_code == 200
map_id = create_resp.json()["id"]
pub_resp = client.put(
f"/api/cma/maps/{map_id}", headers=auth_header(token),
json={"status": "published"},
)
assert pub_resp.status_code == 200
return map_id
class TestMapVersions:
"""战略地图版本 endpoints"""
def test_list_versions_empty_then_publish_has_summary(self, client, db):
create_test_user(db)
token = get_token_for_user(client)
# 未发布空列表
m = client.post("/api/cma/maps/create-with-template", headers=auth_header(token), json={"title": "版本测试A"})
map_id = m.json()["id"]
empty = client.get(f"/api/cma/maps/{map_id}/versions", headers=auth_header(token))
assert empty.status_code == 200
assert empty.json()["data"] == []
# 发布后 v1.0 快照（列表带摘要字段）
client.put(f"/api/cma/maps/{map_id}", headers=auth_header(token), json={"status": "published"})
ver = client.get(f"/api/cma/maps/{map_id}/versions", headers=auth_header(token))
assert ver.status_code == 200
rows = ver.json()["data"]
assert len(rows) == 1
r = rows[0]
assert r["version"] == "v1.0"
assert r["is_current"] is True
assert r["dimension_count"] == 4
assert r["objective_count"] > 0
assert isinstance(r["kpi_count"], int)
assert r["comment"] == "发布 v1.0"
def test_get_version_detail_has_full_content(self, client, db):
create_test_user(db)
token = get_token_for_user(client)
map_id = _create_published_map(client, token, "版本详情B")
versions = client.get(f"/api/cma/maps/{map_id}/versions", headers=auth_header(token)).json()["data"]
vid = versions[0]["id"]
detail = client.get(f"/api/cma/maps/{map_id}/versions/{vid}", headers=auth_header(token))
assert detail.status_code == 200
d = detail.json()
assert isinstance(d["dimensions"], list) and len(d["dimensions"]) == 4
assert isinstance(d["canvas_data"], dict)
assert d["version"] == "v1.0"
def test_rollback_backs_up_current_and_restores_content(self, client, db):
"""核心安全: 回滚前自动备份当前内容为新版本; 回滚后内容=目标版本"""
create_test_user(db)
token = get_token_for_user(client)
map_id = _create_published_map(client, token, "回滚安全C")
# 发布后快照 v1.0（内容=模板初始）
versions = client.get(f"/api/cma/maps/{map_id}/versions", headers=auth_header(token)).json()["data"]
v1_id = versions[0]["id"]
v1_content = client.get(f"/api/cma/maps/{map_id}/versions/{v1_id}", headers=auth_header(token)).json()
# 修改地图内容为全新结构（published 状态下保存，不触发自动快照）
new_dims = [{"key": "finance", "name": "财务层", "icon": "💰", "color": "#F56C6C",
"objectives": [{"name": "回滚前最新目标", "icon": "target",
"kpis": ["F_REVENUE"], "krs": [
{"name": "KR新", "kpi_code": "F_REVENUE", "weight": 100, "target_value": "100"}
]}]}]
upd = client.put(f"/api/cma/maps/{map_id}", headers=auth_header(token),
json={"dimensions": new_dims, "canvas_data": {"connections": [{"from": "a", "to": "b"}]}})
assert upd.status_code == 200
# 回滚到 v1.0 → 应自动备份当前(v1.0→v1.1) 且内容还原为 v1.0
rb = client.post(f"/api/cma/maps/{map_id}/versions/{v1_id}/rollback", headers=auth_header(token))
assert rb.status_code == 200
data = rb.json()
assert data["version"] == "v1.0"
assert data["backup_version"] == "v1.1"
# 列表: 备份(v1.1 最新) + v1.0
after = client.get(f"/api/cma/maps/{map_id}/versions", headers=auth_header(token)).json()["data"]
assert len(after) == 2
assert {x["version"] for x in after} == {"v1.0", "v1.1"}
backup = next(x for x in after if x["version"] == "v1.1")
assert backup["comment"] == "回滚前自动备份"
# 当前地图内容 == v1.0 快照内容（语义还原；列表读路径会经 map_objectives 合并补 description/icon 默认键，按业务字段归一化比较）
maps = client.get("/api/cma/maps", headers=auth_header(token)).json()["data"]
cur = next(x for x in maps if x["id"] == map_id)
assert cur["version"] == "v1.0"
def norm(dims):
return [
{
"key": d.get("key"),
"objectives": [
{k: o.get(k) for k in ("name", "kpis", "krs")}
for o in d.get("objectives", [])
],
}
for d in dims
]
assert norm(cur["dimensions"]) == norm(v1_content["dimensions"])
assert cur["canvas_data"] == v1_content["canvas_data"]
def test_rollback_again_uses_next_free_backup_version(self, client, db):
"""连续两次回滚同一目标: 备份版本号不撞号（v1.1 → v1.2）"""
create_test_user(db)
token = get_token_for_user(client)
map_id = _create_published_map(client, token, "回滚两次D")
versions = client.get(f"/api/cma/maps/{map_id}/versions", headers=auth_header(token)).json()["data"]
v1_id = versions[0]["id"]
client.put(f"/api/cma/maps/{map_id}", headers=auth_header(token),
json={"dimensions": [{"key": "finance", "name": "财务层", "icon": "💰", "color": "#F56C6C",
"objectives": []}]})
rb1 = client.post(f"/api/cma/maps/{map_id}/versions/{v1_id}/rollback", headers=auth_header(token)).json()
assert rb1["backup_version"] == "v1.1"
# 再次改内容再回滚 → 备份 v1.2（不撞 v1.1）
client.put(f"/api/cma/maps/{map_id}", headers=auth_header(token),
json={"dimensions": [{"key": "customer", "name": "客户层", "icon": "👥", "color": "#409EFF",
"objectives": []}]})
rb2 = client.post(f"/api/cma/maps/{map_id}/versions/{v1_id}/rollback", headers=auth_header(token)).json()
assert rb2["backup_version"] == "v1.2"
def test_update_comment(self, client, db):
create_test_user(db)
token = get_token_for_user(client)
map_id = _create_published_map(client, token, "备注编辑E")
versions = client.get(f"/api/cma/maps/{map_id}/versions", headers=auth_header(token)).json()["data"]
vid = versions[0]["id"]
resp = client.put(f"/api/cma/maps/{map_id}/versions/{vid}", headers=auth_header(token),
json={"comment": "2026H2 定稿版"})
assert resp.status_code == 200
assert resp.json()["comment"] == "2026H2 定稿版"
rows = client.get(f"/api/cma/maps/{map_id}/versions", headers=auth_header(token)).json()["data"]
assert rows[0]["comment"] == "2026H2 定稿版"
def test_update_comment_requires_field(self, client, db):
create_test_user(db)
token = get_token_for_user(client)
map_id = _create_published_map(client, token, "备注必填F")
versions = client.get(f"/api/cma/maps/{map_id}/versions", headers=auth_header(token)).json()["data"]
vid = versions[0]["id"]
resp = client.put(f"/api/cma/maps/{map_id}/versions/{vid}", headers=auth_header(token), json={})
assert resp.status_code == 400
def test_cross_entity_all_versions_endpoints_404(self, client, db):
"""跨账套: 列表/详情/备注/回滚 全部 404（地图归属+版本 entity 双重隔离）"""
create_test_user(db)
token1 = get_token_for_user(client)
map_id = _create_published_map(client, token1, "隔离G")
user = db.query(User).filter(User.username == "testadmin").first()
_create_second_entity(db, user)
token2 = _token_for_entity(client, 2)
versions = client.get(f"/api/cma/maps/{map_id}/versions", headers=auth_header(token1)).json()["data"]
vid = versions[0]["id"]
assert client.get(f"/api/cma/maps/{map_id}/versions", headers=auth_header(token2)).status_code == 404
assert client.get(f"/api/cma/maps/{map_id}/versions/{vid}", headers=auth_header(token2)).status_code == 404
assert client.put(f"/api/cma/maps/{map_id}/versions/{vid}", headers=auth_header(token2),
json={"comment": "越权"}).status_code == 404
assert client.post(f"/api/cma/maps/{map_id}/versions/{vid}/rollback", headers=auth_header(token2)).status_code == 404
def test_rollback_unknown_version_404(self, client, db):
create_test_user(db)
token = get_token_for_user(client)
map_id = _create_published_map(client, token, "不存在H")
resp = client.post(f"/api/cma/maps/{map_id}/versions/99999/rollback", headers=auth_header(token))
assert resp.status_code == 404
def test_rollback_other_maps_version_404(self, client, db):
"""vid 属于另一张地图 → 404（map_id + vid 双条件匹配）"""
create_test_user(db)
token = get_token_for_user(client)
map_a = _create_published_map(client, token, "地图A-I")
map_b = _create_published_map(client, token, "地图B-I")
va = client.get(f"/api/cma/maps/{map_a}/versions", headers=auth_header(token)).json()["data"][0]
resp = client.post(f"/api/cma/maps/{map_b}/versions/{va['id']}/rollback", headers=auth_header(token))
assert resp.status_code == 404