diff --git a/backend/scripts/kpi_value_collector.py b/backend/scripts/kpi_value_collector.py index 954c4759..c6050c58 100644 --- a/backend/scripts/kpi_value_collector.py +++ b/backend/scripts/kpi_value_collector.py @@ -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 diff --git a/backend/scripts/verify_cashflow_direct_caliber.py b/backend/scripts/verify_cashflow_direct_caliber.py index 43d44a4f..f777d43a 100644 --- a/backend/scripts/verify_cashflow_direct_caliber.py +++ b/backend/scripts/verify_cashflow_direct_caliber.py @@ -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%"], diff --git a/backend/tests/test_void_marker_filter_20260914.py b/backend/tests/test_void_marker_filter_20260914.py new file mode 100644 index 00000000..5a4f2bbc --- /dev/null +++ b/backend/tests/test_void_marker_filter_20260914.py @@ -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()