"""票据合规预审规则引擎（唯一业务内核）。

四步审核（对齐《核心建设内容：各部门"Skill"深度需求规格书》财务处 Skill 001）：
  Step 1 字段完整性校验（F1-F4）
  Step 2 三方交叉验真（V1 真伪状态 / V2 购买方税号白名单 / V3 日期逻辑）
  Step 3 经费标准硬约束（S1 住宿费上限 / S2 交通等级 / S3 项目归属）
  Step 4 风控标记（R1 连号 / R2 同商家集中开票 / R3 节假日开票）

每条规则输出 {rule_id, name, level, detail, rule_ref}；level ∈ pass/warn/reject。
总判定：任一 reject → 驳回；有 warn → 待确认；否则 通过。
预审结果仅作提示，不构成最终报销结论；真伪以人工/官方查验为准。
"""

from __future__ import annotations

import json
import re
import uuid
from datetime import date, datetime, timedelta
from typing import Any, Optional

from db.connection import get_conn, init_db
from util.cny_amount import parse_cny_upper
from util.logging_config import get_skill_logger
from util.timeutil import now_unix

_VERDICT_PASS = "通过"
_VERDICT_WARN = "待确认"
_VERDICT_REJECT = "驳回"

_DATE_PATTERNS = ("%Y-%m-%d", "%Y/%m/%d", "%Y.%m.%d", "%Y年%m月%d日")

# 发票号码已知位数：数电票 20 位；增值税普/专票 8 位；旧版 10/12 位
_KNOWN_NUMBER_LENGTHS = {8, 10, 12, 20}

_INVOICE_KEY_ALIASES = {
    "invoiceType": ("invoice_type", "发票类型", "type"),
    "invoiceCode": ("invoice_code", "发票代码"),
    "invoiceNumber": ("invoice_number", "invoice_no", "发票号码"),
    "issueDate": ("issue_date", "开票日期", "date"),
    "amount": ("amount", "金额", "total_amount", "价税合计"),
    "amountInWords": ("amount_in_words", "大写金额", "amount_words"),
    "taxRate": ("tax_rate", "税率"),
    "buyerName": ("buyer_name", "购买方名称", "buyer"),
    "buyerTaxNo": ("buyer_tax_no", "购买方税号", "buyer_tax_id"),
    "sellerName": ("seller_name", "销售方名称", "seller"),
    "sellerTaxNo": ("seller_tax_no", "销售方税号", "seller_tax_id"),
    "expenseCategory": ("expense_category", "category", "费用类别"),
    "city": ("city", "城市", "消费城市"),
    "days": ("days", "天数", "住宿天数"),
    "ticketClass": ("ticket_class", "座位等级", "席别", "舱位等级"),
    "manualVerifyResult": ("manual_verify_result", "查验结果", "人工查验结果"),
}

_TICKET_CLASS_LEVELS = (
    (("商务座", "头等舱"), 3),
    (("一等座", "软卧", "公务舱", "一等舱"), 2),
    (("二等座", "硬座", "硬卧", "软座", "经济舱"), 1),
)

_CHECK_PLATFORM_HINT = "全国增值税发票查验平台（inv-veri.chinatax.gov.cn）"


def _log():
    try:
        return get_skill_logger()
    except RuntimeError:
        return None


def _norm_invoice(raw: dict[str, Any]) -> dict[str, Any]:
    """将输入发票字段归一化为 camelCase 键（兼容 snake_case 与常见中文键）。"""
    out: dict[str, Any] = {}
    for key, value in raw.items():
        camel = _camel(key)
        if camel in _INVOICE_KEY_ALIASES:
            out.setdefault(camel, value)
            continue
        matched = False
        for canonical, aliases in _INVOICE_KEY_ALIASES.items():
            if key in aliases:
                out.setdefault(canonical, value)
                matched = True
                break
        if not matched:
            out.setdefault(camel, value)
    return out


def _camel(name: str) -> str:
    parts = re.split(r"[\s_]+", str(name).strip())
    if len(parts) <= 1:
        return str(name)
    head, *rest = parts
    return head + "".join(p[:1].upper() + p[1:] for p in rest)


def _to_float(value: Any) -> Optional[float]:
    if value is None or value == "":
        return None
    if isinstance(value, (int, float)):
        return float(value)
    text = str(value).strip()
    text = re.sub(r"[¥￥元人民币\s,，]", "", text)
    try:
        return float(text)
    except ValueError:
        return None


def _parse_date(value: Any) -> Optional[date]:
    if value is None or value == "":
        return None
    if isinstance(value, date):
        return value
    text = str(value).strip()
    for pattern in _DATE_PATTERNS:
        try:
            return datetime.strptime(text, pattern).date()
        except ValueError:
            continue
    return None


def _fmt_date(value: Optional[date]) -> str:
    return value.isoformat() if value else ""


def _load_whitelist_taxnos() -> set[str]:
    conn = get_conn()
    try:
        cur = conn.cursor()
        cur.execute("SELECT tax_no FROM taxid_whitelist")
        return {str(row[0]).strip().upper() for row in cur.fetchall()}
    finally:
        conn.close()


def _load_holiday_dates() -> set[str]:
    conn = get_conn()
    try:
        cur = conn.cursor()
        cur.execute("SELECT holiday_date FROM holiday_calendar")
        return {str(row[0]) for row in cur.fetchall()}
    finally:
        conn.close()


def _resolve_city_tier(city: str) -> Optional[tuple[str, str]]:
    """按城市名归档城市等级，返回 (city_tier, basis)。"""
    if not city:
        return None
    conn = get_conn()
    try:
        cur = conn.cursor()
        cur.execute(
            "SELECT city_tier, city_example FROM expense_standards WHERE category = '住宿费' GROUP BY city_tier, city_example"
        )
        rows = cur.fetchall()
    finally:
        conn.close()
    for city_tier, city_example in rows:
        names = [n.strip() for n in str(city_example or "").split(",") if n.strip()]
        if city in names:
            return city_tier, "标准库城市示例"
    return None


def _lookup_standard(category: str, city_tier: Optional[str], staff_title: Optional[str]) -> Optional[dict[str, Any]]:
    conn = get_conn()
    try:
        cur = conn.cursor()
        sql = "SELECT city_tier, staff_title, max_amount, unit, basis_doc, clause FROM expense_standards WHERE category = ?"
        params: list[Any] = [category]
        if city_tier:
            sql += " AND city_tier = ?"
            params.append(city_tier)
        if staff_title:
            sql += " AND staff_title = ?"
            params.append(staff_title)
        cur.execute(sql + " LIMIT 1", params)
        row = cur.fetchone()
    finally:
        conn.close()
    if not row:
        return None
    return {
        "city_tier": row[0],
        "staff_title": row[1],
        "max_amount": float(row[2]),
        "unit": row[3] or "",
        "basis_doc": row[4] or "",
        "clause": row[5] or "",
    }


def _ref(basis_doc: str, clause: str) -> str:
    if basis_doc and clause:
        return f"{basis_doc}{clause}"
    return basis_doc or clause or ""


def _item(rule_id: str, name: str, level: str, detail: str, rule_ref: str = "") -> dict[str, str]:
    return {"rule_id": rule_id, "name": name, "level": level, "detail": detail, "rule_ref": rule_ref}


# ---------------------------------------------------------------------------
# Step 1 字段完整性校验
# ---------------------------------------------------------------------------

def _check_field_completeness(inv: dict[str, Any], today: date) -> list[dict[str, str]]:
    items: list[dict[str, str]] = []
    missing = [
        label
        for key, label in (
            ("invoiceNumber", "发票号码"),
            ("issueDate", "开票日期"),
            ("amount", "金额"),
            ("sellerName", "销售方名称"),
        )
        if inv.get(key) in (None, "")
    ]
    if missing:
        items.append(
            _item(
                "F1",
                "必填字段齐全",
                "warn",
                f"以下关键字段缺失或为空：{'、'.join(missing)}。请重新拍摄/补录后再提交报销。",
                "票据预审·字段校验",
            )
        )

    number = str(inv.get("invoiceNumber") or "").strip().replace(" ", "")
    if number and not number.isdigit():
        items.append(_item("F2", "发票号码格式", "warn", f"发票号码含非数字字符：{number}", "票据预审·字段校验"))
    elif number and len(number) not in _KNOWN_NUMBER_LENGTHS:
        items.append(
            _item(
                "F2",
                "发票号码格式",
                "warn",
                f"发票号码 {len(number)} 位，不属于已知位数（数电票 20 位、普/专票 8 位、旧版 10/12 位），疑似异常，请人工核验票面。",
                "票据预审·字段校验",
            )
        )

    issue_date = _parse_date(inv.get("issueDate"))
    raw_date = str(inv.get("issueDate") or "").strip()
    if raw_date and issue_date is None:
        items.append(_item("F3", "开票日期合法性", "reject", f"开票日期无法解析：{raw_date}（应为 YYYY-MM-DD）", "票据预审·字段校验"))
    elif issue_date and issue_date > today:
        items.append(_item("F3", "开票日期合法性", "reject", f"开票日期 {issue_date.isoformat()} 晚于今日，票面异常。", "票据预审·字段校验"))
    elif issue_date:
        items.append(_item("F3", "开票日期合法性", "pass", f"开票日期 {issue_date.isoformat()} 合法。", "票据预审·字段校验"))

    amount = _to_float(inv.get("amount"))
    words = str(inv.get("amountInWords") or "").strip()
    if amount is not None and words:
        words_value = parse_cny_upper(words)
        if words_value is None:
            items.append(_item("F4", "大小写金额一致", "warn", f"大写金额“{words}”无法解析，请人工核对大小写金额。", "票据预审·字段校验"))
        elif abs(words_value - amount) > 0.01:
            items.append(
                _item(
                    "F4",
                    "大小写金额一致",
                    "reject",
                    f"大写金额 {words_value:.2f} 元与小写金额 {amount:.2f} 元不一致，票面异常。",
                    "票据预审·字段校验",
                )
            )
        else:
            items.append(_item("F4", "大小写金额一致", "pass", f"大小写金额一致（{amount:.2f} 元）。", "票据预审·字段校验"))
    return items


# ---------------------------------------------------------------------------
# Step 2 三方交叉验真
# ---------------------------------------------------------------------------

def _check_cross_verification(
    inv: dict[str, Any],
    context: dict[str, Any],
    whitelist: set[str],
    today: date,
) -> list[dict[str, str]]:
    items: list[dict[str, str]] = []

    verify = str(inv.get("manualVerifyResult") or "").strip().lower()
    if verify in ("fail", "false", "不符", "假票", "查验不符"):
        items.append(_item("V1", "发票真伪", "reject", "人工查验结果为“不符”，判定为问题票据。", _CHECK_PLATFORM_HINT))
    elif verify in ("pass", "true", "通过", "真票", "查验通过"):
        items.append(_item("V1", "发票真伪", "pass", "人工查验结果为“通过”。", _CHECK_PLATFORM_HINT))
    else:
        items.append(
            _item(
                "V1",
                "发票真伪",
                "warn",
                f"尚未查验真伪：请经办人到{_CHECK_PLATFORM_HINT}录入票面信息核验，并在本报告回填查验结果。",
                _CHECK_PLATFORM_HINT,
            )
        )

    buyer_tax_no = str(inv.get("buyerTaxNo") or "").strip()
    if not buyer_tax_no:
        items.append(_item("V2", "购买方税号白名单", "warn", "票面缺少购买方税号，无法进行白名单比对，请补录抬头信息。", "票据预审·税号白名单"))
    elif buyer_tax_no.strip().upper() in whitelist:
        buyer_name = str(inv.get("buyerName") or "").strip()
        items.append(_item("V2", "购买方税号白名单", "pass", f"购买方税号在我校准予报销白名单内{('（' + buyer_name + '）') if buyer_name else ''}。", "票据预审·税号白名单"))
    else:
        items.append(
            _item(
                "V2",
                "购买方税号白名单",
                "reject",
                f"购买方税号 {buyer_tax_no} 不在我校准予报销税号白名单内，抬头开错，须换开后方可报销。",
                "票据预审·税号白名单",
            )
        )

    issue_date = _parse_date(inv.get("issueDate"))
    trip_start = _parse_date(context.get("trip_start"))
    trip_end = _parse_date(context.get("trip_end"))
    if issue_date and trip_start:
        window_end = trip_end or trip_start
        if issue_date < trip_start:
            items.append(
                _item(
                    "V3",
                    "日期逻辑（先审批后出行）",
                    "warn",
                    f"开票日期 {issue_date.isoformat()} 早于出差开始日 {trip_start.isoformat()}，请核实行程与审批单（先审批后出行）。",
                    "票据预审·日期逻辑",
                )
            )
        elif issue_date > window_end + timedelta(days=1):
            items.append(
                _item(
                    "V3",
                    "日期逻辑（先审批后出行）",
                    "warn",
                    f"开票日期 {issue_date.isoformat()} 晚于出差结束日 {window_end.isoformat()}（含 1 天宽限），请核实行程。",
                    "票据预审·日期逻辑",
                )
            )
        else:
            items.append(
                _item(
                    "V3",
                    "日期逻辑（先审批后出行）",
                    "pass",
                    f"开票日期落在出差区间 [{trip_start.isoformat()}, {(trip_end or trip_start).isoformat()}] 内。",
                    "票据预审·日期逻辑",
                )
            )
    return items


# ---------------------------------------------------------------------------
# Step 3 经费标准硬约束
# ---------------------------------------------------------------------------

def _check_expense_standards(inv: dict[str, Any], context: dict[str, Any]) -> list[dict[str, str]]:
    items: list[dict[str, str]] = []
    category = str(inv.get("expenseCategory") or "").strip()
    amount = _to_float(inv.get("amount"))
    staff_title = str(context.get("staff_title") or "").strip()

    if "住宿" in category and amount is not None:
        city = str(context.get("trip_city") or inv.get("city") or "").strip()
        tier_info = _resolve_city_tier(city) if city else None
        if not city:
            items.append(
                _item(
                    "S1",
                    "住宿费标准比对",
                    "warn",
                    "未提供消费城市，无法匹配住宿费标准，请补录城市后重新预审或人工对照《差旅费管理办法》。",
                    _ref("《差旅费管理办法》", "第十条"),
                )
            )
        elif not tier_info:
            items.append(
                _item(
                    "S1",
                    "住宿费标准比对",
                    "warn",
                    f"城市“{city}”不在标准库城市等级中，无法匹配住宿费标准，请人工对照《差旅费管理办法》或请管理员维护该城市。",
                    _ref("《差旅费管理办法》", "第十条"),
                )
            )
        else:
            standard = _lookup_standard("住宿费", tier_info[0], staff_title or None)
            days = _to_float(inv.get("days")) or 1.0
            days = max(days, 1.0)
            if not standard:
                items.append(
                    _item(
                        "S1",
                        "住宿费标准比对",
                        "warn",
                        f"未匹配到“{tier_info[0]}/{staff_title or '未提供职级'}”的住宿费标准，请人工对照《差旅费管理办法》。",
                        _ref("《差旅费管理办法》", "第十条"),
                    )
                )
            else:
                per_day = amount / days
                limit = standard["max_amount"]
                ref = _ref(standard["basis_doc"], standard["clause"])
                if per_day > limit + 0.001:
                    items.append(
                        _item(
                            "S1",
                            "住宿费标准比对",
                            "reject",
                            (
                                f"住宿费折合 {per_day:.2f} 元/人/天，超过“{standard['city_tier']}/{standard['staff_title']}”"
                                f"上限 {limit:.2f} 元/人/天，超标 {per_day - limit:.2f} 元。"
                            ),
                            ref,
                        )
                    )
                else:
                    items.append(
                        _item(
                            "S1",
                            "住宿费标准比对",
                            "pass",
                            f"住宿费折合 {per_day:.2f} 元/人/天，未超“{standard['city_tier']}/{standard['staff_title']}”上限 {limit:.2f} 元/人/天。",
                            ref,
                        )
                    )

    if "交通" in category or "出租" in str(inv.get("invoiceType") or ""):
        ticket_class = str(inv.get("ticketClass") or "").strip()
        standard = _lookup_standard("交通费", None, staff_title or None)
        if ticket_class and standard:
            class_level = _ticket_class_level(ticket_class)
            ref = _ref(standard["basis_doc"], standard["clause"])
            if class_level is None:
                items.append(_item("S2", "交通席位等级", "warn", f"席位/舱位等级“{ticket_class}”无法识别，请人工对照职级可乘等级。", ref))
            elif class_level > standard["max_amount"]:
                items.append(
                    _item(
                        "S2",
                        "交通席位等级",
                        "warn",
                        f"“{ticket_class}”等级高于“{standard['staff_title']}”可乘等级（代码 {standard['max_amount']:.0f}），请确认是否符合职级规定或已获审批。",
                        ref,
                    )
                )
            else:
                items.append(_item("S2", "交通席位等级", "pass", f"“{ticket_class}”符合“{standard['staff_title']}”可乘等级。", ref))

    project_no = str(context.get("project_no") or "").strip()
    if project_no:
        items.append(
            _item(
                "S3",
                "项目经费归属",
                "pass",
                f"本笔将计入预算项目 {project_no}；项目余额校验待财务系统对接后启用。",
                "票据预审·项目归属",
            )
        )
    return items


def _ticket_class_level(ticket_class: str) -> Optional[int]:
    for keywords, level in _TICKET_CLASS_LEVELS:
        if any(k in ticket_class for k in keywords):
            return level
    return None


# ---------------------------------------------------------------------------
# Step 4 风控标记
# ---------------------------------------------------------------------------

def _check_holiday(inv: dict[str, Any], holidays: set[str]) -> list[dict[str, str]]:
    items: list[dict[str, str]] = []
    issue_date = _parse_date(inv.get("issueDate"))
    if not issue_date:
        return items
    iso = issue_date.isoformat()
    if iso in holidays:
        items.append(
            _item("R3", "节假日开票", "warn", f"开票日期 {iso} 为法定节假日，请核实消费真实性。", "票据预审·风控标记")
        )
    elif issue_date.weekday() >= 5:
        items.append(
            _item("R3", "节假日开票", "warn", f"开票日期 {iso} 为周末（{'周六' if issue_date.weekday() == 5 else '周日'}），请核实消费真实性。", "票据预审·风控标记")
        )
    return items


def _check_batch_risks(invoices: list[dict[str, Any]]) -> dict[int, list[dict[str, str]]]:
    """R1 连号发票 + R2 同商家集中开票（批内）。返回 invoice index → items。"""
    extra: dict[int, list[dict[str, str]]] = {}

    # R1：同发票代码且号码连续（差值 ≤ 2）
    numeric_by_code: dict[str, list[tuple[int, int]]] = {}
    for idx, inv in enumerate(invoices):
        code = str(inv.get("invoiceCode") or "").strip() or "-"
        number = str(inv.get("invoiceNumber") or "").strip()
        if number.isdigit():
            numeric_by_code.setdefault(code, []).append((idx, int(number)))
    for _code, entries in numeric_by_code.items():
        for idx, number in entries:
            peers = [n for j, n in entries if j != idx and abs(n - number) <= 2]
            if peers:
                extra.setdefault(idx, []).append(
                    _item(
                        "R1",
                        "连号发票",
                        "warn",
                        f"与本批发票号码连号（相邻号码：{', '.join(str(p) for p in sorted(set(peers)))}），请核实是否拆分消费。",
                        "票据预审·风控标记",
                    )
                )

    # R2：同商家集中开票（批内按销方税号，缺税号回退销方名称）
    sellers: dict[str, list[int]] = {}
    for idx, inv in enumerate(invoices):
        key = str(inv.get("sellerTaxNo") or "").strip().upper() or str(inv.get("sellerName") or "").strip()
        if key:
            sellers.setdefault(key, []).append(idx)
    for _key, idx_list in sellers.items():
        if len(idx_list) >= 3:
            sample_name = str(invoices[idx_list[0]].get("sellerName") or "同一销售方").strip()
            for idx in idx_list:
                extra.setdefault(idx, []).append(
                    _item(
                        "R2",
                        "同商家集中开票",
                        "warn",
                        f"本批有 {len(idx_list)} 张发票来自“{sample_name}”，存在集中开票情形，请核实消费真实性。",
                        "票据预审·风控标记",
                    )
                )
    return extra


def _check_seller_history(inv: dict[str, Any]) -> Optional[dict[str, str]]:
    """R2 历史扩展：该销方近 30 天（含本张）累计 ≥ 5 张 → 黄灯。"""
    seller_tax_no = str(inv.get("sellerTaxNo") or "").strip()
    if not seller_tax_no:
        return None
    conn = get_conn()
    try:
        cur = conn.cursor()
        cur.execute(
            """
            SELECT COUNT(*) FROM invoice_records
            WHERE seller_tax_no = ? AND created_at >= ?
            """,
            (seller_tax_no, now_unix() - 30 * 24 * 3600),
        )
        history = int(cur.fetchone()[0])
    finally:
        conn.close()
    if history + 1 >= 5:
        seller_name = str(inv.get("sellerName") or seller_tax_no).strip()
        return _item(
            "R2",
            "同商家集中开票",
            "warn",
            f"销售方“{seller_name}”近 30 天累计开票 {history + 1} 张（含本张），存在集中开票情形，请核实消费真实性。",
            "票据预审·风控标记",
        )
    return None


# ---------------------------------------------------------------------------
# 内核入口
# ---------------------------------------------------------------------------

def _decide(items: list[dict[str, str]]) -> str:
    levels = {i["level"] for i in items}
    if "reject" in levels:
        return _VERDICT_REJECT
    if "warn" in levels:
        return _VERDICT_WARN
    return _VERDICT_PASS


def _summarize(verdict: str, invoice_number: str, items: list[dict[str, str]]) -> str:
    rejects = [i for i in items if i["level"] == "reject"]
    warns = [i for i in items if i["level"] == "warn"]
    if verdict == _VERDICT_REJECT:
        return f"发票 {invoice_number} 预审驳回：{'；'.join(i['name'] for i in rejects[:3])}"
    if verdict == _VERDICT_WARN:
        return f"发票 {invoice_number} 待确认：{'；'.join(i['name'] for i in warns[:3])}"
    return f"发票 {invoice_number} 预审通过，全部检查项通过。"


def _validate_input(invoices: list[dict[str, Any]]) -> None:
    if not invoices:
        raise ValueError("发票字段为空：请提供至少一张发票的结构化字段 JSON")
    if len(invoices) > 50:
        raise ValueError(f"单批发票数量过多（{len(invoices)} 张），单批上限 50 张")
    for idx, inv in enumerate(invoices):
        if not isinstance(inv, dict):
            raise ValueError(f"第 {idx + 1} 张发票字段不是 JSON 对象")


def run_prereview(invoices_raw: list[dict[str, Any]], context: dict[str, Any]) -> dict[str, Any]:
    """预审内核：四步审核 → 写库 → 生成《初审意见单》HTML。"""
    from service.report_service import write_report_for_kernel

    log = _log()
    init_db()
    invoices = [_norm_invoice(inv) for inv in invoices_raw]
    _validate_input(invoices)

    batch_no = now_unix().to_bytes(5, "big").hex() + uuid.uuid4().hex[:6]
    today = date.today()
    whitelist = _load_whitelist_taxnos()
    holidays = _load_holiday_dates()

    if log:
        log.info("prereview_kernel_start batch_no=%s count=%s stage=kernel", batch_no, len(invoices))

    batch_extra = _check_batch_risks(invoices)
    results: list[dict[str, Any]] = []

    conn = get_conn()
    try:
        cur = conn.cursor()
        for idx, inv in enumerate(invoices):
            items: list[dict[str, str]] = []
            items.extend(_check_field_completeness(inv, today))
            items.extend(_check_cross_verification(inv, context, whitelist, today))
            items.extend(_check_expense_standards(inv, context))
            items.extend(_check_holiday(inv, holidays))
            items.extend(batch_extra.get(idx, []))

            seller_key = str(inv.get("sellerTaxNo") or "").strip()
            if seller_key:
                history_item = _check_seller_history(inv)
                if history_item and not any(i["rule_id"] == "R2" for i in items):
                    items.append(history_item)

            verdict = _decide(items)
            invoice_number = str(inv.get("invoiceNumber") or "").strip() or f"未提供-{idx + 1}"

            cur.execute(
                """
                INSERT INTO invoice_records
                    (batch_no, invoice_type, invoice_code, invoice_number, issue_date, amount,
                     amount_in_words, tax_rate, buyer_name, buyer_tax_no, seller_name, seller_tax_no,
                     expense_category, city, days, purpose, project_no, trip_context_json)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    batch_no,
                    str(inv.get("invoiceType") or ""),
                    str(inv.get("invoiceCode") or ""),
                    invoice_number,
                    _fmt_date(_parse_date(inv.get("issueDate"))),
                    _to_float(inv.get("amount")),
                    str(inv.get("amountInWords") or ""),
                    _to_float(inv.get("taxRate")),
                    str(inv.get("buyerName") or ""),
                    str(inv.get("buyerTaxNo") or ""),
                    str(inv.get("sellerName") or ""),
                    str(inv.get("sellerTaxNo") or ""),
                    str(inv.get("expenseCategory") or ""),
                    str(inv.get("city") or ""),
                    int(_to_float(inv.get("days")) or 0),
                    str(context.get("purpose") or ""),
                    str(context.get("project_no") or ""),
                    json.dumps(context, ensure_ascii=False),
                ),
            )
            invoice_id = int(cur.lastrowid)

            summary = _summarize(verdict, invoice_number, items)
            cur.execute(
                """
                INSERT INTO prereview_reports
                    (batch_no, invoice_id, invoice_number, seller_name, amount, verdict,
                     pass_json, warn_json, reject_json, summary)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    batch_no,
                    invoice_id,
                    invoice_number,
                    str(inv.get("sellerName") or ""),
                    _to_float(inv.get("amount")),
                    verdict,
                    json.dumps([i for i in items if i["level"] == "pass"], ensure_ascii=False),
                    json.dumps([i for i in items if i["level"] == "warn"], ensure_ascii=False),
                    json.dumps([i for i in items if i["level"] == "reject"], ensure_ascii=False),
                    summary,
                ),
            )
            report_id = int(cur.lastrowid)

            report_view = {
                "id": report_id,
                "batch_no": batch_no,
                "invoice_number": invoice_number,
                "seller_name": str(inv.get("sellerName") or ""),
                "amount": _to_float(inv.get("amount")),
                "verdict": verdict,
                "summary": summary,
                "verify_status": "待查验",
                "created_at": now_unix(),
                "pass_json": json.dumps([i for i in items if i["level"] == "pass"], ensure_ascii=False),
                "warn_json": json.dumps([i for i in items if i["level"] == "warn"], ensure_ascii=False),
                "reject_json": json.dumps([i for i in items if i["level"] == "reject"], ensure_ascii=False),
            }
            invoice_view = {
                "invoice_type": str(inv.get("invoiceType") or ""),
                "invoice_code": str(inv.get("invoiceCode") or ""),
                "invoice_number": invoice_number,
                "issue_date": _fmt_date(_parse_date(inv.get("issueDate"))),
                "amount": _to_float(inv.get("amount")),
                "buyer_name": str(inv.get("buyerName") or ""),
                "buyer_tax_no": str(inv.get("buyerTaxNo") or ""),
                "seller_name": str(inv.get("sellerName") or ""),
                "seller_tax_no": str(inv.get("sellerTaxNo") or ""),
                "expense_category": str(inv.get("expenseCategory") or ""),
                "city": str(inv.get("city") or ""),
            }
            report_file = write_report_for_kernel(
                report_id=report_id,
                report=report_view,
                invoice=invoice_view,
                context=context,
            )
            cur.execute(
                "UPDATE prereview_reports SET report_file = ? WHERE id = ?",
                (report_file, report_id),
            )

            results.append(
                {
                    "report_id": report_id,
                    "batch_no": batch_no,
                    "invoice_id": invoice_id,
                    "invoice_number": invoice_number,
                    "verdict": verdict,
                    "summary": summary,
                    "items": items,
                    "report_file": report_file,
                }
            )
        conn.commit()
    finally:
        conn.close()

    counts = {v: sum(1 for r in results if r["verdict"] == v) for v in (_VERDICT_PASS, _VERDICT_WARN, _VERDICT_REJECT)}
    payload = {
        "ok": True,
        "batch_no": batch_no,
        "count": len(results),
        "verdict_counts": counts,
        "disclaimer": "预审结果仅作提示，不构成最终报销结论；发票真伪以官方查验结果为准。",
        "reports": results,
    }
    if log:
        log.info(
            "prereview_kernel_done batch_no=%s pass=%s warn=%s reject=%s stage=kernel",
            batch_no,
            counts[_VERDICT_PASS],
            counts[_VERDICT_WARN],
            counts[_VERDICT_REJECT],
        )
    return payload