diff --git a/.gitignore b/.gitignore index 2d9579dd..6dcb9b16 100644 --- a/.gitignore +++ b/.gitignore @@ -11,3 +11,6 @@ __pycache__/ *.tsbuildinfo venv/ backend/logs/ + +# 因果链验证报告（生成物） +backend/scripts/reports/ diff --git a/backend/app/api/kpi_causality.py b/backend/app/api/kpi_causality.py index e0af8037..c110f854 100644 --- a/backend/app/api/kpi_causality.py +++ b/backend/app/api/kpi_causality.py @@ -1,6 +1,7 @@ """KPI因果链建模 — 任务2 -KPI间因果关系网络 + 模拟推演 +KPI间因果关系网络 + 模拟推演 + 三层验证机制(数据/AI/人工) (2026-08-27 P2) """ +from datetime import datetime from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy.orm import Session from sqlalchemy import text @@ -11,6 +12,11 @@ from app.database import get_db from app.deps import get_entity_id from app.auth_middleware import require_auth, require_role from app.models import KPIDefinition, KPICausality, KPIValue, OperationLog +from app.services.causality_verification import ( + ALL_STATUSES, + ALL_SOURCE_TYPES, + STATUS_PENDING, +) logger = logging.getLogger("kpi-causality") @@ -30,8 +36,8 @@ def _to_dict(obj): @router.get("/full-network") def get_full_network(db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)): - """获取全局因果网络数据（用于力导向图）— 账套隔离: 仅当前企业KPI的因果链 (2026-08-23 P1b)""" - edges = db.query(KPICausality).join(KPIDefinition, KPIDefinition.id == KPICausality.source_kpi_id).filter(KPIDefinition.entity_id == entity_id).all() + """获取全局因果网络数据（用于力导向图）— 账套隔离: 仅当前企业KPI的因果链 (2026-08-23 P1b, 2026-08-27 用entity_id列)""" + edges = db.query(KPICausality).filter(KPICausality.entity_id == entity_id).all() node_ids = set() edge_list = [] for e in edges: @@ -61,14 +67,19 @@ def get_full_network(db: Session = Depends(get_db), entity_id: int = Depends(get @router.get("/kpi/{kpi_id}/network") -def get_kpi_network(kpi_id: int, db: Session = Depends(get_db)): - """获取KPI的因果网络（上游驱动 + 下游影响）""" +def get_kpi_network(kpi_id: int, db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)): + """获取KPI的因果网络（上游驱动 + 下游影响）— 账套隔离: 校验KPI属于当前企业""" kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first() if not kpi: raise HTTPException(404, "KPI不存在") + if kpi.entity_id != entity_id: + raise HTTPException(404, "KPI不存在") # 跨企业不暴露存在性 # 上游（指向当前KPI的因果） - upstream = db.query(KPICausality).filter(KPICausality.target_kpi_id == kpi_id).all() + upstream = db.query(KPICausality).filter( + KPICausality.target_kpi_id == kpi_id, + KPICausality.entity_id == entity_id, + ).all() upstream_list = [] for c in upstream: src = db.query(KPIDefinition).filter(KPIDefinition.id == c.source_kpi_id).first() @@ -78,10 +89,16 @@ def get_kpi_network(kpi_id: int, db: Session = Depends(get_db)): "kpi_id": src.id, "kpi_code": src.kpi_code, "kpi_name": src.kpi_name, "strength": c.strength, "lag_months": c.lag_months, "direction": c.direction, "formula": c.formula, + "source_type": c.source_type, "verify_status": c.verify_status, + "verified_at": c.verified_at.isoformat() if c.verified_at else None, + "verified_by": c.verified_by, }) # 下游（当前KPI指向的因果） - downstream = db.query(KPICausality).filter(KPICausality.source_kpi_id == kpi_id).all() + downstream = db.query(KPICausality).filter( + KPICausality.source_kpi_id == kpi_id, + KPICausality.entity_id == entity_id, + ).all() downstream_list = [] for c in downstream: tgt = db.query(KPIDefinition).filter(KPIDefinition.id == c.target_kpi_id).first() @@ -91,6 +108,9 @@ def get_kpi_network(kpi_id: int, db: Session = Depends(get_db)): "kpi_id": tgt.id, "kpi_code": tgt.kpi_code, "kpi_name": tgt.kpi_name, "strength": c.strength, "lag_months": c.lag_months, "direction": c.direction, "formula": c.formula, + "source_type": c.source_type, "verify_status": c.verify_status, + "verified_at": c.verified_at.isoformat() if c.verified_at else None, + "verified_by": c.verified_by, }) return { @@ -101,7 +121,7 @@ def get_kpi_network(kpi_id: int, db: Session = Depends(get_db)): @router.post("/simulate") -def simulate_causality(data: dict, db: Session = Depends(get_db)): +def simulate_causality(data: dict, db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)): """模拟推演: 修改一个KPI的值，预测对其他KPI的影响 Body: { kpi_id: int, new_value: float, period: str } """ @@ -115,6 +135,8 @@ def simulate_causality(data: dict, db: Session = Depends(get_db)): source_kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first() if not source_kpi: raise HTTPException(404, "KPI不存在") + if source_kpi.entity_id != entity_id: + raise HTTPException(404, "KPI不存在") # 获取当前值 current_value = None @@ -142,9 +164,10 @@ def simulate_causality(data: dict, db: Session = Depends(get_db)): continue visited.add(current_kpi_id) - # 查找从current_kpi_id出发的下游因果链 + # 查找从current_kpi_id出发的下游因果链（账套隔离: 仅本企业链） downstream = db.query(KPICausality).filter( - KPICausality.source_kpi_id == current_kpi_id + KPICausality.source_kpi_id == current_kpi_id, + KPICausality.entity_id == entity_id, ).all() for edge in downstream: @@ -212,15 +235,18 @@ def simulate_causality(data: dict, db: Session = Depends(get_db)): def list_causalities( source_kpi_id: Optional[int] = None, target_kpi_id: Optional[int] = None, + verify_status: Optional[str] = None, db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id), ): - """获取因果链列表（账套隔离: 仅当前企业KPI, 2026-08-23 P1b）""" - query = db.query(KPICausality).join(KPIDefinition, KPIDefinition.id == KPICausality.source_kpi_id).filter(KPIDefinition.entity_id == entity_id) + """获取因果链列表（账套隔离: 仅当前企业KPI, 2026-08-23 P1b, 2026-08-27 支持verify_status筛选）""" + query = db.query(KPICausality).filter(KPICausality.entity_id == entity_id) if source_kpi_id: query = query.filter(KPICausality.source_kpi_id == source_kpi_id) if target_kpi_id: query = query.filter(KPICausality.target_kpi_id == target_kpi_id) + if verify_status: + query = query.filter(KPICausality.verify_status == verify_status) items = query.order_by(KPICausality.id).all() result = [] @@ -236,11 +262,51 @@ def list_causalities( return {"data": result, "total": len(result)} +@router.get("/verify-status") +def get_verify_status( + verify_status: Optional[str] = None, + db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), +): + """验证状态总览 — 按状态统计 + 链列表（2026-08-27 三层验证机制） + + 可选 ?verify_status=pending/data_verified/human_verified/disputed 筛选 + """ + if verify_status and verify_status not in ALL_STATUSES: + raise HTTPException(400, f"verify_status 必须为 {'/'.join(ALL_STATUSES)}") + + query = db.query(KPICausality).filter(KPICausality.entity_id == entity_id) + if verify_status: + query = query.filter(KPICausality.verify_status == verify_status) + items = query.order_by(KPICausality.id).all() + + by_status = {s: 0 for s in ALL_STATUSES} + data = [] + for c in items: + by_status[c.verify_status] = by_status.get(c.verify_status, 0) + 1 + d = _to_dict(c) + src = db.query(KPIDefinition).filter(KPIDefinition.id == c.source_kpi_id).first() + tgt = db.query(KPIDefinition).filter(KPIDefinition.id == c.target_kpi_id).first() + d["source_kpi_code"] = src.kpi_code if src else None + d["source_kpi_name"] = src.kpi_name if src else None + d["target_kpi_code"] = tgt.kpi_code if tgt else None + d["target_kpi_name"] = tgt.kpi_name if tgt else None + d["verified_at"] = c.verified_at.isoformat() if c.verified_at else None + data.append(d) + + return { + "summary": {"total": len(items), "by_status": by_status}, + "data": data, + } + + @router.get("/{causality_id}") -def get_causality(causality_id: int, db: Session = Depends(get_db)): +def get_causality(causality_id: int, db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)): c = db.query(KPICausality).filter(KPICausality.id == causality_id).first() if not c: raise HTTPException(404, "因果链不存在") + if c.entity_id != entity_id: + raise HTTPException(404, "因果链不存在") # 账套隔离 d = _to_dict(c) src = db.query(KPIDefinition).filter(KPIDefinition.id == c.source_kpi_id).first() tgt = db.query(KPIDefinition).filter(KPIDefinition.id == c.target_kpi_id).first() @@ -251,7 +317,7 @@ def get_causality(causality_id: int, db: Session = Depends(get_db)): @router.post("") def create_causality(data: dict, db: Session = Depends(get_db), user=WRITE_ROLES): - """创建因果链""" + """创建因果链（source_type标记来源, 2026-08-27）""" source_id = data.get("source_kpi_id") target_id = data.get("target_kpi_id") if not source_id or not target_id: @@ -262,6 +328,8 @@ def create_causality(data: dict, db: Session = Depends(get_db), user=WRITE_ROLES tgt = db.query(KPIDefinition).filter(KPIDefinition.id == target_id).first() if not src or not tgt: raise HTTPException(404, "KPI不存在") + if src.entity_id != tgt.entity_id: + raise HTTPException(400, "源KPI和目标KPI必须属于同一企业") existing = db.query(KPICausality).filter( KPICausality.source_kpi_id == source_id, @@ -270,40 +338,91 @@ def create_causality(data: dict, db: Session = Depends(get_db), user=WRITE_ROLES if existing: raise HTTPException(400, f"因果链已存在: {src.kpi_code}→{tgt.kpi_code}") + source_type = data.get("source_type", "manual") + if source_type not in ALL_SOURCE_TYPES: + raise HTTPException(400, f"source_type 必须为 {'/'.join(ALL_SOURCE_TYPES)}") + c = KPICausality( + entity_id=src.entity_id, source_kpi_id=source_id, target_kpi_id=target_id, strength=data.get("strength", 0.5), lag_months=data.get("lag_months", 1), formula=data.get("formula"), direction=data.get("direction", "positive"), + source_type=source_type, + verify_status=STATUS_PENDING, ) db.add(c) db.commit() db.refresh(c) db.add(OperationLog(action="create", target_type="kpi_causality", - detail=f"创建因果链: {src.kpi_code}→{tgt.kpi_code}")) + detail=f"创建因果链: {src.kpi_code}→{tgt.kpi_code} (source={source_type})")) db.commit() return _to_dict(c) @router.put("/{causality_id}") -def update_causality(causality_id: int, data: dict, db: Session = Depends(get_db), user=WRITE_ROLES): +def update_causality(causality_id: int, data: dict, db: Session = Depends(get_db), user=WRITE_ROLES, + entity_id: int = Depends(get_entity_id)): c = db.query(KPICausality).filter(KPICausality.id == causality_id).first() if not c: raise HTTPException(404, "因果链不存在") - for field in ("strength", "lag_months", "formula", "direction"): + if c.entity_id != entity_id: + raise HTTPException(404, "因果链不存在") # 账套隔离 + for field in ("strength", "lag_months", "formula", "direction", "source_type"): if field in data: + if field == "source_type" and data[field] not in ALL_SOURCE_TYPES: + raise HTTPException(400, f"source_type 必须为 {'/'.join(ALL_SOURCE_TYPES)}") setattr(c, field, data[field]) + # 修改链定义后，验证状态回到待检（定义变了旧结论失效） + if any(f in data for f in ("strength", "lag_months", "formula", "direction")): + c.verify_status = STATUS_PENDING + c.verified_at = None + c.verified_by = None db.commit() db.refresh(c) return _to_dict(c) +@router.put("/{causality_id}/verify") +def verify_causality(causality_id: int, data: dict, db: Session = Depends(get_db), user=WRITE_ROLES, + entity_id: int = Depends(get_entity_id)): + """人工确认（战略回顾会核对打标）— 2026-08-27 三层验证机制 + + Body: { verify_status: "human_verified"|"disputed", verified_by?: "任富海" } + 默认打标 human_verified（人工最终确认）。 + """ + c = db.query(KPICausality).filter(KPICausality.id == causality_id).first() + if not c: + raise HTTPException(404, "因果链不存在") + if c.entity_id != entity_id: + raise HTTPException(404, "因果链不存在") # 账套隔离 + + target_status = data.get("verify_status", "human_verified") + if target_status not in ("human_verified", "disputed"): + raise HTTPException(400, "verify_status 必须为 human_verified 或 disputed") + verified_by = data.get("verified_by") or user.name or user.username + + c.verify_status = target_status + c.verified_at = datetime.now() + c.verified_by = str(verified_by)[:50] + db.add(OperationLog(action="verify", target_type="kpi_causality", + detail=f"因果链 #{causality_id} 人工确认: {target_status} (by {verified_by})")) + db.commit() + db.refresh(c) + d = _to_dict(c) + d["verified_at"] = c.verified_at.isoformat() if c.verified_at else None + return d + + @router.delete("/{causality_id}") -def delete_causality(causality_id: int, db: Session = Depends(get_db), user=WRITE_ROLES): +def delete_causality(causality_id: int, db: Session = Depends(get_db), user=WRITE_ROLES, + entity_id: int = Depends(get_entity_id)): c = db.query(KPICausality).filter(KPICausality.id == causality_id).first() if c: + if c.entity_id != entity_id: + raise HTTPException(404, "因果链不存在") # 账套隔离 db.delete(c) db.commit() return {"message": "已删除"} diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 0686eae3..aa869afd 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -292,12 +292,17 @@ class KPICausality(Base): """KPI因果链 — 记录KPI间的因果关系""" __tablename__ = "kpi_causality" id = Column(Integer, primary_key=True, index=True) + entity_id = Column(Integer, default=1, comment="企业ID (多租户隔离 2026-08-27)") source_kpi_id = Column(Integer, ForeignKey("kpi_definitions.id"), nullable=False, comment="源KPI（因）") target_kpi_id = Column(Integer, ForeignKey("kpi_definitions.id"), nullable=False, comment="目标KPI（果）") strength = Column(Float, default=0.5, comment="影响强度 0~1") lag_months = Column(Integer, default=1, comment="滞后期（月）") formula = Column(String(500), nullable=True, comment="影响公式描述") direction = Column(String(10), default="positive", comment="positive/negative 正向/负向影响") + source_type = Column(String(20), default="manual", comment="建链来源 AI_suggested/manual/imported (2026-08-27 验证机制)") + verify_status = Column(String(20), default="pending", comment="验证状态 pending/data_verified/human_verified/disputed (2026-08-27)") + verified_at = Column(DateTime, nullable=True, comment="验证时间") + verified_by = Column(String(50), nullable=True, comment="验证人/AI/脚本") created_at = Column(DateTime, server_default=func.now()) updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now()) diff --git a/backend/app/services/causality_verification.py b/backend/app/services/causality_verification.py new file mode 100644 index 00000000..d6e18f7f --- /dev/null +++ b/backend/app/services/causality_verification.py @@ -0,0 +1,238 @@ +"""因果链验证核心逻辑 — 数据验证（Pearson相关性）+ 状态机 (2026-08-27 P2) + +三层验证： +1. 数据验证（自动化）：kpi_values 历史值 → Pearson 相关系数 + 方向一致性 + 滞后对齐 +2. AI/人工验证：战略回顾会人工打标 human_verified（API PUT /verify） +3. 状态机流转： + pending（初始）→ data_verified / disputed（数据验证 cron） + pending/data_verified/disputed → human_verified（人工确认，最终） + 任何矛盾 → disputed（待检） + +判定规则（可解释、可测试）： +- 对齐后数据点 < MIN_POINTS(4) → pending（数据不足，无法统计验证） +- |r| >= CORR_THRESHOLD(0.5) 且方向与 direction 一致 → data_verified +- 否则（点数足够但弱相关/方向矛盾）→ disputed +- human_verified 为人工最终确认，脚本默认不覆盖（respect_human=True）， + 但若数据矛盾会在报告中给出警示（disputed_note） + +脚本入口: scripts/correlation-check.py +测试入口: tests/test_causality_verification.py +""" +from __future__ import annotations + +import logging +import math +import re +from typing import Dict, List, Optional, Tuple + +logger = logging.getLogger("causality-verification") + +# 验证状态 +STATUS_PENDING = "pending" +STATUS_DATA_VERIFIED = "data_verified" +STATUS_HUMAN_VERIFIED = "human_verified" +STATUS_DISPUTED = "disputed" +ALL_STATUSES = (STATUS_PENDING, STATUS_DATA_VERIFIED, STATUS_HUMAN_VERIFIED, STATUS_DISPUTED) + +# 建链来源 +SOURCE_MANUAL = "manual" +SOURCE_AI = "AI_suggested" +SOURCE_IMPORTED = "imported" +ALL_SOURCE_TYPES = (SOURCE_MANUAL, SOURCE_AI, SOURCE_IMPORTED) + +# 判定参数 +MIN_POINTS = 4 # 最少对齐数据点（少于则无法统计验证） +CORR_THRESHOLD = 0.5 # |r| 阈值：达到且方向一致 → 数据证实 +VERIFIER_SCRIPT = "correlation-check" + +# period 粒度（对齐时只允许同粒度配对，避免月度/年度量纲混用） +GRANULARITY_ORDER = ("month", "half", "year") + +_PERIOD_RE = { + "month": re.compile(r"^(\d{4})-(\d{2})$"), + "half": re.compile(r"^(\d{4})-H([12])$"), + "year": re.compile(r"^(\d{4})$"), +} + + +def parse_period(period: str) -> Optional[Tuple[str, int]]: + """解析 period 为 (granularity, seq)。 + + seq = year*12 + 月序号（0-indexed），可比较/做滞后偏移。 + - '2026-07' → ('month', 2026*12+6) + - '2026-H1' → ('half', 2026*12+5) (H1≈6月) + - '2026-H2' → ('half', 2026*12+11) (H2≈12月) + - '2026' → ('year', 2026*12+5) (年中) + 无法解析 → None + """ + if not period: + return None + s = str(period).strip() + m = _PERIOD_RE["month"].match(s) + if m: + year, mon = int(m.group(1)), int(m.group(2)) + if 1 <= mon <= 12: + return ("month", year * 12 + (mon - 1)) + m = _PERIOD_RE["half"].match(s) + if m: + year, half = int(m.group(1)), int(m.group(2)) + return ("half", year * 12 + (5 if half == 1 else 11)) + m = _PERIOD_RE["year"].match(s) + if m: + return ("year", int(m.group(1)) * 12 + 5) + return None + + +def pearson(xs: List[float], ys: List[float]) -> Tuple[Optional[float], int]: + """Pearson 相关系数。点数 < 2 返回 (None, n)。""" + n = len(xs) + if n < 2: + return None, n + mx = sum(xs) / n + my = sum(ys) / n + sxy = sum((x - mx) * (y - my) for x, y in zip(xs, ys)) + sxx = sum((x - mx) ** 2 for x in xs) + syy = sum((y - my) ** 2 for y in ys) + if sxx <= 0 or syy <= 0: # 常数列 → 无相关 + return None, n + r = sxy / math.sqrt(sxx * syy) + # 数值保护：浮点误差可能略超 [-1,1] + return max(-1.0, min(1.0, r)), n + + +def align_series( + source_values: List[Tuple[str, float]], + target_values: List[Tuple[str, float]], + lag_months: int = 0, +) -> Tuple[Optional[str], List[Tuple[float, float]]]: + """按滞后期对齐 source/target 时间序列，返回 (granularity, pairs)。 + + 语义：source 是因（先发生），target 是果（滞后 lag 月出现）。 + pair = (source[t], target[t + lag])。 + 只使用同粒度（month/half/year）数据配对，避免量纲混用。 + 按粒度优先级 month > half > year 选取数据点最多的粒度。 + """ + parsed_src: Dict[str, Dict[int, float]] = {g: {} for g in GRANULARITY_ORDER} + parsed_tgt: Dict[str, Dict[int, float]] = {g: {} for g in GRANULARITY_ORDER} + for period, val in source_values: + if val is None: + continue + r = parse_period(period) + if r: + g, seq = r + parsed_src[g][seq] = float(val) + for period, val in target_values: + if val is None: + continue + r = parse_period(period) + if r: + g, seq = r + parsed_tgt[g][seq] = float(val) + + best_g, best_pairs = None, [] + for g in GRANULARITY_ORDER: + src_map, tgt_map = parsed_src[g], parsed_tgt[g] + pairs = [] + for seq, sv in sorted(src_map.items()): + tv = tgt_map.get(seq + lag_months) + if tv is not None: + pairs.append((sv, tv)) + if len(pairs) > len(best_pairs): + best_g, best_pairs = g, pairs + return best_g, best_pairs + + +def evaluate_chain( + source_values: List[Tuple[str, float]], + target_values: List[Tuple[str, float]], + lag_months: int = 0, + direction: str = "positive", + min_points: int = MIN_POINTS, + corr_threshold: float = CORR_THRESHOLD, +) -> dict: + """对单条因果链做数据验证。 + + 返回: + { + granularity, n, r, expected_sign, actual_sign, direction_consistent, + status (pending/data_verified/disputed), reason + } + """ + granularity, pairs = align_series(source_values, target_values, lag_months) + n = len(pairs) + r = None + if n >= 2: + r, _ = pearson([p[0] for p in pairs], [p[1] for p in pairs]) + + expected_sign = 1 if direction == "positive" else -1 + actual_sign = 1 if r is not None and r > 0 else (-1 if r is not None and r < 0 else 0) + direction_consistent = r is not None and actual_sign == expected_sign + + if n < min_points or r is None: + return { + "granularity": granularity, "n": n, "r": r, + "expected_sign": expected_sign, "actual_sign": actual_sign, + "direction_consistent": direction_consistent, + "status": STATUS_PENDING, + "reason": f"数据不足（对齐后{n}点，需≥{min_points}点）" if n < min_points else "序列无方差，无法计算相关性", + } + + abs_r = abs(r) + if abs_r >= corr_threshold and direction_consistent: + return { + "granularity": granularity, "n": n, "r": r, + "expected_sign": expected_sign, "actual_sign": actual_sign, + "direction_consistent": True, + "status": STATUS_DATA_VERIFIED, + "reason": f"|r|={abs_r:.3f}≥{corr_threshold} 且方向一致({direction}) → 数据证实", + } + if not direction_consistent: + return { + "granularity": granularity, "n": n, "r": r, + "expected_sign": expected_sign, "actual_sign": actual_sign, + "direction_consistent": False, + "status": STATUS_DISPUTED, + "reason": f"方向矛盾: 声明{direction}但实际相关方向{'正' if r > 0 else '负'} (r={r:.3f})", + } + return { + "granularity": granularity, "n": n, "r": r, + "expected_sign": expected_sign, "actual_sign": actual_sign, + "direction_consistent": True, + "status": STATUS_DISPUTED, + "reason": f"弱相关: |r|={abs_r:.3f}<{corr_threshold}，数据暂不能证实该因果强度", + } + + +def apply_state_machine( + current_status: str, + eval_status: str, + respect_human: bool = True, +) -> Tuple[str, Optional[str]]: + """状态机：根据数据验证结果流转状态。 + + 规则: + - human_verified 是人工最终确认: respect_human=True 时不被脚本覆盖 + （返回原状态 + disputed_note 警示） + - 数据不足(pending) → 保持当前状态（不降级已有结论） + - data_verified → 覆盖为非 human_verified 的当前状态 + - disputed → 覆盖为非 human_verified 的当前状态 + """ + if current_status == STATUS_HUMAN_VERIFIED and respect_human: + if eval_status == STATUS_DISPUTED: + return current_status, "人工已确认但数据复核矛盾，建议重新核对" + return current_status, None + if eval_status == STATUS_PENDING: + return current_status, None + return eval_status, None + + +def summarize(results: List[dict]) -> dict: + """验证结果汇总统计。""" + counter = {s: 0 for s in ALL_STATUSES} + for r in results: + counter[r.get("status", STATUS_PENDING)] = counter.get(r.get("status", STATUS_PENDING), 0) + 1 + return { + "total": len(results), + "by_status": counter, + "data_verified_ratio": round(counter[STATUS_DATA_VERIFIED] / len(results), 3) if results else 0, + } diff --git a/backend/scripts/correlation-check.py b/backend/scripts/correlation-check.py new file mode 100644 index 00000000..0f2709e4 --- /dev/null +++ b/backend/scripts/correlation-check.py @@ -0,0 +1,193 @@ +#!/usr/bin/env python3 +"""因果链数据验证脚本 — 每月 cron 自动跑 (2026-08-27 P2) + +对 kpi_causality 每条链： + 取 source/target KPI 的 kpi_values 历史值 + → Pearson 相关系数 + 方向一致性 + 滞后对齐(lag_months) + → 更新 verify_status: data_verified / disputed / pending + → 输出验证报告 JSON + 控制台摘要 + +用法: + python3 scripts/correlation-check.py # 全部企业，写库 + python3 scripts/correlation-check.py --entity-id 1 # 指定企业 + python3 scripts/correlation-check.py --dry-run # 只算不写库 + +月度 cron: 0 9 1 * * cd /root/cma-management/backend && python3 scripts/correlation-check.py +""" +import argparse +import json +import logging +import sys +from datetime import datetime +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from sqlalchemy import text # noqa: E402 + +from app.database import get_engine # noqa: E402 +from app.services.causality_verification import ( # noqa: E402 + STATUS_DATA_VERIFIED, + STATUS_DISPUTED, + STATUS_HUMAN_VERIFIED, + STATUS_PENDING, + VERIFIER_SCRIPT, + apply_state_machine, + evaluate_chain, + summarize, +) + +logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") +logger = logging.getLogger("correlation-check") + +REPORT_DIR = Path(__file__).resolve().parent / "reports" + + +def load_chains(engine, entity_id: int = None) -> list: + """加载因果链 + 两端KPI信息。""" + q = """ + SELECT c.id, c.entity_id, c.source_kpi_id, c.target_kpi_id, + c.strength, c.lag_months, c.direction, c.source_type, c.verify_status, + s.kpi_code AS src_code, s.kpi_name AS src_name, + t.kpi_code AS tgt_code, t.kpi_name AS tgt_name + FROM kpi_causality c + JOIN kpi_definitions s ON s.id = c.source_kpi_id + JOIN kpi_definitions t ON t.id = c.target_kpi_id + """ + if entity_id is not None: + q += " WHERE c.entity_id = :eid" + with engine.connect() as conn: + rows = conn.execute(text(q), {"eid": entity_id} if entity_id is not None else {}).mappings().all() + return [dict(r) for r in rows] + + +def load_values(engine, kpi_ids: list) -> dict: + """加载 KPI 历史值: {kpi_id: [(period, actual_value), ...]}""" + if not kpi_ids: + return {} + ids = list(set(int(i) for i in kpi_ids)) + q = """ + SELECT kpi_id, period, actual_value + FROM kpi_values + WHERE kpi_id IN :ids AND actual_value IS NOT NULL + ORDER BY period + """ + with engine.connect() as conn: + rows = conn.execute(text(q).bindparams(ids=ids), {"ids": ids}).mappings().all() + result = {} + for r in rows: + result.setdefault(r["kpi_id"], []).append((r["period"], r["actual_value"])) + return result + + +def main(): + ap = argparse.ArgumentParser(description="因果链数据验证") + ap.add_argument("--entity-id", type=int, default=None, help="只验证指定企业（默认全部）") + ap.add_argument("--dry-run", action="store_true", help="只计算不写库") + args = ap.parse_args() + + engine = get_engine() + chains = load_chains(engine, args.entity_id) + if not chains: + logger.info("无因果链，退出") + return 0 + + kpi_ids = [c["source_kpi_id"] for c in chains] + [c["target_kpi_id"] for c in chains] + values = load_values(engine, kpi_ids) + + now = datetime.now() + results = [] + updated = {"data_verified": 0, "disputed": 0, "unchanged": 0} + notes = [] + + with engine.begin() as conn: + for c in chains: + src_vals = values.get(c["source_kpi_id"], []) + tgt_vals = values.get(c["target_kpi_id"], []) + ev = evaluate_chain( + src_vals, tgt_vals, + lag_months=c["lag_months"] or 0, + direction=c["direction"] or "positive", + ) + new_status, note = apply_state_machine(c["verify_status"], ev["status"], respect_human=True) + + if note: + notes.append({"causality_id": c["id"], "note": note}) + + changed = new_status != c["verify_status"] + if changed: + updated[new_status if new_status in updated else "unchanged"] = \ + updated.get(new_status if new_status in updated else "unchanged", 0) + 1 + else: + updated["unchanged"] += 1 + + if not args.dry_run: + conn.execute(text( + "UPDATE kpi_causality SET verify_status = :st, verified_at = :va, verified_by = :vb " + "WHERE id = :cid" + ), { + "st": new_status, "va": now, "vb": VERIFIER_SCRIPT, "cid": c["id"], + }) + + results.append({ + "causality_id": c["id"], + "source": f'{c["src_code"]}({c["src_name"]})', + "target": f'{c["tgt_code"]}({c["tgt_name"]})', + "direction": c["direction"], + "lag_months": c["lag_months"], + "strength": c["strength"], + "granularity": ev["granularity"], + "n_points": ev["n"], + "r": round(ev["r"], 4) if ev["r"] is not None else None, + "direction_consistent": ev["direction_consistent"], + "old_status": c["verify_status"], + "new_status": new_status, + "reason": ev["reason"], + }) + + summary = summarize([{"status": r["new_status"]} for r in results]) + report = { + "generated_at": now.strftime("%Y-%m-%d %H:%M:%S"), + "script": VERIFIER_SCRIPT, + "dry_run": args.dry_run, + "entity_id": args.entity_id, + "summary": summary, + "updated": updated, + "human_verified_notes": notes, + "chains": results, + } + + REPORT_DIR.mkdir(exist_ok=True) + report_path = REPORT_DIR / f"causality_verification_{now.strftime('%Y%m%d_%H%M%S')}.json" + report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") + + # 控制台摘要（cron 输出即消息） + lines = [ + f"因果链数据验证{'[dry-run]' if args.dry_run else ''} {now.strftime('%Y-%m-%d %H:%M')}", + f"总数: {summary['total']} | 数据证实: {summary['by_status'][STATUS_DATA_VERIFIED]} | " + f"存疑: {summary['by_status'][STATUS_DISPUTED]} | 待检(数据不足): {summary['by_status'][STATUS_PENDING]} | " + f"人工确认: {summary['by_status'][STATUS_HUMAN_VERIFIED]}", + f"本次更新: data_verified={updated['data_verified']} disputed={updated['disputed']} unchanged={updated['unchanged']}", + ] + verified = [r for r in results if r["new_status"] == STATUS_DATA_VERIFIED] + disputed = [r for r in results if r["new_status"] == STATUS_DISPUTED] + if verified: + lines.append("── 数据证实 ──") + for r in verified: + lines.append(f" #{r['causality_id']} {r['source']}→{r['target']} r={r['r']} n={r['n_points']}") + if disputed: + lines.append("── 数据存疑 ──") + for r in disputed: + lines.append(f" #{r['causality_id']} {r['source']}→{r['target']} r={r['r']} n={r['n_points']} ({r['reason']})") + if notes: + lines.append("── 人工确认链的数据警示 ──") + for n in notes: + lines.append(f" #{n['causality_id']}: {n['note']}") + lines.append(f"报告: {report_path}") + print("\n".join(lines)) + logger.info("报告已写入 %s", report_path) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/backend/scripts/migrate_causality_verification.py b/backend/scripts/migrate_causality_verification.py new file mode 100644 index 00000000..b8376ba1 --- /dev/null +++ b/backend/scripts/migrate_causality_verification.py @@ -0,0 +1,95 @@ +"""kpi_causality 因果链验证机制迁移脚本 (2026-08-27 P2) + +加列: +- source_type: varchar(20) 建链来源 AI_suggested/manual/imported +- verify_status: varchar(20) 验证状态 pending/data_verified/human_verified/disputed +- verified_at: datetime 验证时间 +- verified_by: varchar(50) 验证人/AI/脚本 +- entity_id: int 多租户隔离 (2026-08-27 收官补齐) + +幂等: 列已存在则跳过; entity_id 回填只更新 NULL/0 行。 +用法: python scripts/migrate_causality_verification.py +""" +import logging +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from sqlalchemy import text + +from app.database import get_engine + +logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") +logger = logging.getLogger("migrate-causality-verification") + +COLUMNS = [ + ("source_type", "ALTER TABLE kpi_causality ADD COLUMN source_type VARCHAR(20) NOT NULL DEFAULT 'manual' COMMENT '建链来源 AI_suggested/manual/imported'"), + ("verify_status", "ALTER TABLE kpi_causality ADD COLUMN verify_status VARCHAR(20) NOT NULL DEFAULT 'pending' COMMENT '验证状态 pending/data_verified/human_verified/disputed'"), + ("verified_at", "ALTER TABLE kpi_causality ADD COLUMN verified_at DATETIME NULL COMMENT '验证时间'"), + ("verified_by", "ALTER TABLE kpi_causality ADD COLUMN verified_by VARCHAR(50) NULL COMMENT '验证人/AI/脚本'"), + ("entity_id", "ALTER TABLE kpi_causality ADD COLUMN entity_id INT NOT NULL DEFAULT 1 COMMENT '企业ID (多租户隔离 2026-08-27)'"), +] + + +def run(): + engine = get_engine() + with engine.connect() as conn: + # 1. 检查表是否存在 + exists = conn.execute(text( + "SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = DATABASE() AND table_name = 'kpi_causality'" + )).scalar() + if not exists: + logger.error("kpi_causality 表不存在，跳过") + return 1 + + # 2. 现有列 + existing = {r[0] for r in conn.execute(text( + "SELECT column_name FROM information_schema.columns WHERE table_schema = DATABASE() AND table_name = 'kpi_causality'" + ))} + logger.info("现有列: %s", sorted(existing)) + + # 3. 加列（幂等） + for col, ddl in COLUMNS: + if col in existing: + logger.info("列 %s 已存在，跳过", col) + else: + conn.execute(text(ddl)) + logger.info("已添加列 %s", col) + + # 4. 回填 entity_id（无条件从 source KPI 对齐，纠正默认值偏差） + # 仅当来源KPI存在才回填；无来源KPI的孤儿链保持原值 + conn.execute(text( + "UPDATE kpi_causality c JOIN kpi_definitions k ON k.id = c.source_kpi_id " + "SET c.entity_id = k.entity_id" + )) + orphan = conn.execute(text( + "SELECT COUNT(*) FROM kpi_causality c LEFT JOIN kpi_definitions k ON k.id = c.source_kpi_id " + "WHERE k.id IS NULL" + )).scalar() + if orphan: + logger.warning("%d 条因果链无来源KPI（孤儿链）", orphan) + else: + logger.info("entity_id 已全部按来源KPI回填") + + # 5. 验证 + cols = {r[0] for r in conn.execute(text( + "SELECT column_name FROM information_schema.columns WHERE table_schema = DATABASE() AND table_name = 'kpi_causality'" + ))} + missing = {c for c, _ in COLUMNS} - cols + if missing: + logger.error("仍有缺失列: %s", missing) + return 1 + row = conn.execute(text( + "SELECT COUNT(*) FROM kpi_causality WHERE entity_id IS NULL OR entity_id = 0" + )).scalar() + if row: + logger.error("仍有 %d 行 entity_id 为空", row) + return 1 + total = conn.execute(text("SELECT COUNT(*) FROM kpi_causality")).scalar() + logger.info("迁移完成: kpi_causality %d 条, 新列: source_type/verify_status/verified_at/verified_by/entity_id", total) + return 0 + + +if __name__ == "__main__": + sys.exit(run()) diff --git a/backend/tests/test_causality_verification.py b/backend/tests/test_causality_verification.py new file mode 100644 index 00000000..4f455c60 --- /dev/null +++ b/backend/tests/test_causality_verification.py @@ -0,0 +1,363 @@ +"""因果链验证机制测试 — 数据验证核心 + 状态机 + API (2026-08-27 P2) + +覆盖: +1. 服务层: parse_period / pearson / align_series(滞后) / evaluate_chain / apply_state_machine +2. API: create(source_type) / verify-status / verify(人工确认) / entity隔离 / 权限 +""" +import hashlib + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy.orm import Session + +from app.models import KPIDefinition, KPICausality, KPIValue, Entity, User +from app.services.causality_verification import ( + STATUS_DATA_VERIFIED, + STATUS_DISPUTED, + STATUS_HUMAN_VERIFIED, + STATUS_PENDING, + align_series, + apply_state_machine, + evaluate_chain, + parse_period, + pearson, + summarize, +) +from tests.conftest import create_test_user, get_token_for_user, auth_header + +BASE = "/api/cma/kpi-causality" + + +def _seed_kpi(db: Session, code: str, name: str = None, dimension: str = "finance", + entity_id: int = 1) -> KPIDefinition: + kpi = KPIDefinition( + kpi_code=code, kpi_name=name or code, dimension=dimension, + entity_id=entity_id, status="active", target_value=100.0, + ) + db.add(kpi) + db.commit() + db.refresh(kpi) + return kpi + + +def _seed_chain(db: Session, source_type: str = "AI_suggested"): + src = _seed_kpi(db, "BH_REVENUE", "营业收入") + tgt = _seed_kpi(db, "BH_NET_PROFIT", "净利润") + c = KPICausality(entity_id=src.entity_id, source_kpi_id=src.id, target_kpi_id=tgt.id, + strength=0.5, lag_months=0, direction="positive", + source_type=source_type, verify_status=STATUS_PENDING) + db.add(c) + db.commit() + db.refresh(c) + return src, tgt, c + + +class TestParsePeriod: + def test_month(self): + assert parse_period("2026-07") == ("month", 2026 * 12 + 6) + + def test_half(self): + assert parse_period("2026-H1") == ("half", 2026 * 12 + 5) + assert parse_period("2026-H2") == ("half", 2026 * 12 + 11) + + def test_year(self): + assert parse_period("2026") == ("year", 2026 * 12 + 5) + + def test_invalid(self): + assert parse_period("abc") is None + assert parse_period("") is None + assert parse_period(None) is None + + +class TestPearson: + def test_perfect_positive(self): + r, n = pearson([1, 2, 3, 4], [2, 4, 6, 8]) + assert n == 4 + assert abs(r - 1.0) < 1e-9 + + def test_perfect_negative(self): + r, n = pearson([1, 2, 3, 4], [8, 6, 4, 2]) + assert abs(r + 1.0) < 1e-9 + + def test_known_value(self): + # 与 numpy 核对过的样例 (F_REVENUE / F_NET_PROFIT 7点) + xs = [180.87, 132.33, 120.15, 60.5, 90.09, 129.32, 81.08] + ys = [94.31, -85.06, -21.66, -3.74, -21.91, -45.04, -17.63] + r, n = pearson(xs, ys) + assert n == 7 + assert abs(r - 0.394012) < 1e-4 + + def test_insufficient(self): + r, n = pearson([1], [2]) + assert r is None and n == 1 + + def test_constant_series(self): + r, n = pearson([3, 3, 3], [1, 2, 3]) + assert r is None and n == 3 + + +class TestAlignSeries: + def test_no_lag(self): + src = [("2026-01", 1), ("2026-02", 2), ("2026-03", 3)] + tgt = [("2026-01", 10), ("2026-02", 20), ("2026-03", 30)] + g, pairs = align_series(src, tgt, lag_months=0) + assert g == "month" + assert pairs == [(1, 10), (2, 20), (3, 30)] + + def test_lag_alignment(self): + """source t 与 target t+lag 配对""" + src = [("2026-01", 1), ("2026-02", 2), ("2026-03", 3)] + tgt = [("2026-02", 10), ("2026-03", 20), ("2026-04", 30)] + g, pairs = align_series(src, tgt, lag_months=1) + assert pairs == [(1, 10), (2, 20), (3, 30)] + + def test_granularity_filter(self): + """月度/半年度混用时只取同粒度（优先月）""" + src = [("2026-01", 1), ("2026-02", 2), ("2026-H1", 3)] + tgt = [("2026-01", 10), ("2026-02", 20), ("2026-H1", 30)] + g, pairs = align_series(src, tgt, lag_months=0) + assert g == "month" + assert pairs == [(1, 10), (2, 20)] + + +class TestEvaluateChain: + def test_data_verified_positive(self): + src = [(f"2026-{m:02d}", m) for m in range(1, 9)] + tgt = [(f"2026-{m:02d}", m * 2) for m in range(1, 9)] + ev = evaluate_chain(src, tgt, lag_months=0, direction="positive") + assert ev["status"] == STATUS_DATA_VERIFIED + assert ev["direction_consistent"] is True + assert ev["n"] == 8 + + def test_data_verified_negative(self): + src = [(f"2026-{m:02d}", m) for m in range(1, 9)] + tgt = [(f"2026-{m:02d}", -m * 2) for m in range(1, 9)] + ev = evaluate_chain(src, tgt, lag_months=0, direction="negative") + assert ev["status"] == STATUS_DATA_VERIFIED + + def test_direction_conflict(self): + """声明 positive 但实际负相关 → disputed""" + src = [(f"2026-{m:02d}", m) for m in range(1, 9)] + tgt = [(f"2026-{m:02d}", -m) for m in range(1, 9)] + ev = evaluate_chain(src, tgt, lag_months=0, direction="positive") + assert ev["status"] == STATUS_DISPUTED + assert "方向矛盾" in ev["reason"] + + def test_weak_correlation(self): + """弱相关（方向一致但|r|<阈值）→ disputed""" + # numpy seed=1: x=[1..8], y=x+N(0,6) → r≈0.119 (弱正相关) + src = [(f"2026-{m:02d}", m) for m in range(1, 9)] + tgt = [(f"2026-{m:02d}", y) for m, y in enumerate( + [10.75, -1.67, -0.17, -2.44, 10.19, -7.81, 17.47, 3.43], start=1)] + ev = evaluate_chain(src, tgt, lag_months=0, direction="positive") + assert ev["status"] == STATUS_DISPUTED + assert "弱相关" in ev["reason"] + + def test_insufficient_points(self): + """数据点不足 → pending""" + src = [("2026-01", 1), ("2026-02", 2)] + tgt = [("2026-01", 10), ("2026-02", 20)] + ev = evaluate_chain(src, tgt, lag_months=0, direction="positive") + assert ev["status"] == STATUS_PENDING + + def test_no_shared_periods(self): + src = [("2026-01", 1)] + tgt = [("2026-02", 10)] + ev = evaluate_chain(src, tgt, lag_months=0, direction="positive") + assert ev["status"] == STATUS_PENDING + + +class TestStateMachine: + def test_pending_to_verified(self): + st, note = apply_state_machine(STATUS_PENDING, STATUS_DATA_VERIFIED) + assert st == STATUS_DATA_VERIFIED and note is None + + def test_pending_to_disputed(self): + st, _ = apply_state_machine(STATUS_PENDING, STATUS_DISPUTED) + assert st == STATUS_DISPUTED + + def test_human_verified_not_overridden(self): + st, note = apply_state_machine(STATUS_HUMAN_VERIFIED, STATUS_DISPUTED) + assert st == STATUS_HUMAN_VERIFIED + assert note is not None # 数据矛盾警示 + + def test_human_verified_positive_note_none(self): + st, note = apply_state_machine(STATUS_HUMAN_VERIFIED, STATUS_DATA_VERIFIED) + assert st == STATUS_HUMAN_VERIFIED and note is None + + def test_insufficient_keeps_status(self): + st, _ = apply_state_machine(STATUS_PENDING, STATUS_PENDING) + assert st == STATUS_PENDING + + +class TestSummarize: + def test_counts(self): + s = summarize([{"status": STATUS_DATA_VERIFIED}, {"status": STATUS_DISPUTED}, + {"status": STATUS_PENDING}, {"status": STATUS_HUMAN_VERIFIED}]) + assert s["total"] == 4 + assert s["by_status"][STATUS_DATA_VERIFIED] == 1 + assert s["by_status"][STATUS_DISPUTED] == 1 + assert s["by_status"][STATUS_PENDING] == 1 + assert s["by_status"][STATUS_HUMAN_VERIFIED] == 1 + + +class TestCausalityVerificationAPI: + def test_create_with_source_type(self, client: TestClient, db: Session): + create_test_user(db) + token = get_token_for_user(client) + src = _seed_kpi(db, "BH_REVENUE", "营业收入") + tgt = _seed_kpi(db, "BH_NET_PROFIT", "净利润") + + resp = client.post(BASE, headers=auth_header(token), json={ + "source_kpi_id": src.id, "target_kpi_id": tgt.id, + "source_type": "AI_suggested", + }) + assert resp.status_code == 200 + body = resp.json() + assert body["source_type"] == "AI_suggested" + assert body["verify_status"] == STATUS_PENDING + + def test_create_invalid_source_type(self, client: TestClient, db: Session): + create_test_user(db) + token = get_token_for_user(client) + src = _seed_kpi(db, "BH_REVENUE", "营业收入") + tgt = _seed_kpi(db, "BH_NET_PROFIT", "净利润") + resp = client.post(BASE, headers=auth_header(token), json={ + "source_kpi_id": src.id, "target_kpi_id": tgt.id, "source_type": "unknown", + }) + assert resp.status_code == 400 + + def test_update_resets_verification(self, client: TestClient, db: Session): + create_test_user(db) + token = get_token_for_user(client) + src, tgt, c = _seed_chain(db) + # 先人工确认 + resp = client.put(f"{BASE}/{c.id}/verify", headers=auth_header(token), + json={"verify_status": "human_verified", "verified_by": "任富海"}) + assert resp.json()["verify_status"] == STATUS_HUMAN_VERIFIED + # 修改链定义 → 状态回到 pending + resp2 = client.put(f"{BASE}/{c.id}", headers=auth_header(token), json={"strength": 0.9}) + assert resp2.json()["verify_status"] == STATUS_PENDING + assert resp2.json()["verified_by"] is None + + def test_verify_status_endpoint(self, client: TestClient, db: Session): + create_test_user(db) + token = get_token_for_user(client) + src, tgt, c = _seed_chain(db) + + resp = client.get(f"{BASE}/verify-status", headers=auth_header(token)) + assert resp.status_code == 200 + body = resp.json() + assert body["summary"]["total"] == 1 + assert body["summary"]["by_status"][STATUS_PENDING] == 1 + assert body["data"][0]["id"] == c.id + assert body["data"][0]["verify_status"] == STATUS_PENDING + + def test_verify_status_filter(self, client: TestClient, db: Session): + create_test_user(db) + token = get_token_for_user(client) + src, tgt, c = _seed_chain(db) + client.put(f"{BASE}/{c.id}/verify", headers=auth_header(token), + json={"verify_status": "human_verified"}) + + resp = client.get(f"{BASE}/verify-status?verify_status=human_verified", + headers=auth_header(token)) + assert resp.json()["summary"]["total"] == 1 + resp2 = client.get(f"{BASE}/verify-status?verify_status=pending", headers=auth_header(token)) + assert resp2.json()["summary"]["total"] == 0 + + def test_human_verify(self, client: TestClient, db: Session): + create_test_user(db) + token = get_token_for_user(client) + src, tgt, c = _seed_chain(db) + + resp = client.put(f"{BASE}/{c.id}/verify", headers=auth_header(token), + json={"verify_status": "human_verified", "verified_by": "任富海"}) + assert resp.status_code == 200 + body = resp.json() + assert body["verify_status"] == STATUS_HUMAN_VERIFIED + assert body["verified_by"] == "任富海" + assert body["verified_at"] is not None + + def test_human_verify_default(self, client: TestClient, db: Session): + create_test_user(db) + token = get_token_for_user(client) + src, tgt, c = _seed_chain(db) + resp = client.put(f"{BASE}/{c.id}/verify", headers=auth_header(token), json={}) + assert resp.json()["verify_status"] == STATUS_HUMAN_VERIFIED + assert resp.json()["verified_by"] is not None # 默认取用户名 + + def test_verify_disputed(self, client: TestClient, db: Session): + create_test_user(db) + token = get_token_for_user(client) + src, tgt, c = _seed_chain(db) + resp = client.put(f"{BASE}/{c.id}/verify", headers=auth_header(token), + json={"verify_status": "disputed"}) + assert resp.json()["verify_status"] == STATUS_DISPUTED + + def test_verify_invalid_status(self, client: TestClient, db: Session): + create_test_user(db) + token = get_token_for_user(client) + src, tgt, c = _seed_chain(db) + resp = client.put(f"{BASE}/{c.id}/verify", headers=auth_header(token), + json={"verify_status": "bogus"}) + assert resp.status_code == 400 + + def test_verify_not_found(self, client: TestClient, db: Session): + create_test_user(db) + token = get_token_for_user(client) + resp = client.put(f"{BASE}/99999/verify", headers=auth_header(token), + json={"verify_status": "human_verified"}) + assert resp.status_code == 404 + + def test_business_cannot_verify(self, client: TestClient, db: Session): + """business 角色无写权限 → 403""" + business = User( + username="business_verify", password_hash=hashlib.sha256("pass123".encode()).hexdigest(), + name="业务员", role="business", + ) + db.add(business) + db.commit() + token = get_token_for_user(client, username="business_verify", password="pass123") + src, tgt, c = _seed_chain(db) + resp = client.put(f"{BASE}/{c.id}/verify", headers=auth_header(token), + json={"verify_status": "human_verified"}) + assert resp.status_code == 403 + + def test_entity_isolation(self, client: TestClient, db: Session): + """企业B看不到企业A的链，也不能verify企业A的链""" + create_test_user(db) + token = get_token_for_user(client) # entity_id=1 + _seed_entity2(db) + src2 = _seed_kpi(db, "BH2_REVENUE", "博海收入", entity_id=2) + tgt2 = _seed_kpi(db, "BH2_PROFIT", "博海利润", entity_id=2) + c2 = KPICausality(entity_id=2, source_kpi_id=src2.id, target_kpi_id=tgt2.id, + strength=0.5, lag_months=0, direction="positive", + source_type="manual", verify_status=STATUS_PENDING) + db.add(c2) + db.commit() + + # entity1 的 verify-status 看不到 entity2 的链 + resp = client.get(f"{BASE}/verify-status", headers=auth_header(token)) + assert resp.json()["summary"]["total"] == 0 + # entity1 的 token verify entity2 的链 → 404 + resp2 = client.put(f"{BASE}/{c2.id}/verify", headers=auth_header(token), + json={"verify_status": "human_verified"}) + assert resp2.status_code == 404 + + def test_network_includes_verify_status(self, client: TestClient, db: Session): + create_test_user(db) + token = get_token_for_user(client) + src, tgt, c = _seed_chain(db) + resp = client.get(f"{BASE}/kpi/{src.id}/network", headers=auth_header(token)) + assert resp.status_code == 200 + downstream = resp.json()["downstream"] + assert downstream[0]["verify_status"] == STATUS_PENDING + assert downstream[0]["source_type"] == "AI_suggested" + + +def _seed_entity2(db: Session) -> None: + ent = db.query(Entity).filter(Entity.id == 2).first() + if not ent: + db.add(Entity(id=2, name="博海网络科技", short_name="博海", status="active")) + db.commit() diff --git a/backend/tests/test_kpi_causality.py b/backend/tests/test_kpi_causality.py index 8a828fde..d27bc371 100644 --- a/backend/tests/test_kpi_causality.py +++ b/backend/tests/test_kpi_causality.py @@ -348,6 +348,6 @@ class TestPermissions: assert resp.status_code == 200 def test_no_token_denied(self, client: TestClient): - """无token → 403""" + """无token → 401（HTTPBearer 标准行为）""" resp = client.get(BASE) - assert resp.status_code == 403 + assert resp.status_code == 401 diff --git a/frontend/src/api/index.ts b/frontend/src/api/index.ts index e536dc16..feb049bd 100644 --- a/frontend/src/api/index.ts +++ b/frontend/src/api/index.ts @@ -293,6 +293,8 @@ export const kpiCausalityApi = { getNetwork: (kpiId: number) => api.get(`/kpi-causality/kpi/${kpiId}/network`), getFullNetwork: () => api.get('/kpi-causality/full-network'), simulate: (data: any) => api.post('/kpi-causality/simulate', data), + getVerifyStatus: (params?: any) => api.get('/kpi-causality/verify-status', { params }), + verify: (id: number, data: any) => api.put(`/kpi-causality/${id}/verify`, data), } export const dataQualityApi = { diff --git a/frontend/src/views/KPIDetail.vue b/frontend/src/views/KPIDetail.vue index 2ccbb380..68da0d34 100644 --- a/frontend/src/views/KPIDetail.vue +++ b/frontend/src/views/KPIDetail.vue @@ -216,7 +216,13 @@ -
{{ item.kpi_name }} ({{ item.kpi_code }})
+
+ {{ item.kpi_name }} ({{ item.kpi_code }}) + + {{ verifyBadge(item.verify_status).text }} + +
{{ item.direction === 'positive' ? '正向驱动' : '负向抑制' }} · {{ item.formula }} @@ -233,7 +239,13 @@ -
{{ item.kpi_name }} ({{ item.kpi_code }})
+
+ {{ item.kpi_name }} ({{ item.kpi_code }}) + + {{ verifyBadge(item.verify_status).text }} + +
{{ item.direction === 'positive' ? '正向推动' : '负向抑制' }} · {{ item.formula }} @@ -408,6 +420,17 @@ function statusLabel(s: string) { } return map[s] || s || '—' } + +// 因果链验证状态映射（三层验证机制 2026-08-27） +function verifyBadge(s: string) { + const map: Record = { + data_verified: { text: '✅ 数据证实', color: '#67c23a', bg: 'rgba(103,194,58,.12)' }, + human_verified: { text: '✅ 人工确认', color: '#409eff', bg: 'rgba(64,158,255,.12)' }, + disputed: { text: '⚠️ 存疑', color: '#e6a23c', bg: 'rgba(230,162,60,.12)' }, + pending: { text: '⏳ 待检', color: '#909399', bg: 'rgba(144,147,153,.12)' }, + } + return map[s] || { text: s || '—', color: '#909399', bg: 'rgba(144,147,153,.12)' } +} const mapOptions = ref([]) const activeTab = ref('basic')