Hermes CI Fix
9b61c48051
安全: - ⑦ update_map 字段白名单(禁entity_id篡改, status限值域) - ⑧ objectives 全端点补entity校验(原括号错位导致隔离失效) + update白名单 - ⑨ strategic_map_versions 加entity_id列+回填, list/snapshot/rollback过滤 - ⑩ get_map_review KPI查询加entity(同编码跨企业混入) - ⑪ _sync_map_objectives 强制继承地图entity 逻辑: - ③ MapCanvas 自动保存 currentMap.value 解包(30s自动保存修复) - ④ 乐观锁字段 version_num→version 对齐 - ⑤ 红黄绿灯: 越低越好型反向 + 负值修正 + 月目标优先 - ⑥ review 补 action_plan_summary 字段 验证: 跨企业写目标404✓ entity_id注入被拒✓ pytest 531 passed✓
90 lines
3.1 KiB
Python
90 lines
3.1 KiB
Python
"""战略地图版本管理 API"""
|
||
from fastapi import APIRouter, Depends, HTTPException
|
||
from sqlalchemy.orm import Session
|
||
from app.database import get_db
|
||
from app.deps import get_entity_id
|
||
from app.auth_middleware import require_role
|
||
from app.models import StrategicMap, StrategicMapVersion
|
||
|
||
router = APIRouter(prefix="/api/cma/maps", tags=["战略地图版本"],
|
||
dependencies=[Depends(require_role("ceo", "finance"))],
|
||
)
|
||
|
||
|
||
@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）"""
|
||
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]}
|
||
|
||
|
||
@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, "战略地图不存在")
|
||
|
||
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)
|
||
|
||
# 自动版本号
|
||
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"
|
||
|
||
snapshot = StrategicMapVersion(
|
||
map_id=map_id,
|
||
entity_id=entity_id, # 继承地图企业
|
||
version=new_ver,
|
||
dimensions=dims,
|
||
canvas_data=canvas,
|
||
comment=data.get("comment", f"手动快照 {new_ver}"),
|
||
)
|
||
db.add(snapshot)
|
||
db.commit()
|
||
db.refresh(snapshot)
|
||
return v_to_dict(snapshot)
|
||
|
||
|
||
@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, "战略地图不存在")
|
||
|
||
v = db.query(StrategicMapVersion).filter(
|
||
StrategicMapVersion.id == ver_id,
|
||
StrategicMapVersion.map_id == map_id,
|
||
).first()
|
||
if not v:
|
||
raise HTTPException(404, "版本不存在")
|
||
|
||
m.dimensions = v.dimensions
|
||
m.canvas_data = v.canvas_data
|
||
m.version = f"rollback-{v.version}"
|
||
m.status = "draft"
|
||
db.commit()
|
||
return {"message": f"已回滚到 {v.version}", "version": m.version}
|
||
|
||
|
||
def v_to_dict(v):
|
||
return {c.name: getattr(v, c.name) for c in v.__table__.columns}
|