feat(cash): 智享通凭证JSON直传导入端点 vouchers-json (zxt-cma-import-v1)
- 新端点 POST /api/cma/cash/import/vouchers-json (X-BOT-KEY鉴权, 独立router避开require_role)
- 兼容三种body: 裸rows数组 / {rows} / {datasets[0].rows} 整包
- 凭证级借贷平衡(替代批级, 已知单边30694记异常拒收), 结转行summary识别
- 日期从summary前缀解析, 无日期行pending拒收(不冒充文件时间)
- 科目解析: 行内subject → DB zxt_account_dict → body subject_map, 未命中拒收
- 幂等: 内容指纹batch + import_logs去重 + 行级业务键防重复
- entity默认1(陕西酣客), active校验; 不触发现金流联动(凭证未证实权威)
- 验收脚本 scripts/verify_zxt_vouchers_json.py 12/12 PASS (SQLite本地)
This commit is contained in:
Hermes CI Fix
@@ -0,0 +1,411 @@
|
||||
"""
|
||||
智享通凭证 JSON 直传导入端点 (zxt-cma-import-v1, 2026-09-06)
|
||||
============================================================
|
||||
背景：智享通推送链路每10分钟落盘 2000 行凭证分录 JSON，需进 CMA voucher_details。
|
||||
现有 /api/cma/cash/import/vouchers 只收 Excel multipart，本模块提供 JSON 直传版本。
|
||||
|
||||
链路：
|
||||
接收服务 8500 落盘 incoming/{ts}.json → hook(zxt-cma-import-hook.py) → POST 本端点
|
||||
|
||||
鉴权：X-BOT-KEY: cma-bot-admin-2026（verify_bot_key，与 /api/cma/bot/* 共享密钥）。
|
||||
注意：本 router 独立于 cash.py 的 router —— cash.router 带 require_role 需用户
|
||||
Bearer token，Bot hook 只有 X-BOT-KEY 到不了，故同前缀另挂本 router（FastAPI 允许多 router 同前缀）。
|
||||
|
||||
校验口径（对齐财务BOT映射说明 cma_field_mapping_finance_20260906.md + 任务单验收）：
|
||||
1. 行级：voucher_no 必有；debit/credit 字符串→float round2，非负，借贷不同时0
|
||||
2. 结转行：summary 含"结转" → carry_forward=1（zxt 113行"期间损益结转"100%命中）
|
||||
3. 凭证级借贷平衡（每 Voucher_ID Σ借=Σ贷 容差0.01）—— 批级平衡对 zxt 快照不适用
|
||||
（已知单边凭证 30694 仅贷5479.65），不平衡凭证整张拒收记异常清单，不整批拒绝
|
||||
4. 日期/period：zxt 行无 voucher_date 字段，日期从 summary 前缀 YYYY-MM-DD 解析；
|
||||
解析不到的行（期间损益结转等）按 pending 拒收，不冒充文件生成时间
|
||||
5. 科目：subject_code/subject_name 全 null（智享通侧未带）。解析优先级：
|
||||
a) 行内 subject_code/subject_name 有值 → 直用
|
||||
b) 无 → 查请求体 subject_map {CwAccountID: {subject_code, subject_name}}
|
||||
或 DB 表 zxt_account_dict(CwAccountID, subject_code, subject_name)（若已建）
|
||||
c) 未命中 → 该行拒收 reason="科目字典未命中 CwAccountID=x"（不写脏数据）
|
||||
6. 幂等：快照每10分钟全量重推同一内容。batch = zxt-{日期}-{行集sha1前10}，
|
||||
导入前查 import_logs 已存在同 batch → 直接返回 already_imported（不再入库）；
|
||||
行级再以 (entity_id, voucher_no, subject_code, debit, credit, summary, period) 防重复。
|
||||
7. entity_id：默认 1（陕西酣客，source=zxt-sxzxb20210228），Bot 通道可传 X-Entity-Id 或 body entity_id。
|
||||
8. 入库后不触发现金流/ledger 联动（凭证完整性未证实前不作为权威来源，见财务口径警示）；
|
||||
也不影响 Excel multipart 导入（原端点不动）。
|
||||
"""
|
||||
import hashlib
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
from datetime import datetime
|
||||
from typing import Any, Optional, Union
|
||||
|
||||
from fastapi import APIRouter, Body, Depends, HTTPException, Request
|
||||
from sqlalchemy import text as sa_text
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.database import get_db
|
||||
from app.deps import resolve_entity_for_request
|
||||
from app.models import Entity, ImportLog, VoucherDetail
|
||||
from app.api.bot_bridge import verify_bot_key
|
||||
|
||||
logger = logging.getLogger("cma.cash_zxt")
|
||||
|
||||
router = APIRouter(prefix="/api/cma/cash", tags=["资金管理-智享通JSON导入"])
|
||||
|
||||
_DATE_RE = re.compile(r"(20\d{2})[-/.](\d{1,2})[-/.](\d{1,2})")
|
||||
_AMOUNT_RE = re.compile(r"^-?\d[\d,]*\.?\d*$")
|
||||
|
||||
|
||||
def _norm_str(raw) -> str:
|
||||
"""干净字符串：None/空 → ''"""
|
||||
if raw is None:
|
||||
return ""
|
||||
return str(raw).strip()
|
||||
|
||||
|
||||
def _parse_amount(raw) -> Optional[float]:
|
||||
"""金额：字符串数字（可带逗号）→ float；空/None → 0.0；非法 → None"""
|
||||
if raw is None:
|
||||
return 0.0
|
||||
s = str(raw).strip()
|
||||
if s == "":
|
||||
return 0.0
|
||||
if not _AMOUNT_RE.match(s):
|
||||
return None
|
||||
try:
|
||||
return round(float(s.replace(",", "")), 2)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def _parse_date_from_summary(summary: str):
|
||||
"""summary 前缀提取日期 (YYYY-MM-DD)。zxt 格式='业务类型-YYYY-MM-DD-序号_往来_业务员'"""
|
||||
m = _DATE_RE.search(summary or "")
|
||||
if not m:
|
||||
return None
|
||||
try:
|
||||
return datetime(int(m.group(1)), int(m.group(2)), int(m.group(3)))
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def _is_carry_forward(summary: str) -> bool:
|
||||
"""结转行识别（subject_name 缺失只能靠 summary）：zxt 期间损益结转 100% 命中"""
|
||||
return "结转" in (summary or "")
|
||||
|
||||
|
||||
def _content_fingerprint(rows: list) -> str:
|
||||
"""行集内容指纹：每行规范化 JSON 排序后 sha1 前 10 位（快照幂等用）"""
|
||||
canon = json.dumps(rows, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
||||
return hashlib.sha1(canon.encode("utf-8")).hexdigest()[:10]
|
||||
|
||||
|
||||
def _load_zxt_account_dict(db: Session) -> dict:
|
||||
"""读 DB 科目字典表（若已建）：zxt_account_dict(CwAccountID, subject_code, subject_name)。
|
||||
表不存在 → 返回空 dict（由调用方决定走请求体 subject_map 或拒收）"""
|
||||
try:
|
||||
rows = db.execute(
|
||||
sa_text("SELECT CwAccountID, subject_code, subject_name FROM zxt_account_dict")
|
||||
).fetchall()
|
||||
return {
|
||||
str(r[0]): {"subject_code": str(r[1]), "subject_name": str(r[2])}
|
||||
for r in rows if r[1]
|
||||
}
|
||||
except Exception:
|
||||
return {}
|
||||
|
||||
|
||||
def _resolve_subject(row: dict, db_account_dict: dict, req_subject_map: dict):
|
||||
"""科目解析优先级：行内 subject_code → DB 字典 → 请求体 subject_map。返回 (code, name) 或 None"""
|
||||
code = _norm_str(row.get("subject_code"))
|
||||
name = _norm_str(row.get("subject_name"))
|
||||
if code:
|
||||
return code, name
|
||||
cw_id = _norm_str(row.get("CwAccountID"))
|
||||
if not cw_id:
|
||||
return None
|
||||
hit = db_account_dict.get(cw_id) or req_subject_map.get(cw_id)
|
||||
if not hit:
|
||||
return None
|
||||
c = _norm_str(hit.get("subject_code")) if isinstance(hit, dict) else ""
|
||||
if not c:
|
||||
return None
|
||||
n = _norm_str(hit.get("subject_name")) if isinstance(hit, dict) else ""
|
||||
return c, n
|
||||
|
||||
|
||||
def _entity_active(db: Session, entity_id: int) -> bool:
|
||||
try:
|
||||
ent = db.query(Entity).filter(Entity.id == entity_id).first()
|
||||
return bool(ent and ent.status == "active")
|
||||
except Exception:
|
||||
return entity_id == 1 # Entity 模型不可用兜底（正常不会走到）
|
||||
|
||||
|
||||
@router.post("/import/vouchers-json")
|
||||
def api_import_vouchers_json(
|
||||
request: Request,
|
||||
payload: Union[dict, list] = Body(...),
|
||||
bot: dict = Depends(verify_bot_key),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""JSON 直传凭证导入（智享通推送通道）。三种 body 形态均兼容：
|
||||
A. 裸 rows 数组: [{Voucher_ID, CwAccountID, summary, debit, credit, ...}]
|
||||
B. {source?, rows: [...]}
|
||||
C. 完整落盘包: {schema, source, datasets: [{dataset, mode, rows}]} → 取 datasets[0].rows
|
||||
附加字段（形态 B/C）: entity_id(默认1) / subject_map({CwAccountID: {subject_code, subject_name}})
|
||||
"""
|
||||
# ── body 形态归一 → rows ──
|
||||
rows_raw: list = []
|
||||
source = ""
|
||||
req_entity = None
|
||||
req_subject_map: dict = {}
|
||||
if isinstance(payload, list):
|
||||
rows_raw = payload
|
||||
elif isinstance(payload, dict):
|
||||
if "datasets" in payload and isinstance(payload.get("datasets"), list):
|
||||
ds = payload["datasets"]
|
||||
if not ds or not isinstance(ds[0], dict):
|
||||
raise HTTPException(400, "datasets 结构不符: 缺少 datasets[0]")
|
||||
rows_raw = ds[0].get("rows") or []
|
||||
source = _norm_str(payload.get("source"))
|
||||
elif "rows" in payload:
|
||||
rows_raw = payload.get("rows") or []
|
||||
source = _norm_str(payload.get("source"))
|
||||
else:
|
||||
raise HTTPException(400, "无法识别的请求体: 期望 rows 数组 / {rows:[...]} / {datasets:[{rows:[...]}]}")
|
||||
req_entity = payload.get("entity_id")
|
||||
if isinstance(payload.get("subject_map"), dict):
|
||||
req_subject_map = payload["subject_map"]
|
||||
else:
|
||||
raise HTTPException(400, "请求体必须为 JSON 对象或数组")
|
||||
|
||||
if not isinstance(rows_raw, list) or not rows_raw:
|
||||
raise HTTPException(400, "rows 为空或非数组")
|
||||
|
||||
# 兼容 row key 大小写（智享通字段为大写 Voucher_ID/CwAccountID）
|
||||
norm_rows = []
|
||||
for i, r in enumerate(rows_raw):
|
||||
if not isinstance(r, dict):
|
||||
raise HTTPException(400, f"rows[{i}] 不是JSON对象")
|
||||
nr = {}
|
||||
for k, v in r.items():
|
||||
nr[k] = v
|
||||
if k == "Voucher_ID":
|
||||
nr["voucher_no"] = v
|
||||
elif k == "CwAccountID":
|
||||
nr["CwAccountID"] = v
|
||||
norm_rows.append(nr)
|
||||
|
||||
# ── entity 解析 + active 校验 ──
|
||||
entity_id = resolve_entity_for_request(request, req_entity if req_entity not in (None, "") else 1)
|
||||
try:
|
||||
entity_id = int(entity_id)
|
||||
except Exception:
|
||||
entity_id = 1
|
||||
if not _entity_active(db, entity_id):
|
||||
raise HTTPException(403, f"企业ID={entity_id} 不存在或未激活")
|
||||
|
||||
db_account_dict = _load_zxt_account_dict(db)
|
||||
|
||||
# ── 行级解析 + 校验 ──
|
||||
errors = []
|
||||
parsed = []
|
||||
pending_undated = 0
|
||||
pending_dict = 0
|
||||
for idx, raw in enumerate(norm_rows):
|
||||
voucher_no = _norm_str(raw.get("voucher_no") or raw.get("Voucher_ID"))
|
||||
if not voucher_no:
|
||||
errors.append({"row": idx + 1, "field": "voucher_no", "reason": "凭证号为空"})
|
||||
continue
|
||||
|
||||
date_obj = _parse_date_from_summary(_norm_str(raw.get("summary")))
|
||||
if date_obj is None:
|
||||
pending_undated += 1
|
||||
errors.append({"row": idx + 1, "field": "voucher_date",
|
||||
"reason": f"summary无日期(待日期来源): {_norm_str(raw.get('summary'))[:40]}"})
|
||||
continue
|
||||
|
||||
debit = _parse_amount(raw.get("debit"))
|
||||
credit = _parse_amount(raw.get("credit"))
|
||||
if debit is None:
|
||||
errors.append({"row": idx + 1, "field": "debit", "reason": f"借方金额不是有效数字: {raw.get('debit')}"})
|
||||
continue
|
||||
if credit is None:
|
||||
errors.append({"row": idx + 1, "field": "credit", "reason": f"贷方金额不是有效数字: {raw.get('credit')}"})
|
||||
continue
|
||||
if debit < 0 or credit < 0:
|
||||
errors.append({"row": idx + 1, "field": "amount", "reason": "金额不能为负"})
|
||||
continue
|
||||
if debit == 0 and credit == 0:
|
||||
errors.append({"row": idx + 1, "field": "amount", "reason": "借贷金额不能同时为0"})
|
||||
continue
|
||||
|
||||
summary = _norm_str(raw.get("summary"))
|
||||
subject = _resolve_subject(raw, db_account_dict, req_subject_map)
|
||||
if subject is None:
|
||||
pending_dict += 1
|
||||
errors.append({"row": idx + 1, "field": "subject_code",
|
||||
"reason": f"科目字典未命中 CwAccountID={_norm_str(raw.get('CwAccountID')) or '空'}"})
|
||||
continue
|
||||
|
||||
parsed.append({
|
||||
"voucher_no": voucher_no,
|
||||
"date": date_obj,
|
||||
"period": date_obj.strftime("%Y-%m"),
|
||||
"debit_amount": debit,
|
||||
"credit_amount": credit,
|
||||
"summary": summary,
|
||||
"carry_forward": 1 if _is_carry_forward(summary) else 0,
|
||||
"subject_code": subject[0],
|
||||
"subject_name": subject[1],
|
||||
})
|
||||
|
||||
total = len(rows_raw)
|
||||
if not parsed:
|
||||
db.rollback()
|
||||
return {
|
||||
"success": False,
|
||||
"message": "无有效行可入库（科目字典未建立/日期缺失/行级校验全失败），全部拒绝",
|
||||
"total": total,
|
||||
"success_rows": 0,
|
||||
"failed_rows": len(errors),
|
||||
"errors": errors[:100],
|
||||
"balance_check": None,
|
||||
"carry_forward_count": 0,
|
||||
"batch": None,
|
||||
"entity_id": entity_id,
|
||||
"pending_undated": pending_undated,
|
||||
"pending_dict_miss": pending_dict,
|
||||
"hint": "科目字典：请协调智享通推送方补 subject_code/subject_name，或在 body 带 subject_map，或建 zxt_account_dict 表",
|
||||
}
|
||||
|
||||
# ── 凭证级借贷平衡（每 Voucher_ID Σ借=Σ贷 容差0.01）──
|
||||
balance_by_voucher = {}
|
||||
for r in parsed:
|
||||
bv = balance_by_voucher.setdefault(r["voucher_no"], {"debit_total": 0.0, "credit_total": 0.0})
|
||||
bv["debit_total"] = round(bv["debit_total"] + r["debit_amount"], 2)
|
||||
bv["credit_total"] = round(bv["credit_total"] + r["credit_amount"], 2)
|
||||
abnormal_vouchers = []
|
||||
voucher_pass = set()
|
||||
for vno, bv in balance_by_voucher.items():
|
||||
diff = round(bv["debit_total"] - bv["credit_total"], 2)
|
||||
if abs(diff) > 0.01:
|
||||
abnormal_vouchers.append({
|
||||
"voucher_no": vno,
|
||||
"debit_total": bv["debit_total"],
|
||||
"credit_total": bv["credit_total"],
|
||||
"diff": diff,
|
||||
})
|
||||
else:
|
||||
voucher_pass.add(vno)
|
||||
|
||||
if abnormal_vouchers:
|
||||
abnormal_set = {a["voucher_no"] for a in abnormal_vouchers}
|
||||
kept = []
|
||||
for r in parsed:
|
||||
if r["voucher_no"] in abnormal_set:
|
||||
errors.append({"row": 0, "field": "balance",
|
||||
"reason": f"凭证 {r['voucher_no']} 借贷不平衡（整张凭证拒绝入库）"})
|
||||
else:
|
||||
kept.append(r)
|
||||
parsed = kept
|
||||
|
||||
valid_rows = [r for r in parsed if r["voucher_no"] in voucher_pass]
|
||||
success = len(valid_rows)
|
||||
failed = len(errors)
|
||||
|
||||
# ── 幂等：batch 指纹去重 ──
|
||||
fingerprint = _content_fingerprint(rows_raw)
|
||||
batch = f"zxt-{datetime.now().strftime('%Y%m%d%H%M%S')}-{fingerprint}"
|
||||
existing_batch = db.query(ImportLog).filter(
|
||||
ImportLog.batch == batch,
|
||||
ImportLog.import_type == "vouchers-json",
|
||||
).first()
|
||||
if existing_batch:
|
||||
db.rollback()
|
||||
return {
|
||||
"success": True,
|
||||
"already_imported": True,
|
||||
"message": "该快照内容已导入过（同一指纹 batch），跳过重复入库",
|
||||
"total": total,
|
||||
"success_rows": 0,
|
||||
"failed_rows": 0,
|
||||
"batch": batch,
|
||||
"entity_id": entity_id,
|
||||
}
|
||||
|
||||
# ── 入库（行级业务键防重复）──
|
||||
inserted = 0
|
||||
dup_skipped = 0
|
||||
for r in valid_rows:
|
||||
dup = db.query(VoucherDetail).filter(
|
||||
VoucherDetail.entity_id == entity_id,
|
||||
VoucherDetail.voucher_no == r["voucher_no"],
|
||||
VoucherDetail.subject_code == r["subject_code"],
|
||||
VoucherDetail.debit_amount == r["debit_amount"],
|
||||
VoucherDetail.credit_amount == r["credit_amount"],
|
||||
VoucherDetail.summary == r["summary"],
|
||||
VoucherDetail.period == r["period"],
|
||||
).first()
|
||||
if dup:
|
||||
dup_skipped += 1
|
||||
continue
|
||||
db.add(VoucherDetail(
|
||||
entity_id=entity_id,
|
||||
voucher_no=r["voucher_no"],
|
||||
voucher_date=r["date"],
|
||||
subject_code=r["subject_code"],
|
||||
subject_name=r["subject_name"],
|
||||
debit_amount=r["debit_amount"],
|
||||
credit_amount=r["credit_amount"],
|
||||
summary=r["summary"],
|
||||
carry_forward=r["carry_forward"],
|
||||
period=r["period"],
|
||||
batch=batch,
|
||||
))
|
||||
inserted += 1
|
||||
|
||||
period_totals = {}
|
||||
for r in valid_rows:
|
||||
pt = period_totals.setdefault(r["period"], {"debit_total": 0.0, "credit_total": 0.0})
|
||||
pt["debit_total"] = round(pt["debit_total"] + r["debit_amount"], 2)
|
||||
pt["credit_total"] = round(pt["credit_total"] + r["credit_amount"], 2)
|
||||
|
||||
carry_forward_count = sum(1 for r in valid_rows if r["carry_forward"])
|
||||
db.add(ImportLog(
|
||||
entity_id=entity_id,
|
||||
filename=f"zxt-push-voucher-snapshot.json" + (f"({source})" if source else ""),
|
||||
batch=batch,
|
||||
total_rows=total,
|
||||
success_rows=inserted,
|
||||
failed_rows=failed,
|
||||
errors=errors[:100] or None,
|
||||
period=valid_rows[0]["period"] if valid_rows else None,
|
||||
import_type="vouchers-json",
|
||||
created_by="zxt-cma-import-hook",
|
||||
))
|
||||
db.commit()
|
||||
logger.info(
|
||||
f"智享通凭证JSON导入[{batch}] entity={entity_id}: 总{total}/入库{inserted}/行级失败{failed}/重复跳过{dup_skipped}, "
|
||||
f"凭证级异常{len(abnormal_vouchers)}"
|
||||
)
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"total": total,
|
||||
"success_rows": inserted,
|
||||
"failed_rows": failed,
|
||||
"dup_skipped": dup_skipped,
|
||||
"errors": errors[:100] or None,
|
||||
"balance_check": {
|
||||
"mode": "voucher_level",
|
||||
"passed": len(abnormal_vouchers) == 0,
|
||||
"abnormal_vouchers": abnormal_vouchers,
|
||||
"tolerance": 0.01,
|
||||
},
|
||||
"period_totals": period_totals,
|
||||
"carry_forward_count": carry_forward_count,
|
||||
"pending_undated": pending_undated,
|
||||
"pending_dict_miss": pending_dict,
|
||||
"batch": batch,
|
||||
"entity_id": entity_id,
|
||||
}
|
||||
Reference in New Issue
Block a user