From 293df6df8c2b75f4c99a6efb4a9b5339dc4197a7 Mon Sep 17 00:00:00 2001 From: Hermes CI Fix Date: Thu, 17 Sep 2026 00:01:58 +0800 Subject: [PATCH] feat(cma): cashflow NA annotation + dimension guard + threshold unit migration script dev-level prep only: no prod write, no push, no pipeline deploy. --- backend/app/utils/dimension_guard.py | 209 +++++++++++++++++ backend/scripts/import_cashflow_statements.py | 81 ++++++- .../migrate_kpi_threshold_unit_20260916.py | 216 ++++++++++++++++++ 3 files changed, 503 insertions(+), 3 deletions(-) create mode 100644 backend/app/utils/dimension_guard.py create mode 100644 backend/scripts/migrate_kpi_threshold_unit_20260916.py diff --git a/backend/app/utils/dimension_guard.py b/backend/app/utils/dimension_guard.py new file mode 100644 index 00000000..dcd07acc --- /dev/null +++ b/backend/app/utils/dimension_guard.py @@ -0,0 +1,209 @@ +# -*- coding: utf-8 -*- +"""量纲守卫（Dimension Guard）—— 阈值与实测值跨量级比较时拒绝判定 + +依据：财务口径裁定 `finance-20260916-cma-caliber-ruling.md` §四（裁定③） + · 同量纲铁律：阈值表达式须与 kpi_definitions.unit 同量纲 + · 量纲守卫：|log10(阈值)| 与 |log10(实测)| 相差 ≥1 个数量级 → **拒绝判定**， + 标记「口径错误」，**不得输出红/黄/绿** + · 典型：流动比率实测 1.29（倍）vs 阈值 `<150`（百分数写法）→ 跨 2 个量级 → 假红 + +设计原则（**additive / opt-in**）： + · 本模块**不修改** `threshold_parser` 的任何既有行为，也不改现有调用方； + · 判定侧改用 `calc_level_guarded()` 即得守卫（返回 dict，含 level 与守卫明细）； + · 装配到 alert_generator / kpis.py 属生产行为变更，需审批后单独执行（本单只备代码）。 + +单位维度表（裁定 §4.1）： + · 倍（无量纲）→ ratio ：流动比率、速动比率、现金比率 + · % → percent ：资产负债率、毛利率、净利率、保本覆盖率、返利率 + · 天 / 次 → days / times：周转天数、周转率 + · 万元 / 元 → currency_wan / currency_yuan +""" +from __future__ import annotations + +import math +from typing import Optional, Tuple, Union + +from app.utils.threshold_parser import LEVEL_PRIORITY, eval_threshold, parse_clauses + +#: 单位 → 维度 +UNIT_DIMENSION = { + "倍": "ratio", + "": "ratio", + "%": "percent", + "百分比": "percent", + "万元": "currency_wan", + "元": "currency_yuan", + "天": "days", + "次": "times", + "人": "count", + "个": "count", + "家": "count", + "分": "score", +} + +#: 默认拒绝阈值：量级差 ≥ 1.0 → 口径错误 +DEFAULT_ORDERS_GAP = 1.0 + +#: 判定返回：口径错误时的 level 常量（不出红/黄/绿） +LEVEL_DIMENSION_ERROR = "口径错误" + + +def dimension_of(unit) -> str: + """单位字符串 → 维度名（未知单位返回 'unknown'）。""" + if unit is None: + return "unknown" + return UNIT_DIMENSION.get(str(unit).strip(), "unknown") + + +def _orders(v) -> Optional[float]: + """数值量级 = log10(|v|)；0 / None / 非数值 → None（不可判）。""" + if v is None: + return None + try: + f = float(v) + except (TypeError, ValueError): + return None + if f == 0: + return None + return math.log10(abs(f)) + + +def _clause_values(clauses) -> list: + """子句 → 候选比较值列表（区间取两端）。""" + vals = [] + for _op, val in clauses: + if isinstance(val, tuple): + vals.extend(list(val)) + else: + vals.append(val) + return vals + + +def check_dimension(actual, expr, unit=None, orders_gap: float = DEFAULT_ORDERS_GAP) -> dict: + """检查「实测值 vs 阈值表达式」是否同量级。 + + 返回 dict： + ok True=可判定；False=口径错误（拒绝判定） + verdict 'ok' | 'dimension_error' | 'undecidable' + gap 与最接近子句的量级差（None=不可判） + nearest 最接近的子句值 + reason 人类可读说明 + """ + if actual is None or expr is None or not str(expr).strip(): + return {"ok": True, "verdict": "undecidable", "gap": None, "nearest": None, + "reason": "无实测值或无阈值 → 守卫不介入", "dimension": dimension_of(unit)} + + a = _orders(actual) + clauses = parse_clauses(expr) + if a is None: + return {"ok": True, "verdict": "undecidable", "gap": None, "nearest": None, + "reason": "实测值为 0（量级不可判）→ 守卫不介入", "dimension": dimension_of(unit)} + if not clauses: + return {"ok": True, "verdict": "undecidable", "gap": None, "nearest": None, + "reason": "阈值表达式无法解析 → 守卫不介入（解析器自身会返回 False）", + "dimension": dimension_of(unit)} + + best_gap, nearest = None, None + for v in _clause_values(clauses): + o = _orders(v) + if o is None: + continue + g = abs(a - o) + if best_gap is None or g < best_gap: + best_gap, nearest = g, v + + if best_gap is None: + return {"ok": True, "verdict": "undecidable", "gap": None, "nearest": None, + "reason": "阈值子句全为 0 → 守卫不介入", "dimension": dimension_of(unit)} + + if best_gap >= orders_gap: + return {"ok": False, "verdict": "dimension_error", "gap": round(best_gap, 3), + "nearest": nearest, + "reason": ("量级差 %.2f ≥ %.1f 个数量级：实测 %s（单位 %s）vs 阈值 %r（最近子句 %s）" + " → 跨量级比较，判定作废" % (best_gap, orders_gap, actual, unit, expr, nearest)), + "dimension": dimension_of(unit)} + return {"ok": True, "verdict": "ok", "gap": round(best_gap, 3), "nearest": nearest, + "reason": "量级差 %.2f < %.1f → 同量级可判" % (best_gap, orders_gap), + "dimension": dimension_of(unit)} + + +def calc_level_guarded(actual, kpi, orders_gap: float = DEFAULT_ORDERS_GAP) -> dict: + """带量纲守卫的红黄绿判定（不改 threshold_parser 的原行为）。 + + kpi 需含 threshold_green/yellow/red 与 unit 属性（ORM 对象或同名 duck type）。 + 返回 {"level": 'red'|'yellow'|'green'|'info'|'口径错误', "guard": {...}, "expr": 命中的表达式} + 口径错误：level = LEVEL_DIMENSION_ERROR，不出红/黄/绿（裁定 §4.2）。 + """ + exprs = { + "green": getattr(kpi, "threshold_green", None), + "yellow": getattr(kpi, "threshold_yellow", None), + "red": getattr(kpi, "threshold_red", None), + } + unit = getattr(kpi, "unit", None) + if actual is None: + return {"level": "info", "guard": {"ok": True, "verdict": "undecidable", + "reason": "无实测值", "dimension": dimension_of(unit)}, + "expr": None} + + # 逐等级守卫：任一等级被守卫生效（口径错误）→ 不进红/黄/绿 + for level in LEVEL_PRIORITY: + expr = exprs[level] + if not expr: + continue + g = check_dimension(actual, expr, unit, orders_gap) + if not g["ok"]: + return {"level": LEVEL_DIMENSION_ERROR, "guard": g, "expr": expr, + "level_blocked": level} + + for level in LEVEL_PRIORITY: + expr = exprs[level] + if expr and eval_threshold(actual, expr): + return {"level": level, "guard": check_dimension(actual, expr, unit, orders_gap), + "expr": expr} + return {"level": "info", "guard": {"ok": True, "verdict": "ok", "reason": "未命中任何阈值", + "dimension": dimension_of(unit)}, "expr": None} + + +class _K: + """self-test 用的最小 KPI 载体。""" + + def __init__(self, unit, green=None, yellow=None, red=None): + self.unit, self.threshold_green, self.threshold_yellow, self.threshold_red = \ + unit, green, yellow, red + + +def _self_test() -> int: + """守卫自测（零库连接）：假红必拦、真红必放、真黄必放。""" + cases: Tuple[Tuple[str, float, "_K", str], ...] = ( + # (说明, 实测, KPI, 期望 level) + ("假红·流动比率 1.29倍 vs <150（跨2量级）", 1.29, _K("%", ">=200", ">=150", "<150"), + LEVEL_DIMENSION_ERROR), + ("假红·速动比率 0.54倍 vs <80（跨2量级）", 0.54, _K("%", ">=100", ">=80", "<80"), + LEVEL_DIMENSION_ERROR), + ("真红·返利率 48.44% vs >12（同量级）", 48.44, _K("%", "<=10", ">10", ">12"), "red"), + ("真红·保本覆盖率 60.4% vs <85（同量级）", 60.4, _K("%", ">=100", ">=85", "<85"), "red"), + ("真红·应收周转天数 610.2天 vs >91（同量级）", 610.2, _K("天", None, None, ">91"), "red"), + ("真红·收入增长率 -82.2% vs <15（同量级）", -82.2, _K("%", None, None, "<15"), "red"), + ("修正后·流动比率 1.29倍 vs 红<1.0/黄>=1.0/绿>=2.0", 1.29, + _K("倍", ">=2.0", ">=1.0", "<1.0"), "yellow"), + ("修正后·速动比率 0.54倍 vs 红<0.5/黄>=0.5/绿>=1.0", 0.54, + _K("倍", ">=1.0", ">=0.5", "<0.5"), "yellow"), + ("修正后·流动比率 0.8倍 → 红", 0.8, _K("倍", ">=2.0", ">=1.0", "<1.0"), "red"), + ("修正后·流动比率 2.5倍 → 绿", 2.5, _K("倍", ">=2.0", ">=1.0", "<1.0"), "green"), + ("零值不误拦·实测 0 无守卫介入", 0.0, _K("%", ">=100", ">=85", "<85"), "red"), + ("无阈值 → info", 5.0, _K("倍"), "info"), + ) + ok = 0 + print("=== 量纲守卫 self-test ===") + for desc, actual, kpi, expect in cases: + r = calc_level_guarded(actual, kpi) + flag = "PASS" if r["level"] == expect else "FAIL" + ok += (r["level"] == expect) + print(" [%s] %-46s → %-6s (期望 %s) gap=%s" % ( + flag, desc, r["level"], expect, r["guard"].get("gap"))) + print("=== %d/%d PASS ===" % (ok, len(cases))) + return 0 if ok == len(cases) else 1 + + +if __name__ == "__main__": + raise SystemExit(_self_test()) diff --git a/backend/scripts/import_cashflow_statements.py b/backend/scripts/import_cashflow_statements.py index 7a4eaf08..7f0c7340 100644 --- a/backend/scripts/import_cashflow_statements.py +++ b/backend/scripts/import_cashflow_statements.py @@ -12,19 +12,25 @@ 口径： - 经营/投资/筹资 净额 = Σ(流入) − Σ(流出)，单位 元 → 万元（÷10000，保留 2 位） - - 某活动无项目行 → 记 0.00（表列为 NOT NULL；同时写 remark 说明"无该项目=0"） + - **NA（无发生额）标注（本版本新增，裁定 finance-20260916-cma-caliber-ruling §2.1）**： + 某活动**无任何项目行**时净额记 0.00（三净额列为 NOT NULL，表本身表达不了 NA）， + 同时在 `remark` 写入**固定短语**「<活动>活动本期无发生额（无项目行）」， + 供判定层用 `parse_na_activities(remark)` 取回 → 判定时该维度**不参与**（降维）， + 避免把「无发生额」与「收付相抵的 0」混为一谈（后者才参与判定）。 - source='official'（官方报表口径）；期初/期末现金 留空（主表采集缺陷，见报告 §项1-②） - 同 (entity_id, period) 重复执行 = 覆盖更新（幂等） 用法： python3 scripts/import_cashflow_statements.py # dry-run（默认，只算不写） python3 scripts/import_cashflow_statements.py --period 2026-07 # 单期 dry-run + python3 scripts/import_cashflow_statements.py --self-test # NA 标注逻辑自测（零库连接） python3 scripts/import_cashflow_statements.py --apply # 落库（需放行） """ import argparse import csv import glob import os +import re import sys sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) @@ -36,6 +42,29 @@ from app.api.cashflow_structure import classify_enterprise ZKT_BASE = "/root/WB_WS/_data/zxt-hanke/cashflow" CLASSES = ("经营活动", "投资活动", "筹资活动") +#: NA 固定短语模板 —— 判定层与量纲守卫共用，勿改字面量（裁定 §2.1 指定） +NA_PHRASE_TMPL = "{cls}本期无发生额（无项目行）" + +#: 从 remark 反解 NA 活动的正则（与 NA_PHRASE_TMPL 严格同源） +_NA_RE = re.compile(r"(经营活动|投资活动|筹资活动)本期无发生额（无项目行）") + + +def parse_na_activities(remark): + """从 cashflow_statements.remark 取回 NA（无发生额）活动集合。 + + 判定层入口：NA 维度**不参与**八类判定（降维），其 0.00 是"表表达不了 NA"的占位， + 不是"收付相抵的 0"。同时提供 `na_flags(remark)` 返回 {活动: True/False} 供逐维判断。 + """ + if not remark: + return set() + return set(_NA_RE.findall(str(remark))) + + +def na_flags(remark): + """逐维 NA 标志：{'经营活动': bool, '投资活动': bool, '筹资活动': bool}""" + na = parse_na_activities(remark) + return {c: (c in na) for c in CLASSES} + def _read_stat_file(path): """读项目级汇总 CSV → [(分类, 编码, 项目名称, 方向, 金额元)]""" @@ -85,8 +114,11 @@ def collect(period_filter=None): continue nets_yuan, agg = _nets(rows) missing = [c for c in CLASSES if not agg[c]["in"] and not agg[c]["out"]] - remark = ("源: 智享通现金流量统计表 %s（单位元→万元 /10000）；项目行 %d 条；缺项目活动: %s" - % (os.path.basename(src), len(rows), "、".join(missing) or "无")) + na_phrases = [NA_PHRASE_TMPL.format(cls=c) for c in missing] + head = "源: 智享通现金流量统计表 %s（单位元→万元 /10000）；项目行 %d 条" % ( + os.path.basename(src), len(rows)) + remark = (head + "；NA(无发生额，不参与判定): " + "、".join(na_phrases)) if na_phrases \ + else (head + "；三类活动均有项目行（无 NA）") payloads.append({ "period": period, "status": "READY", @@ -95,19 +127,61 @@ def collect(period_filter=None): "nets_yuan": nets_yuan, "nets_wan": {k: round(v / 10000.0, 2) for k, v in nets_yuan.items()}, "missing_classes": missing, + "na_classes": missing, + "na_phrases": na_phrases, + "na_flags": {c: (c in missing) for c in CLASSES}, "remark": remark, }) return payloads +def _self_test(): + """NA 标注逻辑自测（零库连接、零写盘；只读盘上 CSV）——回执证据用。""" + passed = 0 + # 1~3) 短语 → 标志 反解 + for missing, expect in ( + (["投资活动"], {"经营活动": False, "投资活动": True, "筹资活动": False}), + (["投资活动", "筹资活动"], {"经营活动": False, "投资活动": True, "筹资活动": True}), + ([], {"经营活动": False, "投资活动": False, "筹资活动": False}), + ): + phrases = [NA_PHRASE_TMPL.format(cls=c) for c in missing] + remark = ("源: X；项目行 3 条；NA(无发生额，不参与判定): " + "、".join(phrases)) if phrases \ + else "源: X；项目行 3 条；三类活动均有项目行（无 NA）" + assert na_flags(remark) == expect, (missing, remark) + assert parse_na_activities(remark) == set(missing) + passed += 1 + # 4) 空值安全 + assert na_flags(None) == {c: False for c in CLASSES} + assert parse_na_activities("") == set() + passed += 1 + # 5) 真实数据源（盘上 8 期）：投资活动应全为 NA + pay = collect() + ready = [p for p in pay if p["status"] == "READY"] + assert ready, "无可入库期（数据源缺失）" + all_inv_na = all("投资活动" in p["na_classes"] for p in ready) + passed += 1 + print("=== NA 标注 self-test: %d/5 PASS | 期数=%d | 投资活动全NA=%s ===" + % (passed, len(ready), all_inv_na)) + for p in ready: + print(" %-8s NA=%s" % (p["period"], "、".join(p["na_classes"]) or "无")) + print(" remark: %s" % p["remark"]) + assert all_inv_na, "酣客 8 期投资活动应全为 NA（裁定 §2.1 实证）" + return 0 + + def main(): ap = argparse.ArgumentParser() ap.add_argument("--entity-id", type=int, default=1, help="账套：1=酣客(白酒, 智享通通道) 2=博海(IT, 无凭证明细)") ap.add_argument("--period", default=None, help="仅处理某期 YYYY-MM") + ap.add_argument("--self-test", action="store_true", + help="NA 标注逻辑自测（零库连接：只读盘上 CSV）") ap.add_argument("--apply", action="store_true", help="真正写库（缺省=dry-run 只算不写）") args = ap.parse_args() + if args.self_test: + return _self_test() + payloads = collect(args.period) if not payloads: print("未发现任何期数据源（%s/2026.*）" % ZKT_BASE) @@ -129,6 +203,7 @@ def main(): ("类型%d %s" % (cls["type_no"], cls["name"])) if cls.get("classified") else ("未判定: " + cls.get("reason", "")[:28]), os.path.basename(p["src"]))) + print(" NA(无发生额，不参与判定)=%s" % ("、".join(p["na_classes"]) or "无")) if not args.apply: continue rec = db.query(CashFlowStructureValue).filter( diff --git a/backend/scripts/migrate_kpi_threshold_unit_20260916.py b/backend/scripts/migrate_kpi_threshold_unit_20260916.py new file mode 100644 index 00000000..850d1449 --- /dev/null +++ b/backend/scripts/migrate_kpi_threshold_unit_20260916.py @@ -0,0 +1,216 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""阈值单位口径修正 + 量纲守卫装配（迁移脚本·准备态） + +依据：财务口径裁定 finance-20260916-cma-caliber-ruling.md v1.0 §四（裁定③）+ §3.3/§3.4 + · 4.1 单位规范：偿债能力比率=**倍**（无量纲）；资产负债率等=%；返利率=% + · 4.3 具体阈值裁定（贸易型：白酒经销 / IT 经销） + · 3.3 09-09 假红灯 586/587 按「口径错误」作废并重算 + · 3.4 切换顺序不可颠倒：先单位 → 再期间 → 后调度（本脚本只做第 1 步） + +默认 dry-run（只读：列改动 + 影响面复算，不写库）。`--apply` 才写（**需审批**）。 + +用法： + python3 scripts/migrate_kpi_threshold_unit_20260916.py # dry-run + python3 scripts/migrate_kpi_threshold_unit_20260916.py --apply # 写阈值/单位（需放行） + python3 scripts/migrate_kpi_threshold_unit_20260916.py --apply --resolve-false-reds + # 同时作废假红灯 586/587 +说明： + · 影响面口径 = 用阈值复算「最近可用期间(<= 最新已入库月)」的命中条数， + 分别给「修正前 / 修正后」两组数字（裁定 §4.5 要求：影响面以修正后复算值为准）。 + · 返利率（F_REBATE_RATE）**不改阈值**（量纲一致、判定有效），仅标注口径待确认（裁定 §4.4）。 +""" +import argparse +import sys +import os + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from sqlalchemy import text # noqa: E402 +from app.database import get_engine # noqa: E402 +from app.utils.dimension_guard import check_dimension # noqa: E402 +from app.utils.threshold_parser import eval_threshold, LEVEL_PRIORITY # noqa: E402 + +ENTITY_ID = 1 # 酣客（白酒经销）；博海（IT）无阈值 KPI 待核 + +#: 迁移计划：kpi_code → 目标 unit / 三档阈值（None = 不改该字段） +PLAN = { + "F_CURRENT_RATIO": {"name": "流动比率", "unit": "倍", + "green": ">=2.0", "yellow": ">=1.0", "red": "<1.0", + "why": "裁定 §4.3：经验值 2:1；<1 才够紧急（旧 <150 为百分数写法）"}, + "F_QUICK_RATIO": {"name": "速动比率", "unit": "倍", + "green": ">=1.0", "yellow": ">=0.5", "red": "<0.5", + "why": "裁定 §4.3：经验值 1:1（旧 <80 为百分数写法）"}, + "F_DEBT_RATIO": {"name": "资产负债率", "unit": "%", + "green": "<=60", "yellow": ">60", "red": ">70", + "why": "裁定 §4.3：红 >70 / 黄 60–70 / 绿 ≤60"}, +} +#: 明确不改、但需标注的项 +UNCHANGED_NOTES = { + "F_REBATE_RATE": ("返利率", "%", "裁定 §4.4：量纲一致、判定有效，不改阈值；" + "48.44% 口径本身待确认 → 预警保留并标注「口径待确认」"), +} + +#: 假红灯（09-09 单位错配产生）—— 裁定 §3.3：按「口径错误」作废 +FALSE_RED_ALERT_IDS = (586, 587) +VOID_RESOLUTION = "[口径错误·作废] 阈值与实测值跨量级比较（流动/速动比率单位错配），" \ + "非财务事实；由 finance-20260916-cma-caliber-ruling §3.3 裁定作废，待单位修正后重算。" + + +def q(sql, params=None): + with get_engine().connect() as c: + return [dict(r._mapping) for r in c.execute(text(sql), params or {}).fetchall()] + + +def exec_sql(sql, params=None): + with get_engine().begin() as c: + return c.execute(text(sql), params or {}) + + +def fetch_plan_rows(): + codes = list(PLAN.keys()) + list(UNCHANGED_NOTES.keys()) + ph = ",".join(":%s" % ("c%d" % i) for i in range(len(codes))) + return q("SELECT id, kpi_code, kpi_name, unit, threshold_green, threshold_yellow, threshold_red, status " + "FROM kpi_definitions WHERE entity_id=:e AND kpi_code IN (%s) ORDER BY id" % ph, + dict({"e": ENTITY_ID}, **{("c%d" % i): c for i, c in enumerate(codes)})) + + +def latest_period(): + r = q("SELECT MAX(period) AS p FROM kpi_values WHERE period REGEXP '^[0-9]{4}-[0-9]{2}$'") + return r[0]["p"] if r else None + + +def impact(): + """影响面复算：最近可用期间上，各 KPI 在「修正前 / 修正后」的命中条数。""" + p = latest_period() + rows = q( + "SELECT d.id, d.kpi_code, d.kpi_name, d.unit, d.threshold_green AS g, " + "d.threshold_yellow AS y, d.threshold_red AS r, v.actual_value AS actual, v.period " + "FROM kpi_definitions d JOIN kpi_values v ON v.kpi_id = d.id " + "WHERE d.entity_id=:e AND d.status='active' AND v.period = :p AND v.actual_value IS NOT NULL " + "AND (d.threshold_red IS NOT NULL OR d.threshold_yellow IS NOT NULL OR d.threshold_green IS NOT NULL) " + "ORDER BY d.id", {"e": ENTITY_ID, "p": p}) + + def judge(actual, unit, g, y, r): + exprs = {"green": g, "yellow": y, "red": r} + guard_hit = None + for lv in LEVEL_PRIORITY: + if exprs[lv]: + gd = check_dimension(actual, exprs[lv], unit) + if not gd["ok"]: + guard_hit = gd + break + if guard_hit: + return "口径错误" + for lv in LEVEL_PRIORITY: + if exprs[lv] and eval_threshold(actual, exprs[lv]): + return lv + return "info" + + before, after, detail = {"red": 0, "yellow": 0, "green": 0, "info": 0, "口径错误": 0}, \ + {"red": 0, "yellow": 0, "green": 0, "info": 0, "口径错误": 0}, [] + for x in rows: + plan = PLAN.get(x["kpi_code"]) + g2 = plan["green"] if plan else x["g"] + y2 = plan["yellow"] if plan else x["y"] + r2 = plan["red"] if plan else x["r"] + u2 = plan["unit"] if plan else x["unit"] + b = judge(x["actual"], x["unit"], x["g"], x["y"], x["r"]) + a = judge(x["actual"], u2, g2, y2, r2) + before[b] = before.get(b, 0) + 1 + after[a] = after.get(a, 0) + 1 + if b != a: + detail.append({"kpi": "%s/%s" % (x["kpi_code"], x["kpi_name"]), "period": x["period"], + "actual": x["actual"], "unit": x["unit"], + "before": b, "after": a, + "thresholds": "%s|%s|%s → %s|%s|%s" % (x["g"], x["y"], x["r"], g2, y2, r2)}) + return p, before, after, detail + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--apply", action="store_true", help="真正写库（缺省=dry-run）") + ap.add_argument("--resolve-false-reds", action="store_true", + help="同时把假红灯 %s 按口径错误作废（status=resolved）" % (FALSE_RED_ALERT_IDS,)) + args = ap.parse_args() + + print("=== 阈值单位口径修正 %s | entity_id=%d ===" % ("APPLY" if args.apply else "DRY-RUN", ENTITY_ID)) + rows = fetch_plan_rows() + if not rows: + print("!! 未取到目标 KPI 行（kpi_code 不匹配？）") + return 2 + + print("\n-- 计划改动 --") + todo = [] + for x in rows: + plan = PLAN.get(x["kpi_code"]) + if not plan: + note = UNCHANGED_NOTES.get(x["kpi_code"]) + print(" [不改] id=%-4s %-6s %-8s unit=%s 阈值=%s|%s|%s ← %s" + % (x["id"], x["kpi_code"], x["kpi_name"], x["unit"], + x["threshold_green"], x["threshold_yellow"], x["threshold_red"], + note[2] if note else "")) + continue + changed = (x["unit"] != plan["unit"] or x["threshold_green"] != plan["green"] + or x["threshold_yellow"] != plan["yellow"] or x["threshold_red"] != plan["red"]) + print(" [%s] id=%-4s %-6s %-8s unit %s→%s 阈值 %s|%s|%s → %s|%s|%s" + % ("改" if changed else "已符", x["id"], x["kpi_code"], x["kpi_name"], + x["unit"], plan["unit"], x["threshold_green"], x["threshold_yellow"], x["threshold_red"], + plan["green"], plan["yellow"], plan["red"])) + print(" 理由: %s" % plan["why"]) + if changed: + todo.append((x["id"], plan)) + + print("\n-- 影响面复算（最近可用期间） --") + p, before, after, detail = impact() + print(" 期间 P* = %s" % p) + print(" 修正前: 红 %d / 黄 %d / 绿 %d / 无判定 %d / 口径错误 %d" + % (before["red"], before["yellow"], before["green"], before["info"], before["口径错误"])) + print(" 修正后: 红 %d / 黄 %d / 绿 %d / 无判定 %d / 口径错误 %d" + % (after["red"], after["yellow"], after["green"], after["info"], after["口径错误"])) + print(" 条数变化（仅列变化项）:") + for d in detail: + print(" %-22s %s actual=%s unit=%s %s → %s [%s]" + % (d["kpi"], d["period"], d["actual"], d["unit"], d["before"], d["after"], d["thresholds"])) + print(" ※ 影响面以「修正后」为准（裁定 §4.5）；此处红/黄为静态阈值复算上界，" + "实际落库条数还受期间口径与幂等约束。") + + if args.resolve_false_reds: + print("\n-- 假红灯作废 --") + ph = ",".join(":a%d" % i for i in range(len(FALSE_RED_ALERT_IDS))) + al = q("SELECT id, kpi_id, alert_level, status, alert_message FROM kpi_alerts " + "WHERE id IN (%s)" % ph, + {("a%d" % i): v for i, v in enumerate(FALSE_RED_ALERT_IDS)}) + for a in al: + print(" id=%s level=%s status=%s msg=%s" % (a["id"], a["alert_level"], a["status"], + (a["alert_message"] or "")[:70])) + if not al: + print(" （未取到该两条预警，可能已清理）") + + if not args.apply: + print("\n=== dry-run 结束：未写库（%d 项待改，加 --apply 落库 / 需审批）===" % len(todo)) + return 0 + + for kid, plan in todo: + exec_sql("UPDATE kpi_definitions SET unit=:u, threshold_green=:g, threshold_yellow=:y, " + "threshold_red=:r WHERE id=:i AND entity_id=:e", + {"u": plan["unit"], "g": plan["green"], "y": plan["yellow"], + "r": plan["red"], "i": kid, "e": ENTITY_ID}) + if args.resolve_false_reds: + exec_sql("UPDATE kpi_alerts SET status='resolved', resolution=:res, resolved_at=NOW() " + "WHERE id IN (%s)" % ",".join(":a%d" % i for i in range(len(FALSE_RED_ALERT_IDS))), + dict({("a%d" % i): v for i, v in enumerate(FALSE_RED_ALERT_IDS)}, + res=VOID_RESOLUTION)) + print("\n=== 已提交：%d 项阈值/单位%s ===" % (len(todo), + "；已作废假红灯 %s" % (FALSE_RED_ALERT_IDS,) if args.resolve_false_reds else "")) + print("-- 写入后回查 --") + for x in fetch_plan_rows(): + if x["kpi_code"] in PLAN: + print(" id=%-4s %-6s unit=%s 阈值=%s|%s|%s" + % (x["id"], x["kpi_code"], x["unit"], x["threshold_green"], + x["threshold_yellow"], x["threshold_red"])) + return 0 + + +if __name__ == "__main__": + sys.exit(main())