diff --git a/backend/app/api/alert_rules.py b/backend/app/api/alert_rules.py index c56cc3bf..212196b4 100644 --- a/backend/app/api/alert_rules.py +++ b/backend/app/api/alert_rules.py @@ -1136,7 +1136,8 @@ def run_ar_overdue_check( return payload if dry_run: - payload.update({"would_insert": 0 if blockers else len(hits), "write_blockers": blockers}) + payload.update({"would_insert": 0 if blockers else ar_overdue_alert.count_would_insert( + db, entity_id, resolved_kpi_id, result), "write_blockers": blockers}) return payload if blockers: @@ -1147,6 +1148,6 @@ def run_ar_overdue_check( "hit": len(hits), **written}, entity_id=entity_id) db.commit() - payload.update({"would_insert": len(hits), "write_blockers": [], **written, + payload.update({"would_insert": written["inserted"], "write_blockers": [], **written, "generated": written["inserted"]}) return payload diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 216c393b..b7f5ee35 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -167,6 +167,8 @@ class KPIAlert(Base): alert_level = Column(String(20), default="yellow", comment="green/yellow/red") alert_message = Column(String(500), nullable=False) alert_type = Column(String(30), default="actual", comment="actual/forecast — 实际值超限/预测值超限") + dedupe_key = Column(String(160), nullable=True, + comment="生成侧幂等键 period|scope|ident；仅 alert_type=overdue 填充，历史行为 NULL") suggestion = Column(Text, nullable=True, comment="情景建议") action_plan_linked_id = Column(Integer, nullable=True, comment="关联的改善计划ID") status = Column(String(20), default="pending", comment="pending/processing/resolved") diff --git a/backend/app/services/ar_overdue_alert.py b/backend/app/services/ar_overdue_alert.py index dd21218f..bf2d52cb 100644 --- a/backend/app/services/ar_overdue_alert.py +++ b/backend/app/services/ar_overdue_alert.py @@ -4,9 +4,10 @@ * **阈值/责任人/排除名单全部来自 `alert_rules.params`（DB 行）**，本模块不出现任何 业务阈值数字；缺键即该判据关闭（显式失效，不回落隐式默认值）。 -* **只 INSERT**：命中写 `kpi_alerts(alert_type='overdue')`，同 - `(entity_id, kpi_id, alert_type, alert_message)` 已存在则跳过（`alert_message` - 含 period+对象 ⇒ 天然去重键）。**零 UPDATE / 零 DELETE**。 +* **只 INSERT**：命中写 `kpi_alerts(alert_type='overdue')`，去重键为**生成侧语义键** + `dedupe_key = period|scope|ident`（单户 `cust|`、聚合 `agg|<族名>`）， + 由唯一索引兜底；文案口径变化不再产生重复行。历史行 `dedupe_key IS NULL` 走只读反解兼容。 + **零 UPDATE / 零 DELETE**。 * `assignee` 必填（`assignee_by_unit[unit_code]` → `assignee_default`），两者皆空 ⇒ `AssigneeMissing`（调用方转 400，fail-closed）——禁落空 assignee 的行。 * 口径：只读 `借方应收户口径`（`app.utils.ar_caliber`），负档位不抹平。 @@ -15,10 +16,12 @@ from __future__ import annotations import calendar import logging +import re from decimal import Decimal, ROUND_HALF_UP from typing import Any, Dict, List, Optional, Tuple from sqlalchemy import text +from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from app.models import KPIAlert @@ -738,35 +741,178 @@ def hits(result: Dict[str, Any]) -> List[Dict[str, Any]]: return [h for h in all_hits if str(h.get("assignee") or "").strip()] -def write_alerts(db: Session, entity_id: int, kpi_id: int, result: Dict[str, Any]) -> Dict[str, int]: - """把命中集合写成 `kpi_alerts(alert_type='overdue')`；历史行零触碰。""" - inserted = 0 - skipped = 0 +# ============================================================ +# 去重键（生成侧语义键）：period|scope|ident +# ============================================================ +_AGG_KEY_FAMILY = { + "share_90plus": "share_90plus", + "count_mom": "count_mom", + "structure_90_150": "structure", # 族名，不跟随判据代次 + "structure_150p": "structure", +} + +_CUST_MSG_RE = re.compile(r"（([A-Za-z0-9_\-]+)）(\d{4}-\d{2})") +_PERIOD_RE = re.compile(r"(\d{4}-\d{2})") + + +def _agg_family(key: str) -> str: + if key in _AGG_KEY_FAMILY: + return _AGG_KEY_FAMILY[key] + return str(key).split("_")[0] or "unknown" + + +def _item_dedupe_key(item: Dict[str, Any], period: str) -> Optional[str]: + """生成侧幂等键；不解析文案。 + + - 单户：`|cust|` + - 聚合：`|agg|<族名>`（structure_90_150 / structure_150p 同归 structure） + - 无法判定 → None（调用方退回文案逐字比较，方向安全） + """ + p = str(item.get("period") or period or "").strip() + if not p: + return None + unit_code = str(item.get("unit_code") or "").strip() + if unit_code: + return "%s|cust|%s" % (p, unit_code) + key = str(item.get("key") or "").strip() + if key: + return "%s|agg|%s" % (p, _agg_family(key)) + return None + + +def _key_from_message(msg: str) -> Optional[str]: + """历史兼容：从旧文案只读反解 dedupe_key（仅用于 dedupe_key IS NULL 的行）。 + + 反解失败返回 None ⇒ 视为未命中 ⇒ 照常插入（fail-open，多插一条而非吞真实告警）。 + """ + if not msg: + return None + m = _CUST_MSG_RE.search(msg) + if m: + return "%s|cust|%s" % (m.group(2), m.group(1)) + pm = _PERIOD_RE.search(msg) + if not pm: + return None + period = pm.group(1) + if "90 天以上" in msg and "占应收总额" in msg: + return "%s|agg|share_90plus" % period + if "逾期户数" in msg and "环比" in msg: + return "%s|agg|count_mom" % period + if "账龄" in msg and ("结构" in msg or "90–150" in msg or "150 天以上" in msg): + return "%s|agg|structure" % period + return None + + +def _legacy_keys(db: Session, entity_id: int, kpi_id: int) -> List[str]: + """取历史行（dedupe_key IS NULL）反解出的键集合（边界层，量级=历史行数）。""" + rows = db.query(KPIAlert.alert_message).filter( + KPIAlert.entity_id == entity_id, + KPIAlert.kpi_id == kpi_id, + KPIAlert.alert_type == ALERT_TYPE, + KPIAlert.dedupe_key.is_(None), + ).all() + keys = [] + for (m,) in rows: + k = _key_from_message(m or "") + if k: + keys.append(k) + return keys + + +def count_would_insert(db: Session, entity_id: int, kpi_id: Optional[int], + result: Dict[str, Any]) -> int: + """dry-run 用：按与 write_alerts 完全相同的判重口径，算出真正会插入的条数。""" + if kpi_id is None: + return 0 + period = str(result.get("period") or "") + existing = { + k for (k,) in db.query(KPIAlert.dedupe_key).filter( + KPIAlert.entity_id == entity_id, + KPIAlert.kpi_id == kpi_id, + KPIAlert.alert_type == ALERT_TYPE, + KPIAlert.dedupe_key.isnot(None), + ).all() if k + } + existing.update(_legacy_keys(db, entity_id, kpi_id)) + seen = set() + n = 0 for h in hits(result): msg = h.get("alert_message") or "" if not msg: continue - exists = db.query(KPIAlert.id).filter( + dk = _item_dedupe_key(h, period) + if dk is None: + # 无键 ⇒ 退回文案逐字（与旧行为一致） + n += 1 + continue + if dk in existing or dk in seen: + continue + seen.add(dk) + n += 1 + return n + + +def write_alerts(db: Session, entity_id: int, kpi_id: int, result: Dict[str, Any]) -> Dict[str, int]: + """把命中集合写成 `kpi_alerts(alert_type='overdue')`；历史行零触碰。 + + 判重三级：① dedupe_key 命中 ② 历史 NULL 行反解键命中 ③ 无命中 → INSERT。 + 并发兜底：唯一索引 + savepoint 捕获 IntegrityError 计跳过。 + """ + inserted = 0 + skipped = 0 + period = str(result.get("period") or "") + existing = { + k for (k,) in db.query(KPIAlert.dedupe_key).filter( KPIAlert.entity_id == entity_id, KPIAlert.kpi_id == kpi_id, KPIAlert.alert_type == ALERT_TYPE, - KPIAlert.alert_message == msg, - ).first() - if exists: - skipped += 1 + KPIAlert.dedupe_key.isnot(None), + ).all() if k + } + existing.update(_legacy_keys(db, entity_id, kpi_id)) + for h in hits(result): + msg = h.get("alert_message") or "" + if not msg: continue - db.add(KPIAlert( + dk = _item_dedupe_key(h, period) + if dk is not None: + if dk in existing: + skipped += 1 + continue + else: + # 无键（理论上不会发生）：退回逐字判重（旧行为） + dup = db.query(KPIAlert.id).filter( + KPIAlert.entity_id == entity_id, + KPIAlert.kpi_id == kpi_id, + KPIAlert.alert_type == ALERT_TYPE, + KPIAlert.alert_message == msg, + ).first() + if dup: + skipped += 1 + continue + row = KPIAlert( entity_id=entity_id, kpi_id=kpi_id, kpi_value_id=None, alert_level=h.get("level") or "yellow", alert_message=msg, alert_type=ALERT_TYPE, + dedupe_key=dk, suggestion=h.get("suggestion") or None, status="pending", assignee=h.get("assignee") or None, - )) - inserted += 1 + ) + sp = db.begin_nested() + try: + db.add(row) + db.flush() + sp.commit() + inserted += 1 + if dk: + existing.add(dk) + except IntegrityError: + sp.rollback() + skipped += 1 if inserted: db.commit() return {"inserted": inserted, "skipped_duplicate": skipped} diff --git a/backend/scripts/ddl_kpi_alerts_dedupe.py b/backend/scripts/ddl_kpi_alerts_dedupe.py new file mode 100644 index 00000000..a48c7bc2 --- /dev/null +++ b/backend/scripts/ddl_kpi_alerts_dedupe.py @@ -0,0 +1,113 @@ +#!/usr/bin/env python3 +"""kpi_alerts 去重键结构变更（幂等执行器，A2 方案） + +用法： + ddl_kpi_alerts_dedupe.py --check # 只读：打印列/索引是否存在（默认） + ddl_kpi_alerts_dedupe.py --apply # 幂等执行：缺列则加列，缺索引则建索引 + ddl_kpi_alerts_dedupe.py --rollback # 回退：删索引 + 删列（数据无需恢复：本改造无 UPDATE/DELETE） + +特性： + * information_schema 判存在后再执行 ⇒ 重复跑安全（第二次全 NOOP） + * 只加可空列 + 唯一索引；历史行 dedupe_key 全 NULL ⇒ 建索引必然无冲突 + * root 本机 socket 连接（与 migrate_map_connections.py 同口径），不读 .env +""" +from __future__ import annotations + +import argparse +import sys + +from sqlalchemy import create_engine, text + +URL = "mysql+pymysql://root@localhost/cma?unix_socket=/var/run/mysqld/mysqld.sock&charset=utf8mb4" +TABLE = "kpi_alerts" +COLUMN = "dedupe_key" +INDEX = "uk_kpi_alerts_dedupe" + +ADD_COLUMN_SQL = ( + "ALTER TABLE kpi_alerts " + "ADD COLUMN dedupe_key VARCHAR(160) NULL " + "COMMENT '生成侧幂等键 period|scope|ident，仅 alert_type=overdue 填充；历史行为 NULL' " + "AFTER alert_type" +) +ADD_INDEX_SQL = ( + "CREATE UNIQUE INDEX uk_kpi_alerts_dedupe " + "ON kpi_alerts (entity_id, kpi_id, alert_type, dedupe_key)" +) +DROP_INDEX_SQL = "DROP INDEX uk_kpi_alerts_dedupe ON kpi_alerts" +DROP_COLUMN_SQL = "ALTER TABLE kpi_alerts DROP COLUMN dedupe_key" + + +def col_exists(eng) -> bool: + with eng.connect() as c: + n = c.execute(text( + "SELECT COUNT(*) FROM information_schema.columns " + "WHERE table_schema=DATABASE() AND table_name=:t AND column_name=:c" + ), {"t": TABLE, "c": COLUMN}).scalar() + return int(n or 0) > 0 + + +def idx_exists(eng) -> bool: + with eng.connect() as c: + n = c.execute(text( + "SELECT COUNT(*) FROM information_schema.statistics " + "WHERE table_schema=DATABASE() AND table_name=:t AND index_name=:i" + ), {"t": TABLE, "i": INDEX}).scalar() + return int(n or 0) > 0 + + +def main() -> int: + ap = argparse.ArgumentParser() + g = ap.add_mutually_exclusive_group() + g.add_argument("--check", action="store_true", help="只读检查（默认）") + g.add_argument("--apply", action="store_true", help="幂等执行") + g.add_argument("--rollback", action="store_true", help="回退结构") + a = ap.parse_args() + if not (a.check or a.apply or a.rollback): + a.check = True + + eng = create_engine(URL, future=True) + col, idx = col_exists(eng), idx_exists(eng) + print("[state] %s.%s=%s index %s=%s" % (TABLE, COLUMN, col, INDEX, idx)) + + if a.check: + print("[check] %s" % ("结构就绪（无操作）" if (col and idx) else + ("缺列" if not col else "") + ("缺索引" if not idx else ""))) + return 0 + + if a.rollback: + with eng.begin() as c: + if idx: + c.execute(text(DROP_INDEX_SQL)) + print("[rollback] DROP INDEX %s ✅" % INDEX) + if col: + c.execute(text(DROP_COLUMN_SQL)) + print("[rollback] DROP COLUMN %s ✅" % COLUMN) + print("[state] 回退后 col=%s idx=%s" % (col_exists(eng), idx_exists(eng))) + return 0 + + # --apply + done = [] + if not col: + with eng.begin() as c: + c.execute(text(ADD_COLUMN_SQL)) + done.append("ADD COLUMN %s" % COLUMN) + print("[apply] ADD COLUMN %s ✅" % COLUMN) + else: + print("[apply] column %s 已存在 → NOOP" % COLUMN) + if not idx: + with eng.begin() as c: + c.execute(text(ADD_INDEX_SQL)) + done.append("CREATE INDEX %s" % INDEX) + print("[apply] CREATE UNIQUE INDEX %s ✅" % INDEX) + else: + print("[apply] index %s 已存在 → NOOP" % INDEX) + + col2, idx2 = col_exists(eng), idx_exists(eng) + print("[state] 执行后 col=%s idx=%s" % (col2, idx2)) + payload = {"changed": done, "column_exists": col2, "index_exists": idx2} + print("[result] %s" % payload) + return 0 if (col2 and idx2) else 5 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/backend/tests/test_dedupe_key.py b/backend/tests/test_dedupe_key.py new file mode 100644 index 00000000..f816c444 --- /dev/null +++ b/backend/tests/test_dedupe_key.py @@ -0,0 +1,162 @@ +"""A2 幂等键：kpi_alerts.dedupe_key（任务 project-20261021-ar-overdue-dedupe-key-fullstack） + +覆盖： + T1 历史旧文案行（dedupe_key=NULL）零改动 + 同语义口径重跑 → 全跳过 + T2 重复跑不增行（幂等） + T3 负测：新单户 → 插入 1 条（证明不是「一律跳过」） + T4 负测：同户金额变化 → 不插（语义幂等，键不含金额） + T5 同批两条同键 → 1 插 1 跳（批内去重 + 唯一索引兜底） + T6 dry-run count_would_insert 与真实 inserted 一致 + T7 键生成/反解单测（结构 + 聚合族归一化） +""" +import pytest +from sqlalchemy import text + +from app.models import KPIAlert +from app.services import ar_overdue_alert as A + +EID, KPI = 1, 517 +P = "2026-09" + +LEGACY = [ + # 1 条聚合（旧文案，能被反解成 agg|share_90plus） + "【应收账款逾期】2026-09 借方应收户口径 90 天以上 ¥1,000.00，占应收总额 15.56%（红线 >25.00%）｜建议：见规则配置", + # 7 条单户（旧文案，反解成 cust|） + "【应收账款逾期】王敏（HKJJ053）2026-09 账龄最老 151 天，逾期合计 ¥666,537.00（占该户 100.00%）｜建议：催收｜责任人：李世昌", + "【应收账款逾期】西安酣君品创（HKJJ120）2026-09 账龄最老 151 天，逾期合计 ¥300,120.00（占该户 85.14%）｜建议：催收｜责任人：李世昌", + "【应收账款逾期】吉林酣客（HKJJ156）2026-09 账龄最老 151 天，逾期合计 ¥75,640.00（占该户 100.00%）｜建议：催收｜责任人：李世昌", + "【应收账款逾期】荣升酒窖（HKJJ166）2026-09 账龄最老 151 天，逾期合计 ¥36,586.00（占该户 100.00%）｜建议：催收｜责任人：兀晨博", + "【应收账款逾期】宝鸡裕腾（HKJJ175）2026-09 账龄最老 91 天，逾期合计 ¥32,560.00（占该户 100.00%）｜建议：催收｜责任人：兀晨博", + "【应收账款逾期】酣聚酒窖（HKJJ122）2026-09 账龄最老 151 天，逾期合计 ¥32,538.54（占该户 100.00%）｜建议：催收｜责任人：兀晨博", + "【应收账款逾期】孙飞龙客户（007）2026-09 账龄最老 45 天，逾期合计 ¥10,000.00（占该户 50.00%）｜建议：催收｜责任人：孙飞龙", +] + + +def _seed_legacy(db): + for m in LEGACY: + db.add(KPIAlert(entity_id=EID, kpi_id=KPI, alert_level="red", alert_message=m, + alert_type="overdue", suggestion="催收", status="pending", + assignee="李世昌")) + db.commit() + return len(LEGACY) + + +def _cust(code, amt="1.00", name="客户", assignee="李世昌"): + return {"unit_code": code, "unit_name": name, "overdue_amt": amt, "total_amt": "1.00", + "oldest_days": 151, "pct_of_unit": 1.0, "level": "red", + "alert_message": "【应收账款逾期】%s（%s）%s 账龄最老 151 天，逾期合计 ¥%s（占该户 100.00%%）" + "｜建议：催收｜责任人：%s" % (name, code, P, amt, assignee), + "suggestion": "催收", "assignee": assignee, "assignee_source": "unit", "emp_id": "E1"} + + +def _agg(key="share_90plus", msg=None): + return {"key": key, "level": "red", + "metrics": {"share_90plus": 0.1556}, + "alert_message": msg or ("【应收账款逾期】%s 借方应收户口径 90 天以上 ¥1,000.00，" + "占应收总额 15.56%%（红线 >25.00%%）｜建议：见规则配置" % P), + "suggestion": "见规则配置", "assignee": "李世昌", "assignee_source": "aggregate"} + + +def _result(customers, aggs=None): + return {"period": P, "aggregates": aggs or [], "per_customer": customers} + + +def _total(db): + return db.execute(text("select count(*) from kpi_alerts")).scalar() + + +# ── T7 键生成/反解（纯函数）──────────────────────────────────────────── +def test_t7_key_gen_and_parse(): + assert A._item_dedupe_key({"unit_code": "HKJJ053"}, P) == "2026-09|cust|HKJJ053" + assert A._item_dedupe_key({"key": "share_90plus"}, P) == "2026-09|agg|share_90plus" + assert A._item_dedupe_key({"key": "count_mom"}, P) == "2026-09|agg|count_mom" + assert A._item_dedupe_key({"key": "structure_90_150"}, P) == "2026-09|agg|structure" + assert A._item_dedupe_key({"key": "structure_150p"}, P) == "2026-09|agg|structure" + assert A._item_dedupe_key({}, P) is None + # 反解（历史兼容） + assert A._key_from_message(LEGACY[0]) == "2026-09|agg|share_90plus" + for m in LEGACY[1:]: + code = m.split("（")[1].split("）")[0] + assert A._key_from_message(m) == "2026-09|cust|%s" % code + assert A._key_from_message("完全无关的文案") is None + + +# ── T1/T2 历史行零改 + 重跑幂等 ──────────────────────────────────────── +def test_t1_t2_legacy_untouched_and_idempotent(db): + n0 = _seed_legacy(db) + ids_before = [r[0] for r in db.execute(text("select id from kpi_alerts order by id")).all()] + snap_before = [tuple(r) for r in db.execute(text( + "select id, alert_message, dedupe_key, status, coalesce(assignee,'') from kpi_alerts order by id")).all()] + + # 同语义口径 8 条（1 聚合 + 7 单户），文案由新代码生成（措辞与历史略有差异也可） + res = _result([_cust(c) for c in ("HKJJ053", "HKJJ120", "HKJJ156", "HKJJ166", "HKJJ175", "HKJJ122", "007")], + [_agg()]) + w1 = A.write_alerts(db, EID, KPI, res) + assert w1 == {"inserted": 0, "skipped_duplicate": 8}, w1 + + # 再跑一次（T2） + w2 = A.write_alerts(db, EID, KPI, res) + assert w2 == {"inserted": 0, "skipped_duplicate": 8}, w2 + + # 历史行零改动（R1 红线） + snap_after = [tuple(r) for r in db.execute(text( + "select id, alert_message, dedupe_key, status, coalesce(assignee,'') from kpi_alerts order by id")).all()] + assert snap_after == snap_before + assert _total(db) == n0 + assert [r[0] for r in db.execute(text("select id from kpi_alerts order by id")).all()] == ids_before + + +# ── T3 新户插入（负测）──────────────────────────────────────────────── +def test_t3_new_customer_inserts(db): + _seed_legacy(db) + n0 = _total(db) + res = _result([_cust("HKJJ999", amt="88,888.00", name="新户")]) + w = A.write_alerts(db, EID, KPI, res) + assert w == {"inserted": 1, "skipped_duplicate": 0}, w + assert _total(db) == n0 + 1 + row = db.query(KPIAlert).filter(KPIAlert.dedupe_key == "2026-09|cust|HKJJ999").one() + assert row.alert_type == "overdue" and row.assignee == "李世昌" + + +# ── T4 同户金额变化不插（语义幂等）──────────────────────────────────── +def test_t4_amount_change_does_not_insert(db): + _seed_legacy(db) + res = _result([_cust("HKJJ053", amt="700,000.00", name="王敏")]) + w = A.write_alerts(db, EID, KPI, res) + assert w == {"inserted": 0, "skipped_duplicate": 1}, w + + +# ── T5 同批两条同键 → 1 插 1 跳 ─────────────────────────────────────── +def test_t5_same_key_in_one_batch(db): + n0 = _total(db) + res = _result([_cust("HKJJ888", amt="1.00"), _cust("HKJJ888", amt="2.00")]) + w = A.write_alerts(db, EID, KPI, res) + assert w == {"inserted": 1, "skipped_duplicate": 1}, w + assert _total(db) == n0 + 1 + + +# ── T6 dry-run 与实插一致 ───────────────────────────────────────────── +def test_t6_dry_run_matches_insert(db): + _seed_legacy(db) + res = _result([_cust("HKJJ053"), _cust("HKJJ777", amt="3.00")], [_agg()]) + n_pred = A.count_would_insert(db, EID, KPI, res) + w = A.write_alerts(db, EID, KPI, res) + assert n_pred == w["inserted"] == 1, (n_pred, w) + # 写后再算 → 0 + assert A.count_would_insert(db, EID, KPI, res) == 0 + + +# ── T7b 唯一索引语义：手工插同键第二次必须失败（并发兜底）───────────── +def test_t7b_unique_index_enforced(db): + # conftest 的 create_all 不含 DDL 索引 ⇒ 测试内补建（生产由 ddl 脚本建） + db.execute(text("CREATE UNIQUE INDEX IF NOT EXISTS uk_kpi_alerts_dedupe_test " + "ON kpi_alerts (entity_id, kpi_id, alert_type, dedupe_key)")) + db.commit() + db.add(KPIAlert(entity_id=EID, kpi_id=KPI, alert_message="a", alert_type="overdue", + dedupe_key="2026-09|cust|DUP")) + db.commit() + db.add(KPIAlert(entity_id=EID, kpi_id=KPI, alert_message="b", alert_type="overdue", + dedupe_key="2026-09|cust|DUP")) + with pytest.raises(Exception): + db.commit() + db.rollback()