wecom-fullstack
9bbe3001bf
ci/woodpecker/push/main Pipeline was successful
- 列 dedupe_key VARCHAR(160) NULL + 唯一索引 (entity_id,kpi_id,alert_type,dedupe_key) - write_alerts 三级判重：dedupe_key 命中 / 历史 NULL 行反解兼容 / 无命中插入 - 并发兜底：savepoint 捕获 IntegrityError 计跳过 - dry-run would_insert 改用 count_would_insert（修复 len(hits) 虚高） - DDL 幂等执行器 scripts/ddl_kpi_alerts_dedupe.py（check/apply/rollback） - tests/test_dedupe_key.py 7 passed（T1-T7b）；alert/overdue 回归 173 passed
163 lines
8.8 KiB
Python
163 lines
8.8 KiB
Python
"""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|<code>）
|
||
"【应收账款逾期】王敏（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()
|