Hermes CI Fix
3bb55189b1
ci/woodpecker/push/main Pipeline failed
依据：wecom-finance/reports/CMA_口径裁定_20260917.md（任总 2026-09-17 23:14「按建议执行」） 裁定1（现金流「投资活动净额=0」参与判定） - classify_enterprise()：0 为合法输入；投资恒为 0 → 经营(±)×筹资(±) 二维映射出 2 个 候选类型（candidate_type_nos/candidates），不再整体判「未判定」；投资栏 investing_note 标注「本期无投资活动」；经营/筹资亦为 0 → 回退「未判定（无投资活动）」，不报错不静默 - /analysis 增 classification_dimension / investing_note / nets.investing.note；前端出 二维横幅 + 候选行高亮 + 投资栏「本期无投资活动」（不给单一类型名） - import_cashflow_statements.py：打印层修 2D（type_no=None）TypeError 崩溃；NA 措辞改为 「按 0 参与判定」（机读短语 NA_PHRASE_TMPL/parse_na_activities 未动） 裁定2a（比率类单位统一「百分比(%)」口径，先做） - 新增 app/utils/caliber_units.py：逐 KPI 对照表 RATIO_CALIBER_TABLE（唯一事实源）+ 判定侧 唯一入口 resolve_caliber()；3 项需归一 ×100（流动/速动/毛利率），12 项取数已百分数不归一； 表外且命中「疑似错配」窄形态（阈值子句量级≥100 且 0<|实测|<10 且单位属比率类）→ dimension_suspect：不落预警、不推送（唯一规则，防 586/587 假红灯复发） - alert_generator：判定前归一，文案写「口径归一 ×100（原值 1.29 → 129%）」留审计线索 - 新增 scripts/kpi_unit_crosscheck.py（只读对照表产出器） 裁定2b（期间由「当月硬编码」→「最近可用期间」，后做） - resolve_latest_period()：取该 KPI 自己的最新 YYYY-MM（禁全局 MAX，防演示账套劫持）； 伪期间（2026-H1/H2、2025-FY）在 Python 侧过滤（保「单测=生产」语义） - 过渡期覆盖度自检扩展：记录回落后的期间分布 + 当月 0 条归因提示 - --dry-run（零写入零推送）作放行前抽查证据；generate_and_push/run_alert_check 支持 dry_run 测试：backend/tests/test_caliber_rulings_20260917.py 17 项；全量 1158 passed/1 xfailed
100 lines
3.9 KiB
Python
100 lines
3.9 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""比率类 KPI 单位对照表（只读）—— 财务口径裁定 裁定2a 的产出器
|
||
|
||
用途：逐个比率类 KPI 核对「定义 / 声明单位 / 阈值量纲 / 实测值量纲」，出对照表。
|
||
判定侧的事实源是 `app/utils/caliber_units.RATIO_CALIBER_TABLE`（本脚本只核对与呈现，
|
||
不写库、不改阈值）。
|
||
|
||
用法：
|
||
python3 scripts/kpi_unit_crosscheck.py # 全账套
|
||
python3 scripts/kpi_unit_crosscheck.py --entity-id 1 # 单账套
|
||
python3 scripts/kpi_unit_crosscheck.py --json # JSON（机读）
|
||
"""
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
|
||
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.caliber_units import resolve_caliber # noqa: E402
|
||
|
||
PERIOD_RE = re.compile(r"^[0-9]{4}-[0-9]{2}$")
|
||
|
||
|
||
def q(sql, **params):
|
||
with get_engine().connect() as c:
|
||
return [dict(r._mapping) for r in c.execute(text(sql), params).fetchall()]
|
||
|
||
|
||
class _K(dict):
|
||
"""把 kpi 行包成 duck type（resolve_caliber 只读属性）"""
|
||
|
||
def __getattr__(self, item):
|
||
return self.get(item)
|
||
|
||
|
||
def latest_value(kpi_id):
|
||
rows = q("SELECT period, actual_value FROM kpi_values WHERE kpi_id=:k AND actual_value IS NOT NULL "
|
||
"ORDER BY period DESC, calculated_at DESC LIMIT 20", k=kpi_id)
|
||
for r in rows:
|
||
if PERIOD_RE.fullmatch(str(r["period"] or "")):
|
||
return r["period"], r["actual_value"]
|
||
return None, None
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--entity-id", type=int, default=None)
|
||
ap.add_argument("--json", action="store_true")
|
||
args = ap.parse_args()
|
||
|
||
where = "WHERE d.status='active' AND (d.threshold_red IS NOT NULL OR d.threshold_yellow IS NOT NULL " \
|
||
"OR d.threshold_green IS NOT NULL)"
|
||
params = {}
|
||
if args.entity_id:
|
||
where += " AND d.entity_id=:e"
|
||
params["e"] = args.entity_id
|
||
rows = q("SELECT d.id, d.entity_id, d.kpi_code, d.kpi_name, d.unit, d.threshold_green, "
|
||
"d.threshold_yellow, d.threshold_red FROM kpi_definitions d %s "
|
||
"ORDER BY d.entity_id, d.id" % where, **params)
|
||
|
||
out = []
|
||
for r in rows:
|
||
p, v = latest_value(r["id"])
|
||
cal = resolve_caliber(v, _K(r))
|
||
out.append({
|
||
"kpi_code": r["kpi_code"], "kpi_name": r["kpi_name"], "entity_id": r["entity_id"],
|
||
"declared_unit": r["unit"],
|
||
"thresholds": "%s|%s|%s" % (r["threshold_green"], r["threshold_yellow"], r["threshold_red"]),
|
||
"latest_period": p, "actual": v,
|
||
"caliber_status": cal["status"], "norm_factor": cal["factor"],
|
||
"normalized": cal["value"], "note": cal["note"], "reason": cal["reason"],
|
||
})
|
||
|
||
if args.json:
|
||
print(json.dumps(out, ensure_ascii=False, indent=1, default=str))
|
||
return 0
|
||
|
||
print("=== 比率类 KPI 单位对照表（只读；裁定2a）===")
|
||
print("%-4s %-8s %-22s %-6s %-10s %-12s %-8s %-6s %s" % (
|
||
"eid", "code", "name", "unit", "latest", "actual", "status", "factor", "thresholds"))
|
||
for x in out:
|
||
print("%-4s %-8s %-22s %-6s %-10s %-12s %-18s %-6s %s" % (
|
||
x["entity_id"], x["kpi_code"], x["kpi_name"][:20], x["declared_unit"] or "-",
|
||
x["latest_period"] or "-", x["actual"] if x["actual"] is not None else "-",
|
||
x["caliber_status"], x["norm_factor"], x["thresholds"]))
|
||
n_norm = sum(1 for x in out if x["norm_factor"] != 1)
|
||
n_sus = sum(1 for x in out if x["caliber_status"] == "dimension_suspect")
|
||
print("=== 合计 %d 项：口径归一 ×100 %d 项 / 同量纲直判 %d 项 / 疑似错配降级待核 %d 项 ===" % (
|
||
len(out), n_norm, len(out) - n_norm - n_sus, n_sus))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|