feat(kpi): 归集侧接作废/撤单标记过滤（voucher_void_marks）

- kpi_value_collector: 所有消费 voucher_details 的归集路径统一排除已作废凭证 (NOT EXISTS voucher_void_marks, entity_id+voucher_no+voucher_date+revoked_at IS NULL), 与 DB 视图 v_voucher_details_active 定义一致；标记表不存在时自动跳过（兼容旧库） - 禁硬编码：凭证号不出现在代码/配置，撤单只需写标记表 - verify_cashflow_direct_caliber: 2026-08 从已知差异放行改为硬断言（8 期逐期一致） - tests/test_void_marker_filter_20260914.py: 标记表在/不在两分支 3 例
2026-09-14 18:44:39 +08:00
parent 4d918f5786
commit a8a82ff837
3 changed files with 234 additions and 3 deletions
@@ -31,7 +31,7 @@ if __name__ == "__main__":
    import os
    sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

-from sqlalchemy import and_, func, not_, or_
+from sqlalchemy import MetaData, Table, and_, exists, func, not_, or_
 from sqlalchemy.orm import aliased
 from app.database import get_session_local
 from app.models import (
@@ -67,6 +67,70 @@ def _field_expression(model, field: str, aggregate: str = "sum"):
    return func.sum(col)


+# ── 作废/撤单标记过滤（2026-09-14 接入；单一事实来源 = cma.voucher_void_marks）──────
+# 口径: 本模块**所有**消费 voucher_details 的归集路径统一排除「已作废/被更正」凭证。
+#   判定键 = (entity_id, voucher_no, voucher_date) 且 revoked_at IS NULL —— 与 DB 视图
+#   `cma.v_voucher_details_active` 的定义逐字一致（视图供探针/报表等其它消费方使用）。
+# 纪律: 禁硬编码凭证号；标记表是唯一数据源 —— 今后任何撤单只需写标记表，代码零改动。
+# 兼容: 标记表不存在（测试库/旧库）→ 跳过过滤，不阻断采集（告警一次）。
+_VOID_MARK_TABLE = "voucher_void_marks"
+_void_mark_table_ref = None      # 反射缓存: Table（存在）| False（不存在）
+_void_mark_warned = False
+
+
+def _void_mark_table(db):
+    """作废标记表（反射 + 缓存）；表不存在/不可读 → None。"""
+    global _void_mark_table_ref, _void_mark_warned
+    if _void_mark_table_ref is not None:
+        # 注意: Table 对象无真值（__bool__ 未定义），必须显式判 False，不能写 `ref or None`
+        return None if _void_mark_table_ref is False else _void_mark_table_ref
+    try:
+        # 走 Session 自身连接（而非 engine）: 测试库 StaticPool 下两者是同一 DBAPI 连接，
+        # 另开 engine 连接会在归还时回滚掉 Session 未提交的写入；此处保持同一事务。
+        bind = db.connection() if hasattr(db, "connection") else db.get_bind()
+        _void_mark_table_ref = Table(_VOID_MARK_TABLE, MetaData(), autoload_with=bind)
+        logger.info("作废标记过滤已接入: %s（撤销标记后自动重新计入）", _VOID_MARK_TABLE)
+    except Exception as e:  # noqa: BLE001 - 兼容无标记表的环境
+        _void_mark_table_ref = False
+        if not _void_mark_warned:
+            logger.warning("作废标记表 %s 不可用 → 本次不做作废过滤: %s", _VOID_MARK_TABLE, e)
+            _void_mark_warned = True
+    return None if _void_mark_table_ref is False else _void_mark_table_ref
+
+
+def _void_mark_table_exists(db) -> bool:
+    return _void_mark_table(db) is not None
+
+
+def _has_col(src, name: str) -> bool:
+    cols = getattr(src, "c", None)
+    return (name in cols) if cols is not None else hasattr(src, name)
+
+
+def _exclude_void(db, query, src):
+    """给凭证明细查询挂上 NOT EXISTS(voucher_void_marks) 过滤（本模块唯一注入点）。
+
+    src = VoucherDetail（或 aliased(VoucherDetail)）；标记表不存在 → 原样返回。
+    """
+    marks = _void_mark_table(db)
+    if marks is None:
+        return query
+    clause = ~exists().where(and_(
+        marks.c.entity_id == src.entity_id,
+        marks.c.voucher_no == src.voucher_no,
+        marks.c.voucher_date == src.voucher_date,
+        marks.c.revoked_at.is_(None),
+    ))
+    return query.filter(clause)
+
+
+def _apply_voucher_guard(db, query, src):
+    """凡指向凭证明细的查询一律过作废过滤（非凭证明细源表原样返回）。"""
+    if _has_col(src, "voucher_no") and _has_col(src, "voucher_date"):
+        return _exclude_void(db, query, src)
+    return query
+
+
 def collect_for_mapping(db, mapping: KPIValueSource, period: str, write_kpi: bool = True) -> Tuple[Optional[float], str]:
    """执行单条取数映射，返回 (采集值, 说明)。write_kpi=False 时为试跑模式（不写库）。"""
    table = mapping.source_table
@@ -111,6 +175,9 @@ def collect_for_mapping(db, mapping: KPIValueSource, period: str, write_kpi: boo
            return None, f"期间字段 {period_field} 不存在于表 {table}"
        q = q.filter(pf == period)

+    # 作废/撤单标记过滤（消费 voucher_details 的统一守卫）
+    q = _apply_voucher_guard(db, q, model)
+
    # 过滤规则: subject_code / direction / plan_type / carry_forward
    subject_code = filter_rule.get("subject_code")
    if subject_code and hasattr(model, "subject_code"):
@@ -125,6 +192,7 @@ def collect_for_mapping(db, mapping: KPIValueSource, period: str, write_kpi: boo
                q = q.filter(func.date_format(model.voucher_date, "%Y-%m") == period)
            else:
                q = q.filter(getattr(model, period_field) == period)
+            q = _apply_voucher_guard(db, q, model)
            field = "credit_amount"
        elif direction == "debit" and hasattr(model, "debit_amount"):
            q = db.query(_field_expression(model, "debit_amount", aggregate))
@@ -133,6 +201,7 @@ def collect_for_mapping(db, mapping: KPIValueSource, period: str, write_kpi: boo
                q = q.filter(func.date_format(model.voucher_date, "%Y-%m") == period)
            else:
                q = q.filter(getattr(model, period_field) == period)
+            q = _apply_voucher_guard(db, q, model)
            field = "debit_amount"
    plan_type = filter_rule.get("plan_type")
    if plan_type and hasattr(model, "plan_type"):
@@ -239,6 +308,7 @@ def _subject_amount(db, entity_id: int, pattern: str, period: str, measure: str)
        q = q.filter(VoucherDetail.subject_code == pattern)
    if period:
        q = q.filter(VoucherDetail.period == period)
+    q = _exclude_void(db, q, VoucherDetail)      # 作废/撤单标记过滤
    credit, debit, rows = q.first()
    credit, debit = float(credit or 0), float(debit or 0)
    if measure == "debit":
@@ -319,6 +389,7 @@ def _cashflow_voucher_subquery(db, entity_id: int, period: Optional[str], cash_s
    if period:
        q = q.filter(VoucherDetail.period == period)
    q = q.filter(or_(*[VoucherDetail.subject_code.like(p) for p in cash_subjects]))
+    q = _exclude_void(db, q, VoucherDetail)      # 作废/撤单标记过滤（凭证键集合）
    voucher_cnt = q.distinct().count()
    return q.distinct().subquery(), voucher_cnt

@@ -5,7 +5,8 @@

 判据（全部只读，write_kpi=False）:
  1. 生产同一代码路径试算 2026-01..08，与现网 zxt-sync 权威值对账（容差 0.001）
-     —— 2026-08 因上游已撤单凭证仍在库内（智享通快照不含）按已知差异放行（WARN，不算 FAIL）
+     —— 2026-08 差异已随「作废/撤单标记过滤」接入归集器（2026-09-14）消除：
+        归集侧统一按 cma.voucher_void_marks 排除已作废凭证，8 期逐期一致（无放行期）
  2. 恒等式: 经营净额 + 筹资净额 + 投资净额 = 现金总净额（每期 |Δ| ≤ 0.001）
  3. 负例对照: 若按「凭证级整单分类」（错误做法），2026-06 应为 453,697.63 元
     （行级 = 434,997.63 元，差 +18,700 = 凭证 31226 同单内的银行手续费）→ 证明必须行级拆分
@@ -25,7 +26,7 @@ from scripts.kpi_value_collector import collect_for_mapping
 ENTITY = 1
 PERIODS = ["2026-0%d" % i for i in range(1, 9)]
 TOL = 0.001
-KNOWN_DIFF_PERIODS = {"2026-08"}          # 上游已撤单凭证未标记 → 已知差异（挂账，非回归）
+KNOWN_DIFF_PERIODS = set()   # 2026-09-14 起为空集：作废标记过滤已接入归集器，8 期逐期硬断言
 RULE = {
    "calc": "cashflow_direct", "measure": "net", "version": 1, "bucket": "operating",
    "cash_subjects": ["1001%", "1002%", "1012%"],
@@ -0,0 +1,159 @@
+"""作废/撤单标记过滤（kpi_value_collector.py）单测 — 2026-09-14
+
+任务: dama-void-marker-collector-fullstack-20260914（采集侧接作废标记过滤）
+
+覆盖两条分支：
+  1. 标记表缺失（SQLite 测试库 / 旧库）→ 过滤自动跳过，归集结果与无标记时一致（不阻断采集）；
+  2. 标记表存在 → 被标记凭证从归集口径中排除，且过滤由表内容驱动
+     （换一条标记即换结果 / 软撤回即恢复 → 无硬编码凭证号）。
+
+口径说明: 生产 MySQL 上 voucher_details.voucher_date 与 voucher_void_marks.voucher_date
+均为 DATE（日期对日期比较）；SQLite 无 DATE 亲和类型，ORM 侧按 DateTime 存储，
+故本测试的标记表 DDL 用 DATETIME + 统一传 datetime，保证两侧字面量同形。
+生产侧真实过滤效果由 `scripts/verify_cashflow_direct_caliber.py`（MySQL 8 期硬断言）覆盖。
+"""
+import datetime
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from sqlalchemy import String, bindparam, text  # noqa: E402
+
+from app.models import VoucherDetail  # noqa: E402
+from scripts import kpi_value_collector as C  # noqa: E402
+
+D = datetime.datetime(2026, 8, 31)          # 统一日期对象（两侧同形）
+
+MARKS_DDL = """
+CREATE TABLE IF NOT EXISTS voucher_void_marks (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    entity_id INTEGER NOT NULL,
+    voucher_no VARCHAR(32) NOT NULL,
+    voucher_date DATETIME NOT NULL,
+    mark_type VARCHAR(16) NOT NULL DEFAULT 'void',
+    replaced_by VARCHAR(32),
+    reason VARCHAR(500) NOT NULL DEFAULT '',
+    source_ref VARCHAR(255),
+    marked_by VARCHAR(64) NOT NULL,
+    marked_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+    revoked_at DATETIME,
+    revoked_by VARCHAR(64)
+)
+"""
+
+
+def _reset_cache():
+    C._void_mark_table_ref = None
+    C._void_mark_warned = False
+
+
+def _ensure_marks_table(db):
+    """建同结构标记表并清空（本表不在 Base.metadata，setup_db 不负责它）。"""
+    db.execute(text(MARKS_DDL))
+    db.execute(text("DELETE FROM voucher_void_marks"))
+    db.flush()
+
+
+def _drop_marks_table(db):
+    db.execute(text("DROP TABLE IF EXISTS voucher_void_marks"))
+
+
+# 标记行的 voucher_date 取「凭证行在库内的实际字面量」（SQLite 无 DATE 亲和类型，
+# 只有保证两侧同形才能比较；生产 MySQL 是 DATE 对 DATE，见模块头说明）。
+_INS_MARK = text("""INSERT INTO voucher_void_marks
+    (entity_id, voucher_no, voucher_date, marked_by, reason, revoked_at)
+    VALUES (1, :no, :d, 'test', '单测', :rev)""").bindparams(
+    bindparam("d", type_=String), bindparam("rev", type_=String))
+
+
+def _mark(db, no, revoked_at=None):
+    ts = db.execute(text("SELECT voucher_date FROM voucher_details "
+                         "WHERE entity_id=1 AND voucher_no=:n LIMIT 1"),
+                    {"n": str(no)}).scalar()
+    assert ts is not None, f"凭证 {no} 未入库"
+    db.execute(_INS_MARK, {"no": str(no), "d": str(ts), "rev": revoked_at})
+    db.flush()
+
+
+def _add_voucher(db, no, code, credit=0.0, debit=0.0, period="2026-08"):
+    db.add(VoucherDetail(entity_id=1, voucher_no=str(no), voucher_date=D,
+                         subject_code=code, subject_name="测试科目",
+                         credit_amount=credit, debit_amount=debit,
+                         summary="单测", carry_forward=0, period=period))
+    db.flush()
+
+
+def test_filter_skipped_when_marks_table_missing(db):
+    """标记表不存在 → 不做过滤（兼容旧库/测试库），归集照常出数。"""
+    _reset_cache()
+    try:
+        assert C._void_mark_table_exists(db) is False
+        _add_voucher(db, 90001, "5001", credit=120000.0)
+        amount, rows = C._subject_amount(db, 1, "5001", "2026-08", "credit")
+        assert rows == 1 and abs(amount - 120000.0) < 1e-6
+    finally:
+        _reset_cache()
+
+
+def test_marked_voucher_excluded(db):
+    """标记表存在 → 被标记凭证整体排除；软撤回后恢复计入（表是唯一数据源）。"""
+    _reset_cache()
+    _ensure_marks_table(db)
+    try:
+        _add_voucher(db, 90011, "5001", credit=100000.0)
+        _add_voucher(db, 90012, "5001", credit=20000.0)
+        _add_voucher(db, 90012, "100201", debit=20000.0)
+
+        # 未标记：两张凭证都参与
+        amount, rows = C._subject_amount(db, 1, "5001", "2026-08", "credit")
+        assert rows == 2 and abs(amount - 120000.0) < 1e-6, (amount, rows)
+
+        # 标记 90011 作废 → 只剩 90012 的 20000
+        _mark(db, 90011)
+        amount2, rows2 = C._subject_amount(db, 1, "5001", "2026-08", "credit")
+        assert rows2 == 1 and abs(amount2 - 20000.0) < 1e-6, (amount2, rows2)
+
+        # 软撤回 revoked_at 非空 → 该凭证重新计入
+        db.execute(text("UPDATE voucher_void_marks SET revoked_at = :d WHERE voucher_no = '90011'"),
+                   {"d": str(D)})
+        db.flush()
+        amount3, rows3 = C._subject_amount(db, 1, "5001", "2026-08", "credit")
+        assert rows3 == 2 and abs(amount3 - 120000.0) < 1e-6, (amount3, rows3)
+
+        # 换一条标记（90012）→ 结果随表变化，证明无硬编码
+        _mark(db, 90012)
+        amount4, rows4 = C._subject_amount(db, 1, "5001", "2026-08", "credit")
+        assert rows4 == 1 and abs(amount4 - 100000.0) < 1e-6, (amount4, rows4)
+    finally:
+        db.rollback()
+        _drop_marks_table(db)
+        _reset_cache()
+
+
+def test_cashflow_path_excludes_marked_voucher(db):
+    """现金流直接法路径同样过过滤：被标记凭证从「现金凭证集合」中剔除。"""
+    _reset_cache()
+    _ensure_marks_table(db)
+    try:
+        # 现金凭证 A：收 30000（经营对手方）；现金凭证 B：收 5000
+        _add_voucher(db, 90021, "100201", debit=30000.0)
+        _add_voucher(db, 90021, "5001", credit=30000.0)
+        _add_voucher(db, 90022, "100201", debit=5000.0)
+        _add_voucher(db, 90022, "5001", credit=5000.0)
+
+        mapping = type("M", (), {"filter_rule": {"calc": "cashflow_direct", "bucket": "operating",
+                                                 "cash_subjects": ["1001%", "1002%"]},
+                                 "unit_conversion": 1, "entity_id": 1})()
+        v_all, msg_all = C.collect_cashflow_direct(db, mapping, "2026-08", write_kpi=False)
+        assert abs(v_all - 35000.0) < 1e-6, (v_all, msg_all)
+        assert "现金凭证2张" in msg_all, msg_all
+
+        _mark(db, 90021)
+        v_f, msg_f = C.collect_cashflow_direct(db, mapping, "2026-08", write_kpi=False)
+        assert abs(v_f - 5000.0) < 1e-6, (v_f, msg_f)
+        assert "现金凭证1张" in msg_f, msg_f
+    finally:
+        db.rollback()
+        _drop_marks_table(db)
+        _reset_cache()