feat(receivables): customer-level rows only + register hanke_ar as balance snapshot batch

ASK-20261001-33: api_receivables counts only rows with a non-blank counterparty (monthly aggregate rows stay in table but out of the collection page cards/aging/detail/owners); BALANCE_SNAPSHOT_SOURCES gains hanke_ar (default skip, no per-customer overdue alerts). New pytest file covers aggregate-row exclusion, e2 baseline digitwise, snapshot registration.
(cherry picked from commit 475f78bc8c5f67f2716c721172a88ff89ec58227)

backend/app/utils/cash_forecast_engine.py

@@ -55,13 +55,17 @@ KPI_CODE_SUFFIXES = {
 # ── 余额快照批次（F4，2026-09-13）────────────────────────────
 # 这些 source 是「期末余额导入」批次：plan_date 是**导入基准日**（如 2026-06-30 期末），
 # 不是真实到期日 → 逐户生成"到期未收款"会永远 overdue，属伪逾期（DAMA 审计 fin-alert-flood-e2）。
+# hanke_ar（酣客 2026-07 期末客户级应收，ASK-20261001-33）语义与 bohai_ar_20260630 完全相同，
+# 2026-10-01 按设计件 §6.2「快照批次登记」加入名单：默认 skip ⇒ 不逐户生成伪逾期告警。
 BALANCE_SNAPSHOT_SOURCES = {
     "bohai_ar_20260630",
     "bohai_ap_20260630",
     "bohai_prepay_20260630",
     "receivables_migrate",
+    "hanke_ar",
 }
 
+
 # 处理开关：per_customer(逐户，旧行为) / skip(不生成逐户，默认) / aggregate(按批次聚合1条)
 _SNAPSHOT_MODE_DEFAULT = "skip"
 _SNAPSHOT_MODES = ("per_customer", "skip", "aggregate")

backend/tests/test_receivables_customer_level_20261001.py

@@ -0,0 +1,248 @@
+"""应收催收页「只认客户级行」口径 + 酣客快照批次登记（ASK-20261001-33，2026-10-01）
+
+背景：酣客 67 行 `source='zxt-sync'` 月度汇总行（counterparty 空/空白）与落库后的 18 行
+客户级 `hanke_ar` 同表并存。旧口径把两者求和 ⇒ 催收页出现 ¥1.86 亿假「应收总额」。
+本组用例锁定两条契约：
+
+  1. `api_receivables` 只统计**客户级行**（counterparty 非空且非空白）：汇总行不计入
+     total / summary / aging / owners；边界含 None / "" / 空格 / 制表符 / 全角空格。
+  2. 无汇总行的账套（博海 e2 形态：无空 counterparty）逐位不变 —— 判据收窄对既有账套
+     零影响（同 8 项基线数值）。
+  3. `hanke_ar` 已登记为余额快照批次（默认 skip ⇒ 不逐户生成伪逾期告警）。
+"""
+from datetime import datetime, timedelta
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy.orm import Session
+
+from app.models import CashPlan, KPIAlert
+from app.utils.cash_forecast_engine import (
+    BALANCE_SNAPSHOT_SOURCES,
+    check_cash_alerts,
+    get_balance_snapshot_mode,
+    split_snapshot_plans,
+)
+from tests.conftest import create_test_kpi, create_test_user, get_token_for_user, auth_header
+
+BASE = "/api/cma/cash"
+HANK_END = datetime(2026, 7, 31)
+FUTURE = datetime.now() + timedelta(days=90)
+
+
+def _add_plan(db: Session, **kwargs) -> CashPlan:
+    defaults = dict(
+        entity_id=1,
+        plan_type="receive",
+        amount=0.0,
+        plan_date=HANK_END,
+        status="pending",
+        source="manual",
+        paid_amount=0,
+    )
+    defaults.update(kwargs)
+    plan = CashPlan(**defaults)
+    db.add(plan)
+    db.commit()
+    db.refresh(plan)
+    return plan
+
+
+def _receivables(client: TestClient, token: str) -> dict:
+    resp = client.get(f"{BASE}/receivables", headers=auth_header(token))
+    assert resp.status_code == 200
+    return resp.json()
+
+
+class TestAggregateRowsExcluded:
+    """① 月度汇总行被排除（客户级行照常计入）"""
+
+    def test_aggregate_rows_excluded_from_cards_rows_and_owners(self, client, db):
+        create_test_user(db)
+        token = get_token_for_user(client)
+
+        # 月度汇总行三形态：NULL / 空白 / 空串（source 均为 zxt-sync 实收历史）
+        _add_plan(db, counterparty=None, amount=440.61, status="completed",
+                  plan_date=datetime(2021, 3, 31), source="zxt-sync", paid_amount=0)
+        _add_plan(db, counterparty="   ", amount=100.0, status="completed",
+                  plan_date=datetime(2021, 4, 30), source="zxt-sync", paid_amount=0)
+        _add_plan(db, counterparty="", amount=50.0, status="pending",
+                  plan_date=HANK_END, source="zxt-sync", owner="", paid_amount=0)
+        # 客户级行：落库后的酣客 18 行同形态（期末日 2026-07-31 ⇒ 60-90天档）
+        hanke = _add_plan(db, counterparty="陕西弦鸠商贸有限公司", amount=15.5538,
+                          plan_date=HANK_END, source="hanke_ar")
+        own = _add_plan(db, counterparty="某客户", amount=10.0, plan_date=FUTURE,
+                        source="manual", owner="张三")
+
+        body = _receivables(client, token)
+        ids = [row["id"] for row in body["data"]]
+
+        assert body["total"] == 2
+        assert ids == [hanke.id, own.id], "汇总行不得出现在明细里"
+        assert body["summary"]["total_amount"] == 25.55
+        assert body["summary"]["pending_balance"] == 25.55
+        assert body["summary"]["completed_amount"] == 0.0, "汇总行不得计入已回款"
+        assert body["summary"]["completed_count"] == 0
+        assert body["summary"]["aging"] == {"60-90天": 15.55, "未到期": 10.0}
+        assert body["owners"] == ["张三"], "汇总行不得贡献负责人筛选项"
+
+    @pytest.mark.parametrize("blank", [None, "", "   ", "\t", "\u3000"])
+    def test_blank_counterparty_never_counted(self, client, db, blank):
+        create_test_user(db)
+        token = get_token_for_user(client)
+        _add_plan(db, counterparty=blank, amount=100.0, status="pending",
+                  plan_date=HANK_END, source="zxt-sync")
+
+        body = _receivables(client, token)
+
+        assert body["total"] == 0
+        assert body["summary"]["total_amount"] == 0.0
+        assert body["summary"]["pending_balance"] == 0.0
+        assert body["summary"]["overdue_count"] == 0
+        assert body["summary"]["aging"] == {}
+        assert body["owners"] == []
+
+    def test_hanke_before_ingest_is_empty_not_error(self, client, db):
+        """酣客未落库现状（仅 67 行汇总、无客户级行）：卡片/明细如实为空且不报错。"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+        for month in range(1, 4):
+            _add_plan(db, counterparty=None, amount=440.61, status="completed",
+                      plan_date=datetime(2021, month, 28), source="zxt-sync")
+
+        body = _receivables(client, token)
+
+        assert body["total"] == 0
+        assert body["data"] == []
+        assert body["summary"]["pending_balance"] == 0.0
+        assert body["summary"]["completed_amount"] == 0.0
+        assert body["owners"] == []
+
+    def test_customer_rows_only_response_matches_hanke_expectation(self, client, db):
+        """酣客落库后（18 行 hanke_ar，期末日 2026-07-31）催收页期望值：单档 60-90天。"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+        yuan = [155538, 125858, 120320, 100791, 66230, 44400, 27860, 18510,
+                10290, 10219, 10000, 8160, 6940, 5290, 3410, 2000, 530, 500]
+        assert sum(yuan) == 716846
+        for idx, amount in enumerate(yuan):
+            _add_plan(db, counterparty="客户%02d" % idx, amount=round(amount / 10000.0, 4),
+                      plan_date=HANK_END, source="hanke_ar")
+
+        body = _receivables(client, token)
+
+        assert body["total"] == 18
+        assert body["summary"]["total_amount"] == 71.68
+        assert body["summary"]["pending_balance"] == 71.68
+        assert body["summary"]["completed_amount"] == 0.0
+        assert body["summary"]["completed_count"] == 0
+        assert body["summary"]["overdue_count"] == 18
+        assert body["summary"]["overdue_amount"] == 71.68
+        assert body["summary"]["aging"] == {"60-90天": 71.68}
+
+
+class TestE2BaselineUnchanged:
+    """② 无汇总行账套（博海 e2 形态）逐位不变 —— 8 项基线数值"""
+
+    def test_customer_rows_all_kept_digitwise(self, client, db):
+        create_test_user(db)
+        token = get_token_for_user(client)
+        # 复刻 e2 形态：counterparty 全非空（NULL/空 = 0 行），含快照批次来源与已完成行
+        _add_plan(db, counterparty="XX学校", amount=1.26, paid_amount=1.26,
+                  status="completed", plan_date=datetime(2026, 4, 20),
+                  source="receivables_migrate")
+        _add_plan(db, counterparty="博海客户A", amount=100.0, plan_date=datetime(2026, 6, 30),
+                  source="bohai_ar_20260630")
+        _add_plan(db, counterparty="博海客户B", amount=100.72, plan_date=datetime(2026, 6, 1),
+                  source="bohai_ar")
+        _add_plan(db, counterparty="客户C", amount=194.5, plan_date=FUTURE,
+                  source="manual", owner="王婧")
+
+        body = _receivables(client, token)
+
+        assert body["total"] == 4, "无空 counterparty 时一行都不得被排除"
+        assert body["summary"]["total_amount"] == 396.48
+        assert body["summary"]["pending_balance"] == 395.22
+        assert body["summary"]["overdue_count"] == 2
+        assert body["summary"]["overdue_amount"] == 200.72
+        assert body["summary"]["completed_amount"] == 1.26
+        assert body["summary"]["completed_count"] == 1
+        assert body["summary"]["aging"] == {"90天以上": 200.72, "未到期": 194.5}
+        assert body["owners"] == ["王婧"]
+
+    def test_filter_is_row_shape_not_source_whitelist(self, client, db):
+        """判据是行形态：客户级行即便来源是 zxt-sync 也计入（source 不是白名单）。"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+        _add_plan(db, counterparty="智享通客户", amount=12.34, plan_date=FUTURE,
+                  source="zxt-sync")
+        _add_plan(db, counterparty=None, amount=999.0, plan_date=FUTURE, source="manual")
+
+        body = _receivables(client, token)
+
+        assert body["total"] == 1
+        assert body["data"][0]["counterparty"] == "智享通客户"
+        assert body["summary"]["total_amount"] == 12.34
+
+
+class TestHankeSnapshotRegistration:
+    """③ 快照批次登记（默认 skip ⇒ 不逐户生成伪逾期告警）"""
+
+    def test_hanke_ar_registered_and_default_skip(self, monkeypatch):
+        monkeypatch.delenv("CMA_CASH_SNAPSHOT_SOURCES", raising=False)
+        monkeypatch.delenv("CMA_CASH_SNAPSHOT_MODE", raising=False)
+        assert "hanke_ar" in BALANCE_SNAPSHOT_SOURCES
+        assert get_balance_snapshot_mode() == "skip"
+
+    def test_split_snapshot_classifies_hanke_ar(self, monkeypatch):
+        monkeypatch.delenv("CMA_CASH_SNAPSHOT_SOURCES", raising=False)
+
+        class _P:
+            def __init__(self, source):
+                self.source = source
+                self.plan_type = "receive"
+
+        rows = [_P("hanke_ar"), _P("manual"), _P("bohai_ar_20260630")]
+        business, snapshot = split_snapshot_plans(rows)
+
+        assert [p.source for p in business] == ["manual"]
+        assert sorted(p.source for p in snapshot) == ["bohai_ar_20260630", "hanke_ar"]
+
+    def _seed_overdue_hanke(self, db: Session, n: int = 3):
+        for idx in range(n):
+            _add_plan(db, counterparty="酣客客户%d" % idx, amount=10.0,
+                      plan_date=HANK_END, source="hanke_ar")
+
+    def test_default_skip_no_per_customer_overdue_alerts(self, client, db, monkeypatch):
+        """默认 skip：18 行形 hanke_ar 快照行不逐户生成伪逾期告警（告警计数等值判据的单元层代理）。"""
+        monkeypatch.delenv("CMA_CASH_SNAPSHOT_MODE", raising=False)
+        monkeypatch.delenv("CMA_CASH_SNAPSHOT_SOURCES", raising=False)
+        create_test_kpi(db)
+        self._seed_overdue_hanke(db, 3)
+
+        res = check_cash_alerts(db, entity_id=1)
+
+        assert res["balance_snapshot"]["mode"] == "skip"
+        assert res["balance_snapshot"]["skipped_rows"] == 3
+        assert "hanke_ar" in res["balance_snapshot"]["sources"]
+        assert [a for a in res["alerts"] if a["type"] == "overdue_receive"] == []
+        assert db.query(KPIAlert).filter(
+            KPIAlert.alert_type == "cash_plan",
+            KPIAlert.status.in_(["pending", "processing"]),
+        ).count() == 0
+
+    def test_per_customer_mode_control_does_alert(self, client, db, monkeypatch):
+        """负对照：显式切 per_customer（旧行为）时同 3 行会逐户告警 —— 证明上一条不是「没数据」。"""
+        monkeypatch.setenv("CMA_CASH_SNAPSHOT_MODE", "per_customer")
+        monkeypatch.delenv("CMA_CASH_SNAPSHOT_SOURCES", raising=False)
+        create_test_kpi(db)
+        self._seed_overdue_hanke(db, 3)
+
+        res = check_cash_alerts(db, entity_id=1)
+
+        assert res["balance_snapshot"]["mode"] == "per_customer"
+        assert len([a for a in res["alerts"] if a["type"] == "overdue_receive"]) == 3
+        assert db.query(KPIAlert).filter(
+            KPIAlert.alert_type == "cash_plan",
+            KPIAlert.status.in_(["pending", "processing"]),
+        ).count() == 3