#!/usr/bin/env python3
"""Convert fixed Opera RES_DETAIL XML into XLSX reports and a finance-ready audit payload."""

from __future__ import annotations

import argparse
import copy
import hashlib
import json
import os
import re
import subprocess
import sys
import traceback
import xml.etree.ElementTree as ET
from dataclasses import dataclass
from datetime import date, datetime
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple

from openpyxl import Workbook, load_workbook
from openpyxl.cell.cell import Cell
from openpyxl.styles import Alignment, Font
from openpyxl.utils import get_column_letter


VERSION = "1.0"
DAILY_RESULT_VERSION = "2.0"
PROCESSOR_VERSION = "2.2.0"
STRUCTURED_RESULT_SCHEMA_VERSION = "1.0"
DAILY_STRUCTURED_RESULT_SCHEMA_VERSION = "2.0"
MODE_DAILY = "daily"
MODE_DAILY_MONTHLY = "daily-monthly"
PROCESSING_MODES = {MODE_DAILY, MODE_DAILY_MONTHLY}
SKILL_ROOT = Path(__file__).resolve().parent.parent
PRICE_REFERENCE = SKILL_ROOT / "references" / "价格对照.xlsx"
DAILY_TEMPLATE = SKILL_ROOT / "assets" / "daily-template.xlsx"
MONTHLY_TEMPLATE = SKILL_ROOT / "assets" / "channel-report-template.xlsx"
STRUCTURED_RESULT_SCHEMA = SKILL_ROOT / "references" / "structured-result.schema.json"

RULE_SET_PATHS = (
    SKILL_ROOT / "SKILL.md",
    SKILL_ROOT / "scripts" / "process_reports.py",
    SKILL_ROOT / "scripts" / "validate_outputs.py",
    SKILL_ROOT / "references" / "business-rules.md",
    SKILL_ROOT / "references" / "field-contracts.md",
    SKILL_ROOT / "references" / "error-contract.md",
    SKILL_ROOT / "references" / "codex-result.schema.json",
    STRUCTURED_RESULT_SCHEMA,
    PRICE_REFERENCE,
    DAILY_TEMPLATE,
    MONTHLY_TEMPLATE,
)

FINAL_OUTCOMES = {
    "retained",
    "excluded_rate_code",
    "duplicate",
    "validation_failed",
    "price_unmatched",
}

DAILY_HEADERS = [
    "BLOCK_CODE",
    "ADULTS",
    "CHILDREN",
    "COMPANY_NAME",
    "CONFIRMATION_NO",
    "DISP_ROOM_NO",
    "EFFECTIVE_RATE_AMOUNT",
    "FULL_NAME",
    "RES_COMMENT",
    "TRACE_TEXT",
    "NO_OF_ROOMS",
    "PRODUCTS",
    "RATE_CODE",
    "ROOM_CATEGORY_LABEL",
    "ARRIVAL",
    "DEPARTURE",
    "NIGHTS",
    "REAL PRICE",
    "TOTAL PRICE",
]

CHANNEL_HEADERS = [
    "ARRIVAL",
    "DEPARTURE",
    "NIGHTS",
    "ADULTS",
    "CHILDREN",
    "BLOCK_CODE",
    "NO_OF_ROOMS",
    "COMPANY_NAME",
    "CONFIRMATION_NO",
    "DISP_ROOM_NO",
    "RATE_AMOUNT",
    "FULL_NAME",
    "RES_COMMENT",
    "TRACE_TEXT",
    "PRODUCTS",
    "RATE_CODE",
    "ROOM_CATEGORY_LABEL",
    "REAL PRICE",
    "TOTAL PRICE",
]

KB_HEADER = "KB（100/晚/间）"
KB_SHEET = "DY-AI-Easy-KB"
KB_CHANNEL_HEADERS = CHANNEL_HEADERS + [KB_HEADER]

RATE_WHITELIST = {
    "GRPA1",
    "GRPA2",
    "GRPA3",
    "GRPA4",
    "GRP1",
    "WHO1",
    "WHO2",
    "WHO3",
    "WHO4",
    "LTLT",
    "LBLT",
    "LBSM",
    "LBMS",
    "LBW1",
    "LBKB",
    "LBLS",
    "WHKR2100B",
    "GL2100B",
    "GL2200KR",
    "GLSPCB",
}

STANDARD_SHEETS = ["LIANTAI-GROUP", "LIANTAI-FIT", "QBD", KB_SHEET, "FENGRUN"]
COMPANY_KEYWORD_GROUPS = (
    ("LIAN TAI", ("LIANTAI",)),
    ("QBD", ("QBD",)),
    ("RAINBOW/AI", ("RAINBOW",)),
    ("FENGRUN", ("FENGRUN",)),
    ("HANA TOUR", ("HANATOUR", "HANA")),
    ("HONGTAI", ("HONGTAI",)),
    ("GUANGZHOU GO EASY", ("GUANGZHOUGOEASY", "GOEASY")),
)
KB_CHANNEL_COMPANIES = {"RAINBOW/AI", "GUANGZHOU GO EASY"}
ZERO_TOTAL_COMPANIES = KB_CHANNEL_COMPANIES
ZERO_TOTAL_RATE_CODES = {"LBMS", "LBSM"}
COMPANY_COMPACT_RE = re.compile(r"[^A-Z0-9]+")
INVALID_SHEET_CHARS = re.compile(r"[:\\/?*\[\]]")
EXCEPTION_HEADERS = [
    "ERROR_CODE",
    "STAGE",
    "SOURCE_LOCATION",
    "COMPANY_NAME",
    "RATE_CODE",
    "EFFECTIVE_RATE_AMOUNT",
    "CONFIRMATION_NO",
    "MESSAGE",
]


@dataclass
class ErrorItem:
    code: str
    stage: str
    message: str
    source_location: Optional[str] = None
    company_name: Optional[str] = None
    rate_code: Optional[str] = None
    effective_rate_amount: Optional[Decimal] = None
    confirmation_no: Optional[str] = None

    def to_dict(self) -> Dict[str, Any]:
        return {
            "code": self.code,
            "stage": self.stage,
            "source_location": self.source_location,
            "company_name": self.company_name,
            "rate_code": self.rate_code,
            "effective_rate_amount": decimal_to_number(self.effective_rate_amount),
            "confirmation_no": self.confirmation_no,
            "message": self.message,
        }


class ProcessingFailure(Exception):
    def __init__(self, errors: Sequence[ErrorItem], exit_code: int = 2):
        self.errors = list(errors)
        self.exit_code = exit_code
        super().__init__(self.errors[0].message if self.errors else "处理失败")


class CompanyKeywordAmbiguity(ValueError):
    def __init__(self, company_name: str, matches: Sequence[str]):
        self.company_name = company_name
        self.matches = tuple(matches)
        super().__init__(f"公司名同时命中多个关键词组：{', '.join(self.matches)}")


def text_or_blank(value: Any) -> str:
    return "" if value is None else str(value).strip()


def decimal_to_number(value: Optional[Decimal]) -> Optional[Any]:
    if value is None:
        return None
    if value == value.to_integral_value():
        return int(value)
    return float(value)


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def rule_set_sha256() -> str:
    """Hash the exact package resources that determine processing or validation."""

    digest = hashlib.sha256()
    for path in sorted(RULE_SET_PATHS, key=lambda item: item.relative_to(SKILL_ROOT).as_posix()):
        relative = path.relative_to(SKILL_ROOT).as_posix().encode("utf-8")
        digest.update(len(relative).to_bytes(4, "big"))
        digest.update(relative)
        content_hash = bytes.fromhex(sha256_file(path))
        digest.update(content_hash)
    return digest.hexdigest()


def normalize_group_code(res_comment: Any) -> Optional[str]:
    value = text_or_blank(res_comment).upper()
    return value or None


def append_decision(record: Dict[str, Any], code: str) -> None:
    decisions = record.setdefault("_DECISION_CODES", [])
    if code not in decisions:
        decisions.append(code)


def parse_decimal(value: Any) -> Decimal:
    if value is None or (isinstance(value, str) and not value.strip()):
        raise ValueError("missing numeric value")
    try:
        number = Decimal(str(value).replace(",", "").strip())
    except (InvalidOperation, AttributeError, ValueError) as exc:
        raise ValueError("invalid numeric value") from exc
    if not number.is_finite():
        raise ValueError("non-finite numeric value")
    return number.normalize()


def parse_integer(value: Any) -> int:
    number = parse_decimal(value)
    if number != number.to_integral_value():
        raise ValueError("expected integer")
    return int(number)


def parse_date_value(value: Any) -> date:
    if isinstance(value, datetime):
        return value.date()
    if isinstance(value, date):
        return value
    raw = text_or_blank(value)
    if not raw:
        raise ValueError("missing date")
    for fmt in ("%d-%b-%y", "%d-%m-%y", "%Y%m%d", "%Y-%m-%d", "%m/%d/%Y"):
        try:
            return datetime.strptime(raw, fmt).date()
        except ValueError:
            continue
    raise ValueError("unsupported date format")


def first_nonempty(nodes: Iterable[ET.Element]) -> str:
    for node in nodes:
        value = text_or_blank(node.text)
        if value:
            return value
    return ""


def safe_text_cell(cell: Cell, value: Any) -> None:
    if value is None or value == "":
        cell.value = None
        return
    cell.value = str(value)
    cell.data_type = "s"


def numeric_cell(cell: Cell, value: Any) -> None:
    cell.value = value
    cell.data_type = "n"


def is_within(child: Path, parent: Path) -> bool:
    try:
        return os.path.commonpath([str(child.resolve()), str(parent.resolve())]) == str(parent.resolve())
    except ValueError:
        return False


def ensure_absolute_file(path: Path, suffix: str, code: str, label: str) -> None:
    if not path.is_absolute():
        raise ProcessingFailure(
            [ErrorItem(code, "input", f"{label}必须使用绝对路径", str(path))], exit_code=3
        )
    if not path.is_file():
        raise ProcessingFailure(
            [ErrorItem(code, "input", f"找不到{label}", str(path))], exit_code=3
        )
    if path.suffix.lower() != suffix.lower():
        raise ProcessingFailure(
            [ErrorItem(code, "input", f"{label}扩展名必须为 {suffix}", str(path))], exit_code=3
        )


def validate_invocation(
    xml_path: Path,
    monthly_base: Optional[Path],
    output_dir: Path,
    result_json: Path,
    structured_result_json: Path,
    processing_mode: str,
) -> None:
    if processing_mode not in PROCESSING_MODES:
        raise ProcessingFailure(
            [
                ErrorItem(
                    "INPUT_MODE_INVALID",
                    "input",
                    f"mode 必须是 {MODE_DAILY} 或 {MODE_DAILY_MONTHLY}",
                    processing_mode,
                )
            ],
            exit_code=3,
        )
    ensure_absolute_file(xml_path, ".xml", "INPUT_XML_INVALID", "XML文件")
    if monthly_base is not None:
        if processing_mode == MODE_DAILY:
            raise ProcessingFailure(
                [
                    ErrorItem(
                        "INPUT_MONTHLY_FORBIDDEN",
                        "input",
                        "daily 模式不接受 monthly-base；月报必须由数据库程序生成",
                        str(monthly_base),
                    )
                ],
                exit_code=3,
            )
        ensure_absolute_file(monthly_base, ".xlsx", "INPUT_MONTHLY_INVALID", "月报文件")
    if not output_dir.is_absolute():
        raise ProcessingFailure(
            [ErrorItem("INPUT_OUTPUT_DIR_INVALID", "input", "输出目录必须使用绝对路径", str(output_dir))],
            exit_code=3,
        )
    output_dir.mkdir(parents=True, exist_ok=True)
    if not output_dir.is_dir():
        raise ProcessingFailure(
            [ErrorItem("INPUT_OUTPUT_DIR_INVALID", "input", "输出路径不是目录", str(output_dir))],
            exit_code=3,
        )
    if not result_json.is_absolute() or not is_within(result_json, output_dir):
        raise ProcessingFailure(
            [
                ErrorItem(
                    "INPUT_RESULT_PATH_INVALID",
                    "input",
                    "result-json 必须是输出目录内的绝对路径",
                    str(result_json),
                )
            ],
            exit_code=3,
        )
    if result_json.suffix.lower() != ".json":
        raise ProcessingFailure(
            [ErrorItem("INPUT_RESULT_PATH_INVALID", "input", "result-json 扩展名必须为 .json", str(result_json))],
            exit_code=3,
        )
    if (
        not structured_result_json.is_absolute()
        or not is_within(structured_result_json, output_dir)
        or structured_result_json.suffix.lower() != ".json"
        or structured_result_json.resolve() == result_json.resolve()
    ):
        raise ProcessingFailure(
            [
                ErrorItem(
                    "INPUT_STRUCTURED_RESULT_PATH_INVALID",
                    "input",
                    "structured-result-json 必须是输出目录内且不同于result-json的绝对 .json 路径",
                    str(structured_result_json),
                )
            ],
            exit_code=3,
        )


def parse_group_date(group: ET.Element, index: int) -> date:
    sort_value = text_or_blank(group.findtext("GROUPBY1_SORT_COL"))
    display_value = text_or_blank(group.findtext("GROUPBY1_COL"))
    parsed: List[date] = []
    errors: List[ErrorItem] = []
    if sort_value:
        try:
            parsed.append(parse_date_value(sort_value))
        except ValueError:
            errors.append(
                ErrorItem(
                    "XML_GROUP_DATE_INVALID",
                    "xml",
                    "GROUPBY1_SORT_COL 日期无效",
                    f"group[{index}]/GROUPBY1_SORT_COL",
                )
            )
    if display_value:
        try:
            parsed.append(parse_date_value(display_value))
        except ValueError:
            errors.append(
                ErrorItem(
                    "XML_GROUP_DATE_INVALID",
                    "xml",
                    "GROUPBY1_COL 日期无效",
                    f"group[{index}]/GROUPBY1_COL",
                )
            )
    if errors:
        raise ProcessingFailure(errors)
    if not parsed:
        raise ProcessingFailure(
            [
                ErrorItem(
                    "XML_GROUP_DATE_MISSING",
                    "xml",
                    "XML分组缺少业务日期",
                    f"group[{index}]",
                )
            ]
        )
    if len(set(parsed)) != 1:
        raise ProcessingFailure(
            [
                ErrorItem(
                    "XML_GROUP_DATE_CONFLICT",
                    "xml",
                    "同一分组的两个业务日期字段不一致",
                    f"group[{index}]",
                )
            ]
        )
    return parsed[0]


def read_xml(xml_path: Path) -> Tuple[date, List[ET.Element]]:
    raw = xml_path.read_bytes()
    upper = raw.upper()
    if b"<!DOCTYPE" in upper or b"<!ENTITY" in upper:
        raise ProcessingFailure(
            [
                ErrorItem(
                    "INPUT_XML_UNSAFE_DECLARATION",
                    "input",
                    "XML包含不允许的DOCTYPE或ENTITY声明",
                    xml_path.name,
                )
            ],
            exit_code=3,
        )
    try:
        root = ET.fromstring(raw)
    except ET.ParseError as exc:
        raise ProcessingFailure(
            [ErrorItem("XML_PARSE_ERROR", "xml", f"XML无法解析：{exc}", xml_path.name)], exit_code=3
        )
    if root.tag != "RES_DETAIL":
        raise ProcessingFailure(
            [ErrorItem("XML_ROOT_MISMATCH", "xml", "XML根节点必须为 RES_DETAIL", root.tag)]
        )
    groups = root.findall("./LIST_G_GROUP_BY1/G_GROUP_BY1")
    if not groups:
        raise ProcessingFailure(
            [ErrorItem("XML_STRUCTURE_MISMATCH", "xml", "找不到固定的日期分组节点")]
        )
    group_dates = [parse_group_date(group, i) for i, group in enumerate(groups, 1)]
    if len(set(group_dates)) != 1:
        raise ProcessingFailure(
            [ErrorItem("XML_MULTIPLE_BUSINESS_DATES", "xml", "一个XML包含多个业务日期")]
        )
    reservations: List[ET.Element] = []
    for group in groups:
        reservations.extend(group.findall("./LIST_G_RESERVATION/G_RESERVATION"))
    if not reservations:
        raise ProcessingFailure(
            [ErrorItem("XML_NO_RESERVATIONS", "xml", "XML中没有reservation记录")]
        )
    return group_dates[0], reservations


def required_text(record: ET.Element, tag: str, index: int, code: str) -> str:
    value = text_or_blank(record.findtext(tag))
    if not value:
        raise ProcessingFailure(
            [
                ErrorItem(
                    code,
                    "xml",
                    f"白名单记录缺少 {tag}",
                    f"reservation[{index}]/{tag}",
                    confirmation_no=text_or_blank(record.findtext("CONFIRMATION_NO")) or None,
                )
            ]
        )
    return value


def optional_numeric(value: Any, integer: bool = False) -> Optional[Any]:
    try:
        return parse_integer(value) if integer else parse_decimal(value)
    except ValueError:
        return None


def optional_date(value: Any) -> Optional[date]:
    try:
        return parse_date_value(value)
    except ValueError:
        return None


def source_record(node: ET.Element, index: int) -> Dict[str, Any]:
    """Extract one XML reservation without discarding invalid or filtered rows."""

    raw_rate_code = text_or_blank(node.findtext("RATE_CODE"))
    res_comment = first_nonempty(
        node.findall("./LIST_G_COMMENT_RESV_NAME_ID/G_COMMENT_RESV_NAME_ID/RES_COMMENT")
    )
    company = text_or_blank(node.findtext("COMPANY_NAME"))
    company_key: Optional[str] = None
    if company:
        try:
            company_key = price_company_key(company)
        except CompanyKeywordAmbiguity:
            company_key = None
    arrival = optional_date(node.findtext("TRUNC_BEGIN"))
    departure = optional_date(node.findtext("TRUNC_END"))
    nights = None
    if arrival is not None and departure is not None:
        nights = (departure - arrival).days
    return {
        "BLOCK_CODE": text_or_blank(node.findtext("BLOCK_CODE")),
        "ADULTS": optional_numeric(node.findtext("ADULTS"), integer=True),
        "CHILDREN": optional_numeric(node.findtext("CF_CHILDREN"), integer=True),
        "COMPANY_NAME": company,
        "CONFIRMATION_NO": text_or_blank(node.findtext("CONFIRMATION_NO")),
        "DISP_ROOM_NO": text_or_blank(node.findtext("DISP_ROOM_NO")),
        "EFFECTIVE_RATE_AMOUNT": optional_numeric(node.findtext("EFFECTIVE_RATE_AMOUNT")),
        "FULL_NAME": text_or_blank(node.findtext("FULL_NAME")),
        "RES_COMMENT": res_comment,
        "TRACE_TEXT": first_nonempty(node.findall("./LIST_G_DEPT_ID/G_DEPT_ID/TRACE_TEXT")),
        "NO_OF_ROOMS": optional_numeric(node.findtext("NO_OF_ROOMS"), integer=True),
        "PRODUCTS": text_or_blank(node.findtext("PRODUCTS")),
        "RATE_CODE": raw_rate_code.upper(),
        "ROOM_CATEGORY_LABEL": text_or_blank(node.findtext("ROOM_CATEGORY_LABEL")),
        "ARRIVAL": arrival,
        "DEPARTURE": departure,
        "NIGHTS": nights,
        "REAL PRICE": None,
        "TOTAL PRICE": None,
        "_SOURCE_INDEX": index,
        "_SOURCE_LOCATION": f"reservation[{index}]",
        "_SOURCE_WORKSHEET": None,
        "_SOURCE_ROW_NO": None,
        "_RAW_RATE_CODE": raw_rate_code,
        "_NORMALIZED_RATE_CODE": raw_rate_code.upper(),
        "_GROUP_CODE_KEY": normalize_group_code(res_comment),
        "_COMPANY_KEY": company_key,
        "_OUTCOME": "pending",
        "_DECISION_CODES": [],
        "_DUPLICATE_OF_SOURCE_SEQUENCE": None,
        "_CHANNEL_KEY": None,
        "_PRICING_METHOD": None,
        "_KB_AMOUNT": None,
    }


def xml_record_error(
    record: Dict[str, Any], code: str, message: str, field: Optional[str] = None
) -> ErrorItem:
    location = record["_SOURCE_LOCATION"]
    if field:
        location = f"{location}/{field}"
    return ErrorItem(
        code,
        "xml",
        message,
        location,
        record["COMPANY_NAME"] or None,
        record["_NORMALIZED_RATE_CODE"] or None,
        record["EFFECTIVE_RATE_AMOUNT"],
        record["CONFIRMATION_NO"] or None,
    )


def validate_whitelisted_record(
    record: Dict[str, Any], business_date: date
) -> List[ErrorItem]:
    errors: List[ErrorItem] = []
    required_text_fields = (
        ("COMPANY_NAME", "XML_COMPANY_MISSING"),
        ("CONFIRMATION_NO", "XML_CONFIRMATION_MISSING"),
        ("DISP_ROOM_NO", "XML_ROOM_MISSING"),
        ("FULL_NAME", "XML_FULL_NAME_MISSING"),
    )
    for field, code in required_text_fields:
        if not record[field]:
            errors.append(
                xml_record_error(record, code, f"白名单记录缺少 {field}", field)
            )
    numeric_rules = (
        ("ADULTS", "ADULTS", "XML_ADULTS_INVALID", lambda value: value >= 0),
        ("CHILDREN", "CF_CHILDREN", "XML_CHILDREN_INVALID", lambda value: value >= 0),
        ("NO_OF_ROOMS", "NO_OF_ROOMS", "XML_ROOM_COUNT_INVALID", lambda value: value > 0),
        (
            "EFFECTIVE_RATE_AMOUNT",
            "EFFECTIVE_RATE_AMOUNT",
            "XML_RATE_AMOUNT_INVALID",
            lambda value: value >= 0,
        ),
    )
    for field, source_field, code, predicate in numeric_rules:
        value = record[field]
        if value is None or not predicate(value):
            errors.append(
                xml_record_error(
                    record,
                    code,
                    f"白名单记录的 {source_field} 缺失或无效",
                    source_field,
                )
            )
    arrival = record["ARRIVAL"]
    departure = record["DEPARTURE"]
    if arrival is None:
        errors.append(
            xml_record_error(
                record,
                "XML_ARRIVAL_INVALID",
                "白名单记录的 ARRIVAL 缺失或无效",
                "TRUNC_BEGIN",
            )
        )
    elif arrival != business_date:
        errors.append(
            xml_record_error(
                record,
                "XML_BUSINESS_DATE_MISMATCH",
                "reservation ARRIVAL 与XML分组业务日期不一致",
                "TRUNC_BEGIN",
            )
        )
    if departure is None:
        errors.append(
            xml_record_error(
                record,
                "XML_DEPARTURE_INVALID",
                "白名单记录的 DEPARTURE 缺失或无效",
                "TRUNC_END",
            )
        )
    elif arrival is not None and departure < arrival:
        errors.append(
            xml_record_error(
                record,
                "XML_NEGATIVE_NIGHTS",
                "DEPARTURE 不得早于 ARRIVAL",
            )
        )
    return errors


def build_record(node: ET.Element, index: int, business_date: date, rate_code: str) -> Dict[str, Any]:
    """Compatibility wrapper used by callers that require one valid whitelist record."""

    record = source_record(node, index)
    record["RATE_CODE"] = rate_code
    record["_NORMALIZED_RATE_CODE"] = rate_code
    errors = validate_whitelisted_record(record, business_date)
    if errors:
        raise ProcessingFailure(errors)
    return record


def classify_source_records(
    reservations: Sequence[ET.Element], business_date: date
) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]], int, int, List[ErrorItem]]:
    """Classify every XML reservation and retain source order and duplicate lineage."""

    all_records = [source_record(node, index) for index, node in enumerate(reservations, 1)]
    candidates: List[Dict[str, Any]] = []
    errors: List[ErrorItem] = []
    removed_by_rate = 0
    for record in all_records:
        normalized_rate = record["_NORMALIZED_RATE_CODE"]
        if not normalized_rate:
            error = xml_record_error(
                record,
                "XML_RATE_CODE_MISSING",
                "reservation缺少RATE_CODE，无法判断白名单",
                "RATE_CODE",
            )
            record["_OUTCOME"] = "validation_failed"
            append_decision(record, error.code)
            errors.append(error)
            continue
        if normalized_rate not in RATE_WHITELIST:
            record["_OUTCOME"] = "excluded_rate_code"
            append_decision(record, "RATE_CODE_NOT_WHITELISTED")
            removed_by_rate += 1
            continue
        append_decision(record, "RATE_CODE_WHITELISTED")
        record_errors = validate_whitelisted_record(record, business_date)
        if record_errors:
            record["_OUTCOME"] = "validation_failed"
            for error in record_errors:
                append_decision(record, error.code)
            errors.extend(record_errors)
            continue
        candidates.append(record)

    deduplicated: List[Dict[str, Any]] = []
    seen: Dict[Tuple[str, date], int] = {}
    removed_duplicates = 0
    for record in candidates:
        key = (record["DISP_ROOM_NO"], record["ARRIVAL"])
        duplicate_of = seen.get(key)
        if duplicate_of is not None:
            record["_OUTCOME"] = "duplicate"
            record["_DUPLICATE_OF_SOURCE_SEQUENCE"] = duplicate_of
            append_decision(record, "DUPLICATE_DISP_ROOM_ARRIVAL")
            removed_duplicates += 1
            continue
        seen[key] = record["_SOURCE_INDEX"]
        append_decision(record, "UNIQUE_DISP_ROOM_ARRIVAL")
        deduplicated.append(record)
    if not deduplicated and not errors:
        errors.append(ErrorItem("XML_NO_ELIGIBLE_ROWS", "xml", "XML中没有费率白名单内的有效记录"))
    return all_records, deduplicated, removed_by_rate, removed_duplicates, errors


def filter_and_deduplicate(
    reservations: Sequence[ET.Element], business_date: date
) -> Tuple[List[Dict[str, Any]], int, int]:
    _all_records, deduplicated, removed_by_rate, removed_duplicates, errors = (
        classify_source_records(reservations, business_date)
    )
    if errors:
        raise ProcessingFailure(errors)
    return deduplicated, removed_by_rate, removed_duplicates


def price_company_key(company_name: str) -> str:
    compact = COMPANY_COMPACT_RE.sub("", text_or_blank(company_name).upper())
    matches = [
        canonical
        for canonical, keywords in COMPANY_KEYWORD_GROUPS
        if any(keyword in compact for keyword in keywords)
    ]
    if len(matches) > 1:
        raise CompanyKeywordAmbiguity(company_name, matches)
    return matches[0] if matches else compact


def load_price_map(price_path: Path) -> Dict[Tuple[str, str, Decimal], Decimal]:
    ensure_absolute_file(price_path, ".xlsx", "INPUT_PRICE_INVALID", "价格对照文件")
    try:
        workbook = load_workbook(price_path, read_only=True, data_only=True)
    except Exception as exc:
        raise ProcessingFailure(
            [ErrorItem("PRICE_WORKBOOK_UNREADABLE", "price", f"价格表无法读取：{exc}", price_path.name)]
        )
    try:
        sheet = workbook["Sheet1"] if "Sheet1" in workbook.sheetnames else workbook.active
        headers = [text_or_blank(sheet.cell(1, col).value) for col in range(1, 5)]
        expected = ["COMPANY'S NAME", "RATE CODE", "Opera展示的价格", "总价"]
        if headers != expected:
            raise ProcessingFailure(
                [
                    ErrorItem(
                        "PRICE_HEADER_MISMATCH",
                        "price",
                        f"价格表表头必须为 {expected}",
                        f"{price_path.name}!{sheet.title}!A1:D1",
                    )
                ]
            )
        mapping: Dict[Tuple[str, str, Decimal], Decimal] = {}
        key_rows: Dict[Tuple[str, str, Decimal], List[int]] = {}
        errors: List[ErrorItem] = []
        for row in range(2, sheet.max_row + 1):
            values = [sheet.cell(row, col).value for col in range(1, 5)]
            if all(value is None or text_or_blank(value) == "" for value in values):
                continue
            company = text_or_blank(values[0])
            rate = text_or_blank(values[1]).upper()
            try:
                opera_amount = parse_decimal(values[2])
                total = parse_decimal(values[3])
            except ValueError:
                errors.append(
                    ErrorItem(
                        "PRICE_VALUE_INVALID",
                        "price",
                        "价格表必填字段缺失或金额不是数字",
                        f"{price_path.name}!{sheet.title}!A{row}:D{row}",
                        company or None,
                        rate or None,
                    )
                )
                continue
            if opera_amount < 0 or total < 0:
                errors.append(
                    ErrorItem(
                        "PRICE_VALUE_INVALID",
                        "price",
                        "价格表的Opera展示价格和总价不得为负数",
                        f"{price_path.name}!{sheet.title}!A{row}:D{row}",
                        company or None,
                        rate or None,
                        opera_amount,
                    )
                )
                continue
            if not company or not rate:
                errors.append(
                    ErrorItem(
                        "PRICE_VALUE_INVALID",
                        "price",
                        "价格表公司名或RATE CODE为空",
                        f"{price_path.name}!{sheet.title}!A{row}:D{row}",
                        company or None,
                        rate or None,
                        opera_amount,
                    )
                )
                continue
            try:
                company_key = price_company_key(company)
            except CompanyKeywordAmbiguity as exc:
                errors.append(
                    ErrorItem(
                        "PRICE_COMPANY_AMBIGUOUS",
                        "price",
                        str(exc),
                        f"{price_path.name}!{sheet.title}!A{row}",
                        company,
                        rate or None,
                        opera_amount,
                    )
                )
                continue
            key = (company_key, rate, opera_amount)
            key_rows.setdefault(key, []).append(row)
            if key not in mapping:
                mapping[key] = total
        for key, rows in key_rows.items():
            if len(rows) > 1:
                errors.append(
                    ErrorItem(
                        "PRICE_DUPLICATE_KEY",
                        "price",
                        f"价格表标准化三键重复，行号：{', '.join(map(str, rows))}",
                        f"{price_path.name}!{sheet.title}!rows[{','.join(map(str, rows))}]",
                        key[0],
                        key[1],
                        key[2],
                    )
                )
        if errors:
            raise ProcessingFailure(errors)
        if not mapping:
            raise ProcessingFailure(
                [ErrorItem("PRICE_NO_RULES", "price", "价格表没有有效规则", price_path.name)]
            )
        return mapping
    finally:
        workbook.close()


def apply_prices_classified(
    records: List[Dict[str, Any]], price_map: Dict[Tuple[str, str, Decimal], Decimal]
) -> List[ErrorItem]:
    errors: List[ErrorItem] = []
    for record in records:
        rate_code = record["RATE_CODE"].strip().upper()
        try:
            company_key = price_company_key(record["COMPANY_NAME"])
        except CompanyKeywordAmbiguity as exc:
            errors.append(
                ErrorItem(
                    "PRICE_COMPANY_AMBIGUOUS",
                    "price",
                    str(exc),
                    f"reservation[{record['_SOURCE_INDEX']}]",
                    record["COMPANY_NAME"],
                    rate_code,
                    record["EFFECTIVE_RATE_AMOUNT"],
                    record["CONFIRMATION_NO"],
                )
            )
            record["_OUTCOME"] = "validation_failed"
            append_decision(record, "PRICE_COMPANY_AMBIGUOUS")
            continue
        record["_COMPANY_KEY"] = company_key
        if company_key in ZERO_TOTAL_COMPANIES and rate_code in ZERO_TOTAL_RATE_CODES:
            real_price = Decimal(0)
            record["_PRICING_METHOD"] = "zero_price_exception"
            append_decision(record, "ZERO_PRICE_EXCEPTION")
        else:
            key = (
                company_key,
                rate_code,
                record["EFFECTIVE_RATE_AMOUNT"],
            )
            real_price = price_map.get(key)
            if real_price is None:
                errors.append(
                    ErrorItem(
                        "PRICE_UNMATCHED",
                        "price",
                        "日报记录在固定价格表中没有唯一匹配",
                        f"reservation[{record['_SOURCE_INDEX']}]",
                        record["COMPANY_NAME"],
                        record["RATE_CODE"],
                        record["EFFECTIVE_RATE_AMOUNT"],
                        record["CONFIRMATION_NO"],
                    )
                )
                record["_OUTCOME"] = "price_unmatched"
                append_decision(record, "PRICE_UNMATCHED")
                continue
            record["_PRICING_METHOD"] = "price_reference_exact"
            append_decision(record, "PRICE_REFERENCE_MATCHED")
        record["REAL PRICE"] = real_price
        record["TOTAL PRICE"] = real_price * record["NO_OF_ROOMS"] * record["NIGHTS"]
    return errors


def apply_prices(records: List[Dict[str, Any]], price_map: Dict[Tuple[str, str, Decimal], Decimal]) -> None:
    errors = apply_prices_classified(records, price_map)
    if errors:
        raise ProcessingFailure(errors)


def normalized_headers(sheet: Any, count: int = len(DAILY_HEADERS)) -> List[str]:
    return [text_or_blank(sheet.cell(1, col).value) for col in range(1, count + 1)]


def copy_row_style(sheet: Any, source_row: int, target_row: int, column_count: int = 18) -> None:
    if target_row == source_row:
        return
    for col in range(1, column_count + 1):
        source = sheet.cell(source_row, col)
        target = sheet.cell(target_row, col)
        if source.has_style:
            target._style = copy.copy(source._style)
        if source.number_format:
            target.number_format = source.number_format
    if source_row in sheet.row_dimensions:
        sheet.row_dimensions[target_row].height = sheet.row_dimensions[source_row].height


def clear_data_cells(sheet: Any, column_count: int = 18) -> None:
    for row in sheet.iter_rows(
        min_row=2,
        max_row=max(sheet.max_row, 2),
        min_col=1,
        max_col=column_count,
    ):
        for cell in row:
            cell.value = None


def update_table_ranges(sheet: Any, last_row: int, column_count: int = 18) -> None:
    final_row = max(last_row, 2)
    last_column = get_column_letter(column_count)
    for table in sheet.tables.values():
        table.ref = f"A1:{last_column}{final_row}"


def write_record_row(sheet: Any, row_number: int, headers: Sequence[str], record: Dict[str, Any]) -> None:
    text_fields = {
        "BLOCK_CODE",
        "COMPANY_NAME",
        "CONFIRMATION_NO",
        "DISP_ROOM_NO",
        "FULL_NAME",
        "RES_COMMENT",
        "TRACE_TEXT",
        "PRODUCTS",
        "RATE_CODE",
        "ROOM_CATEGORY_LABEL",
    }
    date_fields = {"ARRIVAL", "DEPARTURE"}
    for col, header in enumerate(headers, 1):
        cell = sheet.cell(row_number, col)
        value = record.get(header)
        if header in text_fields:
            safe_text_cell(cell, value)
        elif header in date_fields:
            cell.value = value
            cell.number_format = "DD-MMM-YY"
        else:
            numeric_cell(cell, decimal_to_number(value) if isinstance(value, Decimal) else value)


def write_daily(records: Sequence[Dict[str, Any]], business_date: date, output_path: Path) -> None:
    workbook = load_workbook(DAILY_TEMPLATE)
    try:
        sheet = workbook.active
        if normalized_headers(sheet, len(DAILY_HEADERS)) != DAILY_HEADERS:
            raise ProcessingFailure(
                [ErrorItem("OUTPUT_DAILY_TEMPLATE_HEADER", "output", "日报模板表头不符合契约")]
            )
        sheet.title = f"{business_date.month}.{business_date.day}"
        clear_data_cells(sheet, len(DAILY_HEADERS))
        for offset, record in enumerate(records, 2):
            copy_row_style(sheet, 2, offset, len(DAILY_HEADERS))
            write_record_row(sheet, offset, DAILY_HEADERS, record)
        update_table_ranges(sheet, len(records) + 1, len(DAILY_HEADERS))
        workbook.save(output_path)
    finally:
        workbook.close()


def route_standard(company: str, rate_code: str) -> Optional[str]:
    try:
        company_key = price_company_key(company)
    except CompanyKeywordAmbiguity as exc:
        raise ProcessingFailure(
            [
                ErrorItem(
                    "MONTHLY_COMPANY_AMBIGUOUS",
                    "monthly",
                    str(exc),
                    company_name=company,
                    rate_code=rate_code,
                )
            ]
        ) from exc
    normalized_rate = text_or_blank(rate_code).upper()
    if company_key in KB_CHANNEL_COMPANIES:
        return KB_SHEET
    if company_key == "QBD":
        return "QBD"
    if company_key == "LIAN TAI":
        return "LIANTAI-FIT" if normalized_rate in {"LBLT", "LTLT"} else "LIANTAI-GROUP"
    if company_key == "FENGRUN":
        return "FENGRUN"
    return None


def channel_headers_for_sheet(sheet_name: str) -> List[str]:
    return KB_CHANNEL_HEADERS if sheet_name == KB_SHEET else CHANNEL_HEADERS


def sanitized_sheet_base(company: str) -> str:
    cleaned = INVALID_SHEET_CHARS.sub("", company.strip())
    return cleaned[:31]


def valid_unknown_sheet_title(company: str, title: str) -> bool:
    base = sanitized_sheet_base(company)
    if not base:
        return False
    if title == base:
        return True
    match = re.search(r"-(\d+)$", title)
    if not match or int(match.group(1)) < 2:
        return False
    suffix = match.group(0)
    return title == f"{base[:31 - len(suffix)]}{suffix}"


def read_existing_monthly(
    workbook: Any, business_date: date
) -> Tuple[Dict[str, List[Dict[str, Any]]], Dict[str, str]]:
    errors: List[ErrorItem] = []
    for required in STANDARD_SHEETS:
        if required not in workbook.sheetnames:
            errors.append(
                ErrorItem("MONTHLY_STANDARD_SHEET_MISSING", "monthly", f"月报缺少标准工作表 {required}")
            )
    data: Dict[str, List[Dict[str, Any]]] = {}
    unknown_company_sheet: Dict[str, str] = {}
    for sheet in workbook.worksheets:
        headers = channel_headers_for_sheet(sheet.title)
        column_count = len(headers)
        last_column = get_column_letter(column_count)
        if normalized_headers(sheet, column_count) != headers:
            errors.append(
                ErrorItem(
                    "MONTHLY_HEADER_MISMATCH",
                    "monthly",
                    f"工作表 {sheet.title} 的{column_count}列表头不符合契约",
                    f"{sheet.title}!A1:{last_column}1",
                )
            )
            data[sheet.title] = []
            continue
        rows: List[Dict[str, Any]] = []
        companies: set = set()
        previous_arrival: Optional[date] = None
        for row_number in range(2, sheet.max_row + 1):
            values = [sheet.cell(row_number, col).value for col in range(1, column_count + 1)]
            if all(value is None or text_or_blank(value) == "" for value in values):
                continue
            row = dict(zip(headers, values))
            location = f"{sheet.title}!A{row_number}:{last_column}{row_number}"
            try:
                row["ARRIVAL"] = parse_date_value(row["ARRIVAL"])
                row["DEPARTURE"] = parse_date_value(row["DEPARTURE"])
                row["NIGHTS"] = parse_integer(row["NIGHTS"])
                row["ADULTS"] = parse_integer(row["ADULTS"])
                row["CHILDREN"] = parse_integer(row["CHILDREN"])
                row["NO_OF_ROOMS"] = parse_integer(row["NO_OF_ROOMS"])
                row["RATE_AMOUNT"] = parse_decimal(row["RATE_AMOUNT"])
                row["REAL PRICE"] = parse_decimal(row["REAL PRICE"])
                row["TOTAL PRICE"] = parse_decimal(row["TOTAL PRICE"])
                if sheet.title == KB_SHEET:
                    row[KB_HEADER] = parse_decimal(row[KB_HEADER])
            except ValueError:
                errors.append(
                    ErrorItem("MONTHLY_ROW_INVALID", "monthly", "月报已有行包含无效日期或数字", location)
                )
                continue
            for text_field in (
                "BLOCK_CODE",
                "COMPANY_NAME",
                "CONFIRMATION_NO",
                "DISP_ROOM_NO",
                "FULL_NAME",
                "RES_COMMENT",
                "TRACE_TEXT",
                "PRODUCTS",
                "RATE_CODE",
                "ROOM_CATEGORY_LABEL",
            ):
                row[text_field] = text_or_blank(row[text_field])
            if not row["COMPANY_NAME"] or not row["CONFIRMATION_NO"] or not row["DISP_ROOM_NO"]:
                errors.append(
                    ErrorItem("MONTHLY_ROW_INVALID", "monthly", "月报已有行缺少关键文本字段", location)
                )
                continue
            if (row["ARRIVAL"].year, row["ARRIVAL"].month) != (business_date.year, business_date.month):
                errors.append(
                    ErrorItem(
                        "MONTHLY_WRONG_MONTH",
                        "monthly",
                        "月报包含不属于XML业务月份的数据",
                        location,
                        row["COMPANY_NAME"],
                        row["RATE_CODE"],
                        row["RATE_AMOUNT"],
                        row["CONFIRMATION_NO"],
                    )
                )
            if row["DEPARTURE"] < row["ARRIVAL"] or row["NIGHTS"] != (
                row["DEPARTURE"] - row["ARRIVAL"]
            ).days:
                errors.append(
                    ErrorItem("MONTHLY_NIGHTS_INVALID", "monthly", "月报已有行的晚数与日期不一致", location)
                )
            expected_total = row["REAL PRICE"] * row["NO_OF_ROOMS"] * row["NIGHTS"]
            if row["TOTAL PRICE"] != expected_total:
                errors.append(
                    ErrorItem(
                        "MONTHLY_TOTAL_PRICE_INVALID",
                        "monthly",
                        "月报已有行的TOTAL PRICE不等于REAL PRICE×NO_OF_ROOMS×NIGHTS",
                        location,
                        row["COMPANY_NAME"],
                        row["RATE_CODE"],
                        row["RATE_AMOUNT"],
                        row["CONFIRMATION_NO"],
                    )
                )
            if sheet.title == KB_SHEET and row[KB_HEADER] != Decimal(row["NO_OF_ROOMS"] * 100):
                errors.append(
                    ErrorItem(
                        "MONTHLY_KB_INVALID",
                        "monthly",
                        f"月报已有行的{KB_HEADER}不等于NO_OF_ROOMS×100",
                        location,
                        row["COMPANY_NAME"],
                        row["RATE_CODE"],
                        row["RATE_AMOUNT"],
                        row["CONFIRMATION_NO"],
                    )
                )
            if previous_arrival and row["ARRIVAL"] < previous_arrival:
                errors.append(
                    ErrorItem("MONTHLY_DATE_ORDER_INVALID", "monthly", "月报已有日期未按升序排列", location)
                )
            previous_arrival = row["ARRIVAL"]
            expected_sheet = route_standard(row["COMPANY_NAME"], row["RATE_CODE"])
            if expected_sheet and expected_sheet != sheet.title:
                errors.append(
                    ErrorItem(
                        "MONTHLY_ROUTE_INVALID",
                        "monthly",
                        f"已有记录应位于 {expected_sheet}，实际位于 {sheet.title}",
                        location,
                        row["COMPANY_NAME"],
                        row["RATE_CODE"],
                        row["RATE_AMOUNT"],
                        row["CONFIRMATION_NO"],
                    )
                )
            if expected_sheet is None and sheet.title in STANDARD_SHEETS:
                errors.append(
                    ErrorItem(
                        "MONTHLY_ROUTE_INVALID",
                        "monthly",
                        f"非标准公司记录不得位于标准工作表 {sheet.title}",
                        location,
                        row["COMPANY_NAME"],
                        row["RATE_CODE"],
                        row["RATE_AMOUNT"],
                        row["CONFIRMATION_NO"],
                    )
                )
            companies.add(row["COMPANY_NAME"])
            rows.append(row)
        if sheet.title not in STANDARD_SHEETS:
            if len(companies) > 1:
                errors.append(
                    ErrorItem(
                        "MONTHLY_MIXED_COMPANIES",
                        "monthly",
                        f"非标准工作表 {sheet.title} 包含多个实际公司",
                        sheet.title,
                    )
                )
            elif len(companies) == 1:
                company = next(iter(companies))
                if not valid_unknown_sheet_title(company, sheet.title):
                    errors.append(
                        ErrorItem(
                            "MONTHLY_ROUTE_INVALID",
                            "monthly",
                            f"非标准工作表 {sheet.title} 不符合实际公司名及碰撞后缀规则",
                            sheet.title,
                            company_name=company,
                        )
                    )
                unknown_company_sheet[company] = sheet.title
        data[sheet.title] = rows
    if errors:
        raise ProcessingFailure(errors)
    return data, unknown_company_sheet


def create_unknown_sheet(workbook: Any, title: str) -> Any:
    source = workbook["QBD"]
    sheet = workbook.copy_worksheet(source)
    sheet.title = title
    clear_data_cells(sheet, len(CHANNEL_HEADERS))
    return sheet


def allocate_channel_key(
    company: str,
    rate_code: str,
    used_sheet_names: set[str],
    unknown_company_sheet: Dict[str, str],
) -> str:
    standard = route_standard(company, rate_code)
    if standard:
        return standard
    if company in unknown_company_sheet:
        return unknown_company_sheet[company]
    base = sanitized_sheet_base(company)
    if not base:
        raise ProcessingFailure(
            [
                ErrorItem(
                    "MONTHLY_SHEET_NAME_EMPTY",
                    "monthly",
                    "公司名清除Excel禁用字符后为空，无法创建工作表",
                    company,
                    company_name=company,
                    rate_code=rate_code,
                )
            ]
        )
    candidate = base
    suffix_number = 2
    while candidate in used_sheet_names:
        candidate_suffix = f"-{suffix_number}"
        candidate = f"{base[:31 - len(candidate_suffix)]}{candidate_suffix}"
        suffix_number += 1
    used_sheet_names.add(candidate)
    unknown_company_sheet[company] = candidate
    return candidate


def assign_channels(
    records: Sequence[Dict[str, Any]],
    used_sheet_names: Optional[Iterable[str]] = None,
    unknown_company_sheet: Optional[Dict[str, str]] = None,
) -> Dict[str, int]:
    """Assign deterministic channel keys without requiring a monthly workbook."""

    used = set(used_sheet_names or STANDARD_SHEETS)
    unknown_mapping = unknown_company_sheet if unknown_company_sheet is not None else {}
    channel_counts: Dict[str, int] = {}
    for record in records:
        channel_key = allocate_channel_key(
            record["COMPANY_NAME"],
            record["RATE_CODE"],
            used,
            unknown_mapping,
        )
        record["_CHANNEL_KEY"] = channel_key
        record["_KB_AMOUNT"] = (
            Decimal(record["NO_OF_ROOMS"] * 100) if channel_key == KB_SHEET else None
        )
        channel_counts[channel_key] = channel_counts.get(channel_key, 0) + 1
    return channel_counts


def route_sheet(
    workbook: Any,
    company: str,
    rate_code: str,
    unknown_company_sheet: Dict[str, str],
) -> str:
    sheet_name = allocate_channel_key(
        company,
        rate_code,
        set(workbook.sheetnames),
        unknown_company_sheet,
    )
    if sheet_name not in workbook.sheetnames:
        create_unknown_sheet(workbook, sheet_name)
    return sheet_name


def to_channel_record(record: Dict[str, Any], sheet_name: Optional[str] = None) -> Dict[str, Any]:
    channel_record = {
        "ARRIVAL": record["ARRIVAL"],
        "DEPARTURE": record["DEPARTURE"],
        "NIGHTS": record["NIGHTS"],
        "ADULTS": record["ADULTS"],
        "CHILDREN": record["CHILDREN"],
        "BLOCK_CODE": record["BLOCK_CODE"],
        "NO_OF_ROOMS": record["NO_OF_ROOMS"],
        "COMPANY_NAME": record["COMPANY_NAME"],
        "CONFIRMATION_NO": record["CONFIRMATION_NO"],
        "DISP_ROOM_NO": record["DISP_ROOM_NO"],
        "RATE_AMOUNT": record["EFFECTIVE_RATE_AMOUNT"],
        "FULL_NAME": record["FULL_NAME"],
        "RES_COMMENT": record["RES_COMMENT"],
        "TRACE_TEXT": record["TRACE_TEXT"],
        "PRODUCTS": record["PRODUCTS"],
        "RATE_CODE": record["RATE_CODE"],
        "ROOM_CATEGORY_LABEL": record["ROOM_CATEGORY_LABEL"],
        "REAL PRICE": parse_decimal(record["REAL PRICE"]),
        "TOTAL PRICE": parse_decimal(record["TOTAL PRICE"]),
    }
    if sheet_name == KB_SHEET:
        channel_record[KB_HEADER] = Decimal(record["NO_OF_ROOMS"] * 100)
    return channel_record


def write_monthly(
    records: Sequence[Dict[str, Any]],
    business_date: date,
    monthly_base: Optional[Path],
    output_dir: Path,
) -> Tuple[Path, Dict[str, int]]:
    source = monthly_base if monthly_base is not None else MONTHLY_TEMPLATE
    try:
        workbook = load_workbook(source)
    except Exception as exc:
        raise ProcessingFailure(
            [ErrorItem("MONTHLY_WORKBOOK_UNREADABLE", "monthly", f"月报无法读取：{exc}", source.name)]
        )
    try:
        existing, unknown_company_sheet = read_existing_monthly(workbook, business_date)
        for sheet_name in list(existing):
            existing[sheet_name] = [
                row for row in existing[sheet_name] if row["ARRIVAL"] != business_date
            ]
        channel_counts: Dict[str, int] = {}
        for record in records:
            sheet_name = route_sheet(
                workbook, record["COMPANY_NAME"], record["RATE_CODE"], unknown_company_sheet
            )
            record["_CHANNEL_KEY"] = sheet_name
            record["_KB_AMOUNT"] = (
                Decimal(record["NO_OF_ROOMS"] * 100) if sheet_name == KB_SHEET else None
            )
            existing.setdefault(sheet_name, []).append(to_channel_record(record, sheet_name))
            channel_counts[sheet_name] = channel_counts.get(sheet_name, 0) + 1
        for sheet in list(workbook.worksheets):
            if sheet.title not in STANDARD_SHEETS and not existing.get(sheet.title):
                existing.pop(sheet.title, None)
                workbook.remove(sheet)
        all_arrivals: List[date] = []
        for sheet in workbook.worksheets:
            headers = channel_headers_for_sheet(sheet.title)
            column_count = len(headers)
            rows = existing.setdefault(sheet.title, [])
            rows.sort(key=lambda row: row["ARRIVAL"])
            clear_data_cells(sheet, column_count)
            for row_number, record in enumerate(rows, 2):
                copy_row_style(sheet, 2, row_number, column_count)
                write_record_row(sheet, row_number, headers, record)
                all_arrivals.append(record["ARRIVAL"])
            update_table_ranges(sheet, len(rows) + 1, column_count)
        if not all_arrivals:
            raise ProcessingFailure(
                [ErrorItem("MONTHLY_NO_ROWS", "monthly", "处理后月报没有任何数据行")]
            )
        max_arrival = max(all_arrivals)
        filename = (
            f"各渠道情况-{business_date.year:04d}年{business_date.month:02d}月-"
            f"更新至{max_arrival.month}.{max_arrival.day}.xlsx"
        )
        output_path = output_dir / filename
        if monthly_base is not None and output_path.resolve() == monthly_base.resolve():
            raise ProcessingFailure(
                [
                    ErrorItem(
                        "INPUT_MONTHLY_OUTPUT_COLLISION",
                        "input",
                        "月报输入与候选输出不得是同一文件；请使用隔离的输出目录",
                        str(monthly_base),
                    )
                ],
                exit_code=3,
            )
        try:
            workbook.save(output_path)
        except Exception:
            if output_path.is_file():
                output_path.unlink()
            raise
        return output_path, channel_counts
    finally:
        workbook.close()


def write_exception_workbook(errors: Sequence[ErrorItem], output_path: Path) -> None:
    workbook = Workbook()
    sheet = workbook.active
    sheet.title = "异常清单"
    for col, header in enumerate(EXCEPTION_HEADERS, 1):
        cell = sheet.cell(1, col)
        safe_text_cell(cell, header)
        cell.font = Font(bold=True)
        cell.alignment = Alignment(horizontal="center", vertical="center")
    for row_number, error in enumerate(errors, 2):
        values = [
            error.code,
            error.stage,
            error.source_location,
            error.company_name,
            error.rate_code,
            decimal_to_number(error.effective_rate_amount),
            error.confirmation_no,
            error.message,
        ]
        for col, value in enumerate(values, 1):
            cell = sheet.cell(row_number, col)
            if col == 6 and value is not None:
                numeric_cell(cell, value)
            else:
                safe_text_cell(cell, value)
    widths = [22, 14, 42, 28, 18, 24, 22, 56]
    for col, width in enumerate(widths, 1):
        sheet.column_dimensions[sheet.cell(1, col).column_letter].width = width
    workbook.save(output_path)
    workbook.close()


def empty_metrics() -> Dict[str, Any]:
    return {
        "source_rows": 0,
        "removed_by_rate_code": 0,
        "removed_as_duplicates": 0,
        "output_rows": 0,
        "channels": [],
    }


def channel_metrics(channel_counts: Dict[str, int]) -> List[Dict[str, Any]]:
    ordered_names = [name for name in STANDARD_SHEETS if channel_counts.get(name, 0) > 0]
    ordered_names.extend(
        sorted(name for name, count in channel_counts.items() if name not in STANDARD_SHEETS and count > 0)
    )
    return [{"worksheet": name, "rows": channel_counts[name]} for name in ordered_names]


def result_object(
    status: str,
    business_date: Optional[date],
    message: str,
    metrics: Dict[str, Any],
    daily: Optional[Path] = None,
    monthly: Optional[Path] = None,
    exception: Optional[Path] = None,
    errors: Sequence[ErrorItem] = (),
    processing_mode: str = MODE_DAILY_MONTHLY,
) -> Dict[str, Any]:
    payload = {
        "version": DAILY_RESULT_VERSION if processing_mode == MODE_DAILY else VERSION,
        "status": status,
        "business_date": business_date.isoformat() if business_date else None,
        "month_key": f"{business_date.year:04d}-{business_date.month:02d}" if business_date else None,
        "message": message,
        "metrics": metrics,
        "outputs": {
            "daily_report": daily.name if daily else None,
            "monthly_report": monthly.name if monthly else None,
            "exception_report": exception.name if exception else None,
        },
        "errors": [error.to_dict() for error in errors],
    }
    if processing_mode == MODE_DAILY:
        payload["processing_mode"] = MODE_DAILY
    return payload


def write_result(path: Path, result: Dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")


def artifact_object(path: Optional[Path], file_kind: str, mime_type: str) -> Optional[Dict[str, Any]]:
    if path is None or not path.is_file():
        return None
    resolved = path.resolve()
    return {
        "file_kind": file_kind,
        "original_filename": resolved.name,
        "path": str(resolved),
        "sha256": sha256_file(resolved),
        "byte_size": resolved.stat().st_size,
        "mime_type": mime_type,
    }


def structured_record(record: Dict[str, Any]) -> Dict[str, Any]:
    group_code_key = normalize_group_code(record.get("RES_COMMENT"))
    return {
        "source_sequence": record["_SOURCE_INDEX"],
        "source_location": record["_SOURCE_LOCATION"],
        "source_worksheet": record["_SOURCE_WORKSHEET"],
        "source_row_no": record["_SOURCE_ROW_NO"],
        "outcome": record["_OUTCOME"],
        "decision_codes": list(record["_DECISION_CODES"]),
        "duplicate_of_source_sequence": record["_DUPLICATE_OF_SOURCE_SEQUENCE"],
        "adults": record.get("ADULTS"),
        "children": record.get("CHILDREN"),
        "block_code": record.get("BLOCK_CODE", ""),
        "no_of_rooms": record.get("NO_OF_ROOMS"),
        "company_name": record.get("COMPANY_NAME", ""),
        "company_key": record.get("_COMPANY_KEY"),
        "confirmation_no": record.get("CONFIRMATION_NO", ""),
        "disp_room_no": record.get("DISP_ROOM_NO", ""),
        "effective_rate_amount": decimal_to_number(record.get("EFFECTIVE_RATE_AMOUNT")),
        "full_name": record.get("FULL_NAME", ""),
        "res_comment": record.get("RES_COMMENT", ""),
        "group_code_key": group_code_key,
        "booking_source_match_status": (
            "missing_group_code" if group_code_key is None else "not_checked"
        ),
        "trace_text": record.get("TRACE_TEXT", ""),
        "products": record.get("PRODUCTS", ""),
        "rate_code": record.get("_RAW_RATE_CODE", ""),
        "normalized_rate_code": record.get("_NORMALIZED_RATE_CODE") or None,
        "room_category_label": record.get("ROOM_CATEGORY_LABEL", ""),
        "arrival": record["ARRIVAL"].isoformat() if record.get("ARRIVAL") else None,
        "departure": record["DEPARTURE"].isoformat() if record.get("DEPARTURE") else None,
        "nights": record.get("NIGHTS"),
        "real_price": decimal_to_number(record.get("REAL PRICE")),
        "total_price": decimal_to_number(record.get("TOTAL PRICE")),
        "kb_amount": decimal_to_number(record.get("_KB_AMOUNT")),
        "channel_key": record.get("_CHANNEL_KEY"),
        "pricing_method": record.get("_PRICING_METHOD"),
    }


def finalize_failure_outcomes(records: Sequence[Dict[str, Any]]) -> None:
    for record in records:
        if record.get("_OUTCOME") in {"pending", "retained"}:
            record["_OUTCOME"] = "validation_failed"
            append_decision(record, "BATCH_NOT_VALIDATED")


def finalize_success_outcomes(records: Sequence[Dict[str, Any]]) -> None:
    for record in records:
        if record.get("_OUTCOME") == "pending":
            record["_OUTCOME"] = "retained"


def structured_counts(records: Sequence[Dict[str, Any]]) -> Dict[str, int]:
    return {
        outcome: sum(1 for record in records if record.get("_OUTCOME") == outcome)
        for outcome in sorted(FINAL_OUTCOMES)
    }


def build_structured_result(
    status: str,
    business_date: Optional[date],
    records: Sequence[Dict[str, Any]],
    channels: Sequence[Dict[str, Any]],
    xml_path: Optional[Path],
    result_json: Optional[Path],
    daily_path: Optional[Path] = None,
    monthly_path: Optional[Path] = None,
    exception_path: Optional[Path] = None,
    errors: Sequence[ErrorItem] = (),
    processing_mode: str = MODE_DAILY_MONTHLY,
) -> Dict[str, Any]:
    counts = structured_counts(records)
    payload = {
        "result_schema_version": (
            DAILY_STRUCTURED_RESULT_SCHEMA_VERSION
            if processing_mode == MODE_DAILY
            else STRUCTURED_RESULT_SCHEMA_VERSION
        ),
        "status": status,
        "activation_eligible": status == "success",
        "ingestion_mode": "opera_xml",
        "business_date": business_date.isoformat() if business_date else None,
        "processor_version": PROCESSOR_VERSION,
        "rule_set_sha256": rule_set_sha256(),
        "source_rows": len(records),
        "removed_by_rate_code": counts["excluded_rate_code"],
        "removed_as_duplicates": counts["duplicate"],
        "output_rows": counts["retained"],
        "outcome_counts": counts,
        "channels": list(channels) if status == "success" else [],
        "artifacts": {
            "source_xml": artifact_object(xml_path, "opera_xml", "application/xml"),
            "daily_report": artifact_object(
                daily_path,
                "daily_xlsx",
                "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
            ),
            "monthly_report": artifact_object(
                monthly_path,
                "monthly_xlsx",
                "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
            ),
            "result_json": artifact_object(result_json, "result_json", "application/json"),
            "exception_report": artifact_object(
                exception_path,
                "exception_xlsx",
                "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
            ),
        },
        "records": [structured_record(record) for record in records],
        "errors": [error.to_dict() for error in errors],
    }
    if processing_mode == MODE_DAILY:
        payload["processing_mode"] = MODE_DAILY
    return payload


def validate_structured_completeness(payload: Dict[str, Any]) -> None:
    failures: List[str] = []
    schema_version = payload.get("result_schema_version")
    if schema_version == STRUCTURED_RESULT_SCHEMA_VERSION:
        processing_mode = MODE_DAILY_MONTHLY
        if "processing_mode" in payload:
            failures.append("v1 structured payload不得包含processing_mode")
    elif schema_version == DAILY_STRUCTURED_RESULT_SCHEMA_VERSION:
        processing_mode = payload.get("processing_mode")
        if processing_mode != MODE_DAILY:
            failures.append("v2 structured payload的processing_mode必须为daily")
    else:
        processing_mode = None
        failures.append("result_schema_version不受支持")
    records = payload.get("records")
    if not isinstance(records, list):
        failures.append("records必须是数组")
        records = []
    source_rows = payload.get("source_rows")
    if source_rows != len(records):
        failures.append("source_rows必须等于records长度")
    sequences = [record.get("source_sequence") for record in records if isinstance(record, dict)]
    if sequences != list(range(1, len(records) + 1)):
        failures.append("source_sequence必须从1开始连续且保持XML顺序")
    counts = payload.get("outcome_counts", {})
    if not isinstance(counts, dict) or set(counts) != FINAL_OUTCOMES:
        failures.append("outcome_counts必须覆盖五种固定outcome")
        counts = {}
    for outcome in FINAL_OUTCOMES:
        actual = sum(1 for record in records if record.get("outcome") == outcome)
        if counts.get(outcome) != actual:
            failures.append(f"outcome_counts.{outcome}与记录不一致")
    if payload.get("removed_by_rate_code") != counts.get("excluded_rate_code"):
        failures.append("removed_by_rate_code与outcome计数不一致")
    if payload.get("removed_as_duplicates") != counts.get("duplicate"):
        failures.append("removed_as_duplicates与outcome计数不一致")
    if payload.get("output_rows") != counts.get("retained"):
        failures.append("output_rows与outcome计数不一致")
    status = payload.get("status")
    if status == "success":
        if not payload.get("activation_eligible"):
            failures.append("成功payload必须允许激活")
        if counts.get("validation_failed") or counts.get("price_unmatched"):
            failures.append("成功payload不得含校验或定价失败行")
        if source_rows != (
            payload.get("removed_by_rate_code", 0)
            + payload.get("removed_as_duplicates", 0)
            + payload.get("output_rows", 0)
        ):
            failures.append("成功payload的源行分解不平衡")
    elif status == "failed":
        if payload.get("activation_eligible"):
            failures.append("失败payload不得允许激活")
        if payload.get("output_rows") != 0:
            failures.append("失败payload不得包含正式输出行")
    else:
        failures.append("status必须为success或failed")

    artifacts = payload.get("artifacts", {})
    if not isinstance(artifacts, dict):
        failures.append("artifacts必须是对象")
        artifacts = {}
    expected_artifact_names = {
        "source_xml",
        "daily_report",
        "monthly_report",
        "result_json",
        "exception_report",
    }
    if set(artifacts) != expected_artifact_names:
        failures.append("artifacts字段集不符合契约")
    if status == "success":
        required_success_artifacts = ["source_xml", "daily_report", "result_json"]
        if processing_mode == MODE_DAILY_MONTHLY:
            required_success_artifacts.append("monthly_report")
        for name in required_success_artifacts:
            if not artifacts.get(name):
                failures.append(f"成功payload缺少{name}产物哈希")
        if processing_mode == MODE_DAILY and artifacts.get("monthly_report") is not None:
            failures.append("daily成功payload不得引用月报")
        if artifacts.get("exception_report") is not None:
            failures.append("成功payload不得包含异常清单")
    if status == "failed":
        if artifacts.get("daily_report") is not None or artifacts.get("monthly_report") is not None:
            failures.append("失败payload不得引用日报或月报")
        if artifacts.get("exception_report") is None:
            failures.append("失败payload必须引用异常清单")

    for record in records:
        if not isinstance(record, dict):
            failures.append("每条record必须是对象")
            continue
        if record.get("outcome") not in FINAL_OUTCOMES:
            failures.append("记录outcome不在固定集合内")
            continue
        sequence = record.get("source_sequence")
        if record.get("source_location") != f"reservation[{sequence}]":
            failures.append("记录source_location与source_sequence不一致")
        if record.get("source_worksheet") is not None or record.get("source_row_no") is not None:
            failures.append("直接XML记录不得伪造Excel源坐标")
        decisions = record.get("decision_codes")
        if (
            not isinstance(decisions, list)
            or not decisions
            or not all(isinstance(code, str) and code for code in decisions)
            or len(decisions) != len(dict.fromkeys(decisions))
        ):
            failures.append("记录decision_codes必须是非空且不重复的数组")
        if record.get("normalized_rate_code") != (
            text_or_blank(record.get("rate_code")).upper() or None
        ):
            failures.append("记录normalized_rate_code派生错误")
        expected_group = normalize_group_code(record.get("res_comment"))
        if record.get("group_code_key") != expected_group:
            failures.append("记录group_code_key派生错误")
        expected_match = "missing_group_code" if expected_group is None else "not_checked"
        if record.get("booking_source_match_status") != expected_match:
            failures.append("记录booking_source_match_status派生错误")
        company_name = text_or_blank(record.get("company_name"))
        expected_company_key: Optional[str] = None
        if company_name:
            try:
                expected_company_key = price_company_key(company_name)
            except CompanyKeywordAmbiguity:
                expected_company_key = None
        if record.get("company_key") != expected_company_key:
            failures.append("记录company_key派生错误")
        if record.get("outcome") == "duplicate":
            duplicate_of = record.get("duplicate_of_source_sequence")
            if not isinstance(duplicate_of, int) or duplicate_of >= record.get("source_sequence", 0):
                failures.append("重复记录必须指向更早源序号")
        elif record.get("duplicate_of_source_sequence") is not None:
            failures.append("非重复记录不得包含duplicate_of_source_sequence")
        if record.get("outcome") == "retained":
            required_retained = (
                "company_name",
                "company_key",
                "confirmation_no",
                "disp_room_no",
                "full_name",
                "normalized_rate_code",
                "arrival",
                "departure",
                "no_of_rooms",
                "nights",
                "effective_rate_amount",
                "real_price",
                "total_price",
                "channel_key",
                "pricing_method",
            )
            if any(record.get(field) in (None, "") for field in required_retained):
                failures.append("保留记录缺少必填派生字段")
                continue
            expected_total = (
                parse_decimal(record["real_price"])
                * record["no_of_rooms"]
                * record["nights"]
            )
            if parse_decimal(record["total_price"]) != expected_total:
                failures.append("保留记录total_price派生错误")
            if record["channel_key"] == KB_SHEET:
                if parse_decimal(record.get("kb_amount")) != Decimal(record["no_of_rooms"] * 100):
                    failures.append("保留记录kb_amount派生错误")
            elif record.get("kb_amount") is not None:
                failures.append("非KB渠道不得填写kb_amount")
            if record.get("pricing_method") == "zero_price_exception":
                if (
                    record.get("company_key") not in ZERO_TOTAL_COMPANIES
                    or record.get("normalized_rate_code") not in ZERO_TOTAL_RATE_CODES
                    or parse_decimal(record.get("real_price")) != Decimal(0)
                ):
                    failures.append("归零定价方法与公司/费率/金额条件不一致")
    if failures:
        raise ProcessingFailure(
            [
                ErrorItem(
                    "STRUCTURED_RESULT_INVALID",
                    "structured",
                    "；".join(dict.fromkeys(failures)),
                )
            ]
        )


def write_structured_result(path: Path, payload: Dict[str, Any]) -> None:
    validate_structured_completeness(payload)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")


def run_independent_validator(
    xml_path: Path,
    daily_path: Path,
    monthly_path: Optional[Path],
    result_json: Path,
    structured_result_json: Path,
    processing_mode: str = MODE_DAILY_MONTHLY,
) -> None:
    validator = Path(__file__).resolve().parent / "validate_outputs.py"
    command = [
        sys.executable,
        str(validator),
        "--mode",
        processing_mode,
        "--xml",
        str(xml_path),
        "--daily",
        str(daily_path),
        "--result-json",
        str(result_json),
        "--structured-result-json",
        str(structured_result_json),
        "--price-reference",
        str(PRICE_REFERENCE),
    ]
    if monthly_path is not None:
        command.extend(["--monthly", str(monthly_path)])
    completed = subprocess.run(command, capture_output=True, text=True, check=False)
    if completed.returncode == 0:
        return
    errors: List[ErrorItem] = []
    try:
        payload = json.loads(completed.stdout)
        for item in payload.get("errors", []):
            amount = item.get("effective_rate_amount")
            errors.append(
                ErrorItem(
                    item.get("code", "OUTPUT_VALIDATION_FAILED"),
                    item.get("stage", "output"),
                    item.get("message", "独立输出校验失败"),
                    item.get("source_location"),
                    item.get("company_name"),
                    item.get("rate_code"),
                    parse_decimal(amount) if amount is not None else None,
                    item.get("confirmation_no"),
                )
            )
    except Exception:
        errors = []
    if not errors:
        diagnostic = text_or_blank(completed.stderr) or text_or_blank(completed.stdout)
        errors = [
            ErrorItem(
                "OUTPUT_VALIDATION_FAILED",
                "output",
                f"独立输出校验失败：{diagnostic[:500]}",
            )
        ]
    raise ProcessingFailure(errors)


def remove_candidates(paths: Iterable[Optional[Path]]) -> None:
    for path in paths:
        if path is not None and path.is_file():
            path.unlink()


def process(args: argparse.Namespace) -> int:
    xml_path = Path(args.xml)
    monthly_base_arg = getattr(args, "monthly_base", None)
    monthly_base = Path(monthly_base_arg) if monthly_base_arg else None
    processing_mode = getattr(args, "mode", MODE_DAILY_MONTHLY) or MODE_DAILY_MONTHLY
    output_dir = Path(args.output_dir)
    result_json = Path(args.result_json)
    structured_arg = getattr(args, "structured_result_json", None)
    structured_result_json = (
        Path(structured_arg) if structured_arg else output_dir / "structured-result.json"
    )
    metrics = empty_metrics()
    business_date: Optional[date] = None
    daily_path: Optional[Path] = None
    monthly_path: Optional[Path] = None
    all_records: List[Dict[str, Any]] = []
    try:
        validate_invocation(
            xml_path,
            monthly_base,
            output_dir,
            result_json,
            structured_result_json,
            processing_mode,
        )
        business_date, reservations = read_xml(xml_path)
        metrics["source_rows"] = len(reservations)
        (
            all_records,
            records,
            removed_rate,
            removed_duplicates,
            classification_errors,
        ) = classify_source_records(
            reservations, business_date
        )
        metrics["removed_by_rate_code"] = removed_rate
        metrics["removed_as_duplicates"] = removed_duplicates
        if classification_errors:
            raise ProcessingFailure(classification_errors)
        price_map = load_price_map(PRICE_REFERENCE)
        pricing_errors = apply_prices_classified(records, price_map)
        if pricing_errors:
            raise ProcessingFailure(pricing_errors)

        if processing_mode == MODE_DAILY:
            channel_counts = assign_channels(records)
        daily_path = output_dir / f"{business_date.month}.{business_date.day}.xlsx"
        write_daily(records, business_date, daily_path)
        if processing_mode == MODE_DAILY_MONTHLY:
            monthly_path, channel_counts = write_monthly(
                records, business_date, monthly_base, output_dir
            )
        metrics["channels"] = channel_metrics(channel_counts)
        metrics["output_rows"] = len(records)
        finalize_success_outcomes(all_records)

        success = result_object(
            "success",
            business_date,
            "日报处理成功" if processing_mode == MODE_DAILY else "处理成功",
            metrics,
            daily=daily_path,
            monthly=monthly_path,
            processing_mode=processing_mode,
        )
        write_result(result_json, success)
        structured_success = build_structured_result(
            "success",
            business_date,
            all_records,
            metrics["channels"],
            xml_path,
            result_json,
            daily_path=daily_path,
            monthly_path=monthly_path,
            processing_mode=processing_mode,
        )
        write_structured_result(structured_result_json, structured_success)
        run_independent_validator(
            xml_path,
            daily_path,
            monthly_path,
            result_json,
            structured_result_json,
            processing_mode,
        )
        print(json.dumps(success, ensure_ascii=False))
        return 0
    except ProcessingFailure as exc:
        finalize_failure_outcomes(all_records)
        metrics["source_rows"] = len(all_records) or metrics["source_rows"]
        metrics["removed_by_rate_code"] = sum(
            1 for record in all_records if record.get("_OUTCOME") == "excluded_rate_code"
        )
        metrics["removed_as_duplicates"] = sum(
            1 for record in all_records if record.get("_OUTCOME") == "duplicate"
        )
        metrics["output_rows"] = 0
        metrics["channels"] = []
        remove_candidates((daily_path, monthly_path, structured_result_json))
        try:
            output_dir.mkdir(parents=True, exist_ok=True)
            exception_path = output_dir / "异常清单.xlsx"
            write_exception_workbook(exc.errors, exception_path)
            failed = result_object(
                "failed",
                business_date,
                exc.errors[0].message if exc.errors else "处理失败",
                metrics,
                exception=exception_path,
                errors=exc.errors,
                processing_mode=processing_mode,
            )
            if result_json.is_absolute() and is_within(result_json, output_dir):
                write_result(result_json, failed)
            if (
                structured_result_json.is_absolute()
                and is_within(structured_result_json, output_dir)
                and structured_result_json.resolve() != result_json.resolve()
            ):
                structured_failed = build_structured_result(
                    "failed",
                    business_date,
                    all_records,
                    [],
                    xml_path,
                    result_json if result_json.is_file() else None,
                    exception_path=exception_path,
                    errors=exc.errors,
                    processing_mode=processing_mode,
                )
                write_structured_result(structured_result_json, structured_failed)
            print(json.dumps(failed, ensure_ascii=False))
        except Exception as reporting_error:
            print(f"无法写入失败结果：{reporting_error}", file=sys.stderr)
        return exc.exit_code
    except Exception as exc:
        finalize_failure_outcomes(all_records)
        metrics["source_rows"] = len(all_records) or metrics["source_rows"]
        metrics["removed_by_rate_code"] = sum(
            1 for record in all_records if record.get("_OUTCOME") == "excluded_rate_code"
        )
        metrics["removed_as_duplicates"] = sum(
            1 for record in all_records if record.get("_OUTCOME") == "duplicate"
        )
        metrics["output_rows"] = 0
        metrics["channels"] = []
        remove_candidates((daily_path, monthly_path, structured_result_json))
        error = ErrorItem(
            "INTERNAL_ERROR",
            "internal",
            f"内部处理错误：{type(exc).__name__}: {exc}",
        )
        try:
            output_dir.mkdir(parents=True, exist_ok=True)
            exception_path = output_dir / "异常清单.xlsx"
            write_exception_workbook([error], exception_path)
            failed = result_object(
                "failed",
                business_date,
                error.message,
                metrics,
                exception=exception_path,
                errors=[error],
                processing_mode=processing_mode,
            )
            if result_json.is_absolute() and is_within(result_json, output_dir):
                write_result(result_json, failed)
            if (
                structured_result_json.is_absolute()
                and is_within(structured_result_json, output_dir)
                and structured_result_json.resolve() != result_json.resolve()
            ):
                structured_failed = build_structured_result(
                    "failed",
                    business_date,
                    all_records,
                    [],
                    xml_path,
                    result_json if result_json.is_file() else None,
                    exception_path=exception_path,
                    errors=[error],
                    processing_mode=processing_mode,
                )
                write_structured_result(structured_result_json, structured_failed)
            print(json.dumps(failed, ensure_ascii=False))
        except Exception:
            traceback.print_exc(file=sys.stderr)
        return 4


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--xml", required=True, help="Absolute path to the Opera XML")
    parser.add_argument(
        "--mode",
        choices=sorted(PROCESSING_MODES),
        default=MODE_DAILY_MONTHLY,
        help="daily produces only the daily XLSX; daily-monthly preserves the legacy bundle",
    )
    parser.add_argument("--monthly-base", help="Absolute path to the current monthly XLSX")
    parser.add_argument("--output-dir", required=True, help="Absolute isolated output directory")
    parser.add_argument("--result-json", required=True, help="Absolute result JSON path inside output-dir")
    parser.add_argument(
        "--structured-result-json",
        help="Absolute finance-ready structured JSON path inside output-dir; defaults to structured-result.json",
    )
    return parser


def main() -> int:
    return process(build_parser().parse_args())


if __name__ == "__main__":
    raise SystemExit(main())
