Hermes CI Fix
1613f7f961
ci/woodpecker/push/main Pipeline failed
Customer revenue/concentration now filters bill_type IN (1,4,14) (XSD/XSTHD/XSHHD). Non-sales docs (JHD purchase, YFKD prepay, YSKD accept, QTRK, SKD, YBFYD) carry SUPPLIER unit_id and were polluting 'customer concentration' (old Top1/Top2 were our biggest suppliers; root cause of 6.04x base gap). Unify sign filter to (sum_money+sum_tax_money)>0 (equivalent to money_only on current data, diff 0 rows) and correct the withdrawn 407-row/17.2% wording in cpa_caliber docstring. Deriver: anchors from cpa_caliber single source; new --concentration overwrite and --only concentration (minimal 9-row write face) with pre-overwrite backup. Tests: fixture gains bill_type/sum_tax_money; +5 whitelist cases (44 passed targeted).
434 lines
22 KiB
Python
434 lines
22 KiB
Python
#!/usr/bin/env python3
|
||
"""CPA 客户获利性派生器 —— 客户级收入快照 + 汇总口径 KPI 值（任务 finance-20261003-cpa-... P0-a）
|
||
|
||
产物
|
||
----
|
||
1) 明细表 `customer_profitability_snapshot`（每客户每期一行，幂等 upsert；见
|
||
`scripts/migrate_cpa_snapshot_20261004.py`）
|
||
2) `kpi_values` 汇总口径（**只存汇总，避免 284 客户 × 期数污染 KPI 宽表**，任务 §2.1）：
|
||
* `C_CUST_REVENUE` 期内客户收入合计（万元）
|
||
* `C_CUST_COUNT` 期内活跃客户数（个）
|
||
* `C_CUST_CONCENTRATION` 现网定义 = 前5大客户收入 ÷ 期内合计 ×100。
|
||
`--concentration`（默认 `fill-gaps`）控制**已有期次**的处置：
|
||
* `fill-gaps` = 仅补库内尚无该期行的期次（2026-01~09 九期原样保留）
|
||
* `overwrite` = **重算并覆盖**指定期次（P0 口径纠错后重算九期用；
|
||
属生产写，须走放行）
|
||
|
||
口径（2026-10-04 P0 纠错后）
|
||
单据类型白名单 = **销售类 `bill_type IN (1,4,14)`**（XSD 销售单 / XSTHD 销售退货 /
|
||
XSHHD 销售换货）；非销售单据（进货/验收/入库/预付款/收款/费用）的 unit_id 是**供应商**，
|
||
不得计入客户收入。取数口径单一事实源 = `app/utils/cpa_caliber.py`（与
|
||
`alert_rules._customer_revenue_shares` 同一实现）。
|
||
2026-01~09 对拍锚点：65 户 / 1,207.74 万 / Top1 22.5150%(u152) / Top2 13.9734%(u155) /
|
||
Top4 46.9407% / Top10 69.7511%（`cpa_caliber.ANCHORS_2026_01_09`）。
|
||
|
||
fail-closed（本器**不写**以下列/值；理由见 `app/utils/cpa_caliber.py::FAIL_CLOSED_REASONS`）
|
||
`cogs_alloc` / `chan_sub` / `contrib` / `contrib_rate` / `C_CUST_COGS_ALLOC` /
|
||
`C_CUST_CHAN_SUB` / `C_CUST_CONTRIB` / `C_CUST_CONTRIB_RATE` / 客户平均贡献值
|
||
⇒ 一律不落库（禁 0、禁分摊值），只在快照行 `fail_closed_reason` 留痕。
|
||
|
||
口径单一事实源
|
||
客户收入聚合 = `app/utils/cpa_caliber.py`（与 `alert_rules._customer_revenue_shares` 同一实现）。
|
||
|
||
用法
|
||
----
|
||
python3 scripts/kpi_customer_profitability_deriver.py --check # 只读预检 + 对拍锚点（默认）
|
||
python3 scripts/kpi_customer_profitability_deriver.py --apply # 单事务幂等写
|
||
python3 scripts/kpi_customer_profitability_deriver.py --verify # 只读写后回查
|
||
python3 scripts/kpi_customer_profitability_deriver.py --anchor # 只跑对拍锚点断言
|
||
|
||
选项
|
||
--entity N 账套（默认 1；本单主体范围 = 陕西酣客 e1）
|
||
--db-url URL 显式连接串（离线演练 `sqlite:///<scratch>.db`）
|
||
--since/--until 仅处理该区间（YYYY-MM）
|
||
--batch NAME source_batch 名（默认 cpa-20261004）
|
||
|
||
退出码
|
||
0 成功 ｜ 1 fail-closed 前置不满足 / 校验失败 ｜ 2 用法错误
|
||
"""
|
||
import argparse
|
||
import datetime as _dt
|
||
import json
|
||
import os
|
||
import sys
|
||
from urllib.parse import quote_plus
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
BACKEND = os.path.dirname(HERE)
|
||
if BACKEND not in sys.path:
|
||
sys.path.insert(0, BACKEND)
|
||
|
||
DEFAULT_BATCH = "cpa-20261004"
|
||
SNAPSHOT_TABLE = "customer_profitability_snapshot"
|
||
FAIL_CLOSED_NOTE = (
|
||
"cogs_alloc/chan_sub/contrib/contrib_rate = P1 fail-closed（未裁定口径/未打标），禁静默 0 或分摊值"
|
||
)
|
||
|
||
# 对拍锚点**单一事实源** = `cpa_caliber.ANCHORS_2026_01_09`（销售类白名单口径，2026-10-04 P0 纠错后）
|
||
def _anchors():
|
||
cal = _load_caliber()
|
||
return cal.ANCHORS_2026_01_09, cal.ANCHOR_TOL
|
||
|
||
|
||
def _engine(args):
|
||
from sqlalchemy import create_engine
|
||
if args.db_url:
|
||
url = args.db_url
|
||
else:
|
||
try:
|
||
from dotenv import load_dotenv
|
||
load_dotenv(os.path.join(BACKEND, ".env"), override=False)
|
||
except Exception:
|
||
pass
|
||
user = os.environ.get("CMA_DB_USER", "cma_api")
|
||
pwd = os.environ.get("CMA_DB_PASS", "")
|
||
host = os.environ.get("CMA_DB_HOST", "127.0.0.1")
|
||
port = os.environ.get("CMA_DB_PORT", "3306")
|
||
name = os.environ.get("CMA_DB_NAME", "cma")
|
||
url = f"mysql+pymysql://{user}:{quote_plus(pwd)}@{host}:{port}/{name}?charset=utf8mb4"
|
||
kw = {"connect_args": {"check_same_thread": False}} if url.startswith("sqlite") else {}
|
||
return create_engine(url, **kw), url
|
||
|
||
|
||
def _dialect(url):
|
||
return "sqlite" if url.startswith("sqlite") else "mysql"
|
||
|
||
|
||
def _table_exists(conn, dialect):
|
||
from sqlalchemy import text
|
||
if dialect == "sqlite":
|
||
return bool(conn.execute(text("SELECT name FROM sqlite_master WHERE type='table' AND name=:t"),
|
||
{"t": SNAPSHOT_TABLE}).fetchall())
|
||
return bool(conn.execute(text("SELECT TABLE_NAME FROM information_schema.TABLES "
|
||
"WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=:t"),
|
||
{"t": SNAPSHOT_TABLE}).fetchall())
|
||
|
||
|
||
def _grants_ok(conn, dialect, table):
|
||
"""写前置：应用账号须对 kpi_values 与目标表有 DML。"""
|
||
from sqlalchemy import text
|
||
if dialect == "sqlite":
|
||
return True, "sqlite(离线演练，无授权面)"
|
||
rows = conn.execute(text("SHOW GRANTS FOR CURRENT_USER()")).fetchall()
|
||
joined = "\n".join(r[0] for r in rows)
|
||
has_kv = ("kpi_values" in joined) or ("ON `cma`.*" in joined) or ("ON *.*" in joined)
|
||
has_new = (f"`{table}`" in joined) or ("ON `cma`.*" in joined) or ("ON *.*" in joined)
|
||
return (has_kv and has_new), joined
|
||
|
||
|
||
def _load_caliber():
|
||
from app.utils import cpa_caliber
|
||
return cpa_caliber
|
||
|
||
|
||
def _dump_prior_concentration(engine, entity_id, kpi_id, ts):
|
||
"""**覆盖前备份**现网 `C_CUST_CONCENTRATION` 全部行（回退点），落 `scripts/backups/`。
|
||
|
||
返回备份文件绝对路径（写进结果 JSON，供审批串/回退引用）。
|
||
"""
|
||
import json as _json
|
||
from sqlalchemy import text
|
||
cols = ("id", "period", "actual_value", "data_status", "source_type", "source_batch",
|
||
"remark", "entity_id")
|
||
with engine.connect() as conn:
|
||
rows = conn.execute(text(
|
||
"SELECT id, period, actual_value, data_status, source_type, source_batch, remark, entity_id "
|
||
"FROM kpi_values WHERE entity_id=:e AND kpi_id=:k ORDER BY period, id"),
|
||
{"e": entity_id, "k": kpi_id}).fetchall()
|
||
d = os.path.join(HERE, "backups")
|
||
os.makedirs(d, exist_ok=True)
|
||
path = os.path.join(d, "cpa-concentration-prior-%s.json" % ts.replace(":", "").replace(" ", "-"))
|
||
with open(path, "w", encoding="utf-8") as f:
|
||
_json.dump({"ts": ts, "entity_id": entity_id, "kpi_id": kpi_id,
|
||
"rows": [dict(zip(cols, r)) for r in rows]}, f, ensure_ascii=False,
|
||
indent=2, default=str)
|
||
return path
|
||
|
||
|
||
def compute(engine, dialect, entity_id):
|
||
"""只读计算：{period: {unit_id: amount}} + 汇总。"""
|
||
cal = _load_caliber()
|
||
monthly = cal.fetch_monthly_revenue(engine, entity_id)
|
||
summary = {}
|
||
for p, pairs in monthly.items():
|
||
m = cal.monthly_metrics(pairs)
|
||
m["_total_yuan"] = sum(a for _, a in pairs)
|
||
signature = _rank_signature(pairs) if len(pairs) > 10 else None
|
||
m["_signature"] = signature
|
||
summary[p] = m
|
||
return monthly, summary
|
||
|
||
|
||
def _rank_signature(pairs):
|
||
"""前10名 (unit_id, 分) 指纹（防静默漂移）。"""
|
||
return [(u, round(a, 2)) for u, a in pairs[:10]]
|
||
|
||
|
||
def anchor_check(summary):
|
||
"""2026-01~09 对拍锚点（与任务验收标准逐项比）。"""
|
||
pairs = []
|
||
for p in sorted(summary):
|
||
if "2026-01" <= p <= "2026-09":
|
||
pairs.append(p)
|
||
# 需要按窗口重算 topN（不能逐月相加），故单独取窗口
|
||
return pairs
|
||
|
||
|
||
def run(engine, dialect, entity_id, batch, apply=False, verify=False, since=None, until=None,
|
||
concentration="fill-gaps", only="all"):
|
||
cal = _load_caliber()
|
||
from sqlalchemy import text
|
||
result = {"entity_id": entity_id, "batch": batch, "apply": apply, "verify": verify,
|
||
"concentration_mode": concentration, "only": only}
|
||
if concentration not in ("fill-gaps", "overwrite"):
|
||
raise SystemExit("FAIL-CLOSED: --concentration 仅支持 fill-gaps / overwrite")
|
||
if only not in ("all", "concentration"):
|
||
raise SystemExit("FAIL-CLOSED: --only 仅支持 all / concentration")
|
||
if apply and only == "concentration" and concentration != "overwrite":
|
||
raise SystemExit("FAIL-CLOSED: --only concentration 必须配 --concentration overwrite（否则九期被跳过=空写）")
|
||
|
||
with engine.connect() as conn:
|
||
exists = _table_exists(conn, dialect)
|
||
if apply and only != "concentration" and not exists:
|
||
raise SystemExit("FAIL-CLOSED: 目标表不存在，请先跑 migrate_cpa_snapshot_20261004.py --apply")
|
||
|
||
rows_before = None
|
||
if exists:
|
||
rows_before = conn.execute(text(f"SELECT COUNT(*) FROM {SNAPSHOT_TABLE} WHERE entity_id=:e"),
|
||
{"e": entity_id}).scalar()
|
||
kpi_ids = {}
|
||
for code in ("C_CUST_REVENUE", "C_CUST_COUNT", "C_CUST_CONCENTRATION"):
|
||
r = conn.execute(text("SELECT id FROM kpi_definitions WHERE entity_id=:e AND kpi_code=:c"),
|
||
{"e": entity_id, "c": code}).fetchone()
|
||
kpi_ids[code] = r[0] if r else None
|
||
existing_conc_periods = set()
|
||
if kpi_ids["C_CUST_CONCENTRATION"]:
|
||
for (p,) in conn.execute(text("SELECT period FROM kpi_values WHERE entity_id=:e AND kpi_id=:k"),
|
||
{"e": entity_id, "k": kpi_ids["C_CUST_CONCENTRATION"]}).fetchall():
|
||
existing_conc_periods.add(p)
|
||
|
||
monthly = cal.fetch_monthly_revenue(engine, entity_id)
|
||
periods = sorted(p for p in monthly if (not since or p >= since) and (not until or p <= until))
|
||
result["periods"] = len(periods)
|
||
result["period_range"] = [periods[0], periods[-1]] if periods else None
|
||
result["kpi_ids"] = kpi_ids
|
||
result["existing_concentration_periods"] = sorted(existing_conc_periods)
|
||
|
||
if verify:
|
||
return _verify(engine, dialect, entity_id, monthly, periods, kpi_ids, existing_conc_periods, now_ts=None)
|
||
|
||
snap_rows = []
|
||
kpi_rows = []
|
||
conc_written_periods = []
|
||
for p in periods:
|
||
pairs = monthly[p]
|
||
total_yuan = sum(a for _, a in pairs)
|
||
if only != "concentration":
|
||
for idx, (uid, amount) in enumerate(pairs, start=1):
|
||
snap_rows.append({
|
||
"entity_id": entity_id, "period": p, "unit_id": uid,
|
||
"revenue": cal.to_wan(amount), "rank_no": idx,
|
||
"calib_status": "revenue_only", "fail_closed_reason": FAIL_CLOSED_NOTE,
|
||
})
|
||
kpi_rows.append(("C_CUST_REVENUE", p, cal.to_wan(total_yuan)))
|
||
kpi_rows.append(("C_CUST_COUNT", p, float(cal.active_customer_count(pairs))))
|
||
if concentration == "overwrite" or p not in existing_conc_periods:
|
||
kpi_rows.append(("C_CUST_CONCENTRATION", p, cal.concentration_pct(pairs, 5)))
|
||
if p in existing_conc_periods:
|
||
conc_written_periods.append(p)
|
||
result["only"] = only
|
||
result["snapshot_rows"] = len(snap_rows)
|
||
result["kpi_rows"] = len(kpi_rows)
|
||
result["concentration_overwritten_periods"] = sorted(conc_written_periods)
|
||
result["kpi_rows_skipped_no_definition"] = [c for c, i in kpi_ids.items() if i is None]
|
||
|
||
if not apply:
|
||
result["preview"] = {
|
||
"first_periods": {p: {k: v for k, v in cal.monthly_metrics(monthly[p]).items()} for p in periods[:2]},
|
||
"last_periods": {p: {k: v for k, v in cal.monthly_metrics(monthly[p]).items()} for p in periods[-2:]},
|
||
}
|
||
return result
|
||
|
||
# ---- 写路径（单事务） ----
|
||
with engine.connect() as conn:
|
||
ok, grants = _grants_ok(conn, dialect, SNAPSHOT_TABLE)
|
||
result["grants_ok"] = ok
|
||
if not ok:
|
||
raise SystemExit("FAIL-CLOSED: 应用账号缺 DML 授权（kpi_values / 目标表）。原始 SHOW GRANTS:\n" + grants)
|
||
need = (["C_CUST_CONCENTRATION"] if only == "concentration"
|
||
else ["C_CUST_REVENUE", "C_CUST_COUNT", "C_CUST_CONCENTRATION"])
|
||
missing = [c for c in need if kpi_ids.get(c) is None]
|
||
if missing:
|
||
raise SystemExit("FAIL-CLOSED: KPI 定义缺失 %s —— 请先执行 scripts/sql/cpa_kpi_defs_20261004.sql" % missing)
|
||
|
||
now_ts = _dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
if concentration == "overwrite" and kpi_ids.get("C_CUST_CONCENTRATION"):
|
||
result["prior_backup"] = _dump_prior_concentration(engine, entity_id,
|
||
kpi_ids["C_CUST_CONCENTRATION"], now_ts)
|
||
|
||
inserted_snap = updated_snap = 0
|
||
kpi_inserted = kpi_updated = 0
|
||
with engine.begin() as conn:
|
||
for r in snap_rows:
|
||
row = conn.execute(text(
|
||
f"SELECT id FROM {SNAPSHOT_TABLE} WHERE entity_id=:entity_id AND period=:period AND unit_id=:unit_id"),
|
||
r).fetchone()
|
||
if row:
|
||
conn.execute(text(
|
||
f"UPDATE {SNAPSHOT_TABLE} SET revenue=:revenue, rank_no=:rank_no, "
|
||
"calib_status=:calib_status, fail_closed_reason=:fail_closed_reason, updated_at=:ts "
|
||
"WHERE id=:id"), {**r, "ts": now_ts, "id": row[0]})
|
||
updated_snap += 1
|
||
else:
|
||
conn.execute(text(
|
||
f"INSERT INTO {SNAPSHOT_TABLE} (entity_id, period, unit_id, revenue, rank_no, calib_status, "
|
||
"fail_closed_reason, created_at, updated_at) VALUES (:entity_id, :period, :unit_id, :revenue, "
|
||
":rank_no, :calib_status, :fail_closed_reason, :ts, :ts)"), {**r, "ts": now_ts})
|
||
inserted_snap += 1
|
||
for code, period, value in kpi_rows:
|
||
kid = kpi_ids[code]
|
||
if value is None:
|
||
continue # fail-closed：不给值（不写 NULL 新行，避免污染宽表）
|
||
row = conn.execute(text(
|
||
"SELECT id, actual_value, data_status FROM kpi_values "
|
||
"WHERE entity_id=:e AND kpi_id=:k AND period=:p ORDER BY id DESC LIMIT 1"),
|
||
{"e": entity_id, "k": kid, "p": period}).fetchone()
|
||
conc_note = ("重算覆盖期次（P0 销售类白名单口径）"
|
||
if concentration == "overwrite" and period in existing_conc_periods
|
||
else "补值期次（库内原无该期行）")
|
||
remark = {"C_CUST_REVENUE": "口径=Σzxt_masterbill.sum_money（销售类 bill_type IN (1,4,14)、"
|
||
"期内、unit_id>0、符号过滤(sum_money+sum_tax_money)>0）",
|
||
"C_CUST_COUNT": "口径=期内 zxt_masterbill 销售类单据活跃客户数（unit_id 去重）",
|
||
"C_CUST_CONCENTRATION": f"口径=前5大客户收入/期内客户收入合计×100%（{conc_note}）"}[code]
|
||
|
||
if row:
|
||
# 幂等：值不变则零写入；不降级 data_status
|
||
if abs(float(row[1] or 0) - float(value)) < 1e-9 and row[2]:
|
||
continue
|
||
conn.execute(text("UPDATE kpi_values SET actual_value=:v, source_type='derived', source_batch=:b, "
|
||
"remark=:r WHERE id=:id"),
|
||
{"v": value, "b": batch, "r": remark, "id": row[0]})
|
||
kpi_updated += 1
|
||
else:
|
||
conn.execute(text(
|
||
"INSERT INTO kpi_values (entity_id, kpi_id, period, actual_value, source_type, source_batch, "
|
||
"data_status, calculated_at, remark) VALUES (:e, :k, :p, :v, 'derived', :b, 'pending', :ts, :r)"),
|
||
{"e": entity_id, "k": kid, "p": period, "v": value, "b": batch, "ts": now_ts, "r": remark})
|
||
kpi_inserted += 1
|
||
result.update(snapshot_inserted=inserted_snap, snapshot_updated=updated_snap,
|
||
kpi_inserted=kpi_inserted, kpi_updated=kpi_updated, ts=now_ts)
|
||
return result
|
||
|
||
|
||
def _verify(engine, dialect, entity_id, monthly, periods, kpi_ids, existing_conc_periods, now_ts=None):
|
||
"""只读写后回查：快照行数/值锚点 + kpi_values 三族。"""
|
||
from sqlalchemy import text
|
||
cal = _load_caliber()
|
||
out = {"verify": True, "entity_id": entity_id}
|
||
with engine.connect() as conn:
|
||
out["snapshot_rows"] = conn.execute(text(
|
||
f"SELECT COUNT(*) FROM {SNAPSHOT_TABLE} WHERE entity_id=:e"), {"e": entity_id}).scalar()
|
||
out["snapshot_periods"] = conn.execute(text(
|
||
f"SELECT COUNT(DISTINCT period) FROM {SNAPSHOT_TABLE} WHERE entity_id=:e"), {"e": entity_id}).scalar()
|
||
# 值锚点：2026-01~09 快照合计 == 4,883.52 万
|
||
r = conn.execute(text(
|
||
f"SELECT ROUND(SUM(revenue),2), COUNT(DISTINCT unit_id) FROM {SNAPSHOT_TABLE} "
|
||
"WHERE entity_id=:e AND period>='2026-01' AND period<='2026-09'"), {"e": entity_id}).fetchone()
|
||
out["anchor_2026_01_09_snapshot"] = {"total_wan": float(r[0] or 0), "units": int(r[1] or 0)}
|
||
out["fail_closed_columns_all_null"] = conn.execute(text(
|
||
f"SELECT COUNT(*) FROM {SNAPSHOT_TABLE} WHERE entity_id=:e AND "
|
||
"(cogs_alloc IS NOT NULL OR chan_sub IS NOT NULL OR contrib IS NOT NULL OR contrib_rate IS NOT NULL)"),
|
||
{"e": entity_id}).scalar()
|
||
# kpi_values 回查
|
||
kv = {}
|
||
for code in ("C_CUST_REVENUE", "C_CUST_COUNT", "C_CUST_CONCENTRATION"):
|
||
kid = kpi_ids.get(code)
|
||
if not kid:
|
||
kv[code] = None
|
||
continue
|
||
rows = conn.execute(text("SELECT period, actual_value, data_status FROM kpi_values "
|
||
"WHERE entity_id=:e AND kpi_id=:k ORDER BY period"),
|
||
{"e": entity_id, "k": kid}).fetchall()
|
||
kv[code] = {"n": len(rows), "sample": [[r[0], float(r[1]) if r[1] is not None else None] for r in rows[:3]],
|
||
"last": [[r[0], float(r[1]) if r[1] is not None else None] for r in rows[-2:]]}
|
||
out["kpi_values"] = kv
|
||
# 期次覆盖：快照期次 == 有数据期次
|
||
out["periods_with_data"] = len(periods)
|
||
out["concentration_untouched_note"] = (
|
||
"现网 2026-01~09 九期在写前已存在 ⇒ 本器不覆盖（见 existing_concentration_periods）")
|
||
out["existing_concentration_periods"] = sorted(existing_conc_periods)
|
||
return out
|
||
|
||
|
||
def anchor_assert(engine, entity_id):
|
||
"""2026-01~09 窗口对拍锚点（任务验收标准逐项）。
|
||
|
||
取数**只走** `cpa_caliber` 单一事实源（不另写第二份聚合 SQL）：逐月客户对按月求和
|
||
= 窗口客户合计（每行只属一个 unit，故月度求和与窗口 GROUP BY 等价）。
|
||
"""
|
||
cal = _load_caliber()
|
||
anchors, tol = _anchors()
|
||
monthly = cal.fetch_monthly_revenue(engine, entity_id)
|
||
by_unit = {}
|
||
for p, pairs in monthly.items():
|
||
if "2026-01" <= p <= "2026-09":
|
||
for u, a in pairs:
|
||
by_unit[u] = by_unit.get(u, 0.0) + a
|
||
pairs = sorted(by_unit.items(), key=lambda x: x[1], reverse=True)
|
||
total = sum(a for _, a in pairs)
|
||
cum = lambda n: round(sum(a for _, a in pairs[:n]) / total * 100, 4) if total else None
|
||
got = {"units": len(pairs), "total_wan": round(total / 10000, 2),
|
||
"top1": round(pairs[0][1] / total * 100, 4) if pairs else None,
|
||
"top1_unit": pairs[0][0] if pairs else None,
|
||
"top2": round(pairs[1][1] / total * 100, 4) if len(pairs) > 1 else None,
|
||
"top2_unit": pairs[1][0] if len(pairs) > 1 else None,
|
||
"top4_cum": cum(4), "top10_cum": cum(10)}
|
||
checks = {}
|
||
for k, exp in anchors.items():
|
||
v = got.get(k)
|
||
if isinstance(exp, int) and not isinstance(exp, bool) and k.endswith("_unit"):
|
||
checks[k] = {"expected": exp, "got": v, "pass": (v == exp)}
|
||
continue
|
||
checks[k] = {"expected": exp, "got": v,
|
||
"pass": (v is not None and abs(float(v) - float(exp)) <= tol)}
|
||
return {"anchor": got, "checks": checks, "ALL_PASS": all(c["pass"] for c in checks.values())}
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
g = ap.add_mutually_exclusive_group()
|
||
g.add_argument("--check", action="store_true")
|
||
g.add_argument("--apply", action="store_true")
|
||
g.add_argument("--verify", action="store_true")
|
||
g.add_argument("--anchor", action="store_true")
|
||
ap.add_argument("--entity", type=int, default=1)
|
||
ap.add_argument("--db-url")
|
||
ap.add_argument("--batch", default=DEFAULT_BATCH)
|
||
ap.add_argument("--concentration", choices=("fill-gaps", "overwrite"), default="fill-gaps",
|
||
help="已有期次处置：fill-gaps=只补缺期（默认）/ overwrite=重算并覆盖（P0 纠错后重算九期用）")
|
||
ap.add_argument("--only", choices=("all", "concentration"), default="all",
|
||
help="写面：all=快照+三族汇总（默认）/ concentration=**仅**写 C_CUST_CONCENTRATION 期次"
|
||
"（P0 九期重算最小写面；需配 --concentration overwrite）")
|
||
ap.add_argument("--since")
|
||
ap.add_argument("--until")
|
||
ap.add_argument("--json", action="store_true")
|
||
args = ap.parse_args()
|
||
|
||
engine, url = _engine(args)
|
||
dialect = _dialect(url)
|
||
|
||
if args.anchor:
|
||
res = anchor_assert(engine, args.entity)
|
||
print(json.dumps(res, ensure_ascii=False, indent=2))
|
||
return 0 if res["ALL_PASS"] else 1
|
||
|
||
res = run(engine, dialect, args.entity, args.batch, apply=args.apply,
|
||
verify=args.verify, since=args.since, until=args.until,
|
||
concentration=args.concentration, only=args.only)
|
||
print(json.dumps(res, ensure_ascii=False, indent=2))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|