"""战略地图 · 数据完备度 API(口径唯一真源:CMA-CALIBER-MAP-COMPLETENESS-20261018 v1.0) 主口径(对外唯一主数):仅地图 65(已发布)15/19 = 78.9%,判据 = 被 23-KPI 集挂接。 附注口径(只作附注):全量 4 图 30/31 = 96.8%,判据 = 有KPI挂接(含归档图)。 KPI 级:active 行 117(e1=73 / e2=44),按行计,与目标级分列、不相除。 所有汇总数一律运行时对库内数据现算;判据数据(23-KPI 挂接关系)见 app/data/caliber_map_completeness.py。禁用组合(19/19、83.9%)在本模块无生成路径。 """ from typing import Optional from fastapi import APIRouter, Depends from sqlalchemy.orm import Session from app.auth_middleware import require_auth from app.data import caliber_map_completeness as cal from app.database import get_db from app.deps import get_entity_id from app.models import KPIDefinition, MapObjective, StrategicMap router = APIRouter( prefix="/api/cma/maps/completeness", tags=["战略地图数据完备度"], dependencies=[Depends(require_auth)], ) def _pct(n: int, d: int) -> Optional[float]: """占比只保留 1 位小数;分母为 0 返回 None(展示为「无数据」)。""" if not d: return None return round(n * 100.0 / d, 1) def _entity_scopes(db: Session, entity_id: int): """返回(主口径地图, 同属主的地图列表)。主口径地图仅认 map_id=65 且属于该实体。""" maps = db.query(StrategicMap).filter(StrategicMap.entity_id == entity_id).all() primary = next((m for m in maps if m.id == cal.PRIMARY_MAP_ID), None) return primary, maps def _objective_rows(db: Session, maps): """拉取给定地图的战略目标,并逐行标记 23-KPI 挂接 / 是否有 KPI。""" map_ids = [m.id for m in maps] if not map_ids: return [], {} title_by_map = {m.id: m.title for m in maps} status_by_map = {m.id: (m.status or "") for m in maps} objs = ( db.query(MapObjective) .filter(MapObjective.map_id.in_(map_ids)) .order_by(MapObjective.map_id, MapObjective.sort_order, MapObjective.id) .all() ) rows = [] for o in objs: kpis = o.kpis if isinstance(o.kpis, list) else [] rows.append({ "objective_id": o.id, "map_id": o.map_id, "map_title": title_by_map.get(o.map_id), "map_status": status_by_map.get(o.map_id), "layer": o.dimension_key, "layer_label": cal.LAYER_LABELS.get(o.dimension_key, o.dimension_key), "name": o.name, "kpis": kpis, "has_kpi": len(kpis) > 0, "covered_23kpi": o.id in cal.LINKED_OBJECTIVES, }) return rows, title_by_map @router.get("/summary") def completeness_summary( entity_id: int = Depends(get_entity_id), db: Session = Depends(get_db), ): """数据完备度总览:目标级(主数 + 附注)与 KPI 级(117 行)两层分列。""" primary_map, maps = _entity_scopes(db, entity_id) rows, _ = _objective_rows(db, maps) # ---- 目标级 · 主口径:仅地图 65,判据 = 被 23-KPI 集挂接 ---- if primary_map is not None: main_rows = [r for r in rows if r["map_id"] == cal.PRIMARY_MAP_ID] main_total = len(main_rows) main_covered = sum(1 for r in main_rows if r["covered_23kpi"]) primary = { "available": True, "caliber": cal.PRIMARY_CALIBER, "label": cal.PRIMARY_LABEL, "map_id": primary_map.id, "map_title": primary_map.title, "map_status": primary_map.status, "covered": main_covered, "total": main_total, "pct": _pct(main_covered, main_total), "display": f"{main_covered}/{main_total}", } else: primary = { "available": False, "caliber": cal.PRIMARY_CALIBER, "label": cal.PRIMARY_LABEL, "map_id": cal.PRIMARY_MAP_ID, "covered": None, "total": None, "pct": None, "display": "—", } # ---- 目标级 · 附注口径:全量 4 图,判据 = 有 KPI 挂接 ---- # 判据复现(definition 件 §5 机读件 20261018-c1-spec/map-objectives-dump.csv 逐行): # 23KPI 覆盖 26 + 非23KPI 但 map_objectives.kpis 非空 4 + 缺口 1 = 30/31。 # 即「被 23-KPI 集挂接」与「kpis 非空」取并集;与主口径判据(仅 23-KPI 挂接)不是同一判据。 # 附注与主数同属地图 65 的账套(全量 4 图);无地图 65 的账套不适用,展示「—」而非 0。 footnote_total = len(rows) footnote_covered = sum(1 for r in rows if r["covered_23kpi"] or r["has_kpi"]) footnote_available = primary_map is not None and bool(rows) footnote = { "available": footnote_available, "caliber": cal.FOOTNOTE_CALIBER, "label": cal.FOOTNOTE_LABEL, "covered": footnote_covered if footnote_available else None, "total": footnote_total if footnote_available else None, "pct": _pct(footnote_covered, footnote_total) if footnote_available else None, "display": f"{footnote_covered}/{footnote_total}" if footnote_available else "—", } # ---- KPI 级:active 按行(e1=73 / e2=44,合计 117;去重编码 102 另册)---- kpi_rows = ( db.query(KPIDefinition) .filter(KPIDefinition.status == "active") .all() ) by_entity: dict = {} codes = set() for k in kpi_rows: by_entity[k.entity_id] = by_entity.get(k.entity_id, 0) + 1 codes.add(k.kpi_code) kpi_level = { "caliber": cal.KPI_CALIBER, "label": cal.KPI_LABEL, "total": len(kpi_rows), "by_entity": [ {"entity_id": eid, "count": cnt} for eid, cnt in sorted(by_entity.items()) ], "distinct_codes": len(codes), "note": "按 kpi_code 去重为独立口径,不与行数相除或换算", } return { "caliber_id": cal.CALIBER_ID, "caliber_version": cal.CALIBER_VERSION, "entity_id": entity_id, "applicable": primary_map is not None, "scope_note": "主口径仅覆盖地图 65(已发布);附注口径含归档图,判据不同,不得互换算", "objective_level": {"primary": primary, "footnote": footnote}, "kpi_level": kpi_level, } @router.get("/objectives") def completeness_objectives( map_id: Optional[int] = None, entity_id: int = Depends(get_entity_id), db: Session = Depends(get_db), ): """数据完备度清单(目标级逐行):供空态清单与补齐排期使用。""" _, maps = _entity_scopes(db, entity_id) if map_id is not None: maps = [m for m in maps if m.id == map_id] rows, _ = _objective_rows(db, maps) return { "caliber_id": cal.CALIBER_ID, "caliber_version": cal.CALIBER_VERSION, "entity_id": entity_id, "caliber": cal.FOOTNOTE_CALIBER, "rows": rows, }