cma-management/backend/scripts/verify_cashflow_structure.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
现金流量表结构分析 — 验收验证脚本 (project-cashflow-structure-analysis-20260903 DoD)

覆盖:
  1. POST /api/cma/cashflow-structure/values  录入 O=-30/I=-5/F=+40 → 期望类型6(初创烧钱型🟠)
  2. 同期间覆盖更新 O=+30/I=-20/F=-10 → 期望类型1(成熟健康型🟢)  判定双向验证
  3. GET  /analysis  结构分析响应(三净额符号/判定/瀑布图数据点)
  4. GET  /values /periods /rules 常规查询
  5. 清理验证临时数据(DB 直删 remark 标记行)  → 恢复空态

用法: cd backend && venv/bin/python3 scripts/verify_cashflow_structure.py
依赖: requests (项目 venv 已装)
"""
import json
import sys
import urllib.error
import urllib.request

import os  # noqa: E402  (凭据加载器需要)
# ---- 管理员凭据同源加载（2026-09-14 cma-defaultcred-code-fix）----
# 生产凭据只存 600 密钥文件；禁止在脚本/文档/日志里硬编码口令。
# 顺序：$CMA_ADMIN_ENV → wecom-ops 主副本（ops 轮换落点，权威） → 本 profile 副本。
def _admin_creds():
    """返回 (user, password, 来源文件)；未找到 → 显式报错退出（不静默退回默认口令）"""
    for p in (os.environ.get("CMA_ADMIN_ENV"),
              "/root/.hermes/profiles/wecom-ops/.secrets/cma-admin.env",
              "/root/.hermes/profiles/wecom-fullstack/.secrets/cma-admin.env"):
        if p and os.path.isfile(p):
            kv = {}
            for line in open(p, encoding="utf-8"):
                line = line.strip()
                if "=" in line and not line.startswith("#"):
                    k, v = line.split("=", 1)
                    kv[k.strip()] = v.strip().strip('"').strip("'")
            if kv.get("CMA_ADMIN_USER") and kv.get("CMA_ADMIN_PASS"):
                return kv["CMA_ADMIN_USER"], kv["CMA_ADMIN_PASS"], p
    raise SystemExit("[错误] 未找到 CMA 管理员凭据：请设 $CMA_ADMIN_ENV，或准备 600 权限密钥文件 "
                     ".secrets/cma-admin.env（键名 CMA_ADMIN_USER / CMA_ADMIN_PASS）")


ADMIN_USER, ADMIN_PASS, CRED_SRC = _admin_creds()
print(f"[creds] 来源={CRED_SRC} 账号={ADMIN_USER}（口令不回显）")

BASE = "http://127.0.0.1:8010"
TMP_MARK = "【系统验证临时数据-验证后删除】"


def http(method, path, body=None, token=None, timeout=15):
    url = BASE + path
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(url, data=data, method=method)
    req.add_header("Content-Type", "application/json")
    if token:
        req.add_header("Authorization", "Bearer " + token)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            raw = resp.read().decode()
            return resp.status, json.loads(raw) if raw else None
    except urllib.error.HTTPError as e:
        raw = e.read().decode()
        try:
            return e.code, json.loads(raw)
        except Exception:
            return e.code, {"detail": raw}


def main():
    fails = []

    def check(name, cond, extra=""):
        tag = "✅ PASS" if cond else "❌ FAIL"
        print(f"{tag}  {name}" + (f"  {extra}" if extra else ""))
        if not cond:
            fails.append(name)

    # 1. 登录博海账套 (admin, entity_id=2)
    st, r = http("POST", "/api/cma/auth/login",
                  {"username": ADMIN_USER, "password": ADMIN_PASS, "entity_id": 2})
    check("登录博海(entity_id=2)", st == 200 and "token" in (r or {}),
          f"st={st} keys={sorted((r or {}).keys())}")
    token = (r or {}).get("token", "")
    entity_name = (r or {}).get("entity_name", "")
    print(f"    登录账套: {entity_name}")

    # 2. openapi 端点可见性
    st, spec = http("GET", "/openapi.json")
    paths = [p for p in (spec or {}).get("paths", {}) if "cashflow-structure" in p]
    check("openapi 含 cashflow-structure 端点", st == 200 and len(paths) >= 4,
          f"paths={sorted(paths)}")

    # 3. 空态分析(清理后应先跑一次, 这里先录入再测非空)
    # 4. POST 录入 类型6 用例: O=-30 I=-5 F=+40
    st, r = http("POST", "/api/cma/cashflow-structure/values", {
        "period": "2026-08",
        "operating_net": -30, "investing_net": -5, "financing_net": 40,
        "begin_cash": 10, "end_cash": 15,
        "source": "manual", "remark": TMP_MARK,
    }, token=token)
    cls = (r or {}).get("classification", {})
    check("POST values (O-30/I-5/F+40) → type6", st == 200 and cls.get("type_no") == 6,
          f"st={st} cls={cls.get('type_no')} {cls.get('name')} {cls.get('health_emoji')}")

    # 5. GET analysis 验证
    st, r = http("GET", "/api/cma/cashflow-structure/analysis?period=2026-08", token=token)
    check("GET analysis 200", st == 200 and (r or {}).get("data_available") is True)
    if r:
        wf = r.get("waterfall") or {}
        steps = wf.get("steps") or []
        check("瀑布图数据点≥5(期初→经营→投资→筹资→期末)", len(steps) >= 5,
              f"steps={[s['step'] for s in steps]}")
        check("判定 reason 含符号组合", "经营(-)" in ((r.get("classification") or {}).get("reason") or ""))

    # 6. 覆盖更新 类型1 用例: O=+30 I=-20 F=-10 (同期间 upsert)
    st, r = http("POST", "/api/cma/cashflow-structure/values", {
        "period": "2026-08",
        "operating_net": 30, "investing_net": -20, "financing_net": -10,
        "source": "manual", "remark": TMP_MARK,
    }, token=token)
    cls = (r or {}).get("classification", {})
    check("POST 覆盖更新 (O+30/I-20/F-10) → type1", st == 200 and cls.get("type_no") == 1,
          f"st={st} cls={cls.get('type_no')} {cls.get('name')} {cls.get('health_emoji')}")

    # 7. GET values/periods 常规查询
    st, r = http("GET", "/api/cma/cashflow-structure/values", token=token)
    items = (r or {}).get("items", [])
    check("GET values 返回记录", st == 200 and len(items) >= 1, f"count={len(items)}")
    st, r = http("GET", "/api/cma/cashflow-structure/periods", token=token)
    check("GET periods 含 2026-08", st == 200 and "2026-08" in (r or {}).get("periods", []))
    st, r = http("GET", "/api/cma/cashflow-structure/rules", token=token)
    check("GET rules 8类", st == 200 and len((r or {}).get("rules", [])) == 8)

    # 7b. DoD 验收路径 /api/cma/cashflow/* (兼容别名, 功能与主路径一致)
    st, r = http("GET", "/api/cma/cashflow/statements?period=2026-08", token=token)
    items_c = (r or {}).get("items", [])
    check("GET /api/cma/cashflow/statements 200", st == 200 and len(items_c) == 1,
          f"count={len(items_c)}")
    st, r = http("GET", "/api/cma/cashflow/analysis?period=2026-08", token=token)
    cls_c = (r or {}).get("classification", {})
    check("GET /api/cma/cashflow/analysis 200 type1", st == 200 and cls_c.get("type_no") == 1,
          f"cls={cls_c.get('type_no')} {cls_c.get('name')}")

    # 8. 越权: 无 token 调 analysis 应 401/403 (require_role 拦截)
    st, _ = http("GET", "/api/cma/cashflow-structure/analysis")
    check("无token访问被拦截(401/403)", st in (401, 403), f"st={st}")

    print("\n===== 判定验证输出 =====")
    print("用例1: O=-30/I=-5/F=+40 →", "type6 初创烧钱型 🟠 高风险" if not fails or "POST values" not in "".join(fails) else "待查")
    print("用例2: O=+30/I=-20/F=-10 →", "type1 成熟健康型 🟢 最优")

    print("\n清理验证临时数据...")
    try:
        sys.path.insert(0, ".")
        from app.database import get_db
        from app.models import CashFlowStructureValue
        db = next(get_db())
        rows = db.query(CashFlowStructureValue).filter(
            CashFlowStructureValue.remark == TMP_MARK).all()
        for row in rows:
            db.delete(row)
        db.commit()
        remain = db.query(CashFlowStructureValue).count()
        db.close()
        check("临时数据已清理(表余0)", remain == 0, f"remain={remain}")
    except Exception as e:
        check(f"清理临时数据失败: {e}", False)

    print()
    if fails:
        print(f"❌ 共 {len(fails)} 项失败: {fails}")
        return 1
    print("✅ 全部验收项通过")
    return 0


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