From 424c33c0857f72dfb970d91503f2f0bc0e047b2e Mon Sep 17 00:00:00 2001 From: wecom-fullstack Date: Tue, 13 Oct 2026 20:33:42 +0800 Subject: [PATCH] =?UTF-8?q?feat(kpi,review):=20KPI=20direction=20=E5=88=97?= =?UTF-8?q?=20+=20KPITarget=20=E6=A8=A1=E5=9E=8B=20+=20=E5=A4=8D=E7=9B=98?= =?UTF-8?q?=E4=BA=A7=E5=87=BA=E6=96=B9=E5=85=A5=E5=BA=93=EF=BC=88R1=20?= =?UTF-8?q?=E6=94=B6=E5=8F=A3=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit R1 收口批次（A 侧裁定 p12-ruling-input-r1-batch.RULING.md 主路径）· 单个快进提交： - backend/app/models/__init__.py（+42/−2）：CMA-KPI-20261003 阶段1 `direction`（KPIDefinition/KPITemplate） + ② KPITarget（kpi_targets，11 列 + uk_entity_kpi_period_type + 2 索引） - 复盘产出方三件（review_builder 含 getattr(direction) 硬化 / review_run / 测试件），原始提交件 26dc242 的重铸件 收口边界申报：本提交**不含** D-1 批次（VoucherDetail.row_uid / ImportLog uq_import_logs_entity_batch_type）—— 只读补证显示 prod `cma` 无该列/该唯一键（证据 out/r1-t1-batches.json + out/r1-t1-d1-placement.json）， 携行会导致 deploy 后 voucher_details 整实体 SELECT/INSERT 报 1054；D-1 另单（先 DDL 后 ORM）。 --- backend/app/models/__init__.py | 44 +- backend/app/utils/review_builder.py | 443 ++++++++++++++++++ backend/scripts/review_run.py | 209 +++++++++ .../test_review_records_write_20261012.py | 220 +++++++++ 4 files changed, 914 insertions(+), 2 deletions(-) create mode 100644 backend/app/utils/review_builder.py create mode 100644 backend/scripts/review_run.py create mode 100644 backend/tests/test_review_records_write_20261012.py diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 89cda4ec..c6e6ca47 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -1,5 +1,5 @@ """管理会计OS 数据模型""" -from sqlalchemy import Column, Integer, String, Text, Float, DateTime, ForeignKey, Boolean, JSON, func, UniqueConstraint, Numeric, Date +from sqlalchemy import Column, Integer, String, Text, Float, DateTime, ForeignKey, Boolean, JSON, func, text, UniqueConstraint, Index, Numeric, Date from app.database import Base from app.models.budget_plan import BudgetPlan @@ -93,7 +93,13 @@ class KPIDefinition(Base): data_owner = Column(String(100), default="待指定", comment="数据责任人") frequency = Column(String(20), default="monthly", comment="daily/weekly/monthly/quarterly/yearly") unit = Column(String(50), default="%", comment="单位") - target_value = Column(Float, nullable=True, comment="目标值(兼容旧字段)") + # 2026-10-03 KPI字典分层治理（任务 CMA-KPI-20261003）阶段1：direction 结构列。 + # 枚举 higher_better(越大越好)/lower_better(越小越好)，NULL=未定。 + # 既有行由 migrate_kpi_direction_20261003.py 按 threshold_green 比较符**结构推定**回填 + # （非口径裁定）；口径/阈值语义仍属 finance/DAMA 在途批次。 + direction = Column(String(16), nullable=True, + comment="指标方向: higher_better越大越好/lower_better越小越好/NULL未定") + target_value = Column(Float, nullable=True, comment="目标值(NULL=未设目标；0=真零目标；兼容旧字段)") target_monthly = Column(Float, nullable=True, comment="月度目标值") target_quarterly = Column(Float, nullable=True, comment="季度目标值") target_yearly = Column(Float, nullable=True, comment="年度目标值") @@ -491,6 +497,10 @@ class KPITemplate(Base): formula = Column(Text) formula_desc = Column(String(500)) unit = Column(String(50)) + # 2026-10-03 KPI字典分层治理（任务 CMA-KPI-20261003）阶段1：字典层同加 direction； + # 回填规则 = 继承同 kpi_code 实例的已定方向（见 docs/kpi-dictionary-layering-design.md §2.2）。 + direction = Column(String(16), nullable=True, + comment="指标方向: higher_better越大越好/lower_better越小越好/NULL未定") target_value = Column(Float) description = Column(Text) is_system = Column(Integer, default=0) @@ -864,6 +874,36 @@ class ObjectiveKPI(Base): __table_args__ = (UniqueConstraint("objective_id", "kpi_id", name="uk_obj_kpi"),) +class KPITarget(Base): + """KPI 期间目标（考核型·方案 A）—— ③ 读点收敛 CMA-KPI-20261003 + + 唯一真源（考核型）：`(entity_id, kpi_id, period, target_type)` 唯一； + 承诺型目标只写 `krs`（裁定 §二 边界声明：同一 KPI 不得两处均有 target）。 + + **列定义与 ② DDL 出件包 `up.sql` 逐字一致**（11 列 + uk_entity_kpi_period_type + 2 索引）； + `entity_id` 无 DDL DEFAULT（写入口必须显式传值，防静默默认把实体隔离降级为约定）； + `updated_at` 无 `ON UPDATE CURRENT_TIMESTAMP`（由 ORM/写入口维护，与现网 kpi_definitions 同风格）。 + """ + __tablename__ = "kpi_targets" + id = Column(Integer, primary_key=True, autoincrement=True) + entity_id = Column(Integer, nullable=False, comment="账套(企业ID)；写入口必须显式传值（实体隔离）") + kpi_id = Column(Integer, nullable=False, comment="指标实例 → kpi_definitions.id") + period = Column(String(20), nullable=False, comment="期间: 2026-08(月)/2026Q3(季)/2026(年)") + target_type = Column(String(16), nullable=False, comment="目标粒度: monthly/quarterly/yearly") + target_value = Column(Numeric(15, 2), nullable=True, comment="NULL=该期未设;0=真零") + source = Column(String(20), nullable=False, server_default=text("'manual'"), + comment="manual/okr/budget/import") + note = Column(String(255), nullable=True, comment="备注") + created_by = Column(String(64), nullable=True, comment="写入者(用户/Bot标识)") + created_at = Column(DateTime, nullable=False, server_default=func.now()) + updated_at = Column(DateTime, nullable=False, server_default=func.now()) + __table_args__ = ( + UniqueConstraint("entity_id", "kpi_id", "period", "target_type", name="uk_entity_kpi_period_type"), + Index("ix_kpi_targets_entity_period", "entity_id", "period"), + Index("ix_kpi_targets_kpi", "kpi_id"), + ) + + class KR(Base): """关键结果KR — OKR完整化 (O→KR→KPI)""" __tablename__ = "krs" diff --git a/backend/app/utils/review_builder.py b/backend/app/utils/review_builder.py new file mode 100644 index 00000000..65afdf21 --- /dev/null +++ b/backend/app/utils/review_builder.py @@ -0,0 +1,443 @@ +# -*- coding: utf-8 -*- +"""复盘内容生成器 —— 从 live 数据构建「O/KR/KPI 现值 + 偏差 + 关联行动」（2026-10-12） + +任务: yanxue-20261012-review-records-write（任总 2026-10-12 P0-1 立单） +背景: `review_records` 表 2026-08-27 建、78 天零行 —— 写入链路（API/前端）已通， + 缺的是**产出真实复盘内容的生成器**与**触发口径**。本模块补前者。 + +设计纪律（为什么这么写）: + 1. **只读**：本模块只 SELECT，不产生任何写动作；落库由调用方（scripts/review_run.py） + 经已部署的 `POST /api/cma/maps/{id}/review-records` 完成 —— 写入路径保持单一。 + 2. **不另立口径**： + · KPI 现值取数沿用实测区口径 `app.utils.kpi_value_scope.actual_only_clause()` + （预估行不作实测值），与 `maps.py /review`、看板同源； + · 期间过滤沿用 `period <= 当前月`（与 verify 引擎一致，不取未来期间）； + · 目标值读 `target_monthly or target_value`（与 maps.py /review 一致）。 + 3. **方向/色阶按契约**（references/kpi-direction-and-attainment-contract.md）: + · `direction IS NULL` **不默认正向** —— 标 `未配方向`，不做达成率判定； + · `unit='状态'` 走 1=达成 / 0、NULL=未达成，不走达成率； + · 目标 0 / 缺失 / 现值缺失 ⇒ 一律 `info`（灰），禁除零； + · **偏差 = 现值 − 目标**（有符号），不依赖方向 ⇒ 全量可比，是本复盘的主判据。 + 4. 关联行动按**库内既有挂靠**取（ActionPlan.kpi_id / objective_id / kr_id）， + 不发明新挂靠；挂靠缺失如实计入 `linkage_gaps`（这正是 PDCA 报告指出的「89% 行动未挂 O」）。 +""" +from __future__ import annotations + +from datetime import date, datetime +from typing import Any, Dict, List, Optional + +from sqlalchemy.orm import Session + +from app.models import ActionPlan, KPIDefinition, KPIValue, KR, Objective, StrategicMap +from app.utils.kpi_value_scope import actual_only_clause + +# 色阶阈值（与 maps.py /review 同口径：绿≥90% / 黄60-90% / 红<60%） +GREEN_RATIO = 0.9 +YELLOW_RATIO = 0.6 + +REVIEW_TYPES = ("monthly", "quarterly") +REVIEW_TYPE_LABEL = {"monthly": "月度运营复盘", "quarterly": "季度战略回顾"} + + +def _f(v) -> Optional[float]: + """JSON 列可能回读为 Decimal / str；转不了返回 None（不抛）。""" + if v is None: + return None + try: + return float(v) + except (TypeError, ValueError): + return None + + +def _period_limit(ref: Optional[str] = None) -> str: + """现值期间上限：'YYYY-MM'（当月及以前；不取未来期间）。""" + if ref: + return str(ref)[:7] + return datetime.now().strftime("%Y-%m") + + +def _level_from_ratio(ratio: Optional[float]) -> str: + if ratio is None: + return "info" + if ratio >= GREEN_RATIO: + return "green" + if ratio >= YELLOW_RATIO: + return "yellow" + return "red" + + +def _kpi_reading(kpi: KPIDefinition, actual: Optional[float], target: Optional[float], + period: Optional[str]) -> Dict[str, Any]: + """单个 KPI 的现值/偏差/判定。方向未配不判定（契约 §2）。""" + direction = (getattr(kpi, "direction", None) or "").strip() or None + out: Dict[str, Any] = { + "kpi_id": kpi.id, + "kpi_code": kpi.kpi_code, + "kpi_name": kpi.kpi_name, + "unit": kpi.unit, + "direction": direction or "unset", + "period": period, + "actual_value": actual, + "target_value": target, + "deviation": None, + "deviation_pct": None, + "attainment": None, + "level": "info", + "verdict": "无法判定", + } + if actual is not None and target is not None: + out["deviation"] = round(actual - target, 4) + if target != 0: + out["deviation_pct"] = round((actual - target) / abs(target) * 100, 2) + + # 状态类：不走达成率（契约 §2） + if (kpi.unit or "").strip() == "状态": + if actual == 1: + out.update(level="green", verdict="达成") + else: + out.update(level="red", verdict="未达成") + return out + + # 零点/空值保护（契约 §3） + if actual is None or target is None or target == 0: + out["verdict"] = "数据不足" if actual is None else "目标缺失/为0" + return out + + # 方向未配 ⇒ 不判达成率，只给偏差（契约 §2：禁止默认正向） + if direction is None: + out["verdict"] = "方向未配，仅列现值/偏差" + return out + + ratio = target / actual if direction == "lower_better" else actual / target + if ratio is None or ratio != ratio or ratio in (float("inf"), float("-inf")): + out["verdict"] = "数据异常" + return out + out["attainment"] = round(ratio, 4) + out["level"] = _level_from_ratio(ratio) + out["verdict"] = {"green": "达标", "yellow": "接近达标", "red": "未达标"}[out["level"]] + return out + + +def _latest_kpi_values(db: Session, kpi_ids: List[int], entity_id: int, + limit: str) -> Dict[int, Dict[str, Any]]: + """每个 KPI 取 period <= 当前月 的最新一条**实测区**值（预估行不作实测）。""" + out: Dict[int, Dict[str, Any]] = {} + for kid in kpi_ids: + v = ( + db.query(KPIValue) + .filter( + KPIValue.kpi_id == kid, + KPIValue.entity_id == entity_id, + KPIValue.actual_value.isnot(None), + KPIValue.period <= limit, + actual_only_clause(), + ) + .order_by(KPIValue.period.desc(), KPIValue.calculated_at.desc(), KPIValue.id.desc()) + .first() + ) + if v: + out[kid] = {"actual_value": _f(v.actual_value), "period": v.period, + "source_type": v.source_type, "data_status": v.data_status} + return out + + +def build_review(db: Session, entity_id: int, map_id: int, review_type: str = "quarterly", + period_ref: Optional[str] = None, review_date: Optional[date] = None, + top_n: int = 8) -> Dict[str, Any]: + """构建一条复盘记录载荷（只读）。 + + 返回 {summary, adjustments, next_priorities, snapshot, meta}： + · summary 人类可读复盘结论（落 review_records.summary，复盘页可直接读回） + · adjustments 结构化「O/KR/KPI 现值 + 偏差 + 关联行动」清单 + （落 review_records.adjustments；条目形如 + {type, target, action, current_value, target_value, deviation, level}） + · next_priorities 下期优先级（落 review_records.next_priorities） + · snapshot 本次复盘的只读数据快照（供 --print/留痕；DB 侧由 adjustments 承载） + """ + if review_type not in REVIEW_TYPES: + raise ValueError(f"review_type 必须是 {REVIEW_TYPES} 之一，收到 {review_type!r}") + review_date = review_date or datetime.now().date() + limit = _period_limit(period_ref) + + m = db.query(StrategicMap).filter(StrategicMap.id == map_id, + StrategicMap.entity_id == entity_id).first() + if not m: + raise ValueError(f"战略地图不存在: map_id={map_id} entity_id={entity_id}") + + dims = m.dimensions + if isinstance(dims, str): + import json as _json + dims = _json.loads(dims) + dims = dims or [] + + # ── 1. 战略地图维度节点（O 的战略面）─────────────────────────────── + map_nodes: List[Dict[str, Any]] = [] + map_kpi_codes: set[str] = set() + for dim in dims: + for obj in dim.get("objectives") or []: + map_nodes.append({"dimension": dim.get("key"), "dimension_name": dim.get("name"), + "name": obj.get("name"), "kpis": obj.get("kpis") or []}) + map_kpi_codes.update(obj.get("kpis") or []) + + # ── 2. 目标级 O / KR ───────────────────────────────────────────── + objectives = ( + db.query(Objective).filter(Objective.entity_id == entity_id).order_by(Objective.id).all() + ) + obj_ids = [o.id for o in objectives] + krs = ( + db.query(KR).filter(KR.entity_id == entity_id, KR.objective_id.in_(obj_ids or [0])) + .order_by(KR.id).all() + ) if obj_ids else [] + obj_by_id = {o.id: o for o in objectives} + + kr_kpi_ids = [k.metric_kpi_id for k in krs if k.metric_kpi_id] + + # ── 3. 关联行动（按库内既有挂靠：kpi_id / objective_id / kr_id）───── + plans = db.query(ActionPlan).filter(ActionPlan.entity_id == entity_id).all() + kpi_ids_from_plans = sorted({p.kpi_id for p in plans if p.kpi_id}) + + kpi_defs = ( + db.query(KPIDefinition) + .filter(KPIDefinition.entity_id == entity_id, + (KPIDefinition.id.in_(set(kr_kpi_ids) | set(kpi_ids_from_plans)) if + (kr_kpi_ids or kpi_ids_from_plans) else False)) + .all() + ) if (kr_kpi_ids or kpi_ids_from_plans) else [] + if map_kpi_codes: + kpi_defs = kpi_defs + db.query(KPIDefinition).filter( + KPIDefinition.entity_id == entity_id, + KPIDefinition.kpi_code.in_(map_kpi_codes), + ).all() + seen_kpi_ids: set[int] = set() + kpi_defs = [k for k in kpi_defs if not (k.id in seen_kpi_ids or seen_kpi_ids.add(k.id))] + kpi_ids = [k.id for k in kpi_defs] + kpi_by_id = {k.id: k for k in kpi_defs} + + latest = _latest_kpi_values(db, kpi_ids, entity_id, limit) + + kpi_rows: List[Dict[str, Any]] = [] + for k in sorted(kpi_defs, key=lambda x: (x.kpi_code or "")): + lv = latest.get(k.id) or {} + target = _f(k.target_monthly) if k.target_monthly is not None else _f(k.target_value) + row = _kpi_reading(k, lv.get("actual_value"), target, lv.get("period")) + row["kpi_status"] = k.status + row["kpi_level"] = k.kpi_level + kpi_rows.append(row) + + # 关联行动：kpi_id / objective_id / kr_id 任一命中即关联（不发明挂靠） + in_scope_kpi_ids = {r["kpi_id"] for r in kpi_rows} + kr_ids = {k.id for k in krs} + action_rows: List[Dict[str, Any]] = [] + for p in plans: + hit = [] + if p.kpi_id and p.kpi_id in in_scope_kpi_ids: + hit.append("kpi") + if p.objective_id and p.objective_id in set(obj_ids): + hit.append("objective") + if p.kr_id and p.kr_id in kr_ids: + hit.append("kr") + if not hit: + continue + k = kpi_by_id.get(p.kpi_id) + action_rows.append({ + "plan_id": p.id, + "title": p.title, + "status": p.status, + "progress": p.progress or 0, + "assignee": p.assignee, + "due_date": p.due_date.date().isoformat() if p.due_date else None, + "verify_status": p.verify_status, + "linked_by": hit, + "kpi_code": k.kpi_code if k else None, + "kpi_name": k.kpi_name if k else None, + "objective_id": p.objective_id, + "kr_id": p.kr_id, + }) + + # ── 4. 偏差清单（主判据：现值 − 目标；方向未配者只列偏差不判达标）──── + # 可比偏差：目标已设且非 0 且现值可用。目标为 0/未设者单位间不可比， + # 单列 `unset_target`（这本身是「补齐 KPI 目标」的待办，不混进排名）。 + _rank = {"red": 0, "yellow": 1, "green": 2, "info": 3} + comparable = [r for r in kpi_rows + if r["deviation"] is not None and (r["target_value"] or 0) != 0] + unset_target = [r for r in kpi_rows if (r["target_value"] or 0) == 0] + # 排序：非达标（红/黄/未配方向不可判）在前，各自按 |偏差%| 降序 —— 复盘的"要点先看"。 + _bad = sorted([r for r in comparable if r["level"] != "green"], + key=lambda r: (-abs(r["deviation_pct"] or 0), r["kpi_code"] or "")) + _ok = sorted([r for r in comparable if r["level"] == "green"], + key=lambda r: (-abs(r["deviation_pct"] or 0), r["kpi_code"] or "")) + deviations = _bad + _ok + + linkage_gaps = { + "objectives_without_map": sum(1 for o in objectives if not o.strategic_map_id), + "objectives_without_quarter": sum(1 for o in objectives if not (o.quarter or "").strip()), + "krs_without_kpi": sum(1 for k in krs if not k.metric_kpi_id), + "krs_without_due_date": sum(1 for k in krs if not k.due_date), + "krs_progress_zero": sum(1 for k in krs if not (k.progress or 0)), + "plans_total": len(plans), + "plans_linked_objective": sum(1 for p in plans if p.objective_id), + "plans_linked_kr": sum(1 for p in plans if p.kr_id), + "plans_linked_kpi": sum(1 for p in plans if p.kpi_id), + "plans_with_rule": sum(1 for p in plans if p.auto_verify_rule), + "plans_without_due_date": sum(1 for p in plans if not p.due_date), + } + + # ── 5. 结构化「调整项 / 现值+偏差+行动」清单（落 adjustments）──────── + adjustments: List[Dict[str, Any]] = [] + for r in kpi_rows: + linked = [a["title"] for a in action_rows if a["kpi_code"] == r["kpi_code"]] + adjustments.append({ + "type": "kpi", + "target": r["kpi_code"], + "target_name": r["kpi_name"], + "action": ("；".join(linked) if linked else "未挂行动计划，需补挂责任人与规则"), + "current_value": r["actual_value"], + "target_value": r["target_value"], + "deviation": r["deviation"], + "deviation_pct": r["deviation_pct"], + "period": r["period"], + "unit": r["unit"], + "level": r["level"], + "verdict": r["verdict"], + "linked_plan_ids": [a["plan_id"] for a in action_rows if a["kpi_code"] == r["kpi_code"]], + }) + for k in krs: + o = obj_by_id.get(k.objective_id) + adjustments.append({ + "type": "kr", + "target": k.title, + "target_name": f"{o.title}/{k.title}" if o else k.title, + "action": (f"目标值 {_f(k.target_value)}，现值 {_f(k.current_value)}" + if k.current_value is not None else "尚无实测现值，需接 KPI 或人工填报"), + "current_value": _f(k.current_value), + "target_value": _f(k.target_value), + "deviation": (round(_f(k.current_value) - _f(k.target_value), 4) + if k.current_value is not None and k.target_value is not None else None), + "progress": k.progress or 0, + "status": k.status, + "due_date": k.due_date.date().isoformat() if k.due_date else None, + "level": "info", + "verdict": ("进行中" if (k.progress or 0) > 0 else "无进度"), + }) + for o in objectives: + adjustments.append({ + "type": "objective", + "target": o.title, + "target_name": o.title, + "action": (f"季度 {o.quarter}" if (o.quarter or "").strip() else "未填季度，需补 SMARTER 的 T"), + "dimension": o.dimension, + "current_value": o.progress, + "target_value": 100, + "progress": o.progress or 0, + "status": o.status, + "owner": o.owner, + "level": "info", + "verdict": "未开工" if not (o.progress or 0) else "进行中", + }) + + # ── 6. 下期优先级（确定性规则，不用 LLM）──────────────────────────── + priorities: List[str] = [] + for r in _bad[:3]: + priorities.append( + f"{r['kpi_code']} 偏差 {r['deviation']:+.2f}{r['unit'] or ''}" + f"（现值 {r['actual_value']} / 目标 {r['target_value']}，{r['verdict']}）" + ) + if unset_target: + priorities.append(f"补齐 KPI 目标：{len(unset_target)} 项目标为 0/未设，无目标即无法判达标" + f"（如 {', '.join(r['kpi_code'] for r in unset_target[:4])}）") + if linkage_gaps["krs_progress_zero"]: + priorities.append(f"KR 进度回填：{linkage_gaps['krs_progress_zero']}/{len(krs)} 条进度为 0，接 KPI 自动算进度或人工填报") + if linkage_gaps["plans_linked_kr"] < linkage_gaps["plans_total"]: + priorities.append(f"行动挂靠补齐：{linkage_gaps['plans_total'] - linkage_gaps['plans_linked_kr']} 条行动未挂 KR（现挂 KR {linkage_gaps['plans_linked_kr']} 条）") + if linkage_gaps["krs_without_due_date"]: + priorities.append(f"KR 截止日期补齐：{linkage_gaps['krs_without_due_date']}/{len(krs)} 条无 due_date") + if linkage_gaps["objectives_without_quarter"]: + priorities.append(f"目标季度补齐：{linkage_gaps['objectives_without_quarter']}/{len(objectives)} 条未填 quarter") + + # ── 7. 人读结论（落 review_records.summary，复盘页时间线直接渲染）──── + label = REVIEW_TYPE_LABEL[review_type] + lines: List[str] = [] + lines.append(f"【{label}】{m.title}") + lines.append(f"复盘日期 {review_date.isoformat()}｜口径：KPI 现值取实测区最新值（period ≤ {limit}），" + f"偏差 = 现值 − 目标（有符号）") + lines.append("") + lines.append(f"一、目标级（小 PDCA）现状：O {len(objectives)} 条、KR {len(krs)} 条；" + f"O 进度非零 {len(objectives) - sum(1 for o in objectives if not (o.progress or 0))} 条、" + f"KR 有进度 {len(krs) - linkage_gaps['krs_progress_zero']} 条、" + f"KR 已挂 KPI {len(krs) - linkage_gaps['krs_without_kpi']} 条。") + dims_summary = ", ".join(f"{n['dimension_name']}({sum(1 for x in map_nodes if x['dimension'] == n['dimension'])})" + for n in [{"dimension": d.get("key"), "dimension_name": d.get("name")} for d in dims]) + lines.append(f"　　战略地图节点 {len(map_nodes)} 个：{dims_summary}") + lines.append("") + lines.append(f"二、KPI 现值与偏差（{len(kpi_rows)} 项纳入复盘；未达标/不可判在前，各按 |偏差%| 降序，" + f"列前 {min(top_n, len(deviations))}）：") + if deviations: + for r in deviations[:top_n]: + pct = (f"（{r['deviation_pct']:+.1f}%）" if r["deviation_pct"] is not None else "（目标缺失，无百分比）") + lines.append(f"　· {r['kpi_code']} {r['kpi_name']}：现值 {r['actual_value']}{r['unit'] or ''}" + f"（{r['period']}）/ 目标 {r['target_value']}{r['unit'] or ''}" + f"　偏差 {r['deviation']:+.2f}{r['unit'] or ''}" + f"{pct}　判定：{r['verdict']}") + else: + lines.append("　· 本期无可比偏差（无「目标已设且现值可用」的 KPI）") + if unset_target: + lines.append(f"　（另 {len(unset_target)} 项 KPI 目标为 0 或未设 ⇒ 无可比偏差，需先补目标：" + + ", ".join(r["kpi_code"] for r in unset_target[:6]) + + ("…" if len(unset_target) > 6 else "") + "）") + no_dir = [r for r in kpi_rows if r["direction"] == "unset"] + if no_dir: + lines.append(f"　（其中 {len(no_dir)} 项 KPI 未配方向，按契约只列现值/偏差、不作达标判定：" + + ", ".join(r["kpi_code"] for r in no_dir[:6]) + + ("…" if len(no_dir) > 6 else "") + "）") + lines.append("") + lines.append(f"三、关联行动：命中 {len(action_rows)} 条（挂 KPI {sum(1 for a in action_rows if 'kpi' in a['linked_by'])}、" + f"挂 O {sum(1 for a in action_rows if 'objective' in a['linked_by'])}、" + f"挂 KR {sum(1 for a in action_rows if 'kr' in a['linked_by'])}）；" + f"状态 completed {sum(1 for a in action_rows if a['status'] == 'completed')} / " + f"in_progress {sum(1 for a in action_rows if a['status'] == 'in_progress')} / " + f"pending {sum(1 for a in action_rows if a['status'] == 'pending')}") + lines.append(f"　　全量行动 {linkage_gaps['plans_total']} 条中：挂 O {linkage_gaps['plans_linked_objective']} 条、" + f"挂 KR {linkage_gaps['plans_linked_kr']} 条、已配验证规则 {linkage_gaps['plans_with_rule']} 条、" + f"无截止日期 {linkage_gaps['plans_without_due_date']} 条") + lines.append("") + lines.append("四、为什么没达成 / 关键偏差：") + if deviations: + top = deviations[0] + lines.append(f"　· 最大偏差项 {top['kpi_code']}：{top['verdict']}，偏差 {top['deviation']:+.2f}{top['unit'] or ''}" + f"（现值 {top['actual_value']} vs 目标 {top['target_value']}）") + lines.append(f"　· 结构性问题：O 未挂战略地图 {linkage_gaps['objectives_without_map']} 条、" + f"未填季度 {linkage_gaps['objectives_without_quarter']} 条；KR 无 KPI 挂靠 {linkage_gaps['krs_without_kpi']} 条、" + f"无截止日期 {linkage_gaps['krs_without_due_date']} 条；行动未挂 KR {linkage_gaps['plans_total'] - linkage_gaps['plans_linked_kr']} 条") + lines.append("") + lines.append("五、下期优先级：") + for i, p in enumerate(priorities, 1): + lines.append(f"　{i}. {p}") + + summary = "\n".join(lines) + + snapshot = { + "generated_at": datetime.now().isoformat(timespec="seconds"), + "entity_id": entity_id, + "map_id": map_id, + "map_title": m.title, + "review_type": review_type, + "review_date": review_date.isoformat(), + "period_limit": limit, + "map_nodes": map_nodes, + "kpis": kpi_rows, + "kpis_unset_target": [r["kpi_code"] for r in unset_target], + "objectives": [{"id": o.id, "title": o.title, "dimension": o.dimension, + "quarter": o.quarter, "status": o.status, "progress": o.progress, + "owner": o.owner} for o in objectives], + "krs": [{"id": k.id, "objective_id": k.objective_id, "title": k.title, + "metric_kpi_id": k.metric_kpi_id, "target_value": _f(k.target_value), + "current_value": _f(k.current_value), "progress": k.progress, + "status": k.status} for k in krs], + "actions": action_rows, + "linkage_gaps": linkage_gaps, + } + return {"summary": summary, "adjustments": adjustments, + "next_priorities": priorities, "snapshot": snapshot, + "meta": {"entity_id": entity_id, "map_id": map_id, "review_type": review_type, + "review_date": review_date.isoformat(), "period_limit": limit, + "kpi_count": len(kpi_rows), "action_count": len(action_rows)}} diff --git a/backend/scripts/review_run.py b/backend/scripts/review_run.py new file mode 100644 index 00000000..9bb7649c --- /dev/null +++ b/backend/scripts/review_run.py @@ -0,0 +1,209 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""复盘落地执行器 —— 生成「月度运营复盘 / 季度战略回顾」并经既有 API 落库（2026-10-12） + +任务: yanxue-20261012-review-records-write（任总 2026-10-12 P0-1） +链路: app/utils/review_builder.build_review（只读算） + → POST /api/cma/maps/{map_id}/review-records（**唯一写入口**，已部署端点，不改后端） + → GET /api/cma/maps/{map_id}/review-records（读回校验） + +用法（默认只预览，不写）: + cd /srv/hermes/shared/cma-management/backend + ./venv/bin/python3 scripts/review_run.py --entity 2 --map 49 --type monthly --print + ./venv/bin/python3 scripts/review_run.py --entity 2 --map 49 --type quarterly --write \ + --evidence /path/evidence.json + +写动作纪律: + · **幂等（2026-10-13 P1-1）**：--write 前先按同一读路径列出该 map 的复盘；同日同 review_type 已存在 ⇒ NOOP 零写（`--force` 才强制再写）。 + · 幂等闸读不到（列表接口非 200）⇒ **fail-closed 不写**（rc=6），宁可缺一条也不产重复行。 + · 本脚本不做任何直接 SQL 写；落库一律经 HTTP API（走鉴权/多租户/审计同一路径）。 + · 凭据进程内从 600 权限文件读取（--admin-env 或 $CMA_ADMIN_ENV_FILE），不入命令行/日志。 + · 生产写前先过 approval-gate（调用方负责），本脚本只负责「按放行结果执行一次」。 +""" +from __future__ import annotations + +import argparse +import json +import os +import sys +import urllib.error +import urllib.request +from datetime import date, datetime + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from app.database import get_db # noqa: E402 +from app.utils.review_builder import ( # noqa: E402 + REVIEW_TYPES, build_review, +) + + +def _load_env_file(path: str) -> dict: + d = {} + with open(path) as fh: + for line in fh: + line = line.strip() + if line and not line.startswith("#") and "=" in line: + k, v = line.split("=", 1) + d[k.strip()] = v.strip() + return d + + +def _http(method: str, url: str, token=None, body=None, timeout=30): + data = json.dumps(body).encode() if body is not None else None + req = urllib.request.Request(url, data=data, method=method) + req.add_header("Content-Type", "application/json") + if token: + req.add_header("Authorization", "Bearer " + token) + try: + with urllib.request.urlopen(req, timeout=timeout) as r: + return r.status, json.loads(r.read().decode()) + except urllib.error.HTTPError as e: + return e.code, e.read().decode()[:800] + except Exception as e: # noqa: BLE001 + return -1, repr(e) + + +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("--entity", type=int, required=True) + ap.add_argument("--map", type=int, required=True, dest="map_id") + ap.add_argument("--type", required=True, choices=list(REVIEW_TYPES)) + ap.add_argument("--date", default=None, help="复盘日期 YYYY-MM-DD（缺省今天）") + ap.add_argument("--period", default=None, help="现值期间上限 YYYY-MM（缺省当月）") + ap.add_argument("--api-base", default="http://127.0.0.1:8010") + ap.add_argument("--admin-env", default=os.environ.get("CMA_ADMIN_ENV_FILE", ""), + help="登录凭据文件（600）；仅 --write 需要") + ap.add_argument("--print", dest="do_print", action="store_true", help="打印复盘全文（预览）") + ap.add_argument("--json-out", default=None, help="把生成载荷落 JSON（留痕）") + ap.add_argument("--write", action="store_true", help="经 API 落库（默认不写）") + ap.add_argument("--force", action="store_true", + help="跳过同日同 type 幂等闸（默认关；常态化触发勿用）") + ap.add_argument("--evidence", default=None, help="写模式证据 JSON 落盘路径") + args = ap.parse_args() + + rdate = datetime.strptime(args.date, "%Y-%m-%d").date() if args.date else date.today() + + db = next(get_db()) + try: + payload = build_review(db, entity_id=args.entity, map_id=args.map_id, + review_type=args.type, period_ref=args.period, + review_date=rdate) + finally: + try: + db.close() + except Exception: # noqa: BLE001 + pass + + post_body = { + "review_type": args.type, + "review_date": rdate.isoformat(), + "summary": payload["summary"], + "adjustments": payload["adjustments"], + "next_priorities": payload["next_priorities"], + } + + if args.do_print: + print(payload["summary"]) + print("\n--- next_priorities ---") + for i, p in enumerate(payload["next_priorities"], 1): + print(f"{i}. {p}") + print(f"\n--- meta --- {json.dumps(payload['meta'], ensure_ascii=False)}") + print(f"--- adjustments 条目数 {len(payload['adjustments'])} " + f"(kpi {sum(1 for a in payload['adjustments'] if a['type'] == 'kpi')} / " + f"kr {sum(1 for a in payload['adjustments'] if a['type'] == 'kr')} / " + f"objective {sum(1 for a in payload['adjustments'] if a['type'] == 'objective')})") + + if args.json_out: + with open(args.json_out, "w") as fh: + json.dump({"post_body": post_body, "snapshot": payload["snapshot"]}, + fh, ensure_ascii=False, indent=1, default=str) + print(f"[json-out] {args.json_out}") + + if not args.write: + print("[preview] 未写库（需 --write）") + return 0 + + if not args.admin_env or not os.path.exists(args.admin_env): + print(f"❌ --write 需要凭据文件（--admin-env 或 $CMA_ADMIN_ENV_FILE）：{args.admin_env}") + return 2 + cred = _load_env_file(args.admin_env) + st, res = _http("POST", f"{args.api_base}/api/cma/auth/login", body={ + "username": cred["CMA_ADMIN_USER"], + "password": cred["CMA_ADMIN_PASS"], + "entity_id": int(cred["CMA_ADMIN_ENTITY_ID"]), + }) + if st != 200: + print(f"❌ 登录失败 {st}: {res}") + return 3 + token = res["token"] + + # ---- 幂等闸（P1-1）：同日同 type 已存在 ⇒ NOOP（零写）---- + if not args.force: + st_l0, listed0 = _http("GET", + f"{args.api_base}/api/cma/maps/{args.map_id}/review-records", + token=token) + if st_l0 != 200 or not isinstance(listed0, dict): + print(f"❌ 幂等闸无法读取历史复盘（GET list={st_l0}）⇒ fail-closed 不写，rc=6") + return 6 + existing_id = None + for r0 in listed0.get("data", []): + if (r0.get("review_type") == args.type + and str(r0.get("review_date") or "")[:10] == rdate.isoformat()): + existing_id = r0.get("id") + break + if existing_id is not None: + print(f"[idempotent] {args.type} / {rdate.isoformat()} 已存在 id={existing_id} ⇒ NOOP（零写）") + if args.evidence: + with open(args.evidence, "w") as fh: + json.dump({"task": "yanxue-20261012-review-records-write", + "at": datetime.now().isoformat(timespec="seconds"), + "entity_id": args.entity, "map_id": args.map_id, + "review_type": args.type, "review_date": rdate.isoformat(), + "action": "NOOP", "record_id": existing_id, + "idempotent_guard": True, "list_status": st_l0, + "reason": "same review_type+review_date already present"}, + fh, ensure_ascii=False, indent=1, default=str) + print(f"[evidence] {args.evidence}") + return 0 + st, created = _http("POST", + f"{args.api_base}/api/cma/maps/{args.map_id}/review-records", + token=token, body=post_body) + print(f"[write] POST review-records -> {st}") + if st not in (200, 201): + print(f"❌ 落库失败: {created}") + return 4 + record_id = created.get("id") if isinstance(created, dict) else None + + # 读回校验（同 API，与复盘页同一读路径） + st_l, listed = _http("GET", f"{args.api_base}/api/cma/maps/{args.map_id}/review-records", + token=token) + st_d, detail = _http("GET", + f"{args.api_base}/api/cma/maps/{args.map_id}/review-records/{record_id}", + token=token) + ok = (st_l == 200 and st_d == 200 + and isinstance(listed, dict) and any(r.get("id") == record_id + for r in listed.get("data", []))) + print(f"[read-back] list={st_l} detail={st_d} 命中记录 {record_id}: {ok}") + + if args.evidence: + with open(args.evidence, "w") as fh: + json.dump({ + "task": "yanxue-20261012-review-records-write", + "at": datetime.now().isoformat(timespec="seconds"), + "entity_id": args.entity, "map_id": args.map_id, "review_type": args.type, + "review_date": rdate.isoformat(), + "record_id": record_id, + "post_status": st, "list_status": st_l, "detail_status": st_d, + "read_back_ok": ok, + "meta": payload["meta"], + "created": created, + "read_back": detail if isinstance(detail, dict) else detail, + }, fh, ensure_ascii=False, indent=1, default=str) + print(f"[evidence] {args.evidence}") + + return 0 if ok else 5 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/backend/tests/test_review_records_write_20261012.py b/backend/tests/test_review_records_write_20261012.py new file mode 100644 index 00000000..45febb6e --- /dev/null +++ b/backend/tests/test_review_records_write_20261012.py @@ -0,0 +1,220 @@ +"""复盘内容生成器 + 落库链路测试（任务 yanxue-20261012-review-records-write） + +覆盖: + 1. build_review 只读产出 summary / adjustments / next_priorities，含 KPI 现值+偏差+关联行动 + 2. 方向契约: lower_better 用 target/actual；direction IS NULL **不**默认正向（不作达标判定）； + unit='状态' 走 1=达成；目标 0 / 现值为空 ⇒ 不崩、不除零 + 3. 期间口径: period <= 当前月（不取未来期间）；预估行（estimated/forecast）不作实测现值 + 4. 落库链路: POST /maps/{id}/review-records → GET 读回（同一读路径，复盘页同源） +""" +from __future__ import annotations + +from datetime import date, datetime, timedelta + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy.orm import Session + +from app.models import (ActionPlan, KPIDefinition, KPIValue, KR, Objective, + StrategicMap) +from app.utils.review_builder import build_review +from tests.conftest import auth_header, create_test_user, get_token_for_user + + +def _map(db: Session, entity_id: int = 1) -> StrategicMap: + m = StrategicMap( + entity_id=entity_id, + title="测试战略地图", + version="v1.0", + status="published", + dimensions=[{"key": "finance", "name": "财务层", "icon": "F", "color": "#fff", + "objectives": [{"name": "营收目标", "kpis": ["RB_REV"]}]}], + ) + db.add(m) + db.commit() + db.refresh(m) + return m + + +def _bind_map(db: Session, m: StrategicMap, *codes: str) -> StrategicMap: + """把 KPI code 挂进地图维度节点 —— 复盘范围 = 地图节点 KPI ∪ KR 挂靠 KPI ∪ 行动挂靠 KPI。""" + dims = list(m.dimensions) + dims[0] = dict(dims[0]) + dims[0]["objectives"] = [{"name": dims[0]["objectives"][0]["name"], + "kpis": ["RB_REV"] + list(codes)}] + m.dimensions = dims + db.commit() + db.refresh(m) + return m + + +def _kpi(db: Session, code: str, **kw) -> KPIDefinition: + defaults = dict(entity_id=1, kpi_code=code, kpi_name=code + "名", unit="万元", + direction="higher_better", target_value=100.0, status="active") + defaults.update(kw) + k = KPIDefinition(**defaults) + db.add(k) + db.commit() + db.refresh(k) + return k + + +def _value(db: Session, kpi: KPIDefinition, value, period: str, **kw) -> KPIValue: + v = KPIValue(entity_id=kpi.entity_id, kpi_id=kpi.id, period=period, actual_value=value, **kw) + db.add(v) + db.commit() + db.refresh(v) + return v + + +def _seed_min(db: Session): + m = _map(db) + obj = Objective(entity_id=1, title="扩大收入规模", dimension="finance", quarter="", + status="active", progress=0) + db.add(obj) + db.commit() + db.refresh(obj) + kr = KR(entity_id=1, objective_id=obj.id, title="营收增长率≥20%", target_value=20, + progress=0, status="pending") + db.add(kr) + db.commit() + db.refresh(kr) + kpi = _kpi(db, "RB_REV") + _value(db, kpi, 80.0, datetime.now().strftime("%Y-%m")) + plan = ActionPlan(entity_id=1, kpi_id=kpi.id, title="营收攻坚行动", status="pending", + progress=0, verify_status="pending") + db.add(plan) + db.commit() + db.refresh(plan) + return m, obj, kr, kpi, plan + + +class TestBuildReviewReadonly: + def test_payload_shape_and_content(self, db: Session): + m, obj, kr, kpi, plan = _seed_min(db) + out = build_review(db, entity_id=1, map_id=m.id, review_type="monthly") + assert out["summary"].strip() + assert "RB_REV" in out["summary"] and "偏差" in out["summary"] + types = {a["type"] for a in out["adjustments"]} + assert {"kpi", "kr", "objective"} <= types + kpi_row = next(a for a in out["adjustments"] if a["type"] == "kpi") + assert kpi_row["target"] == "RB_REV" + assert kpi_row["current_value"] == 80.0 + assert kpi_row["target_value"] == 100.0 + assert kpi_row["deviation"] == -20.0 # 现值 − 目标（有符号） + # 关联行动: 按 kpi_id 既有挂靠命中 + assert plan.id in kpi_row["linked_plan_ids"] + assert out["next_priorities"] + assert out["meta"]["kpi_count"] == 1 + + def test_readonly_no_write(self, db: Session): + """生成器只读：不得改动任何行（回查计数与值不变）""" + m, obj, kr, kpi, plan = _seed_min(db) + before = (db.query(KR).count(), db.query(ActionPlan).count(), + db.query(KPIValue).count(), kr.progress, plan.status) + build_review(db, entity_id=1, map_id=m.id, review_type="quarterly") + after = (db.query(KR).count(), db.query(ActionPlan).count(), + db.query(KPIValue).count(), kr.progress, plan.status) + assert before == after + + +class TestDirectionContract: + def test_lower_better_uses_target_over_actual(self, db: Session): + m = _map(db) + _bind_map(db, m, "RB_COST") + k = _kpi(db, "RB_COST", unit="%", direction="lower_better", target_value=20.0) + _value(db, k, 10.0, datetime.now().strftime("%Y-%m")) + out = build_review(db, entity_id=1, map_id=m.id, review_type="monthly") + row = next(a for a in out["adjustments"] if a["target"] == "RB_COST") + assert row["level"] == "green" and row["verdict"] == "达标" + + def test_direction_null_not_defaulted_to_positive(self, db: Session): + """契约 §2：direction IS NULL 不默认正向 —— 不判达标，只列现值/偏差""" + m = _map(db) + _bind_map(db, m, "RB_NULLDIR") + k = _kpi(db, "RB_NULLDIR", direction=None, target_value=150.0) + _value(db, k, 186.0, datetime.now().strftime("%Y-%m")) + out = build_review(db, entity_id=1, map_id=m.id, review_type="monthly") + row = next(a for a in out["adjustments"] if a["target"] == "RB_NULLDIR") + assert row["level"] == "info" + assert "方向未配" in row["verdict"] + assert row["deviation"] == 36.0 + + def test_status_unit_uses_flag(self, db: Session): + """契约 §2：unit='状态' 走 1=达成 / 0=未达成，不走达成率""" + m = _map(db) + _bind_map(db, m, "RB_FLAG") + k = _kpi(db, "RB_FLAG", unit="状态", direction=None, target_value=1) + _value(db, k, 1, datetime.now().strftime("%Y-%m")) + out = build_review(db, entity_id=1, map_id=m.id, review_type="monthly") + row = next(a for a in out["adjustments"] if a["target"] == "RB_FLAG") + assert row["level"] == "green" and row["verdict"] == "达成" + + def test_zero_target_and_missing_actual_no_crash(self, db: Session): + """契约 §3：目标 0 / 现值缺失 ⇒ 不除零、不 NaN，落 info""" + m = _map(db) + _bind_map(db, m, "RB_ZERO", "RB_NOACT") + k0 = _kpi(db, "RB_ZERO", target_value=0.0) + _value(db, k0, 5.0, datetime.now().strftime("%Y-%m")) + k_null = _kpi(db, "RB_NOACT", target_value=10.0) # 无 KPIValue + out = build_review(db, entity_id=1, map_id=m.id, review_type="monthly") + r0 = next(a for a in out["adjustments"] if a["target"] == "RB_ZERO") + rn = next(a for a in out["adjustments"] if a["target"] == "RB_NOACT") + assert r0["level"] == "info" and r0["deviation_pct"] is None + assert rn["current_value"] is None and rn["level"] == "info" + + +class TestPeriodCaliber: + def test_future_period_excluded(self, db: Session): + """期间口径：period <= 当前月（不取未来期间）""" + m = _map(db) + _bind_map(db, m, "RB_FUT") + k = _kpi(db, "RB_FUT", target_value=100.0) + _value(db, k, 50.0, datetime.now().strftime("%Y-%m")) + future = (datetime.now() + timedelta(days=62)).strftime("%Y-%m") + _value(db, k, 150.0, future) + out = build_review(db, entity_id=1, map_id=m.id, review_type="monthly") + row = next(a for a in out["adjustments"] if a["target"] == "RB_FUT") + assert row["current_value"] == 50.0 + + def test_forecast_row_not_used_as_actual(self, db: Session): + """实测区口径：estimated / forecast 行不作实测现值""" + m = _map(db) + _bind_map(db, m, "RB_FC") + k = _kpi(db, "RB_FC", target_value=100.0) + now = datetime.now().strftime("%Y-%m") + _value(db, k, 30.0, now, data_status="verified") + _value(db, k, 999.0, now, data_status="estimated", source_type="forecast", + calculated_at=datetime.now() + timedelta(minutes=5)) + out = build_review(db, entity_id=1, map_id=m.id, review_type="monthly") + row = next(a for a in out["adjustments"] if a["target"] == "RB_FC") + assert row["current_value"] == 30.0 + + +class TestWriteReadChain: + def test_post_then_get_readback(self, db: Session, client: TestClient): + """落库链路: POST 写 → GET 读回（复盘页同一读路径）""" + m, obj, kr, kpi, plan = _seed_min(db) + create_test_user(db) + token = get_token_for_user(client) + out = build_review(db, entity_id=1, map_id=m.id, review_type="quarterly") + body = {"review_type": "quarterly", "review_date": date.today().isoformat(), + "summary": out["summary"], "adjustments": out["adjustments"], + "next_priorities": out["next_priorities"]} + r = client.post(f"/api/cma/maps/{m.id}/review-records", json=body, + headers=auth_header(token)) + assert r.status_code == 200, r.text + rid = r.json()["id"] + + r2 = client.get(f"/api/cma/maps/{m.id}/review-records", headers=auth_header(token)) + assert r2.status_code == 200 + ids = [x["id"] for x in r2.json()["data"]] + assert rid in ids + + r3 = client.get(f"/api/cma/maps/{m.id}/review-records/{rid}", headers=auth_header(token)) + assert r3.status_code == 200 + got = r3.json() + assert got["review_type"] == "quarterly" + assert "RB_REV" in got["summary"] + # 结构化「O/KR/KPI 现值+偏差+关联行动」随记录回读 + assert any(a["type"] == "kpi" and a["target"] == "RB_REV" for a in got["adjustments"])