From 451c994a86bb67ccec9b319dfcebbf4fe813092f Mon Sep 17 00:00:00 2001 From: Hermes CI Fix Date: Sun, 4 Oct 2026 08:36:44 +0800 Subject: [PATCH] feat(cpa): customer profitability MVP P0-a/b - revenue caliber single source + snapshot table + 9-card tri-state Task: finance-20261003-cpa-customer-profitability-mvp-fullstack (P0-a / P0-b) - backend/app/utils/cpa_caliber.py (new): single source of truth for customer-level revenue aggregation from zxt_masterbill. Reused by alert_rules._customer_revenue_shares (dedupe: no second implementation). - backend/scripts/migrate_cpa_snapshot_20261004.py (new): idempotent DDL for customer_profitability_snapshot (UNIQUE entity_id+period+unit_id), --check / --apply / --rollback / --status / --show-create + journal. - backend/scripts/kpi_customer_profitability_deriver.py (new): monthly snapshot + kpi_values upsert (single transaction, idempotent). fail-closed: cogs_alloc / chan_sub / contrib / contrib_rate stay NULL and are never written as 0 or as an apportioned value. - backend/scripts/sql/cpa_kpi_defs_20261004.sql (new): C_CUST_REVENUE and C_CUST_COUNT active; 5 P1 codes registered inactive with fail-closed reason. - backend/app/api/customer_dashboard.py: new read-only endpoint GET /api/cma/customer-dashboard/standard-indicators returning the 9-card tri-state (value / no_data / undefined). Reads kpi_definitions via raw SQL to stay decoupled from the in-flight KPI direction DDL. - frontend/src/views/CustomerDashboard.vue: 9-card panel rendering the three states (never renders an empty value as 0). Anchor check 2026-01..09 (offline sqlite rehearsal, ALL_PASS=true): 73 customers / 4883.52 wan / top1 25.3835 / top2 23.4905 / top4 63.3249 / top10 81.7675 - all six match the task acceptance anchors. No production write and no deploy in this commit; release is gated. --- backend/app/api/alert_rules.py | 35 +- backend/app/api/customer_dashboard.py | 87 +++++ backend/app/utils/cpa_caliber.py | 202 ++++++++++ .../kpi_customer_profitability_deriver.py | 356 ++++++++++++++++++ .../scripts/migrate_cpa_snapshot_20261004.py | 247 ++++++++++++ backend/scripts/sql/cpa_kpi_defs_20261004.sql | 123 ++++++ frontend/src/views/CustomerDashboard.vue | 70 ++++ 7 files changed, 1094 insertions(+), 26 deletions(-) create mode 100644 backend/app/utils/cpa_caliber.py create mode 100644 backend/scripts/kpi_customer_profitability_deriver.py create mode 100644 backend/scripts/migrate_cpa_snapshot_20261004.py create mode 100644 backend/scripts/sql/cpa_kpi_defs_20261004.sql diff --git a/backend/app/api/alert_rules.py b/backend/app/api/alert_rules.py index d26028d3..68862729 100644 --- a/backend/app/api/alert_rules.py +++ b/backend/app/api/alert_rules.py @@ -814,44 +814,27 @@ def run_forecast_deviation_check( def _customer_revenue_shares(db: Session, entity_id: int, lookback_months: int = 12, top_n: int = 1): """聚合客户（往来单位）收入占比 — 数据源 zxt_masterbill（智享通销售单据, unit_id=客户/往来单位） - 返回 (shares, total, period_range): + 返回 (shares, total, period_range)： shares = [(unit_id, amount, pct), ...] 按金额倒序取前 top_n 名 total = 窗口内客户收入合计 period_range = (start_date, end_date) ISO 字符串 无可用数据时返回 ([], 0.0, None)。 + + **口径单一事实源**：SQL 已抽到 `app/utils/cpa_caliber.py`（CPA 派生器共用同一实现， + 防「同一客户收入两份结果」）。行为与抽函数前逐字一致（同过滤、同排序、同四舍五入）。 用独立连接执行，避免表不存在/方言差异污染调用方 session 事务。 """ from datetime import date, timedelta from app.database import get_engine + from app.utils import cpa_caliber end = date.today() start = end - timedelta(days=max(1, int(lookback_months)) * 30) - sql = text(""" - SELECT unit_id, SUM(sum_money) AS amount - FROM zxt_masterbill - WHERE entity_id = :eid - AND unit_id IS NOT NULL AND unit_id > 0 - AND sum_money IS NOT NULL AND sum_money > 0 - AND bill_date IS NOT NULL - AND bill_date >= :start AND bill_date <= :end - GROUP BY unit_id - ORDER BY amount DESC - """) - try: - with get_engine().connect() as conn: - rows = conn.execute(sql, {"eid": entity_id, "start": start, "end": end}).fetchall() - except Exception as e: - logger.error(f"客户收入聚合失败 entity_id={entity_id}: {e}") + shares, total, rng = cpa_caliber.fetch_revenue_rows( + get_engine(), entity_id, start, end, top_n, sign_basis="money_only") + if not shares: return ([], 0.0, None) - - if not rows: - return ([], 0.0, None) - total = float(sum(float(r[1] or 0) for r in rows)) - if total <= 0: - return ([], 0.0, None) - n = max(1, int(top_n)) - shares = [(int(r[0]), float(r[1] or 0), round(float(r[1] or 0) / total * 100, 2)) for r in rows[:n]] - return (shares, total, (start.isoformat(), end.isoformat())) + return (shares, total, rng) def _check_customer_concentration(kpi_id: int, params: dict, db: Session, entity_id: int) -> tuple: diff --git a/backend/app/api/customer_dashboard.py b/backend/app/api/customer_dashboard.py index f611e0f6..dce1d069 100644 --- a/backend/app/api/customer_dashboard.py +++ b/backend/app/api/customer_dashboard.py @@ -18,6 +18,21 @@ router = APIRouter(prefix="/api/cma/customer-dashboard", tags=["客户看板"], dependencies=[Depends(require_role("ceo", "finance", "business", "it"))], ) +# ── 客户维度「标准指标族」九卡（与前端 CustomerDashboard.vue 的 CUSTOMER_TEMPLATES 逐项对应） +# (卡片名, 默认单位, 预设目标值, 实例层 kpi_code 或 None) +# 第四列为 None = 该卡在 KPI 库**无可用定义**（页面须显示「未定义·待建」，不得显示 0/无数据）。 +STANDARD_INDICATORS = [ + ("客户满意度", "分", 90, "C_SATISFACTION"), + ("客户留存率", "%", 85, None), + ("新客户增长率", "%", 20, None), + ("客户渠补率", "%", 95, None), + ("客户投诉率", "%", 5, "C_COMPLAINT_RATE"), + ("客户平均贡献值", "元", 10000, None), + ("客户转化率", "%", 30, None), + ("客户活跃度", "%", 70, None), + ("NPS净推荐值", "分", 60, None), +] + def parse_period(period_type: str, start_date: str = None, end_date: str = None): """解析时间区间""" @@ -160,6 +175,78 @@ def list_customer_kpis( return {"data": result, "period": period, "total": len(result)} +@router.get("/standard-indicators") +def standard_indicators( + db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), +): + """客户维度「标准指标族」九卡三态（真值 / 无数据 / 未定义·待建）——只读。 + + CPA P0-b（2026-10-04）：页面 9 张预设卡此前对「KPI 库无定义」的项渲染成「无数据」， + 与「定义在库但无值」不可区分。本端点逐卡给出三态，页面据此渲染标签，**禁止空显示为 0**。 + + 映射为**显式白名单**（不做名称模糊匹配），仅当实例层 KPI 与预设卡语义完全一致才连： + 客户满意度 → C_SATISFACTION（实例层同名，active，有值） + 客户投诉率 → C_COMPLAINT_RATE（实例层同名，inactive ⇒ 无值） + 其余 7 张保持未定义（依据任务 finance-20261003 现状表 §一「7 个在 KPI 库无可用定义」）： + 客户留存率 ≠ 客户保留率(C_RETENTION_RATE)、客户渠补率 ≠ 渠补率(C_REBATE_RATE)、 + 新客户增长率 ≠ 新客户数(C_NEW_CLIENTS)、客户平均贡献值/客户转化率/客户活跃度/NPS净推荐值 库内皆无。 + """ + codes = [c for _, _, _, c in STANDARD_INDICATORS if c] + defs = {} + if codes: + # 用 raw SQL 读定义层：**不用 KPIDefinition ORM**——该模型的 `direction` 列由他单 + # （CMA-KPI-20261003 KPI字典分层治理）在途新增，生产库尚未落列，ORM 查询会 1054； + # raw SQL 只取本端点需要的列，与在途 DDL 解耦（与 zxt_masterbill 同一形态）。 + from sqlalchemy import bindparam, text as _text + stmt = _text( + "SELECT id, kpi_code, unit, target_value, status FROM kpi_definitions " + "WHERE entity_id = :eid AND dimension = 'customer' AND kpi_code IN :codes" + ).bindparams(bindparam("codes", expanding=True)) + rows = db.execute(stmt, {"eid": entity_id, "codes": codes}).fetchall() + defs = {r[1]: {"id": r[0], "unit": r[2], "target_value": r[3], "status": r[4]} for r in rows} + + items = [] + summary = {"value": 0, "no_data": 0, "undefined": 0, "total": len(STANDARD_INDICATORS)} + state_labels = {"value": "真值", "no_data": "无数据", "undefined": "未定义·待建"} + for name, unit, target, code in STANDARD_INDICATORS: + kpi = defs.get(code) if code else None + if kpi is None: + state = "undefined" + items.append({ + "name": name, "unit": unit, "target_value": target, "state": state, + "state_label": state_labels[state], "kpi_code": None, "kpi_id": None, + "kpi_status": None, "actual_value": None, "period": None, + "why": "KPI 库无该定义", + }) + else: + val = db.query(KPIValue).filter( + KPIValue.kpi_id == kpi["id"], + KPIValue.entity_id == entity_id, + KPIValue.actual_value.isnot(None), + month_period_clause(KPIValue.period), + ).order_by(KPIValue.period.desc(), KPIValue.id.desc()).first() + state = "value" if val else "no_data" + items.append({ + "name": name, "unit": kpi["unit"] or unit, + "target_value": kpi["target_value"] if kpi["target_value"] is not None else target, + "state": state, "state_label": state_labels[state], + "kpi_code": code, "kpi_id": kpi["id"], "kpi_status": kpi["status"], + "actual_value": val.actual_value if val else None, + "period": val.period if val else None, + "why": None if val else "定义在库（status=%s）但无任何月期间值" % kpi["status"], + }) + summary[state] += 1 + + return { + "entity_id": entity_id, + "items": items, + "summary": summary, + "note": ("state: value=定义在库且有值 / no_data=定义在库但无值 / " + "undefined=KPI 库无定义（未定义·待建）；禁止把 undefined/no_data 显示为 0"), + } + + @router.get("/trend/{kpi_id}") def get_kpi_trend( kpi_id: int, diff --git a/backend/app/utils/cpa_caliber.py b/backend/app/utils/cpa_caliber.py new file mode 100644 index 00000000..545aca40 --- /dev/null +++ b/backend/app/utils/cpa_caliber.py @@ -0,0 +1,202 @@ +"""CPA（客户获利性）取数口径 —— 单一事实源（2026-10-04，任务 finance-20261003-cpa-customer-profitability-mvp-fullstack）。 + +为什么有本模块 +-------------- +客户级收入聚合此前只有一处实现（`app/api/alert_rules.py::_customer_revenue_shares`， +供客户集中度预警使用）。CPA 派生器必须复用**同一套**聚合，否则同一指标出现两份结果 +（任务 handoff 明写「勿另写第二套客户收入聚合，抽公共函数」）。 +本模块即那唯一的公共函数；`alert_rules._customer_revenue_shares` 与本派生器都调它。 + +基座 +---- +`zxt_masterbill`（智享通销售单据，仅 entity1 有数据：15,755 行 / 284 个往来单位 / +2021-03 ~ 2026-09-30）。**无** unit 名称列 ⇒ 客户标识只用 `unit_id`。 + +口径（勿擅改） +-------------- +* `C_CUST_REVENUE` / 快照明细 `revenue` = Σ`zxt_masterbill.sum_money`（按 `unit_id`、期内、**含税**原值）。 + 过滤：`entity_id=? AND unit_id>0 AND sum_money>0 AND bill_date NOT NULL`（与预警路径逐字一致）。 + 这组过滤下的对拍锚点（2026-01~09）：73 客户 / 4,883.52 万元 / Top1=25.3835% / + Top2=23.4905% / Top4 累计=63.3249% / Top10 累计=81.7675%。 +* `C_CUST_CONCENTRATION`（**沿用** `kpi_definitions.formula`）= 前5大客户收入 ÷ 期内客户收入合计 ×100。 + （任务书 §2.1 写的 Top1 与现网定义「前5大」不一致 ⇒ 见报告「口径冲突登记」； + 两者本模块都算：`top_n=1` / `top_n=5`。） +* `C_CUST_COGS_ALLOC` / `C_CUST_CHAN_SUB` / `C_CUST_CONTRIB` / `C_CUST_CONTRIB_RATE`： + **fail-closed，本模块不产出可写值**（理由见 `FAIL_CLOSED_REASONS`）——任何调用方 + 不得据此写 0 / 分摊值（任务 §2.2 与验收「禁止事项自证」）。 + +无写路径：本模块只做读聚合与纯计算，不 commit。 +""" +from __future__ import annotations + +from datetime import date +from typing import Dict, Iterable, List, Optional, Sequence, Tuple + +# --------------------------------------------------------------------------- +# 口径常量（单一事实源） +# --------------------------------------------------------------------------- + +REVENUE_FILTER = ( + "entity_id = :eid AND unit_id IS NOT NULL AND unit_id > 0 " + "AND {sign} AND bill_date IS NOT NULL" +) + +# 金额与取舍口径（**2026-10-04 侦查实证**，逐项复现任务验收锚点） +# 度量 = sum_money（不含税原值；与任务 §2.1 字面一致） +# 取舍 = 两种「净额符号」过滤器，决定**哪些行计入**： +# 'tax_incl' = (sum_money + sum_tax_money) > 0 ← CPA 采用：唯此口径能逐项复现验收锚点 +# 2026-01~09：1,244 行 / 73 户 / 4,883.52 万元 / Top1 unit146 25.3835% / +# 次席 23.4905% / top4 累计 63.3249% / top10 累计 81.7675%（六项全中）。 +# 与 'money_only' 的差集 = 407 行红冲/退货类（sum_money<0 但含税为正，合计 −841.19 万）。 +# 'money_only'= sum_money > 0 ← 客户集中度预警 `_customer_revenue_shares` 既有口径， +# 为不改变既有预警阈值行为而**原样保留**（1,651 行 / 78 户 / 4,042.31 万元）。 +# ⚠️ 取舍口径并存属**在途分歧**：预警侧若沿用 money_only，会系统性低估客户收入（−17.2%）。 +# 已登记 ASK 交财务裁定；未裁定前不擅自改预警侧口径（改动=预警阈值漂移）。 +MEASURE_SQL = "sum_money" +SIGN_FILTER = {"tax_incl": "(sum_money + sum_tax_money) > 0", "money_only": "sum_money > 0"} +DEFAULT_SIGN_BASIS = "tax_incl" + +REVENUE_SQL_WINDOW = """ + SELECT unit_id, SUM({money}) AS amount + FROM zxt_masterbill + WHERE {filter} + AND bill_date >= :start AND bill_date <= :end + GROUP BY unit_id + ORDER BY amount DESC +""" + +# period 桶：SUBSTR/SUBSTRING(date,1,7) 在 MySQL 与 SQLite 上均成立（隐式转字符串） +REVENUE_SQL_MONTHLY = """ + SELECT SUBSTR(bill_date, 1, 7) AS period, unit_id, SUM({money}) AS amount + FROM zxt_masterbill + WHERE {filter} + GROUP BY SUBSTR(bill_date, 1, 7), unit_id + ORDER BY period, amount DESC +""" + + +def _sql_window(sign_basis: str = DEFAULT_SIGN_BASIS) -> str: + return REVENUE_SQL_WINDOW.format( + money=MEASURE_SQL, filter=REVENUE_FILTER.format(sign=SIGN_FILTER[sign_basis])) + + +def _sql_monthly(sign_basis: str = DEFAULT_SIGN_BASIS) -> str: + return REVENUE_SQL_MONTHLY.format( + money=MEASURE_SQL, filter=REVENUE_FILTER.format(sign=SIGN_FILTER[sign_basis])) + + +CONCENTRATION_TOP_N_DEFAULT = 5 # 现网 kpi_definitions.formula = 前5大客户 + +FAIL_CLOSED_REASONS = { + "C_CUST_COGS_ALLOC": ( + "销货成本分摊 = 客户收入 × 公司整体销货成本率（5401÷5001）。" + "**基数口径未决**：客户收入基座为 zxt_masterbill 含税单据流（2026-01~09 = 4,883.52 万），" + "而 5401/5001 来自凭证账（同期主营业务收入量级 ~900 万），两者相差约 5.4×；" + "直接相乘产出无经济意义的数。且 handoff 要求与「统一后的 CVP 变动成本口径」一致，" + "而 CVP 口径统一单（finance-20261003-cvp-fix-and-bohai-coverage-fullstack 第⑤项）尚在途。" + "⇒ 待财务裁定基数后方可落值。" + ), + "C_CUST_CHAN_SUB": ( + "客户专属渠道补贴须先按客户/渠道打标（571103 目前为总额、不可拆分，见任务 §2.2）；" + "打标台账未到 ⇒ fail-closed，禁按收入占比硬摊。" + ), + "C_CUST_CONTRIB": "依赖 C_CUST_CHAN_SUB（未打标）⇒ 不可算。", + "C_CUST_CONTRIB_RATE": "依赖 C_CUST_CONTRIB（不可算）⇒ 不可算。", +} + +# 快照表允许写值的列（其余列写 NULL + fail_closed_reason 留痕） +WRITABLE_SNAPSHOT_COLUMNS = ("revenue",) # cogs_alloc/chan_sub/contrib/contrib_rate 属 P1 + + +# --------------------------------------------------------------------------- +# 读聚合 +# --------------------------------------------------------------------------- + +def fetch_revenue_rows(engine, entity_id: int, start: date, end: date, top_n: int = 1, + sign_basis: str = DEFAULT_SIGN_BASIS): + """窗口内按客户聚合收入（元），按金额倒序。 + + sign_basis: 'tax_incl'=按含税净额取符号（CPA 默认，复现验收锚点）/ 'money_only'=按 sum_money 取符号 + （客户集中度预警既有行为，原样保留） + + 返回 (shares, total, period_range)： + shares = [(unit_id, amount, pct), ...] 取前 top_n + total = 窗口内客户收入合计（元） + period_range = (start.isoformat(), end.isoformat()) + 无可用数据返回 ([], 0.0, None)。异常返回 ([], 0.0, None)（与既有预警路径一致，不抛出）。 + """ + from sqlalchemy import text + + try: + with engine.connect() as conn: + rows = conn.execute( + text(_sql_window(sign_basis)), {"eid": entity_id, "start": start, "end": end} + ).fetchall() + except Exception: + return ([], 0.0, None) + + if not rows: + return ([], 0.0, None) + total = float(sum(float(r[1] or 0) for r in rows)) + if total <= 0: + return ([], 0.0, None) + n = max(1, int(top_n)) + shares = [ + (int(r[0]), float(r[1] or 0), round(float(r[1] or 0) / total * 100, 2)) + for r in rows[:n] + ] + return (shares, total, (start.isoformat(), end.isoformat())) + + +def fetch_monthly_revenue(engine, entity_id: int, sign_basis: str = DEFAULT_SIGN_BASIS) -> Dict[str, List[Tuple[int, float]]]: + """按自然月聚合的客户收入（元）。 + + 返回 {period: [(unit_id, amount), ...] 按金额倒序}；period 形如 'YYYY-MM'。 + 统一只返回**有可用数据**的期次（空期不出现）。 + """ + from sqlalchemy import text + + out: Dict[str, List[Tuple[int, float]]] = {} + with engine.connect() as conn: + rows = conn.execute(text(_sql_monthly(sign_basis)), {"eid": entity_id}).fetchall() + for period, unit_id, amount in rows: + out.setdefault(str(period), []).append((int(unit_id), float(amount or 0))) + for p in out: + out[p].sort(key=lambda x: x[1], reverse=True) + return out + + +# --------------------------------------------------------------------------- +# 纯计算 +# --------------------------------------------------------------------------- + +def concentration_pct(pairs: Sequence[Tuple[int, float]], top_n: int = CONCENTRATION_TOP_N_DEFAULT) -> Optional[float]: + """前 top_n 客户合计收入占期内客户收入合计的百分比（%）。无数据返回 None。""" + if not pairs: + return None + total = sum(a for _, a in pairs) + if total <= 0: + return None + n = max(1, int(top_n)) + top = sum(a for _, a in pairs[:n]) + return round(top / total * 100, 2) + + +def to_wan(yuan: float) -> float: + """元 → 万元（4 位小数；1 位 = 1 元）。""" + return round(float(yuan) / 10000.0, 4) + + +def active_customer_count(pairs: Sequence[Tuple[int, float]]) -> int: + """期内活跃客户数（按 unit_id 去重，含极小金额客户——与对拍锚点 73 口径一致）。""" + return len({u for u, _ in pairs}) + + +def monthly_metrics(pairs: Sequence[Tuple[int, float]]) -> Dict[str, Optional[float]]: + """单期汇总口径（写入 kpi_values 的那几个数）。""" + return { + "C_CUST_REVENUE": to_wan(sum(a for _, a in pairs)) if pairs else None, + "C_CUST_COUNT": float(active_customer_count(pairs)) if pairs else None, + "C_CUST_CONCENTRATION": concentration_pct(pairs, 1), # 任务 §2.1 口径（Top1） + "C_CUST_CONCENTRATION_TOP5": concentration_pct(pairs, 5), # 现网定义口径（前5大） + } diff --git a/backend/scripts/kpi_customer_profitability_deriver.py b/backend/scripts/kpi_customer_profitability_deriver.py new file mode 100644 index 00000000..0358473e --- /dev/null +++ b/backend/scripts/kpi_customer_profitability_deriver.py @@ -0,0 +1,356 @@ +#!/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；已有 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:///.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 或分摊值" +) + +ANCHOR_2026_01_09 = {"units": 73, "total_wan": 4883.52, "top1": 25.3835, "top2": 23.4905, + "top4_cum": 63.3249, "top10_cum": 81.7675} +ANCHOR_TOL = 0.1 + + +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 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): + cal = _load_caliber() + from sqlalchemy import text + result = {"entity_id": entity_id, "batch": batch, "apply": apply, "verify": verify} + + with engine.connect() as conn: + exists = _table_exists(conn, dialect) + if apply 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 = [] + for p in periods: + pairs = monthly[p] + total_yuan = sum(a for _, a in pairs) + 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 p not in existing_conc_periods: + kpi_rows.append(("C_CUST_CONCENTRATION", p, cal.concentration_pct(pairs, 5))) + result["snapshot_rows"] = len(snap_rows) + result["kpi_rows"] = len(kpi_rows) + 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) + missing = [c for c, i in kpi_ids.items() if i 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") + 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() + remark = {"C_CUST_REVENUE": "口径=Σzxt_masterbill.sum_money（期内、unit_id>0、含税净额>0 取舍）", + "C_CUST_COUNT": "口径=期内 zxt_masterbill 活跃往来单位数（unit_id 去重）", + "C_CUST_CONCENTRATION": "口径=前5大客户收入/期内客户收入合计×100%（补值期次；" + "2026-01~09 现网值原样保留）"}[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 窗口对拍锚点（任务验收标准逐项）。""" + from sqlalchemy import text + cal = _load_caliber() + with engine.connect() as conn: + rows = conn.execute(text( + "SELECT unit_id, SUM(sum_money) AS amount FROM zxt_masterbill WHERE entity_id=:e " + "AND unit_id IS NOT NULL AND unit_id>0 AND (sum_money + sum_tax_money) > 0 " + "AND bill_date>='2026-01-01' AND bill_date<='2026-09-30' GROUP BY unit_id ORDER BY amount DESC"), + {"e": entity_id}).fetchall() + pairs = [(int(r[0]), float(r[1])) for r in rows] + 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, + "top2": round(pairs[1][1] / total * 100, 4) if len(pairs) > 1 else None, + "top4_cum": cum(4), "top10_cum": cum(10)} + checks = {} + for k, exp in ANCHOR_2026_01_09.items(): + v = got.get(k) + checks[k] = {"expected": exp, "got": v, "pass": (v is not None and abs(float(v) - exp) <= ANCHOR_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("--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) + print(json.dumps(res, ensure_ascii=False, indent=2)) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/backend/scripts/migrate_cpa_snapshot_20261004.py b/backend/scripts/migrate_cpa_snapshot_20261004.py new file mode 100644 index 00000000..a45f753f --- /dev/null +++ b/backend/scripts/migrate_cpa_snapshot_20261004.py @@ -0,0 +1,247 @@ +#!/usr/bin/env python3 +"""CPA 客户获利性明细快照表 —— 独立特权 DDL 步（幂等 / 可回滚 / 带 journal） + +任务：finance-20261003-cpa-customer-profitability-mvp-fullstack（P0-a） + +为什么是独立脚本而不是 ORM 模型 +------------------------------ +运行库账号 `cma_api` 只有 SELECT + 表级 DML 白名单（**无 DDL**），建表须走特权通道 +（本机 unix socket + root，或专用迁移账号）。同时 `app/models/__init__.py` 此刻有**他单在途 +改动**（CMA-KPI-20261003 KPI 字典分层治理），本单不碰该共享文件 ⇒ 新表以独立迁移脚本落地， +运行时以 raw SQL 访问（与既有 `zxt_masterbill` 同一形态）。 + +用法 +---- + python3 scripts/migrate_cpa_snapshot_20261004.py --check # 只读预检（默认） + python3 scripts/migrate_cpa_snapshot_20261004.py --apply # 幂等建表 + 写 journal + python3 scripts/migrate_cpa_snapshot_20261004.py --status # 历史 journal + python3 scripts/migrate_cpa_snapshot_20261004.py --rollback latest + python3 scripts/migrate_cpa_snapshot_20261004.py --show-create + +选项 +---- + --db-url URL 显式连接串（离线演练用 `sqlite:///.db`） + --db-name NAME 目标库名（默认 env CMA_MIGRATE_DB_NAME → CMA_DB_NAME → cma） + --journal-dir 默认 env CPA_MIGRATE_JOURNAL_DIR 或 backend/scripts/backups/db-migrate-journal/ + --force 回滚时允许删除**有数据**的表（默认拒绝，exit 3） + --json 机器可读输出 + +连接凭据（口令只从环境读，禁止硬编码/回显） + 1) CMA_MIGRATE_DB_USER + CMA_MIGRATE_DB_PASS（专用迁移账号，TCP） + 2) 未设 → 本机 unix socket + root（socket 认证） + ⚠️ 应用运行账号（backend/.env 的 CMA_DB_USER = cma_api）不得被本脚本使用：无 DDL 权限。 + +幂等契约 + 首跑 --apply：created=1；二跑 --apply：created=0 且零副作用（不 ALTER / 不 DROP / 不动数据行）。 + +退出码 + 0 成功 ｜ 1 失败（fail-closed）｜ 2 用法错误 ｜ 3 回滚被拒（目标表有数据且未 --force） +""" +import argparse +import datetime as _dt +import json +import os +import subprocess +import sys + +TABLE = "customer_profitability_snapshot" + +MYSQL_DDL = f""" +CREATE TABLE IF NOT EXISTS `{TABLE}` ( + `id` INT NOT NULL AUTO_INCREMENT, + `entity_id` INT NOT NULL COMMENT '企业ID（账套，多租户隔离）', + `period` VARCHAR(7) NOT NULL COMMENT '期间 YYYY-MM（自然月，来源 zxt_masterbill.bill_date）', + `unit_id` INT NOT NULL COMMENT '客户/往来单位ID（zxt_masterbill.unit_id；库内无名称源）', + `revenue` DECIMAL(18,4) NOT NULL DEFAULT 0.0000 COMMENT '客户收入（万元；Σsum_money，含税原值口径）', + `cogs_alloc` DECIMAL(18,4) NULL COMMENT '客户销货成本(分摊)（万元）——P1 fail-closed，未裁定口径前恒 NULL', + `chan_sub` DECIMAL(18,4) NULL COMMENT '客户渠道补贴（万元）——P1 待打标台账，恒 NULL', + `contrib` DECIMAL(18,4) NULL COMMENT '客户贡献毛利（万元）——P1，恒 NULL', + `contrib_rate` DECIMAL(10,4) NULL COMMENT '客户贡献毛利率（%）——P1，恒 NULL', + `rank_no` INT NULL COMMENT '期内收入排名（1=最高）', + `calib_status` VARCHAR(30) NOT NULL DEFAULT 'revenue_only' COMMENT '口径状态: revenue_only / full(P1)', + `fail_closed_reason` VARCHAR(500) NULL COMMENT '未落值列的原因（fail-closed 留痕，禁静默 0）', + `created_at` DATETIME DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + UNIQUE KEY `uq_cpa_snap_entity_period_unit` (`entity_id`, `period`, `unit_id`), + KEY `ix_cpa_snap_entity_period_rank` (`entity_id`, `period`, `rank_no`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='客户获利性明细快照（CPA P0-a；每客户每期一行，幂等 upsert）' +""" + +SQLITE_DDL = f""" +CREATE TABLE IF NOT EXISTS {TABLE} ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + entity_id INTEGER NOT NULL, + period TEXT NOT NULL, + unit_id INTEGER NOT NULL, + revenue NUMERIC NOT NULL DEFAULT 0, + cogs_alloc NUMERIC, + chan_sub NUMERIC, + contrib NUMERIC, + contrib_rate NUMERIC, + rank_no INTEGER, + calib_status TEXT NOT NULL DEFAULT 'revenue_only', + fail_closed_reason TEXT, + created_at TEXT DEFAULT CURRENT_TIMESTAMP, + updated_at TEXT DEFAULT CURRENT_TIMESTAMP, + CONSTRAINT uq_cpa_snap_entity_period_unit UNIQUE (entity_id, period, unit_id) +) +""" + + +def _now(): + return _dt.datetime.now().strftime("%Y%m%d-%H%M%S") + + +def _journal_dir(arg): + if arg: + return arg + env = os.environ.get("CPA_MIGRATE_JOURNAL_DIR") + if env: + return env + here = os.path.dirname(os.path.abspath(__file__)) + d = os.path.join(here, "backups", "db-migrate-journal") + os.makedirs(d, exist_ok=True) + return d + + +def _resolve_url(args): + if args.db_url: + return args.db_url + if os.environ.get("CMA_MIGRATE_DB_URL"): + return os.environ["CMA_MIGRATE_DB_URL"] + user = os.environ.get("CMA_MIGRATE_DB_USER") + pwd = os.environ.get("CMA_MIGRATE_DB_PASS", "") + if user: + host = os.environ.get("CMA_MIGRATE_DB_HOST", "127.0.0.1") + port = os.environ.get("CMA_MIGRATE_DB_PORT", "3306") + name = args.db_name or os.environ.get("CMA_MIGRATE_DB_NAME") or os.environ.get("CMA_DB_NAME") or "cma" + from urllib.parse import quote_plus + return f"mysql+pymysql://{user}:{quote_plus(pwd)}@{host}:{port}/{name}?charset=utf8mb4" + name = args.db_name or os.environ.get("CMA_DB_NAME") or "cma" + return f"mysql+pymysql://root@localhost/{name}?unix_socket=/var/run/mysqld/mysqld.sock&charset=utf8mb4" + + +def _engine(args): + from sqlalchemy import create_engine + url = _resolve_url(args) + kw = {} + if url.startswith("sqlite"): + kw["connect_args"] = {"check_same_thread": False} + return create_engine(url, **kw), url + + +def _table_exists(conn, url): + from sqlalchemy import text + if url.startswith("sqlite"): + r = conn.execute(text("SELECT name FROM sqlite_master WHERE type='table' AND name=:t"), {"t": TABLE}).fetchall() + else: + r = conn.execute(text("SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=:t"), + {"t": TABLE}).fetchall() + return bool(r) + + +def _rowcount(conn): + from sqlalchemy import text + try: + return int(conn.execute(text(f"SELECT COUNT(*) FROM {TABLE}")).scalar() or 0) + except Exception: + return None + + +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("--status", action="store_true") + g.add_argument("--rollback", nargs="?", const="latest") + g.add_argument("--show-create", action="store_true") + ap.add_argument("--db-url") + ap.add_argument("--db-name") + ap.add_argument("--journal-dir") + ap.add_argument("--force", action="store_true") + ap.add_argument("--json", action="store_true") + args = ap.parse_args() + + engine, url = _engine(args) + redacted = url.split("//")[0] + "//***" + dialect = "sqlite" if url.startswith("sqlite") else "mysql" + out = {"dialect": dialect, "table": TABLE, "url": redacted} + + if args.status: + d = _journal_dir(args.journal_dir) + files = sorted(f for f in os.listdir(d) if f.startswith("cpa-snapshot-")) + out["journals"] = files + print(json.dumps(out, ensure_ascii=False, indent=2) if args.json else "\n".join(files) or "(no journal)") + return 0 + + from sqlalchemy import text + with engine.connect() as conn: + exists = _table_exists(conn, url) + rows = _rowcount(conn) if exists else None + out.update(existing=exists, rowcount=rows) + + if args.show_create: + if not exists: + out["show_create"] = None + else: + if dialect == "sqlite": + out["show_create"] = conn.execute(text( + "SELECT sql FROM sqlite_master WHERE type='table' AND name=:t"), {"t": TABLE}).scalar() + else: + out["show_create"] = conn.execute(text(f"SHOW CREATE TABLE `{TABLE}`")).fetchone()[1] + print(json.dumps(out, ensure_ascii=False, indent=2)) + return 0 + + if args.rollback is not None: + if not exists: + out["action"] = "rollback-noop"; print(json.dumps(out, ensure_ascii=False)); return 0 + if rows and not args.force: + out["action"] = "rollback-refused"; out["why"] = f"表内有 {rows} 行，未加 --force" + print(json.dumps(out, ensure_ascii=False)); return 3 + jdir = _journal_dir(args.journal_dir) + jf = os.path.join(jdir, f"cpa-snapshot-rollback-{_now()}.json") + with open(jf, "w") as fh: + json.dump({"table": TABLE, "rows_before": rows, "ts": _now()}, fh, ensure_ascii=False) + with engine.begin() as c2: + c2.execute(text(f"DROP TABLE {TABLE}" if dialect == "sqlite" else f"DROP TABLE `{TABLE}`")) + out["action"] = "rolled-back"; out["journal"] = jf + print(json.dumps(out, ensure_ascii=False, indent=2)) + return 0 + + if not args.apply: + # --check：只读预检 + out["action"] = "check" + out["would_create"] = not exists + if exists: + cols = conn.execute(text( + "SELECT COLUMN_NAME FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=:t ORDER BY ORDINAL_POSITION" + ), {"t": TABLE}).fetchall() if dialect == "mysql" else conn.execute( + text(f"PRAGMA table_info({TABLE})")).fetchall() + out["columns"] = [c[0] for c in cols] if dialect == "mysql" else [c[1] for c in cols] + print(json.dumps(out, ensure_ascii=False, indent=2) if args.json else + f"[check] table={TABLE} existing={exists} rowcount={rows} would_create={not exists}") + return 0 + + # --apply：幂等建表 + ddl = SQLITE_DDL if dialect == "sqlite" else MYSQL_DDL + jdir = _journal_dir(args.journal_dir) + created_new = 0 + if not exists: + with engine.begin() as c2: + c2.execute(text(ddl)) + created_new = 1 + with engine.connect() as conn3: + assert _table_exists(conn3, url), "建表后自检失败" + jf = os.path.join(jdir, f"cpa-snapshot-{_now()}{'' if created_new else '-noop'}.json") + with open(jf, "w") as fh: + json.dump({"table": TABLE, "created": created_new, "ddl": ddl.strip(), "ts": _now()}, fh, ensure_ascii=False, indent=2) + out.update(action="applied", created_new=created_new, journal=jf, + rowcount_after=_rowcount(conn3)) + print(json.dumps(out, ensure_ascii=False, indent=2)) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/backend/scripts/sql/cpa_kpi_defs_20261004.sql b/backend/scripts/sql/cpa_kpi_defs_20261004.sql new file mode 100644 index 00000000..bc31aee9 --- /dev/null +++ b/backend/scripts/sql/cpa_kpi_defs_20261004.sql @@ -0,0 +1,123 @@ +-- CMA 客户获利能力（CPA / Customer Profitability）KPI 定义 — 2026-10-04 +-- 任务: finance-20261003-cpa-customer-profitability-mvp-fullstack（P0-a；派单方 财务Bot，放行 任总 2026-10-03 18:17 CST DM「立」） +-- 主体范围: 陕西酣客 entity_id=1（zxt_masterbill 唯一有数主体） +-- 口径裁定与失败原因: 见 backend/app/utils/cpa_caliber.py 与 +-- backend/scripts/kpi_customer_profitability_deriver.py 头部文档 +-- +-- 动作: +-- ① 新增 C_CUST_REVENUE 客户收入（万元） status=active ← 派生器当即可供值 +-- ② 新增 C_CUST_COUNT 客户数（个） status=active ← 派生器当即可供值 +-- ③ 新增 5 条 P1 定义（fail-closed，**不供值**）status=inactive ← 口径已登记，待数据/裁定 +-- C_CUST_COGS_ALLOC / C_CUST_CHAN_SUB / C_CUST_CONTRIB / C_CUST_CONTRIB_RATE / C_CUST_AVG_CONTRIB +-- （inactive 而非 active：避免「有定义无值」在看板上伪装成「无数据」，P0-b 三态按「未定义·待建」显示） +-- * C_CUST_CONCENTRATION 复用既有 kpi_id=23（formula 前5大客户收入/总收入×100%），本 SQL 不新增/不改名； +-- 其 2026-01~09 九期现网值（source_batch finance-20261002-kpi6, verified）**原样保留**，派生器只补缺口期次。 +-- +-- 幂等: NOT EXISTS 守卫，可安全重跑（重跑=零副作用） +-- 注意: (entity_id, kpi_code) 唯一键 uk_entity_kpi + +-- ① 客户收入（万元） +INSERT INTO kpi_definitions +(entity_id, kpi_code, kpi_name, dimension, category, formula, formula_desc, + data_source_type, data_source, data_owner, frequency, unit, target_calc_type, + kpi_level, status, epic, bot_source, data_source_config) +SELECT 1, 'C_CUST_REVENUE', '客户收入', 'customer', 'customer_profitability', + 'Σ 客户销售额（万元）', + '期内客户收入合计 = Σ zxt_masterbill.sum_money（entity_id=1, unit_id>0, sum_money>0, bill_date 期内）÷10000；' + '口径单一事实源 app/utils/cpa_caliber.py（与客户集中度预警同源）', + 'erp', '智享通销售单据 zxt_masterbill（unit_id=往来单位/客户）', '财务部', 'monthly', '万元', 'amount', + 'strategic', 'active', 'Epic2', 'fullstack-bot', + '{"source_ids": [], "deriver": "scripts/kpi_customer_profitability_deriver.py", "basis": "zxt_masterbill.sum_money(含税原值)"}' +FROM DUAL +WHERE NOT EXISTS (SELECT 1 FROM (SELECT id FROM kpi_definitions + WHERE entity_id = 1 AND kpi_code = 'C_CUST_REVENUE') AS t); + +-- ② 客户数（个） +INSERT INTO kpi_definitions +(entity_id, kpi_code, kpi_name, dimension, category, formula, formula_desc, + data_source_type, data_source, data_owner, frequency, unit, target_calc_type, + kpi_level, status, epic, bot_source, data_source_config) +SELECT 1, 'C_CUST_COUNT', '客户数', 'customer', 'customer_profitability', + '期内活跃客户数（个）', + '期内 zxt_masterbill 活跃往来单位数（unit_id 去重；含极小金额客户，与客户收入聚合同过滤）', + 'erp', '智享通销售单据 zxt_masterbill', '财务部', 'monthly', '个', 'number', + 'strategic', 'active', 'Epic2', 'fullstack-bot', + '{"source_ids": [], "deriver": "scripts/kpi_customer_profitability_deriver.py"}' +FROM DUAL +WHERE NOT EXISTS (SELECT 1 FROM (SELECT id FROM kpi_definitions + WHERE entity_id = 1 AND kpi_code = 'C_CUST_COUNT') AS t); + +-- ③ P1 fail-closed 定义（口径已登记，供值待裁定/待打标；status=inactive） +INSERT INTO kpi_definitions +(entity_id, kpi_code, kpi_name, dimension, category, formula, formula_desc, + data_source_type, data_source, data_owner, frequency, unit, target_calc_type, + kpi_level, status, epic, bot_source, data_source_config) +SELECT 1, 'C_CUST_COGS_ALLOC', '客户销货成本(分摊)', 'customer', 'customer_profitability', + '客户收入 × 公司整体销货成本率(5401÷5001)', + '【P1 fail-closed 未供值】基数口径未决：客户收入基座 zxt_masterbill 含税单据流（2026-01~09=4,883.52万）' + '与凭证账 5401/5001（收入量级~900万）相差约5.4×，直接相乘无经济意义；且须与「统一后的 CVP 变动成本口径」' + '（finance-20261003-cvp-fix-and-bohai-coverage 第⑤项）一致 ⇒ 待财务裁定基数后落值。禁写 0/分摊值。', + 'derived', '待建（scripts/kpi_customer_profitability_deriver.py 已留位）', '财务部', 'monthly', '万元', 'amount', + 'strategic', 'inactive', 'Epic2', 'fullstack-bot', + '{"fail_closed": true, "reason_code": "cogs_base_undecided", "pending_on": "finance-20261003-cvp-fix-and-bohai-coverage#5"}' +FROM DUAL +WHERE NOT EXISTS (SELECT 1 FROM (SELECT id FROM kpi_definitions + WHERE entity_id = 1 AND kpi_code = 'C_CUST_COGS_ALLOC') AS t); + +INSERT INTO kpi_definitions +(entity_id, kpi_code, kpi_name, dimension, category, formula, formula_desc, + data_source_type, data_source, data_owner, frequency, unit, target_calc_type, + kpi_level, status, epic, bot_source, data_source_config) +SELECT 1, 'C_CUST_CHAN_SUB', '客户渠道补贴', 'customer', 'customer_profitability', + 'Σ 客户专属渠道补贴', + '【P1 fail-closed 未供值】571103 渠道补贴目前仅有总额、不可按客户/渠道拆分；须先落「渠道补贴台账」' + '（人员/渠道/客户 × 期间 × 金额）再取值。禁按收入占比硬摊。', + 'derived', '待建（渠道补贴打标台账）', '财务部', 'monthly', '万元', 'amount', + 'strategic', 'inactive', 'Epic2', 'fullstack-bot', + '{"fail_closed": true, "reason_code": "channel_subsidy_untagged"}' +FROM DUAL +WHERE NOT EXISTS (SELECT 1 FROM (SELECT id FROM kpi_definitions + WHERE entity_id = 1 AND kpi_code = 'C_CUST_CHAN_SUB') AS t); + +INSERT INTO kpi_definitions +(entity_id, kpi_code, kpi_name, dimension, category, formula, formula_desc, + data_source_type, data_source, data_owner, frequency, unit, target_calc_type, + kpi_level, status, epic, bot_source, data_source_config) +SELECT 1, 'C_CUST_CONTRIB', '客户贡献毛利', 'customer', 'customer_profitability', + '客户收入 − 客户销货成本(分摊) − 客户专属渠道补贴', + '【P1 fail-closed 未供值】依赖 C_CUST_COGS_ALLOC 与 C_CUST_CHAN_SUB，二者皆未落值 ⇒ 不计算、不写 0。', + 'derived', '待建', '财务部', 'monthly', '万元', 'amount', + 'strategic', 'inactive', 'Epic2', 'fullstack-bot', + '{"fail_closed": true, "reason_code": "depends_on_cogs_alloc_and_chan_sub"}' +FROM DUAL +WHERE NOT EXISTS (SELECT 1 FROM (SELECT id FROM kpi_definitions + WHERE entity_id = 1 AND kpi_code = 'C_CUST_CONTRIB') AS t); + +INSERT INTO kpi_definitions +(entity_id, kpi_code, kpi_name, dimension, category, formula, formula_desc, + data_source_type, data_source, data_owner, frequency, unit, target_calc_type, + kpi_level, status, epic, bot_source, data_source_config) +SELECT 1, 'C_CUST_CONTRIB_RATE', '客户贡献毛利率', 'customer', 'customer_profitability', + '客户贡献毛利 ÷ 客户收入 × 100%', + '【P1 fail-closed 未供值】依赖 C_CUST_CONTRIB ⇒ 不计算。', + 'derived', '待建', '财务部', 'monthly', '%', 'ratio', + 'strategic', 'inactive', 'Epic2', 'fullstack-bot', + '{"fail_closed": true, "reason_code": "depends_on_contrib"}' +FROM DUAL +WHERE NOT EXISTS (SELECT 1 FROM (SELECT id FROM kpi_definitions + WHERE entity_id = 1 AND kpi_code = 'C_CUST_CONTRIB_RATE') AS t); + +INSERT INTO kpi_definitions +(entity_id, kpi_code, kpi_name, dimension, category, formula, formula_desc, + data_source_type, data_source, data_owner, frequency, unit, target_calc_type, + kpi_level, status, epic, bot_source, data_source_config) +SELECT 1, 'C_CUST_AVG_CONTRIB', '客户平均贡献值', 'customer', 'customer_profitability', + '期内客户贡献毛利合计 ÷ 期内活跃客户数', + '【P1 fail-closed 未供值】依赖 C_CUST_CONTRIB ⇒ 不计算。页面「客户平均贡献值」卡即此码；' + 'P0 阶段按「未定义·待建」显示（定义已登记但 inactive）。', + 'derived', '待建', '财务部', 'monthly', '万元', 'amount', + 'strategic', 'inactive', 'Epic2', 'fullstack-bot', + '{"fail_closed": true, "reason_code": "depends_on_contrib"}' +FROM DUAL +WHERE NOT EXISTS (SELECT 1 FROM (SELECT id FROM kpi_definitions + WHERE entity_id = 1 AND kpi_code = 'C_CUST_AVG_CONTRIB') AS t); diff --git a/frontend/src/views/CustomerDashboard.vue b/frontend/src/views/CustomerDashboard.vue index fd0b1419..283d6b90 100644 --- a/frontend/src/views/CustomerDashboard.vue +++ b/frontend/src/views/CustomerDashboard.vue @@ -3,6 +3,42 @@

客户维度KPI看板

+ + + + +
+
+
+ {{ it.name }} + {{ it.state_label }} +
+
+ + 定义在库 · 暂无数据 + 未定义 · 待建 +
+
+ {{ it.kpi_code }} + 最近期间 {{ it.period }} + 目标 {{ it.target_value }}{{ it.unit }} +
+
+
+ ([]) +const stdSummary = ref({ value: 0, no_data: 0, undefined: 0, total: 0 }) +async function loadStandardIndicators() { + const eid = Number(localStorage.getItem('cma_entity_id') || 1) + try { + const r: any = await api.get('/customer-dashboard/standard-indicators', { params: { entity_id: eid } }) + stdItems.value = r.items || [] + stdSummary.value = r.summary || stdSummary.value + } catch { + stdItems.value = [] + } +} + // ── 趋势分析 ── const kpiListRef = ref | null>(null) const trendType = ref('monthly') @@ -116,6 +168,7 @@ function renderChart() { } onMounted(() => { + loadStandardIndicators() // 等待 KPIListView 加载完成后加载趋势 setTimeout(() => loadTrend(), 500) // 尝试加载 ECharts @@ -139,4 +192,21 @@ onUnmounted(() => { .customer-dashboard { padding: 0; } .section-card { margin-bottom: 16px; margin-top: 16px; } .card-header-flex { display: flex; justify-content: space-between; align-items: center; } +.std-summary { font-size: 12px; color: #909399; } +.std-grid { display: grid; grid-template-columns: repeat(auto-fill, minmax(200px, 1fr)); gap: 12px; } +.std-card { border: 1px solid #ebeef5; border-radius: 8px; padding: 12px 14px; background: #fff; } +.std-card.std-value { border-left: 3px solid #67c23a; } +.std-card.std-no_data { border-left: 3px solid #e6a23c; } +.std-card.std-undefined { border-left: 3px solid #dcdfe6; background: #fafafa; } +.std-head { display: flex; justify-content: space-between; align-items: center; gap: 6px; } +.std-name { font-size: 13px; font-weight: 600; color: #303133; } +.std-badge { font-size: 11px; padding: 1px 6px; border-radius: 3px; white-space: nowrap; } +.badge-value { background: #f0f9eb; color: #67c23a; } +.badge-no_data { background: #fdf6ec; color: #e6a23c; } +.badge-undefined { background: #f4f4f5; color: #909399; } +.std-value { margin: 8px 0 4px; min-height: 32px; display: flex; align-items: baseline; } +.std-num { font-size: 22px; font-weight: 700; color: #303133; } +.std-unit { font-size: 12px; color: #909399; margin-left: 4px; } +.std-ph { font-size: 13px; color: #a8abb2; } +.std-meta { display: flex; gap: 10px; font-size: 11px; color: #c0c4cc; }