#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
ARHB-POC 跑批引擎
按 PRD-ARHB-001 v1.1 已拍板规则执行：
  1) 读取源台账（T1 应收 / T2 收款 / T3 客户账期 / T4 汇率）
  2) 红蓝票分区识别（R-01）
  3) 多币种折算（R-16 / R-18）
  4) 红蓝票对冲（R-02~R-06）
  5) 收款核销（R-07~R-09）
  6) 账龄分层 + 逾期天数（R-10~R-13）
  7) 输出结果工作簿 + 分录草稿 + 差异池 + 跑批留痕

参数说明：
    python run_arhb_poc.py --base-date 2026-09-29 \
                           --t1 数据源电子表格/T1_应收明细台账.xlsx \
                           --t2 数据源电子表格/T2_收款明细台账.xlsx \
                           --t3 数据源电子表格/T3_客户账期表.xlsx \
                           --t4 数据源电子表格/T4_汇率表.xlsx \
                           --out 结果工作簿.xlsx
"""

import os
import re
import sys
import argparse
import logging
from datetime import datetime, timedelta
from collections import defaultdict

import pandas as pd
import numpy as np

# -----------------------------------------------------------------------------
# 常量：PRD 已拍板参数
# -----------------------------------------------------------------------------
STD_TERM_DAYS = 30          # HOLD-13
TAIL_TOLERANCE = 0.01       # HOLD-14
BIG_AMOUNT_THRESHOLD = 500000.0  # HOLD-15
COVERAGE_PRECISE = 0.70     # HOLD-16
COVERAGE_FUZZY = 0.90       # HOLD-16
BASE_CURRENCY = "CNY"

# 一级科目大类（分录草稿用，HOLD-17）
ACCOUNT_CATEGORIES = {
    "应收账款": "资产类",
    "银行存款": "资产类",
    "应交税费-应交增值税（销项税额）": "税费类",
    "主营业务收入": "损益类",
    "主营业务成本": "损益类",
    "销售退回与折让": "损益类",
}

logger = logging.getLogger("arhb_poc")


def setup_logging(log_path):
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(message)s",
        handlers=[logging.FileHandler(log_path, encoding="utf-8"), logging.StreamHandler()],
    )


def parse_date(v):
    if pd.isna(v):
        return None
    if isinstance(v, (datetime, pd.Timestamp)):
        return v.date() if hasattr(v, "date") else v
    if isinstance(v, str):
        for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%Y%m%d"):
            try:
                return datetime.strptime(v, fmt).date()
            except ValueError:
                continue
    raise ValueError(f"无法解析日期：{v}")


def to_date_series(s):
    """把可能是字符串或 datetime 的列统一转成 date"""
    return pd.to_datetime(s, errors="coerce").dt.date


def load_tables(args):
    """读取并做最小标准化"""
    t1 = pd.read_excel(args.t1)
    t2 = pd.read_excel(args.t2)
    t3 = pd.read_excel(args.t3)
    t4 = pd.read_excel(args.t4)

    # 日期标准化
    t1["开票日"] = to_date_series(t1["开票日"])
    t2["收款日"] = to_date_series(t2["收款日"])
    t4["日期"] = to_date_series(t4["日期"])

    # 金额数值化
    for col in ["原币金额（含税）", "税额（原币）"]:
        t1[col] = pd.to_numeric(t1[col], errors="coerce")
    t2["收款金额（原币）"] = pd.to_numeric(t2["收款金额（原币）"], errors="coerce")
    t4["中间价（对人民币）"] = pd.to_numeric(t4["中间价（对人民币）"], errors="coerce")

    # 大小写/空格清洗
    for df in (t1, t2):
        for col in ["单据号", "客户编码", "客户名称", "币种", "发票号", "关联原蓝票号"]:
            if col in df.columns:
                df[col] = df[col].astype(str).replace("nan", None).str.strip()
                df[col] = df[col].where(df[col] != "None", None)

    t3["客户编码"] = t3["客户编码"].astype(str).str.strip()
    t4["币种"] = t4["币种"].astype(str).str.strip()

    return t1, t2, t3, t4


def validate(t1, t2, t3, t4):
    """基础校验，返回异常清单"""
    issues = []

    # 必填字段非空
    for col in ["单据号", "客户编码", "原币金额（含税）", "税额（原币）", "开票日", "币种"]:
        nulls = t1[t1[col].isna() | (t1[col] == "")]
        if len(nulls):
            issues.append({"异常类型": "E-01/E-02/E-03", "描述": f"T1 字段 {col} 缺失 {len(nulls)} 行", "样例行号": nulls.index.tolist()[:3]})

    # 客户编码是否都在 T3
    missing_cust = set(t1["客户编码"].dropna()) - set(t3["客户编码"].dropna())
    if missing_cust:
        issues.append({"异常类型": "E-03", "描述": f"T1 中 {len(missing_cust)} 个客户编码不在 T3", "样例客户": list(missing_cust)[:5]})

    # 单据号重复
    dup = t1[t1.duplicated("单据号", keep=False)]
    if len(dup):
        issues.append({"异常类型": "E-10", "描述": f"T1 单据号重复 {len(dup)} 行", "样例": dup["单据号"].unique()[:3].tolist()})

    # 收款金额非空
    null_pay = t2[t2["收款金额（原币）"].isna()]
    if len(null_pay):
        issues.append({"异常类型": "E-01", "描述": f"T2 收款金额缺失 {len(null_pay)} 行", "样例行号": null_pay.index.tolist()[:3]})

    # 汇率表覆盖度
    needed = set(zip(t1["币种"], t1["开票日"])) | set(zip(t2["币种"], t2["收款日"]))
    have = set(zip(t4["币种"], t4["日期"]))
    missing_rates = needed - have
    if missing_rates:
        issues.append({"异常类型": "R-18", "描述": f"汇率表缺失 {len(missing_rates)} 个币种×日期组合", "样例": list(missing_rates)[:5]})

    return pd.DataFrame(issues)


def partition_red_blue(t1):
    """R-01 三级判定：单据类型 -> 关联原蓝票号 -> 金额符号"""
    red_types = {"红字应收单", "销售退货单", "折让单"}
    zones = []
    conflict = []

    for idx, row in t1.iterrows():
        flags = {
            "type": row["单据类型"] in red_types,
            "link": pd.notna(row.get("关联原蓝票号")) and str(row.get("关联原蓝票号")).strip().lower() not in ("", "none", "nan"),
            "sign": row["原币金额（含税）"] < 0,
        }
        # 优先级：单据类型 > 关联原蓝票号 > 金额符号
        if flags["type"]:
            zone = "红区"
            basis = "单据类型"
        elif flags["link"]:
            zone = "红区"
            basis = "关联原蓝票号"
        elif flags["sign"]:
            zone = "红区"
            basis = "金额符号"
        else:
            zone = "蓝区"
            basis = "默认蓝区"

        # 冲突检测：三个标志不一致
        if sum(flags.values()) > 0 and sum(flags.values()) < 3 and flags["type"] is False:
            conflict.append({"单据号": row["单据号"], "判定": zone, "冲突标志": flags})

        zones.append({
            "单据号": row["单据号"],
            "客户编码": row["客户编码"],
            "客户名称": row["客户名称"],
            "发票号": row["发票号"],
            "关联原蓝票号": row.get("关联原蓝票号"),
            "币种": row["币种"],
            "原币金额": row["原币金额（含税）"],
            "税额": row["税额（原币）"],
            "开票日": row["开票日"],
            "判定分区": zone,
            "判定依据": basis,
        })

    return pd.DataFrame(zones), pd.DataFrame(conflict)


def attach_exchange_rate(df, date_col, currency_col, rate_df, amount_col, rmb_amount_col, tax_col=None, rmb_tax_col=None):
    """把汇率拼到 df，缺失汇率的行汇率留空"""
    lookup = rate_df.set_index(["币种", "日期"])["中间价（对人民币）"].to_dict()
    keys = list(zip(df[currency_col], df[date_col]))
    rates = [lookup.get(k, np.nan) for k in keys]
    df = df.copy()
    df["汇率"] = rates
    df[rmb_amount_col] = np.where(
        df["汇率"].notna(),
        (df[amount_col] * df["汇率"]).round(2),
        np.nan,
    )
    if tax_col and rmb_tax_col:
        df[rmb_tax_col] = np.where(
            df["汇率"].notna(),
            (df[tax_col] * df["汇率"]).round(2),
            np.nan,
        )
    return df


def get_term_days(row, customer_terms):
    """R-12 账期优先级：合同 > 客户标准 > 全局 30"""
    if pd.notna(row.get("合同账期（天）")) and row["合同账期（天）"] != "":
        try:
            return int(float(row["合同账期（天）"]))
        except Exception:
            pass
    std = customer_terms.get(row["客户编码"])
    if pd.notna(std):
        return int(std)
    return STD_TERM_DAYS


def offset_match(a, b, tol=TAIL_TOLERANCE):
    """判断两个金额在容差内是否可视为相等"""
    return abs(a + b) <= tol or abs(a - b) <= tol


def run_offset(blue_df, red_df):
    """
    红蓝对冲（R-02~R-06）
    返回：对冲记录，更新后的蓝票/红票余额
    """
    records = []
    blue_remaining = blue_df.set_index("单据号")["人民币金额（含税）"].to_dict()
    red_remaining = red_df.set_index("单据号")["人民币金额（含税）"].to_dict()
    red_tax = red_df.set_index("单据号")["税额（人民币）"].to_dict()
    blue_tax = blue_df.set_index("单据号")["税额（人民币）"].to_dict()

    # 按 R-04 优先级对每张红票找蓝票
    # 预处理：客户 -> 蓝票列表
    cust_blues = defaultdict(list)
    for _, b in blue_df.iterrows():
        cust_blues[b["客户编码"]].append(b)

    red_df_sorted = red_df.sort_values("开票日")
    matched_red = set()

    for _, r in red_df_sorted.iterrows():
        r_id = r["单据号"]
        r_cust = r["客户编码"]
        r_inv = r.get("关联原蓝票号")
        r_amt = abs(red_remaining.get(r_id, 0))
        r_tax = abs(red_tax.get(r_id, 0))
        r_date = r["开票日"]

        candidates = [b for b in cust_blues[r_cust] if b["单据号"] not in matched_red]

        # 优先级 1：原蓝票号精确
        chosen = None
        level = None
        if r_inv:
            for b in candidates:
                if b["发票号"] == r_inv:
                    chosen = b
                    level = "精确-原蓝票号"
                    break

        # 优先级 2：金额绝对值相等 + 税额相等
        if chosen is None:
            for b in candidates:
                if offset_match(b["人民币金额（含税）"], -r_amt) and offset_match(blue_tax[b["单据号"]], -r_tax):
                    chosen = b
                    level = "金额+税额匹配"
                    break

        # 优先级 3：开票日最近
        if chosen is None:
            # 找同客户未冲尽、开票日最接近的蓝票（优先早于红票）
            candidates_with_tax = [b for b in candidates if offset_match(blue_tax[b["单据号"]], -r_tax)]
            if not candidates_with_tax:
                candidates_with_tax = candidates  # 税额也匹配不上时放宽到金额接近
            pool = candidates_with_tax
            if pool:
                pool_sorted = sorted(pool, key=lambda b: abs((pd.to_datetime(b["开票日"]) - pd.to_datetime(r_date)).days))
                chosen = pool_sorted[0]
                level = "开票日最近"

        # 优先级 4：FIFO
        if chosen is None and candidates:
            chosen = min(candidates, key=lambda b: b["开票日"])
            level = "FIFO兜底"

        if chosen is None:
            continue

        b_id = chosen["单据号"]
        b_amt = blue_remaining.get(b_id, 0)
        offset_amt = min(r_amt, b_amt)

        # 税额必须完全相等才能对冲（R-03 硬条件）
        # 若税额不相等，只把金额配对但标记异常——这里严格按硬条件拒绝
        if not offset_match(blue_tax[b_id], -r_tax):
            # 进差异池，不生成对冲记录
            records.append({
                "红票号": r_id,
                "蓝票号": b_id,
                "客户编码": r_cust,
                "对冲金额": 0,
                "对冲税额": 0,
                "匹配级别": "税额不匹配-拒绝对冲",
                "状态": "差异池",
            })
            continue

        # 执行对冲
        red_remaining[r_id] = round(red_remaining[r_id] + offset_amt, 2)  # 负数趋近 0
        blue_remaining[b_id] = round(blue_remaining[b_id] - offset_amt, 2)
        matched_red.add(r_id)

        records.append({
            "红票号": r_id,
            "蓝票号": b_id,
            "客户编码": r_cust,
            "对冲金额": offset_amt,
            "对冲税额": r_tax,
            "匹配级别": level,
            "状态": "已对冲",
        })

        # 简化 POC 逻辑：一张红票只找一张蓝票做主要配对
        continue

    columns = ["红票号", "蓝票号", "客户编码", "对冲金额", "对冲税额", "匹配级别", "状态"]
    return pd.DataFrame(records, columns=columns), blue_remaining, red_remaining


def run_writeoff(blue_df, pay_df):
    """
    收款核销（R-07~R-09）
    发票号精确优先，无发票号按客户 FIFO 拆分
    """
    records = []
    blue_remaining = blue_df.set_index("单据号")["未对冲余额"].to_dict()
    pay_remaining = pay_df.set_index("收款单号")["收款金额（人民币）"].to_dict()

    # 发票号精确匹配
    blue_by_inv = {}
    for _, b in blue_df.iterrows():
        if pd.notna(b["发票号"]):
            blue_by_inv.setdefault((b["客户编码"], b["发票号"]), []).append(b)

    pay_df_sorted = pay_df.sort_values("收款日")

    for _, p in pay_df_sorted.iterrows():
        p_id = p["收款单号"]
        p_cust = p["客户编码"]
        p_inv = p.get("发票号")
        p_amt = pay_remaining.get(p_id, 0)

        if p_amt <= 0:
            continue

        targets = []
        level = None
        if pd.notna(p_inv) and p_inv != "None":
            targets = blue_by_inv.get((p_cust, p_inv), [])
            if targets:
                level = "精确-发票号"

        if not targets:
            # 同客户 FIFO，取未冲尽蓝票
            targets = [b for _, b in blue_df.iterrows() if b["客户编码"] == p_cust and blue_remaining[b["单据号"]] > TAIL_TOLERANCE]
            targets = sorted(targets, key=lambda x: x["开票日"])
            if targets:
                level = "FIFO兜底"

        if not targets:
            records.append({
                "收款单号": p_id,
                "应收单号": None,
                "客户编码": p_cust,
                "核销金额": 0,
                "匹配方式": "无匹配应收单",
                "拆分序号": 0,
            })
            continue

        seq = 0
        for b in targets:
            if p_amt <= TAIL_TOLERANCE:
                break
            b_id = b["单据号"]
            avail = blue_remaining.get(b_id, 0)
            if avail <= TAIL_TOLERANCE:
                continue
            amt = min(p_amt, avail)
            blue_remaining[b_id] = round(avail - amt, 2)
            pay_remaining[p_id] = round(p_amt - amt, 2)
            p_amt = pay_remaining[p_id]
            seq += 1
            records.append({
                "收款单号": p_id,
                "应收单号": b_id,
                "客户编码": p_cust,
                "发票号": p_inv if seq == 1 else None,
                "核销金额": amt,
                "匹配方式": level,
                "拆分序号": seq,
            })

    columns = ["收款单号", "应收单号", "客户编码", "发票号", "核销金额", "匹配方式", "拆分序号"]
    return pd.DataFrame(records, columns=columns), blue_remaining, pay_remaining


def build_aging(blue_df, base_date, customer_terms):
    """R-10~R-13 账龄分层 + 逾期计算"""
    rows = []
    for _, b in blue_df.iterrows():
        term = get_term_days(b, customer_terms)
        inv_date = pd.to_datetime(b["开票日"])
        days_age = (base_date - inv_date).days
        due_date = inv_date + timedelta(days=term)
        due_days = (base_date - due_date).days
        overdue_days = max(due_days, 0)
        is_overdue = due_days > 0

        # 六档
        if days_age <= 30:
            bucket = "0-30天"
        elif days_age <= 60:
            bucket = "31-60天"
        elif days_age <= 90:
            bucket = "61-90天"
        elif days_age <= 180:
            bucket = "91-180天"
        elif days_age <= 365:
            bucket = "181-365天"
        else:
            bucket = "365天以上"

        long_alert = days_age > 180 and b["未核销余额"] > 0

        rows.append({
            "应收单号": b["单据号"],
            "客户编码": b["客户编码"],
            "客户名称": b["客户名称"],
            "发票号": b["发票号"],
            "币种": b["币种"],
            "未核销余额（人民币）": b["未核销余额"],
            "账龄分层": bucket,
            "开票日": b["开票日"],
            "账期（天）": term,
            "到期日": due_date.date(),
            "是否逾期": "是" if is_overdue else "否",
            "逾期天数": overdue_days,
            "长账龄预警": "是" if long_alert else "否",
            "账期来源": "合同" if pd.notna(b.get("合同账期（天）")) and b.get("合同账期（天）") != "" else ("客户标准" if pd.notna(term) else "全局30天"),
        })

    if not rows:
        rows = [{}]
    df = pd.DataFrame(rows)
    cols = ["应收单号", "客户编码", "客户名称", "发票号", "币种", "未核销余额（人民币）",
            "账龄分层", "开票日", "账期（天）", "到期日", "是否逾期", "逾期天数", "长账龄预警", "账期来源"]
    for c in cols:
        if c not in df.columns:
            df[c] = None
    return df[cols]


def build_entry_draft(offset_df, writeoff_df):
    """R-14 分录草稿：只出借贷方向 + 一级科目大类"""
    entries = []

    # 对冲分录：红冲
    for _, r in offset_df[offset_df["状态"] == "已对冲"].iterrows():
        entries.append({
            "日期": datetime.now().date(),
            "摘要": f"红蓝对冲 {r['红票号']} 冲 {r['蓝票号']}",
            "科目大类": "资产类",
            "科目": "应收账款",
            "借方": r["对冲金额"],
            "贷方": 0,
            "来源单据号": r["红票号"],
        })
        entries.append({
            "日期": datetime.now().date(),
            "摘要": f"红蓝对冲 {r['红票号']} 冲 {r['蓝票号']}",
            "科目大类": "资产类",
            "科目": "应收账款",
            "借方": 0,
            "贷方": r["对冲金额"],
            "来源单据号": r["蓝票号"],
        })

    # 核销分录：收款冲应收
    for _, r in writeoff_df.iterrows():
        if pd.isna(r["应收单号"]) or r["核销金额"] <= 0:
            continue
        entries.append({
            "日期": datetime.now().date(),
            "摘要": f"收款核销 {r['收款单号']} 冲 {r['应收单号']}",
            "科目大类": "资产类",
            "科目": "银行存款",
            "借方": r["核销金额"],
            "贷方": 0,
            "来源单据号": r["收款单号"],
        })
        entries.append({
            "日期": datetime.now().date(),
            "摘要": f"收款核销 {r['收款单号']} 冲 {r['应收单号']}",
            "科目大类": "资产类",
            "科目": "应收账款",
            "借方": 0,
            "贷方": r["核销金额"],
            "来源单据号": r["应收单号"],
        })

    cols = ["日期", "摘要", "科目大类", "科目", "借方", "贷方", "来源单据号"]
    return pd.DataFrame(entries, columns=cols)


def main():
    parser = argparse.ArgumentParser(description="ARHB-POC 跑批引擎")
    parser.add_argument("--base-date", type=str, default="2026-09-29")
    parser.add_argument("--t1", type=str, default="数据源电子表格/T1_应收明细台账.xlsx")
    parser.add_argument("--t2", type=str, default="数据源电子表格/T2_收款明细台账.xlsx")
    parser.add_argument("--t3", type=str, default="数据源电子表格/T3_客户账期表.xlsx")
    parser.add_argument("--t4", type=str, default="数据源电子表格/T4_汇率表.xlsx")
    parser.add_argument("--out", type=str, default="结果工作簿_ARHB_POC.xlsx")
    parser.add_argument("--log", type=str, default="arhb_poc_run.log")
    args = parser.parse_args()

    base_date = datetime.strptime(args.base_date, "%Y-%m-%d").date()
    base_dt = datetime.combine(base_date, datetime.min.time())

    setup_logging(args.log)
    logger.info("=" * 60)
    logger.info("ARHB-POC 跑批引擎启动")
    logger.info(f"数据基准日：{base_date}")

    # 1. 读取
    logger.info("[1/6] 读取源台账...")
    t1, t2, t3, t4 = load_tables(args)

    # 2. 校验
    logger.info("[2/6] 基础校验...")
    issues_df = validate(t1, t2, t3, t4)
    if not issues_df.empty:
        logger.warning(f"发现 {len(issues_df)} 类异常，已写入差异池")
        for _, r in issues_df.iterrows():
            logger.warning(f"  - {r['异常类型']}: {r['描述']}")

    # 3. 折算
    logger.info("[3/6] 币种折算...")
    t1 = attach_exchange_rate(
        t1, "开票日", "币种", t4,
        amount_col="原币金额（含税）", rmb_amount_col="人民币金额（含税）",
        tax_col="税额（原币）", rmb_tax_col="税额（人民币）",
    )
    t2 = attach_exchange_rate(
        t2, "收款日", "币种", t4,
        amount_col="收款金额（原币）", rmb_amount_col="收款金额（人民币）",
    )

    missing_rate_rows = t1[t1["汇率"].isna()]
    if not missing_rate_rows.empty:
        logger.warning(f"T1 有 {len(missing_rate_rows)} 行缺少汇率，将进差异池")

    # 4. 分区
    logger.info("[4/6] 红蓝分区...")
    partition_df, conflict_df = partition_red_blue(t1)

    # 合并汇率与折算金额
    t1_enriched = t1.merge(
        partition_df[["单据号", "判定分区", "判定依据"]],
        on="单据号",
        how="left",
    )

    # 客户账期映射
    customer_terms = t3.set_index("客户编码")["标准账期（天）"].to_dict()

    # 5. 对冲
    logger.info("[5/6] 红蓝票对冲...")
    blue_df = t1_enriched[t1_enriched["判定分区"] == "蓝区"].copy()
    red_df = t1_enriched[t1_enriched["判定分区"] == "红区"].copy()

    offset_df, blue_remaining_after_offset, red_remaining_after_offset = run_offset(blue_df, red_df)

    # 更新蓝票余额
    blue_df["未对冲余额"] = blue_df["单据号"].map(blue_remaining_after_offset)
    red_df["未对冲余额"] = red_df["单据号"].map(red_remaining_after_offset)

    # 6. 核销
    logger.info("[6/6] 收款核销...")
    writeoff_df, blue_remaining_after_pay, pay_remaining = run_writeoff(blue_df, t2)
    blue_df["未核销余额"] = blue_df["单据号"].map(blue_remaining_after_pay)

    # 重大单复核
    big_blue = blue_df[blue_df["人民币金额（含税）"] > BIG_AMOUNT_THRESHOLD]

    # 7. 账龄分层
    aging_df = build_aging(blue_df, base_dt, customer_terms)

    # 8. 分录草稿
    entry_draft_df = build_entry_draft(offset_df, writeoff_df)

    # 9. 汇总首页 KPI
    red_total = len(red_df)
    matched_red = set(offset_df[offset_df["状态"] == "已对冲"]["红票号"])
    precise_offset = len(offset_df[offset_df["匹配级别"].isin(["精确-原蓝票号", "金额+税额匹配"]) & (offset_df["状态"] == "已对冲")])

    writeoff_total_rows = len(writeoff_df)
    writeoff_success_rows = len(writeoff_df[writeoff_df["核销金额"] > 0])
    precise_writeoff_rows = len(writeoff_df[writeoff_df["匹配方式"] == "精确-发票号"])

    kpi = {
        "对冲覆盖率（精确级）": round(precise_offset / red_total, 4) if red_total else 0,
        "对冲覆盖率（含模糊）": round(len(matched_red) / red_total, 4) if red_total else 0,
        "核销自动化率（精确级）": round(precise_writeoff_rows / writeoff_total_rows, 4) if writeoff_total_rows else 0,
        "核销自动化率（含模糊）": round(writeoff_success_rows / writeoff_total_rows, 4) if writeoff_total_rows else 0,
        "差异池条目数": len(issues_df) + len(conflict_df) + len(big_blue),
        "长账龄预警数": len(aging_df[aging_df["长账龄预警"] == "是"]),
        "未核销余额合计（人民币）": round(blue_df["未核销余额"].sum(), 2),
        "账龄分层合计（人民币）": round(aging_df["未核销余额（人民币）"].sum(), 2),
    }

    # 勾稽校验
    check = abs(kpi["未核销余额合计（人民币）"] - kpi["账龄分层合计（人民币）"]) < TAIL_TOLERANCE
    kpi["勾稽通过"] = "是" if check else "否"

    if not check:
        logger.error("E-15：账龄分层合计 ≠ 未核销余额合计，阻断出表")
        raise ValueError("勾稽校验未通过，请检查规则实现")

    # 10. 写出结果
    logger.info("正在写出结果工作簿...")
    with pd.ExcelWriter(args.out, engine="openpyxl") as writer:
        # 首页 KPI
        pd.DataFrame([kpi]).T.reset_index().rename(columns={"index": "指标", 0: "值"}).to_excel(writer, sheet_name="00_首页KPI", index=False)
        # 分区
        partition_df.to_excel(writer, sheet_name="01_红蓝分区", index=False)
        # 对冲
        offset_df.to_excel(writer, sheet_name="02_对冲记录", index=False)
        # 核销
        writeoff_df.to_excel(writer, sheet_name="03_收款核销", index=False)
        # 账龄
        aging_df.to_excel(writer, sheet_name="04_账龄分层", index=False)
        # 分录
        entry_draft_df.to_excel(writer, sheet_name="05_分录草稿", index=False)
        # 差异池
        diff_rows = []
        for _, r in issues_df.iterrows():
            diff_rows.append({"类型": "校验异常", "描述": r["描述"], "涉及单据": r.get("样例客户") or r.get("样例") or r.get("样例行号")})
        for _, r in conflict_df.iterrows():
            diff_rows.append({"类型": "分区冲突", "描述": f"单据 {r['单据号']} 红蓝判定冲突", "涉及单据": r["单据号"]})
        for _, r in big_blue.iterrows():
            diff_rows.append({"类型": "重大单复核", "描述": f"单张金额 {r['人民币金额（含税）']:.2f} 超 50 万阈值", "涉及单据": r["单据号"]})
        pd.DataFrame(diff_rows).to_excel(writer, sheet_name="06_差异池", index=False)

    # 跑批留痕
    trace = {
        "跑批时间": datetime.now().isoformat(),
        "数据基准日": str(base_date),
        "应收单行数": len(t1),
        "收款单行数": len(t2),
        "红票数": len(red_df),
        "蓝票数": len(blue_df),
        "KPI": kpi,
    }
    with open(args.log, "a", encoding="utf-8") as f:
        f.write("\n[跑批留痕]\n")
        for k, v in trace.items():
            f.write(f"{k}: {v}\n")

    logger.info(f"跑批完成，结果保存至：{args.out}")
    logger.info(f"首页 KPI：{kpi}")


if __name__ == "__main__":
    main()