wecom-fullstack
0a6525edab
ci/woodpecker/push/main Pipeline was successful
- 表 strategy_charter（一账套一行，entity_id 唯一）：mission/vision/vision_year/vision_measurable/values_text/value_items(JSON) 全 NOT NULL
- 迁移脚本 backend/scripts/migrate_strategy_charter_20261021.py：--check/--apply/--down（回滚先 mysqldump，有数据需 --force），零 INSERT
- API：GET 无行 → filled=false 空态（禁示例数据）；PUT 字段白名单 + 账套隔离三道 + vision_year 4位年份 + value_items [{name,meaning}] 结构校验
- 前端：P 组「战略章程」入口 + MyDialog 编辑（ceo/finance 可写，其余只读）；战略地图页顶部只读章程条
- 用例 43 条 + auth-surface 58 条 = 101 passed；vite build rc=0
228 lines
8.5 KiB
Python
228 lines
8.5 KiB
Python
"""战略基座 — 战略章程 API（/api/cma/charter）
|
||
|
||
任务：`project-20261021-cma-strategy-charter-p0-fullstack`（P0 战略基座）
|
||
------------------------------------------------------------------
|
||
契约：
|
||
GET /api/cma/charter 读当前账套章程（无行 → 200 + filled=false 空态，**不造默认/示例值**）
|
||
PUT /api/cma/charter 写当前账套章程（upsert 单行；字段白名单 + 值域校验）
|
||
|
||
隔离（三道）：
|
||
1. 账套一律取 `deps.get_entity_id` —— token 优先，显式传别的账套 → 403（平台统一语义）；
|
||
2. body 里出现 entity_id 且与解析结果不一致 → 403（防 Bot/脚本越权注入）；
|
||
3. 读写都带 `entity_id` 过滤条件（不靠唯一键兜底），跨账套读不到对方行。
|
||
|
||
白名单（派单要求「白名单」）：
|
||
· 字段白名单 CHARTER_FIELDS —— 出现白名单外字段直接 400，不静默忽略；
|
||
· entity_id 不可改（由登录账套决定，手写一律拒绝）；
|
||
· 不可空：mission / vision / vision_year / vision_measurable / values_text / value_items；
|
||
· vision_year 须 4 位年份；value_items 须为 `[{name, meaning}]` 数组（每条可作裁决依据）。
|
||
|
||
鉴权：GET 已登录即可（含 business 旁观）；PUT 限 ceo / finance（战略口径归经营层）。
|
||
"""
|
||
from datetime import datetime
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.auth_middleware import require_auth, require_role
|
||
from app.database import get_db
|
||
from app.deps import get_entity_id
|
||
from app.models import StrategyCharter, User
|
||
from app.models.strategy_charter import (
|
||
MISSION_MAX_LEN,
|
||
VALUE_ITEM_MEANING_MAX_LEN,
|
||
VALUE_ITEM_NAME_MAX_LEN,
|
||
VALUE_ITEMS_MAX,
|
||
VALUES_TEXT_MAX_LEN,
|
||
VISION_MAX_LEN,
|
||
VISION_MEASURABLE_MAX_LEN,
|
||
VISION_YEAR_RE,
|
||
)
|
||
|
||
import re
|
||
|
||
router = APIRouter(
|
||
prefix="/api/cma/charter",
|
||
tags=["战略基座"],
|
||
dependencies=[Depends(require_auth)],
|
||
)
|
||
|
||
# 可写字段白名单（entity_id / 审计字段不接受客户端输入）
|
||
CHARTER_FIELDS = (
|
||
"mission",
|
||
"vision",
|
||
"vision_year",
|
||
"vision_measurable",
|
||
"values_text",
|
||
"value_items",
|
||
)
|
||
|
||
REQUIRED_FIELDS = CHARTER_FIELDS # 六项均为必填（vision_year / vision_measurable 派单点名不可空）
|
||
|
||
|
||
def _bad(msg: str) -> HTTPException:
|
||
return HTTPException(status_code=400, detail=msg)
|
||
|
||
|
||
def _to_dict(row: StrategyCharter) -> dict:
|
||
return {
|
||
"id": row.id,
|
||
"entity_id": row.entity_id,
|
||
"mission": row.mission,
|
||
"vision": row.vision,
|
||
"vision_year": row.vision_year,
|
||
"vision_measurable": row.vision_measurable,
|
||
"values_text": row.values_text,
|
||
"value_items": row.value_items or [],
|
||
"updated_by": row.updated_by,
|
||
"updated_by_name": row.updated_by_name,
|
||
"created_at": row.created_at.isoformat(sep=" ", timespec="seconds") if row.created_at else None,
|
||
"updated_at": row.updated_at.isoformat(sep=" ", timespec="seconds") if row.updated_at else None,
|
||
}
|
||
|
||
|
||
def _empty_state(entity_id: int) -> dict:
|
||
"""空态：库里没有章程行。
|
||
|
||
明确返回 filled=false —— 前端据此显示「尚未填写」并给录入入口，
|
||
**不用任何默认/示例文案填充**（派单红线：真实数据非 Demo）。
|
||
"""
|
||
return {"entity_id": entity_id, "filled": False, "data": None}
|
||
|
||
|
||
@router.get("")
|
||
def get_charter(
|
||
entity_id: int = Depends(get_entity_id),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
"""读取当前账套的战略章程（无行 → filled=false）。"""
|
||
row = (
|
||
db.query(StrategyCharter)
|
||
.filter(StrategyCharter.entity_id == entity_id)
|
||
.first()
|
||
)
|
||
if not row:
|
||
return _empty_state(entity_id)
|
||
payload = _empty_state(entity_id)
|
||
payload["filled"] = True
|
||
payload["data"] = _to_dict(row)
|
||
return payload
|
||
|
||
|
||
def _clean_text(raw, field: str, max_len: int) -> str:
|
||
if raw is None:
|
||
raise _bad(f"缺少必填字段 {field}")
|
||
if not isinstance(raw, str):
|
||
raise _bad(f"字段 {field} 必须是字符串")
|
||
val = raw.strip()
|
||
if not val:
|
||
raise _bad(f"字段 {field} 不能为空")
|
||
if len(val) > max_len:
|
||
raise _bad(f"字段 {field} 超长（最多 {max_len} 字符，当前 {len(val)}）")
|
||
return val
|
||
|
||
|
||
def _clean_value_items(raw):
|
||
"""价值观条目：[{name, meaning}]，每条须可作决策冲突时的裁决依据。"""
|
||
if raw is None:
|
||
raise _bad("缺少必填字段 value_items（价值观条目数组）")
|
||
if not isinstance(raw, list):
|
||
raise _bad("value_items 必须是数组 [{name, meaning}]")
|
||
if not raw:
|
||
raise _bad("value_items 至少需要 1 条（价值观须可作裁决依据）")
|
||
if len(raw) > VALUE_ITEMS_MAX:
|
||
raise _bad(f"value_items 最多 {VALUE_ITEMS_MAX} 条")
|
||
items = []
|
||
for idx, item in enumerate(raw, start=1):
|
||
if not isinstance(item, dict):
|
||
raise _bad(f"value_items 第 {idx} 条必须是对象 {{name, meaning}}")
|
||
extra = sorted(k for k in item.keys() if k not in ("name", "meaning"))
|
||
if extra:
|
||
raise _bad(f"value_items 第 {idx} 条含非白名单键 {extra}；只允许 name / meaning")
|
||
name = _clean_text(item.get("name"), f"value_items[{idx}].name", VALUE_ITEM_NAME_MAX_LEN)
|
||
meaning = _clean_text(item.get("meaning"), f"value_items[{idx}].meaning",
|
||
VALUE_ITEM_MEANING_MAX_LEN)
|
||
items.append({"name": name, "meaning": meaning})
|
||
return items
|
||
|
||
|
||
@router.put("")
|
||
def put_charter(
|
||
body: dict,
|
||
entity_id: int = Depends(get_entity_id),
|
||
db: Session = Depends(get_db),
|
||
user: User = Depends(require_role("ceo", "finance")),
|
||
):
|
||
"""写入/更新当前账套的战略章程（单行 upsert，字段白名单 + 值域校验）。"""
|
||
if not isinstance(body, dict) or not body:
|
||
raise _bad("请求体不能为空")
|
||
|
||
# ① 越权账套参数：显式 entity_id 与 token 账套不一致 → 403（不静默覆盖）
|
||
raw_entity = body.get("entity_id")
|
||
if raw_entity is not None:
|
||
try:
|
||
claimed = int(raw_entity)
|
||
except (TypeError, ValueError):
|
||
raise _bad(f"entity_id 无效: {raw_entity!r}")
|
||
if claimed != entity_id:
|
||
raise HTTPException(
|
||
status_code=403,
|
||
detail=f"无权写入企业 entity_id={claimed}（当前账套: {entity_id}）",
|
||
)
|
||
|
||
# ② 字段白名单：白名单外字段一律拒绝（含 entity_id / id / updated_by 等审计字段）
|
||
unknown = sorted(k for k in body.keys() if k not in CHARTER_FIELDS)
|
||
if unknown:
|
||
raise _bad(
|
||
f"非白名单字段 {unknown}；允许字段 {list(CHARTER_FIELDS)}"
|
||
f"（entity_id 由登录账套决定，不可手写）"
|
||
)
|
||
|
||
# ③ 必填校验（vision_year / vision_measurable 不可空）
|
||
missing = [f for f in REQUIRED_FIELDS if f not in body or body.get(f) is None]
|
||
if missing:
|
||
raise _bad(f"缺少必填字段 {missing}；vision_year 与 vision_measurable 不可空")
|
||
|
||
mission = _clean_text(body.get("mission"), "mission", MISSION_MAX_LEN)
|
||
vision = _clean_text(body.get("vision"), "vision", VISION_MAX_LEN)
|
||
values_text = _clean_text(body.get("values_text"), "values_text", VALUES_TEXT_MAX_LEN)
|
||
|
||
# ④ vision_year：4 位年份（接受 int 2030 或 "2030"）
|
||
raw_year = body.get("vision_year")
|
||
if isinstance(raw_year, bool) or not isinstance(raw_year, (int, str)):
|
||
raise _bad("vision_year 必须是年份（如 2030）")
|
||
year = str(raw_year).strip()
|
||
if not re.match(VISION_YEAR_RE, year):
|
||
raise _bad(f"vision_year 须为 4 位年份（如 2030），收到 {raw_year!r}")
|
||
|
||
measurable = _clean_text(body.get("vision_measurable"), "vision_measurable",
|
||
VISION_MEASURABLE_MAX_LEN)
|
||
value_items = _clean_value_items(body.get("value_items"))
|
||
|
||
row = (
|
||
db.query(StrategyCharter)
|
||
.filter(StrategyCharter.entity_id == entity_id)
|
||
.first()
|
||
)
|
||
if row is None:
|
||
row = StrategyCharter(entity_id=entity_id)
|
||
db.add(row)
|
||
|
||
row.mission = mission
|
||
row.vision = vision
|
||
row.vision_year = year
|
||
row.vision_measurable = measurable
|
||
row.values_text = values_text
|
||
row.value_items = value_items
|
||
row.updated_by = getattr(user, "id", None)
|
||
row.updated_by_name = getattr(user, "name", None)
|
||
|
||
db.commit()
|
||
db.refresh(row)
|
||
|
||
payload = _empty_state(entity_id)
|
||
payload["filled"] = True
|
||
payload["data"] = _to_dict(row)
|
||
payload["message"] = "战略章程已保存"
|
||
return payload
|