cma-management/backend/scripts/import_cashflow_statements.py

#!/usr/bin/env python3
"""现金流量表（三大活动净额）入库器 —— 酣客(智享通) 现金流量统计表 → cashflow_statements

背景（任务 finance-20260916-cma-health-gaps-3items · 项1）：
  cashflow_statements 表 0 行 = **无导入链路**（此前只有 API 读/写端点，无任何 loader；
  上游 智享通 数据虽已落盘 /root/WB_WS/_data/zxt-hanke/cashflow/<period>/，但从未入库）。

数据源（权威口径 = 智享通「现金流量统计表」项目级汇总，单位：元）：
  /root/WB_WS/_data/zxt-hanke/cashflow/<YYYY.MM>/cashflow_stat.csv
     列：分类,编码,项目名称,方向,金额     分类∈{经营活动,投资活动,筹资活动}
  兼容 2026.08 的另存名 cashflow_stat_<YYYY.MM>.csv

口径：
  - 经营/投资/筹资 净额 = Σ(流入) − Σ(流出)，单位 元 → 万元（÷10000，保留 2 位）
  - **NA（无发生额）标注（本版本新增，裁定 finance-20260916-cma-caliber-ruling §2.1）**：
    某活动**无任何项目行**时净额记 0.00（三净额列为 NOT NULL，表本身表达不了 NA），
    同时在 `remark` 写入**固定短语**「<活动>活动本期无发生额（无项目行）」，
    供判定层用 `parse_na_activities(remark)` 取回 → **仅作标注与机读入口**；
    口径裁定（2026-09-17）已定：该维按 **0 如实参与判定**（八类判定的**二维映射**，
    经营(±)×筹资(±) 出 2 个候选类型），**不再**因 NA/0 整体判「未判定」。
  - source='official'（官方报表口径）；期初/期末现金 留空（主表采集缺陷，见报告 §项1-②）
  - 同 (entity_id, period) 重复执行 = 覆盖更新（幂等）

用法：
  python3 scripts/import_cashflow_statements.py                       # dry-run（默认，只算不写）
  python3 scripts/import_cashflow_statements.py --period 2026-07      # 单期 dry-run
  python3 scripts/import_cashflow_statements.py --self-test           # NA 标注逻辑自测（零库连接）
  python3 scripts/import_cashflow_statements.py --apply               # 落库（需放行）
"""
import argparse
import csv
import glob
import os
import re
import sys

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from app.database import get_session_local
from app.models import CashFlowStructureValue
from app.api.cashflow_structure import classify_enterprise

ZKT_BASE = "/root/WB_WS/_data/zxt-hanke/cashflow"
CLASSES = ("经营活动", "投资活动", "筹资活动")

#: NA 固定短语模板 —— 判定层与量纲守卫共用，勿改字面量（裁定 §2.1 指定）
NA_PHRASE_TMPL = "{cls}本期无发生额（无项目行）"

#: 从 remark 反解 NA 活动的正则（与 NA_PHRASE_TMPL 严格同源）
_NA_RE = re.compile(r"(经营活动|投资活动|筹资活动)本期无发生额（无项目行）")


def parse_na_activities(remark):
    """从 cashflow_statements.remark 取回 NA（无发生额）活动集合。

    判定层入口（**只作标注/机读**）：NA 维度的 0.00 是"表表达不了 NA"的占位，
    不等于"收付相抵的 0"；但按 2026-09-17 口径裁定，两者**都按 0 参与判定**
    （投资活动=0 → 二维映射出 2 个候选类型），NA 短语仅用于页面/报告显式标注。
    `na_flags(remark)` 返回 {活动: True/False} 供逐维判断与文案生成。
    """
    if not remark:
        return set()
    return set(_NA_RE.findall(str(remark)))


def na_flags(remark):
    """逐维 NA 标志：{'经营活动': bool, '投资活动': bool, '筹资活动': bool}"""
    na = parse_na_activities(remark)
    return {c: (c in na) for c in CLASSES}


def _read_stat_file(path):
    """读项目级汇总 CSV → [(分类, 编码, 项目名称, 方向, 金额元)]"""
    rows = []
    with open(path, encoding="utf-8-sig") as fh:
        for r in csv.reader(fh):
            if len(r) < 5 or not r[0].strip() or r[1].strip() == "编码":
                continue
            try:
                amt = float(r[4].strip().replace(",", ""))
            except ValueError:
                continue
            rows.append((r[0].strip(), r[1].strip(), r[2].strip(), r[3].strip(), amt))
    return rows


def _nets(rows):
    agg = {c: {"in": 0.0, "out": 0.0} for c in CLASSES}
    for cls, _code, _name, direction, amt in rows:
        if cls not in agg:
            continue
        agg[cls]["in" if "流入" in direction else "out"] += amt
    return {c: round(v["in"] - v["out"], 2) for c, v in agg.items()}, agg


def collect(period_filter=None):
    """扫描盘上各期 → [payload dict]（万元，未写库）"""
    payloads = []
    for d in sorted(glob.glob(os.path.join(ZKT_BASE, "2026.*"))):
        period = os.path.basename(d).replace(".", "-")
        if period_filter and period != period_filter:
            continue
        cands = sorted(glob.glob(os.path.join(d, "cashflow_stat*.csv")))
        # 优先取当期明细名（cashflow_stat_<period>.csv），其次通用名
        src = None
        for c in cands:
            if os.path.basename(c) != "cashflow_stat.csv":
                src = c
        src = src or (cands[0] if cands else None)
        if not src:
            payloads.append({"period": period, "status": "SKIP_no_source_file", "dir": d})
            continue
        rows = _read_stat_file(src)
        if not rows:
            payloads.append({"period": period, "status": "SKIP_empty",
                             "src": src, "rows": 0})
            continue
        nets_yuan, agg = _nets(rows)
        missing = [c for c in CLASSES if not agg[c]["in"] and not agg[c]["out"]]
        na_phrases = [NA_PHRASE_TMPL.format(cls=c) for c in missing]
        head = "源: 智享通现金流量统计表 %s（单位元→万元 /10000）；项目行 %d 条" % (
            os.path.basename(src), len(rows))
        remark = (head + "；NA(无发生额，按0参与判定): " + "、".join(na_phrases)) if na_phrases \
            else (head + "；三类活动均有项目行（无 NA）")
        payloads.append({
            "period": period,
            "status": "READY",
            "src": src,
            "rows": len(rows),
            "nets_yuan": nets_yuan,
            "nets_wan": {k: round(v / 10000.0, 2) for k, v in nets_yuan.items()},
            "missing_classes": missing,
            "na_classes": missing,
            "na_phrases": na_phrases,
            "na_flags": {c: (c in missing) for c in CLASSES},
            "remark": remark,
        })
    return payloads


def _self_test():
    """NA 标注逻辑自测（零库连接、零写盘；只读盘上 CSV）——回执证据用。"""
    passed = 0
    # 1~3) 短语 → 标志 反解
    for missing, expect in (
        (["投资活动"], {"经营活动": False, "投资活动": True, "筹资活动": False}),
        (["投资活动", "筹资活动"], {"经营活动": False, "投资活动": True, "筹资活动": True}),
        ([], {"经营活动": False, "投资活动": False, "筹资活动": False}),
    ):
        phrases = [NA_PHRASE_TMPL.format(cls=c) for c in missing]
        remark = ("源: X；项目行 3 条；NA(无发生额，按0参与判定): " + "、".join(phrases)) if phrases \
            else "源: X；项目行 3 条；三类活动均有项目行（无 NA）"
        assert na_flags(remark) == expect, (missing, remark)
        assert parse_na_activities(remark) == set(missing)
        passed += 1
    # 4) 空值安全
    assert na_flags(None) == {c: False for c in CLASSES}
    assert parse_na_activities("") == set()
    passed += 1
    # 5) 真实数据源（盘上 8 期）：投资活动应全为 NA
    pay = collect()
    ready = [p for p in pay if p["status"] == "READY"]
    assert ready, "无可入库期（数据源缺失）"
    all_inv_na = all("投资活动" in p["na_classes"] for p in ready)
    passed += 1
    print("=== NA 标注 self-test: %d/5 PASS | 期数=%d | 投资活动全NA=%s ==="
          % (passed, len(ready), all_inv_na))
    for p in ready:
        print("  %-8s NA=%s" % (p["period"], "、".join(p["na_classes"]) or "无"))
        print("           remark: %s" % p["remark"])
    assert all_inv_na, "酣客 8 期投资活动应全为 NA（裁定 §2.1 实证）"
    return 0


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--entity-id", type=int, default=1,
                    help="账套：1=酣客(白酒, 智享通通道) 2=博海(IT, 无凭证明细)")
    ap.add_argument("--period", default=None, help="仅处理某期 YYYY-MM")
    ap.add_argument("--self-test", action="store_true",
                    help="NA 标注逻辑自测（零库连接：只读盘上 CSV）")
    ap.add_argument("--apply", action="store_true", help="真正写库（缺省=dry-run 只算不写）")
    args = ap.parse_args()

    if args.self_test:
        return _self_test()

    payloads = collect(args.period)
    if not payloads:
        print("未发现任何期数据源（%s/2026.*）" % ZKT_BASE)
        return 2

    mode = "APPLY" if args.apply else "DRY-RUN"
    print("=== 现金流量表入库器 %s | entity_id=%d ===" % (mode, args.entity_id))
    db = get_session_local()() if args.apply else None
    written = 0
    try:
        for p in payloads:
            if p["status"] != "READY":
                print("  %-8s %s  (%s)" % (p["period"], p["status"], p.get("src") or p.get("dir")))
                continue
            n = p["nets_wan"]
            cls = classify_enterprise(n["经营活动"], n["投资活动"], n["筹资活动"])
            # 裁定1（2026-09-17）：投资活动=0 → dimension="2D"（type_no=None，候选 2 个）
            # ⇒ 打印层不得假定 classified ⇒ type_no 非空（原实现此处 TypeError 崩掉整轮 dry-run）
            if cls.get("classified"):
                if cls.get("dimension") == "2D":
                    verdict = "二维判定(%s) → 候选 %s" % (
                        cls.get("dimension_label") or "投资活动=0",
                        " / ".join("类型%d %s" % (c["type_no"], c["name"])
                                   for c in cls.get("candidates") or []))
                else:
                    verdict = "类型%d %s" % (cls["type_no"], cls["name"])
            else:
                verdict = "未判定: " + cls.get("reason", "")[:28]
            print("  %-8s O=%8.2f I=%8.2f F=%8.2f 万 | 判定=%s | 源=%s"
                  % (p["period"], n["经营活动"], n["投资活动"], n["筹资活动"],
                     verdict, os.path.basename(p["src"])))
            print("           NA(无发生额)=%s；投资栏标注=%s" % (
                "、".join(p["na_classes"]) or "无", cls.get("investing_note") or "-"))
            if not args.apply:
                continue
            rec = db.query(CashFlowStructureValue).filter(
                CashFlowStructureValue.entity_id == args.entity_id,
                CashFlowStructureValue.period == p["period"],
            ).first()
            created = rec is None
            if rec is None:
                rec = CashFlowStructureValue(entity_id=args.entity_id, period=p["period"])
                db.add(rec)
            rec.operating_net = n["经营活动"]
            rec.investing_net = n["投资活动"]
            rec.financing_net = n["筹资活动"]
            rec.source = "official"
            rec.remark = p["remark"]
            db.flush()
            written += 1
            print("      → %s (id=%s)" % ("入库" if created else "更新", rec.id))
        if args.apply:
            db.commit()
            print("=== 已提交 %d 期 ===" % written)
        else:
            print("=== dry-run 结束：未写库（%d 期可入库，加 --apply 落库）===" % len(
                [p for p in payloads if p["status"] == "READY"]))
    finally:
        if db is not None:
            db.close()
    return 0


if __name__ == "__main__":
    sys.exit(main())