guanlan-src/app_ontology/app_ontology_guanlan/sop/backfill.py

# -*- coding: utf-8 -*-
"""backfill.py — 回灌器: cases.json(生命周期结局) → 判据记分卡 + 误报台账 (运维 agent 柱1 飞轮右半).

飞轮: findings→case(cases.py 左半, 发现) →[人工 工单/修/回验]→ case 结局 →**本模块**聚合成
学习信号 →[人裁]→ 判据 n+1 版本化 → 重部署。本模块 = 附录 §8 G3 离线批式学习的机器件。

★RULE-3 铁律 (禁在线自适应): 本模块**只出报告/提案, 不改任何阈值文件**。
  阈值/判据更新 = 人裁 + thresholds.yaml 版本化 + 重部署 (禁运行中自更新)。
  proposals 段 verdict 全 =【参考】, 是"提请人看"的证据, 非"已应用"的动作。飞轮的"轮"在人手里转。

纯函数 (backfill): 读 cases list → 出 report dict; 无副作用/无时钟 (日期从 history.ts 字符串解析)。
判据归因: 一个 case 的结局记给它 evidence 里每个 distinct discriminator_id (decay 除外) —
  多判据喂同一 case 则各记一次 (确诊=各 TP, 误报=各 FP)。
"""

try:
    from app_common.app_common_guanlan.api import install_root as _install_root
except ImportError:                     # 理论不可达；包结构异常时回退到按位置上跳
    from pathlib import Path as _P
    def _install_root(_f): return _P(_f).resolve().parents[4]
from src import paths as P
from datetime import datetime


def _pdate(s):
    """'YYYY-MM-DD'(或前10字符) → datetime; 失败→None (无时钟依赖)。"""
    try:
        return datetime.strptime(str(s)[:10], '%Y-%m-%d')
    except (ValueError, TypeError):
        return None


def _case_discriminators(case):
    """case 的责任判据 = evidence 里 distinct discriminator_id (排除 decay/field_measurement)。"""
    out = []
    for e in (case.get('evidence') or []):
        did = isinstance(e, dict) and e.get('discriminator_id')
        if did and did not in ('decay', 'field_measurement') and did not in out:
            out.append(did)
    return out


def _confirmed_ts(case):
    """history 里首个 to_state=CONFIRMED 的 ts (确诊时刻); 无→None。"""
    for h in (case.get('history') or []):
        if isinstance(h, dict) and h.get('to_state') == 'CONFIRMED':
            return h.get('ts')
    return None


def backfill(cases, precision_floor=0.7, min_resolved=3):
    """回灌聚合 (纯函数, RULE-3: 出提案不改阈值)。

    cases: cases.json 解析 list。precision_floor/min_resolved: 提案触发阈 (低精度且够样本才提请人裁)。
    结局分类 (per case 终局): CLOSED=真阳(TP, 判据抓对且修复回验过) / REFUTED=误报(FP) /
      RELAPSED=复发 / CONFIRMED·TICKETED·IN_REPAIR·VERIFY=在办(未结) / CANDIDATE·WATCH·NORMAL=未决。
    """
    stats = {}     # did → 计数器

    def _s(did):
        return stats.setdefault(did, {
            'n_total': 0, 'n_tp_closed': 0, 'n_fp_refuted': 0, 'n_relapsed': 0,
            'n_confirmed_open': 0, 'n_unresolved': 0, 'lead_days': [], 'repair_pass': 0, 'repair_fail': 0})

    fp_ledger, relapse_ledger = [], []
    ACTIVE_UNRESOLVED = {'NORMAL', 'WATCH', 'CANDIDATE'}
    OPEN_INFLIGHT = {'CONFIRMED', 'TICKETED', 'IN_REPAIR', 'VERIFY'}

    for c in (cases or []):
        if not isinstance(c, dict):
            continue
        state = c.get('state')
        dids = _case_discriminators(c) or ['(unmapped)']
        for did in dids:
            s = _s(did)
            s['n_total'] += 1
            if state == 'CLOSED':
                s['n_tp_closed'] += 1
            elif state == 'REFUTED':
                s['n_fp_refuted'] += 1
            elif state == 'RELAPSED':
                s['n_relapsed'] += 1
            elif state in OPEN_INFLIGHT:
                s['n_confirmed_open'] += 1
            elif state in ACTIVE_UNRESOLVED:
                s['n_unresolved'] += 1
            # 提前量: opened_at → 确诊 ts
            cts = _confirmed_ts(c)
            d0, d1 = _pdate(c.get('opened_at')), _pdate(cts)
            if d0 and d1 and d1 >= d0:
                s['lead_days'].append((d1 - d0).days)
            # 修复有效性
            vr = (c.get('verify') or {}).get('result')
            if vr == 'pass':
                s['repair_pass'] += 1
            elif vr == 'fail':
                s['repair_fail'] += 1
        # 台账 (case 级, 不按判据展开)
        if state == 'REFUTED':
            fp_ledger.append({'case_id': c.get('case_id'), 'system': c.get('system'),
                              'discriminators': dids, 'refuted_basis': c.get('refuted_basis', '')})
        elif state == 'RELAPSED':
            relapse_ledger.append({'case_id': c.get('case_id'), 'system': c.get('system'),
                                   'relapse_of': c.get('relapse_of'), 'discriminators': dids})

    # 派生指标 + 提案 (RULE-3: 参考级, 提请人裁 n+1)
    by_disc, proposals = {}, []
    for did, s in sorted(stats.items()):
        resolved = s['n_tp_closed'] + s['n_fp_refuted']       # 已结 = TP+FP
        prec = round(s['n_tp_closed'] / resolved, 3) if resolved else None
        lead = sorted(s['lead_days'])
        med_lead = lead[len(lead) // 2] if lead else None
        rep_tot = s['repair_pass'] + s['repair_fail']
        by_disc[did] = {**{k: v for k, v in s.items() if k != 'lead_days'},
                        'n_resolved': resolved, 'precision_proxy': prec, 'median_lead_days': med_lead,
                        'repair_effectiveness': round(s['repair_pass'] / rep_tot, 3) if rep_tot else None}
        if prec is not None and resolved >= min_resolved and prec < precision_floor:
            proposals.append({
                'discriminator_id': did, 'verdict': '参考',
                'reason': f'精度代理 {prec} < {precision_floor} (已结 {resolved}: {s["n_tp_closed"]}真阳/'
                          f'{s["n_fp_refuted"]}误报) → 提请人裁: 该判据阈值/守卫是否需 n+1 收紧',
                'action_required': '人裁 + thresholds.yaml 版本化 + 重部署 (禁自动应用, RULE-3)'})

    return {
        'n_cases': len([c for c in (cases or []) if isinstance(c, dict)]),
        'by_discriminator': by_disc,
        'fp_ledger': fp_ledger,
        'relapse_ledger': relapse_ledger,
        'proposals': proposals,
        '_rule3': 'RULE-3: 本报告=提请人裁的证据链, 非自动应用; 判据 n+1 须人裁+版本化+重部署, 禁在线自适应。',
    }


def render_md(report, farm='?'):
    """报告 → 人读 markdown 摘要 (提案在顶, 台账在中, 记分卡在底)。"""
    L = [f'> ⚠️ **回灌报告 · {farm}** — `src/sop/backfill.py` 生成, 勿手改。'
         f'**{report["_rule3"]}**\n', f'# 回灌记分卡 · {farm} ({report["n_cases"]} cases)\n']
    props = report['proposals']
    L.append(f'## 提请人裁 ({len(props)} 项)' + (' — 无 (无判据触发低精度阈)' if not props else ''))
    for p in props:
        L.append(f'- **{p["discriminator_id"]}** 【{p["verdict"]}】: {p["reason"]}\n  - 须: {p["action_required"]}')
    L.append(f'\n## 误报台账 ({len(report["fp_ledger"])} 条)')
    for f in report['fp_ledger']:
        L.append(f'- `{f["case_id"]}` ({f["system"]}) 判据{f["discriminators"]} → 证伪依据: {f["refuted_basis"]}')
    L.append(f'\n## 复发台账 ({len(report["relapse_ledger"])} 条)')
    for r in report['relapse_ledger']:
        L.append(f'- `{r["case_id"]}` ({r["system"]}) 复发自 {r["relapse_of"]}')
    L.append('\n## 判据记分卡\n| 判据 | 总 | 真阳(销案) | 误报(证伪) | 复发 | 在办 | 未决 | 精度代理 | 中位提前量d | 修复有效 |')
    L.append('|---|--|--|--|--|--|--|--|--|--|')
    for did, s in report['by_discriminator'].items():
        L.append(f'| {did} | {s["n_total"]} | {s["n_tp_closed"]} | {s["n_fp_refuted"]} | {s["n_relapsed"]} '
                 f'| {s["n_confirmed_open"]} | {s["n_unresolved"]} | {s["precision_proxy"]} '
                 f'| {s["median_lead_days"]} | {s["repair_effectiveness"]} |')
    return '\n'.join(L)


# ---------------------------------------------------------------------------
if __name__ == '__main__':
    # 控制台可能是 GBK(中文 Windows 代码页 936): 正文里的 ✔ ✗ ✅ ⚠ 这类字符编不出来会抛
    # UnicodeEncodeError, 脚本干成了事却以退出码 1 结束(同类坑见 src/console.py)。降级为 '?' 而不是崩;
    # 不用 import 是为了兼顾 python -m 与直接当脚本跑两种启动方式。
    import sys as _sys
    for _s in (_sys.stdout, _sys.stderr):
        try: _s.reconfigure(errors='replace')
        except Exception: pass
    # CLI: python -m src.sop.backfill <farm|path>  —— 读 cases.json → 出 backfill_report.{json,md}
    import argparse
    import json
    import sys
    from pathlib import Path
    ROOT = _install_root(__file__)
    ap = argparse.ArgumentParser()
    ap.add_argument('target', help='farm 名 (读 outputs/<farm>/sop/cases.json) 或直接 cases.json 路径')
    ap.add_argument('--dry', action='store_true')
    a = ap.parse_args()
    p = Path(a.target)
    if not p.exists():
        p = P.sop(a.target) / 'cases.json'
    if not p.exists():
        print(f'❌ 找不到 cases.json: {a.target}')
        sys.exit(2)
    cases = json.loads(p.read_text(encoding='utf-8'))
    farm = p.parent.parent.name if p.parent.name == 'sop' else a.target
    rep = backfill(cases)
    print(f'✅ {farm}: {rep["n_cases"]} cases → {len(rep["by_discriminator"])} 判据, '
          f'{len(rep["proposals"])} 提案, {len(rep["fp_ledger"])} 误报, {len(rep["relapse_ledger"])} 复发')
    if not a.dry:
        (p.parent / 'backfill_report.json').write_text(json.dumps(rep, ensure_ascii=False, indent=2), encoding='utf-8')
        (p.parent / 'backfill_report.md').write_text(render_md(rep, farm), encoding='utf-8')
        print(f'   写盘: {(p.parent / "backfill_report.json").relative_to(ROOT)} + .md')