Hermes CI Fix
d13bf1b6e4
ci/woodpecker/push/main Pipeline failed
- import_cashflow_statements.py 新增 _read_cash_balances(): 读 <YYYY.MM>/kmyeb.csv 货币资金(1001/1002/1012) 期初借-贷 / 期末借-贷, ÷10000 留 2 位 → begin_cash/end_cash - 仅 entity_id=1(智享通通道=酣客) 生效, 其它账套保持 NULL; 源缺失/科目缺失保持 NULL 不写 0 - 既有三大活动净额与 remark NA 固定短语逻辑未动; dry-run 默认/--period/幂等语义不变 - --self-test 扩至 9/9: 新增 kmyeb缺失/科目缺失/借贷为负 三类边界 + 真实源自洽与跨期连续
385 lines
19 KiB
Python
385 lines
19 KiB
Python
#!/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
|
||
|
||
期初/期末现金数据源（本版本新增，任务 project-20260923-cashflow-beginend-cash-backfill）：
|
||
/root/WB_WS/_data/zxt-hanke/<YYYY.MM>/kmyeb.csv（科目余额表，只读）
|
||
取 科目编码 ∈ {1001 库存现金, 1002 银行存款, 1012 其他货币资金}
|
||
期初 = Σ(期初借 − 期初贷)；期末 = Σ(期末借 − 期末贷)；元 → 万元（÷10000，2 位）
|
||
依据：科目余额表货币资金逐期变动 == 库内三大活动净额合计（8/8 期吻合 ±0.01，
|
||
2026-09-23 交叉验证）；上游「现金流量表主表」（ReportDesign 1405）金额实测全 0 不可用
|
||
⇒ 主表口径（历史 §项1-② 缺陷）不改，改走科目余额表货币资金。
|
||
该源属**智享通通道 = 酣客账套**（KMYEB_ENTITY_ID），故仅对 entity_id=1 生效；
|
||
其它账套保持 NULL（不得把酣客货币资金写到别家账套）。
|
||
|
||
口径：
|
||
- 经营/投资/筹资 净额 = Σ(流入) − Σ(流出)，单位 元 → 万元（÷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'（官方报表口径）
|
||
- **期初/期末现金**（本版本新增）：取科目余额表货币资金（1001/1002/1012）→ 万元 2 位，
|
||
源文件/科目缺失时保持 NULL（不写 0 造数）；仅 entity_id=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 # 标注+期初期末自测（零库连接；
|
||
# 边界用例只在系统临时目录造样，不碰生产源）
|
||
python3 scripts/import_cashflow_statements.py --apply # 落库（需放行）
|
||
"""
|
||
import argparse
|
||
import csv
|
||
import glob
|
||
import os
|
||
import re
|
||
import sys
|
||
import tempfile
|
||
|
||
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 = ("经营活动", "投资活动", "筹资活动")
|
||
|
||
#: 科目余额表（kmyeb.csv）根目录 —— 与 cashflow/ 同级，期初/期末现金来源
|
||
KMYEB_BASE = "/root/WB_WS/_data/zxt-hanke"
|
||
#: 货币资金科目：库存现金 / 银行存款 / 其他货币资金
|
||
CASH_CODES = ("1001", "1002", "1012")
|
||
#: kmyeb 属智享通通道 = 酣客账套；其它账套不得套用（防跨账套写脏）
|
||
KMYEB_ENTITY_ID = 1
|
||
|
||
#: 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 _read_cash_balances(period, base=KMYEB_BASE):
|
||
"""读科目余额表货币资金 → (begin_wan, end_wan, meta)。
|
||
|
||
期初 = Σ(期初借 − 期初贷)、期末 = Σ(期末借 − 期末贷)，科目 ∈ CASH_CODES，元 → 万元 2 位。
|
||
借方−贷方为**有符号**口径：贷方大于借方即为负（红字/透支/方向反），**不得取绝对值或清零**。
|
||
源文件缺失 / 无任一货币资金科目行 ⇒ 返回 (None, None, meta)，由调用方保持 NULL（不写 0 造数）。
|
||
"""
|
||
day_dir = os.path.join(base, period.replace("-", "."))
|
||
src = os.path.join(day_dir, "kmyeb.csv")
|
||
if not os.path.exists(src):
|
||
return None, None, {"status": "SKIP_no_kmyeb", "src": src, "codes": []}
|
||
beg = 0.0
|
||
end = 0.0
|
||
hits = []
|
||
with open(src, encoding="utf-8-sig") as fh:
|
||
for r in csv.DictReader(fh):
|
||
code = (r.get("科目编码") or "").strip()
|
||
if code not in CASH_CODES:
|
||
continue
|
||
try:
|
||
b = float((r.get("期初借") or "0").strip() or 0) - float((r.get("期初贷") or "0").strip() or 0)
|
||
e = float((r.get("期末借") or "0").strip() or 0) - float((r.get("期末贷") or "0").strip() or 0)
|
||
except ValueError:
|
||
continue
|
||
hits.append(code)
|
||
beg += b
|
||
end += e
|
||
if not hits:
|
||
return None, None, {"status": "SKIP_no_cash_subject", "src": src, "codes": []}
|
||
return (round(beg / 10000.0, 2), round(end / 10000.0, 2),
|
||
{"status": "READY", "src": src, "codes": hits})
|
||
|
||
|
||
def _write_kmyeb(root, period, rows):
|
||
"""自测专用：在临时目录造 kmyeb.csv 样件（表头与生产源同构）。"""
|
||
d = os.path.join(root, period.replace("-", "."))
|
||
os.makedirs(d, exist_ok=True)
|
||
p = os.path.join(d, "kmyeb.csv")
|
||
with open(p, "w", encoding="utf-8-sig", newline="") as fh:
|
||
w = csv.writer(fh)
|
||
w.writerow(["科目编码", "科目名称", "期初借", "期初贷", "本期借", "本期贷",
|
||
"本年借", "本年贷", "当前借", "当前贷", "期末借", "期末贷"])
|
||
w.writerows(rows)
|
||
return p
|
||
|
||
|
||
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）")
|
||
cash_begin, cash_end, cash_meta = _read_cash_balances(period)
|
||
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,
|
||
"begin_cash": cash_begin,
|
||
"end_cash": cash_end,
|
||
"cash_meta": cash_meta,
|
||
})
|
||
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
|
||
assert all_inv_na, "酣客 8 期投资活动应全为 NA（裁定 §2.1 实证）"
|
||
|
||
# 6~8) 期初/期末现金取数边界（临时目录造样，零生产源写）
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
# 6) kmyeb 缺失 → 保持 NULL（不得写 0）
|
||
b6, e6, m6 = _read_cash_balances("2026-01", base=tmp)
|
||
assert (b6, e6) == (None, None) and m6["status"] == "SKIP_no_kmyeb", (b6, e6, m6)
|
||
passed += 1
|
||
# 7) 文件在但无货币资金科目（非现金科目不得混入）
|
||
_write_kmyeb(tmp, "2026-01", [
|
||
["1122", "应收账款", "999", "0", "0", "0", "0", "0", "0", "0", "999", "0"]])
|
||
b7, e7, m7 = _read_cash_balances("2026-01", base=tmp)
|
||
assert (b7, e7) == (None, None) and m7["status"] == "SKIP_no_cash_subject", (b7, e7, m7)
|
||
passed += 1
|
||
# 8) 借贷为负（贷>借）：负号必须保留，不得 abs/清零；非现金科目应被忽略
|
||
_write_kmyeb(tmp, "2026-02", [
|
||
["1001", "库存现金", "0", "5000", "0", "0", "0", "0", "0", "0", "0", "12000"],
|
||
["1002", "银行存款", "20000", "0", "0", "0", "0", "0", "0", "0", "30000", "0"],
|
||
["1122", "应收账款", "999", "0", "0", "0", "0", "0", "0", "0", "999", "0"]])
|
||
b8, e8, m8 = _read_cash_balances("2026-02", base=tmp)
|
||
assert (b8, e8) == (1.5, 1.8), (b8, e8)
|
||
assert m8["status"] == "READY" and sorted(m8["codes"]) == ["1001", "1002"], m8
|
||
_write_kmyeb(tmp, "2026-03", [
|
||
["1001", "库存现金", "0", "12000", "0", "0", "0", "0", "0", "0", "0", "12000"]])
|
||
b8b, e8b, _m = _read_cash_balances("2026-03", base=tmp)
|
||
assert (b8b, e8b) == (-1.2, -1.2), (b8b, e8b)
|
||
passed += 1
|
||
|
||
# 9) 真实源 8 期：期初/期末齐全 + 与三大净额自洽（期末−期初 == Σ净额 ±0.02）+ 跨期连续（±0.01）
|
||
with_cash = [p for p in ready
|
||
if p["begin_cash"] is not None and p["end_cash"] is not None]
|
||
assert len(with_cash) == len(ready), "部分期未取到货币资金：%d/%d" % (len(with_cash), len(ready))
|
||
for p in ready:
|
||
s = round(sum(p["nets_wan"][c] for c in CLASSES), 2)
|
||
d = round(p["end_cash"] - p["begin_cash"], 2)
|
||
assert abs(d - s) <= 0.02, ("期末−期初≠净额合计", p["period"], d, s)
|
||
prev_end = None
|
||
for p in ready:
|
||
if prev_end is not None:
|
||
assert abs(p["begin_cash"] - prev_end) <= 0.01, (
|
||
"跨期不连续", p["period"], p["begin_cash"], prev_end)
|
||
prev_end = p["end_cash"]
|
||
passed += 1
|
||
|
||
print("=== self-test: %d/9 PASS | 期数=%d | 投资活动全NA=%s | 期初/期末齐全=%d/%d ==="
|
||
% (passed, len(ready), all_inv_na, len(with_cash), len(ready)))
|
||
for p in ready:
|
||
print(" %-8s NA=%s | 期初=%.2f 期末=%.2f 万（Δ=%.2f / 净额合计=%.2f）"
|
||
% (p["period"], "、".join(p["na_classes"]) or "无", p["begin_cash"], p["end_cash"],
|
||
round(p["end_cash"] - p["begin_cash"], 2),
|
||
round(sum(p["nets_wan"][c] for c in CLASSES), 2)))
|
||
print(" remark: %s" % p["remark"])
|
||
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 "-"))
|
||
cash_applicable = (args.entity_id == KMYEB_ENTITY_ID)
|
||
has_cash = p["begin_cash"] is not None and p["end_cash"] is not None
|
||
print(" 期初/期末现金 = %s / %s 万（源 %s）%s" % (
|
||
"NULL" if p["begin_cash"] is None else "%.2f" % p["begin_cash"],
|
||
"NULL" if p["end_cash"] is None else "%.2f" % p["end_cash"],
|
||
os.path.basename(p["cash_meta"]["src"]),
|
||
"" if cash_applicable else
|
||
"；账套 %d 非智享通通道 → 不写期初/期末" % args.entity_id))
|
||
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"]
|
||
if cash_applicable and has_cash:
|
||
# 期初/期末现金：科目余额表货币资金（已 ÷10000 留 2 位）
|
||
rec.begin_cash = p["begin_cash"]
|
||
rec.end_cash = p["end_cash"]
|
||
db.flush()
|
||
written += 1
|
||
print(" → %s (id=%s) 期初/期末=%s" % (
|
||
"入库" if created else "更新", rec.id,
|
||
"NULL（源缺失，保持原值）" if not (cash_applicable and has_cash)
|
||
else "%.2f / %.2f 万" % (rec.begin_cash, rec.end_cash)))
|
||
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())
|