#!/usr/bin/env python3
"""Independently validate Opera XLSX outputs and the finance-ready structured payload."""

from __future__ import annotations

import argparse
import json
import sys
import traceback
from datetime import date, datetime
from decimal import Decimal
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple

from openpyxl import load_workbook
from openpyxl.utils import get_column_letter

import process_reports as core


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"}
DAILY_NUMERIC_FIELDS = {
    "ADULTS",
    "CHILDREN",
    "EFFECTIVE_RATE_AMOUNT",
    "NO_OF_ROOMS",
    "NIGHTS",
    "REAL PRICE",
    "TOTAL PRICE",
}
CHANNEL_NUMERIC_FIELDS = {
    "NIGHTS",
    "ADULTS",
    "CHILDREN",
    "NO_OF_ROOMS",
    "RATE_AMOUNT",
    "REAL PRICE",
    "TOTAL PRICE",
}
REQUIRED_TEXT_FIELDS = {
    "COMPANY_NAME",
    "CONFIRMATION_NO",
    "DISP_ROOM_NO",
    "FULL_NAME",
    "RATE_CODE",
}


def validation_error(
    code: str,
    message: str,
    source_location: Optional[str] = None,
    record: Optional[Dict[str, Any]] = None,
) -> core.ErrorItem:
    record = record or {}
    amount = record.get("EFFECTIVE_RATE_AMOUNT", record.get("RATE_AMOUNT"))
    try:
        normalized_amount = core.parse_decimal(amount) if amount is not None else None
    except ValueError:
        normalized_amount = None
    return core.ErrorItem(
        code=code,
        stage="output",
        message=message,
        source_location=source_location,
        company_name=core.text_or_blank(record.get("COMPANY_NAME")) or None,
        rate_code=core.text_or_blank(record.get("RATE_CODE")) or None,
        effective_rate_amount=normalized_amount,
        confirmation_no=core.text_or_blank(record.get("CONFIRMATION_NO")) or None,
    )


def actual_date(value: Any) -> Optional[date]:
    if isinstance(value, datetime):
        return value.date()
    if isinstance(value, date):
        return value
    return None


def is_number(value: Any) -> bool:
    return isinstance(value, (int, float)) and not isinstance(value, bool)


def comparable(value: Any, field: str) -> Any:
    if field in DATE_FIELDS:
        return actual_date(value)
    if field in DAILY_NUMERIC_FIELDS or field in CHANNEL_NUMERIC_FIELDS or field == core.KB_HEADER:
        try:
            return core.parse_decimal(value)
        except ValueError:
            return value
    if field in TEXT_FIELDS:
        return core.text_or_blank(value)
    return value


def workbook_rows(
    sheet: Any,
    headers: Sequence[str],
    workbook_name: str,
    errors: List[core.ErrorItem],
) -> List[Tuple[int, Dict[str, Any]]]:
    rows: List[Tuple[int, Dict[str, Any]]] = []
    last_column = get_column_letter(len(headers))
    for row_number in range(2, sheet.max_row + 1):
        values = [sheet.cell(row_number, col).value for col in range(1, len(headers) + 1)]
        if all(value is None or core.text_or_blank(value) == "" for value in values):
            continue
        location = f"{workbook_name}!{sheet.title}!A{row_number}:{last_column}{row_number}"
        for col, value in enumerate(values, 1):
            if isinstance(value, str) and value.startswith("=") and sheet.cell(row_number, col).data_type == "f":
                errors.append(
                    validation_error(
                        "OUTPUT_FORMULA_FORBIDDEN",
                        "输出数据不得包含公式",
                        f"{workbook_name}!{sheet.title}!{sheet.cell(row_number, col).coordinate}",
                    )
                )
        rows.append((row_number, dict(zip(headers, values))))
    for row in sheet.iter_rows(min_row=1):
        for cell in row:
            if cell.value is not None and cell.data_type == "f":
                location = f"{workbook_name}!{sheet.title}!{cell.coordinate}"
                if not any(error.source_location == location for error in errors):
                    errors.append(
                        validation_error(
                            "OUTPUT_FORMULA_FORBIDDEN", "输出工作簿不得包含公式", location
                        )
                    )
    return rows


def validate_headers(
    sheet: Any,
    expected: Sequence[str],
    workbook_name: str,
    errors: List[core.ErrorItem],
) -> bool:
    last_column = get_column_letter(len(expected))
    actual = [core.text_or_blank(sheet.cell(1, col).value) for col in range(1, len(expected) + 1)]
    if actual != list(expected):
        errors.append(
            validation_error(
                "OUTPUT_HEADER_MISMATCH",
                f"{len(expected)}列表头不符合契约；应为 {list(expected)}",
                f"{workbook_name}!{sheet.title}!A1:{last_column}1",
            )
        )
        return False
    for col in range(len(expected) + 1, sheet.max_column + 1):
        if sheet.cell(1, col).value not in (None, ""):
            errors.append(
                validation_error(
                    "OUTPUT_HEADER_MISMATCH",
                    f"第{len(expected)}列之后不得出现额外表头",
                    f"{workbook_name}!{sheet.title}!{sheet.cell(1, col).coordinate}",
                )
            )
            return False
    return True


def validate_row_types(
    row: Dict[str, Any],
    numeric_fields: Iterable[str],
    location: str,
    errors: List[core.ErrorItem],
) -> None:
    for field in TEXT_FIELDS:
        value = row.get(field)
        if value is not None and not isinstance(value, str):
            errors.append(
                validation_error(
                    "OUTPUT_TEXT_TYPE_MISMATCH",
                    f"{field} 必须以Excel文本类型写入",
                    location,
                    row,
                )
            )
        if field in REQUIRED_TEXT_FIELDS and core.text_or_blank(value) == "":
            errors.append(
                validation_error(
                    "OUTPUT_REQUIRED_TEXT_MISSING", f"{field} 不能为空", location, row
                )
            )
    for field in DATE_FIELDS:
        if actual_date(row.get(field)) is None:
            errors.append(
                validation_error(
                    "OUTPUT_DATE_TYPE_MISMATCH",
                    f"{field} 必须是Excel真实日期，而非文本",
                    location,
                    row,
                )
            )
    for field in numeric_fields:
        if not is_number(row.get(field)):
            errors.append(
                validation_error(
                    "OUTPUT_NUMBER_TYPE_MISMATCH",
                    f"{field} 必须是静态Excel数字",
                    location,
                    row,
                )
            )


def compare_rows(
    actual_rows: Sequence[Tuple[int, Dict[str, Any]]],
    expected_rows: Sequence[Dict[str, Any]],
    headers: Sequence[str],
    workbook_name: str,
    sheet_name: str,
    errors: List[core.ErrorItem],
) -> None:
    if len(actual_rows) != len(expected_rows):
        errors.append(
            validation_error(
                "OUTPUT_ROW_COUNT_MISMATCH",
                f"应有 {len(expected_rows)} 行，实际 {len(actual_rows)} 行",
                f"{workbook_name}!{sheet_name}",
            )
        )
    for offset, (actual_pair, expected) in enumerate(zip(actual_rows, expected_rows), 1):
        row_number, actual = actual_pair
        for field in headers:
            if comparable(actual.get(field), field) != comparable(expected.get(field), field):
                errors.append(
                    validation_error(
                        "OUTPUT_VALUE_OR_ORDER_MISMATCH",
                        f"第 {offset} 条记录的 {field} 与XML推导结果不一致",
                        f"{workbook_name}!{sheet_name}!row[{row_number}]/{field}",
                        actual,
                    )
                )
                break


def expected_from_xml(
    xml_path: Path, price_path: Path
) -> Tuple[date, int, int, int, List[Dict[str, Any]]]:
    business_date, reservations = core.read_xml(xml_path)
    records, removed_by_rate, removed_duplicates = core.filter_and_deduplicate(
        reservations, business_date
    )
    price_map = core.load_price_map(price_path)
    core.apply_prices(records, price_map)
    return business_date, len(reservations), removed_by_rate, removed_duplicates, records


def validate_result_contract(
    payload: Dict[str, Any],
    business_date: date,
    source_rows: int,
    removed_by_rate: int,
    removed_duplicates: int,
    expected_records: Sequence[Dict[str, Any]],
    daily_path: Path,
    monthly_path: Optional[Path],
    processing_mode: str,
    expected_channels: Optional[Sequence[Dict[str, Any]]],
    errors: List[core.ErrorItem],
) -> None:
    required = {
        "version",
        "status",
        "business_date",
        "month_key",
        "message",
        "metrics",
        "outputs",
        "errors",
    }
    if processing_mode == core.MODE_DAILY:
        required.add("processing_mode")
    if set(payload) != required:
        errors.append(
            validation_error(
                "OUTPUT_RESULT_CONTRACT_MISMATCH", "result.json 顶层字段不符合固定Schema"
            )
        )
        return
    expected_version = (
        core.DAILY_RESULT_VERSION
        if processing_mode == core.MODE_DAILY
        else core.VERSION
    )
    if payload.get("version") != expected_version or payload.get("status") != "success":
        errors.append(
            validation_error(
                "OUTPUT_RESULT_CONTRACT_MISMATCH", "成功校验时result版本或status不正确"
            )
        )
    if processing_mode == core.MODE_DAILY and payload.get("processing_mode") != core.MODE_DAILY:
        errors.append(
            validation_error(
                "OUTPUT_RESULT_CONTRACT_MISMATCH",
                "daily result 的 processing_mode 必须为 daily",
            )
        )
    if payload.get("business_date") != business_date.isoformat():
        errors.append(
            validation_error(
                "OUTPUT_RESULT_DATE_MISMATCH", "result业务日期与XML不一致"
            )
        )
    month_key = f"{business_date.year:04d}-{business_date.month:02d}"
    if payload.get("month_key") != month_key:
        errors.append(
            validation_error("OUTPUT_RESULT_DATE_MISMATCH", "result月份与XML不一致")
        )
    metrics = payload.get("metrics")
    expected_metrics = {
        "source_rows": source_rows,
        "removed_by_rate_code": removed_by_rate,
        "removed_as_duplicates": removed_duplicates,
        "output_rows": len(expected_records),
    }
    if not isinstance(metrics, dict):
        errors.append(validation_error("OUTPUT_RESULT_CONTRACT_MISMATCH", "metrics必须是对象"))
    else:
        for key, value in expected_metrics.items():
            if metrics.get(key) != value:
                errors.append(
                    validation_error(
                        "OUTPUT_RESULT_METRIC_MISMATCH",
                        f"metrics.{key} 应为 {value}，实际为 {metrics.get(key)}",
                    )
                )
        if not isinstance(metrics.get("channels"), list):
            errors.append(
                validation_error("OUTPUT_RESULT_CONTRACT_MISMATCH", "metrics.channels必须是数组")
            )
        elif expected_channels is not None and metrics.get("channels") != list(expected_channels):
            errors.append(
                validation_error(
                    "OUTPUT_RESULT_CHANNEL_MISMATCH",
                    "metrics.channels 与XML确定性渠道路由不一致",
                )
            )
    outputs = payload.get("outputs")
    if not isinstance(outputs, dict):
        errors.append(validation_error("OUTPUT_RESULT_CONTRACT_MISMATCH", "outputs必须是对象"))
    else:
        if outputs.get("daily_report") != daily_path.name:
            errors.append(
                validation_error("OUTPUT_RESULT_FILENAME_MISMATCH", "result日报文件名不一致")
            )
        if processing_mode == core.MODE_DAILY:
            if outputs.get("monthly_report") is not None:
                errors.append(
                    validation_error(
                        "OUTPUT_RESULT_CONTRACT_MISMATCH",
                        "daily result 不得包含月报文件名",
                    )
                )
        elif monthly_path is None or outputs.get("monthly_report") != monthly_path.name:
            errors.append(
                validation_error("OUTPUT_RESULT_FILENAME_MISMATCH", "result月报文件名不一致")
            )
        if outputs.get("exception_report") is not None:
            errors.append(
                validation_error(
                    "OUTPUT_RESULT_CONTRACT_MISMATCH", "成功结果不得包含异常清单文件名"
                )
            )
        filename_fields = ["daily_report"]
        if processing_mode == core.MODE_DAILY_MONTHLY:
            filename_fields.append("monthly_report")
        for field in filename_fields:
            value = outputs.get(field)
            if not isinstance(value, str) or Path(value).name != value:
                errors.append(
                    validation_error(
                        "OUTPUT_RESULT_PATH_FORBIDDEN", f"outputs.{field} 必须是相对文件名"
                    )
                )
    if payload.get("errors") != []:
        errors.append(
            validation_error("OUTPUT_RESULT_CONTRACT_MISMATCH", "成功结果的errors必须为空数组")
        )


def validate_artifact(
    actual: Any,
    expected_path: Path,
    file_kind: str,
    mime_type: str,
    label: str,
    errors: List[core.ErrorItem],
) -> None:
    expected = core.artifact_object(expected_path, file_kind, mime_type)
    if actual != expected:
        errors.append(
            validation_error(
                "OUTPUT_STRUCTURED_ARTIFACT_MISMATCH",
                f"structured-result.json 的 {label} 路径、哈希或文件元数据不一致",
                expected_path.name,
            )
        )


def validate_structured_result_contract(
    payload: Dict[str, Any],
    xml_path: Path,
    daily_path: Path,
    monthly_path: Optional[Path],
    result_json: Path,
    business_date: date,
    source_rows: int,
    removed_by_rate: int,
    removed_duplicates: int,
    expected_records: Sequence[Dict[str, Any]],
    result_payload: Dict[str, Any],
    processing_mode: str,
    errors: List[core.ErrorItem],
) -> None:
    required = {
        "result_schema_version",
        "status",
        "activation_eligible",
        "ingestion_mode",
        "business_date",
        "processor_version",
        "rule_set_sha256",
        "source_rows",
        "removed_by_rate_code",
        "removed_as_duplicates",
        "output_rows",
        "outcome_counts",
        "channels",
        "artifacts",
        "records",
        "errors",
    }
    if processing_mode == core.MODE_DAILY:
        required.add("processing_mode")
    if set(payload) != required:
        errors.append(
            validation_error(
                "OUTPUT_STRUCTURED_CONTRACT_MISMATCH",
                "structured-result.json 顶层字段不符合固定Schema",
            )
        )
        return
    try:
        core.validate_structured_completeness(payload)
    except core.ProcessingFailure as exc:
        for item in exc.errors:
            errors.append(
                validation_error(
                    "OUTPUT_STRUCTURED_CONTRACT_MISMATCH",
                    item.message,
                    item.source_location,
                )
            )
    expected_scalars = {
        "result_schema_version": (
            core.DAILY_STRUCTURED_RESULT_SCHEMA_VERSION
            if processing_mode == core.MODE_DAILY
            else core.STRUCTURED_RESULT_SCHEMA_VERSION
        ),
        "status": "success",
        "activation_eligible": True,
        "ingestion_mode": "opera_xml",
        "business_date": business_date.isoformat(),
        "processor_version": core.PROCESSOR_VERSION,
        "rule_set_sha256": core.rule_set_sha256(),
        "source_rows": source_rows,
        "removed_by_rate_code": removed_by_rate,
        "removed_as_duplicates": removed_duplicates,
        "output_rows": len(expected_records),
        "channels": result_payload.get("metrics", {}).get("channels"),
        "errors": [],
    }
    if processing_mode == core.MODE_DAILY:
        expected_scalars["processing_mode"] = core.MODE_DAILY
    for field, expected in expected_scalars.items():
        if payload.get(field) != expected:
            errors.append(
                validation_error(
                    "OUTPUT_STRUCTURED_VALUE_MISMATCH",
                    f"structured-result.json 的 {field} 应为 {expected!r}",
                )
            )
    expected_outcome_counts = {
        "duplicate": removed_duplicates,
        "excluded_rate_code": removed_by_rate,
        "price_unmatched": 0,
        "retained": len(expected_records),
        "validation_failed": 0,
    }
    if payload.get("outcome_counts") != expected_outcome_counts:
        errors.append(
            validation_error(
                "OUTPUT_STRUCTURED_OUTCOME_MISMATCH",
                f"structured outcome计数应为 {expected_outcome_counts}",
            )
        )

    artifacts = payload.get("artifacts")
    if isinstance(artifacts, dict):
        validate_artifact(artifacts.get("source_xml"), xml_path, "opera_xml", "application/xml", "source_xml", errors)
        validate_artifact(
            artifacts.get("daily_report"),
            daily_path,
            "daily_xlsx",
            "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
            "daily_report",
            errors,
        )
        if processing_mode == core.MODE_DAILY:
            if artifacts.get("monthly_report") is not None:
                errors.append(
                    validation_error(
                        "OUTPUT_STRUCTURED_ARTIFACT_MISMATCH",
                        "daily structured-result.json不得引用月报",
                    )
                )
        elif monthly_path is not None:
            validate_artifact(
                artifacts.get("monthly_report"),
                monthly_path,
                "monthly_xlsx",
                "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
                "monthly_report",
                errors,
            )
        validate_artifact(
            artifacts.get("result_json"),
            result_json,
            "result_json",
            "application/json",
            "result_json",
            errors,
        )
        if artifacts.get("exception_report") is not None:
            errors.append(
                validation_error(
                    "OUTPUT_STRUCTURED_ARTIFACT_MISMATCH",
                    "成功structured-result.json不得引用异常清单",
                )
            )
    else:
        errors.append(
            validation_error(
                "OUTPUT_STRUCTURED_CONTRACT_MISMATCH", "structured artifacts必须是对象"
            )
        )

    _date, reservations = core.read_xml(xml_path)
    all_records, retained, _removed_rate, _removed_duplicates, classification_errors = (
        core.classify_source_records(reservations, business_date)
    )
    if classification_errors:
        errors.append(
            validation_error(
                "OUTPUT_STRUCTURED_SOURCE_REPLAY_FAILED",
                "成功批次的XML独立重放不应出现行级校验错误",
            )
        )
        return
    price_map = core.load_price_map(Path(core.PRICE_REFERENCE).resolve())
    pricing_errors = core.apply_prices_classified(retained, price_map)
    if pricing_errors:
        errors.append(
            validation_error(
                "OUTPUT_STRUCTURED_SOURCE_REPLAY_FAILED",
                "成功批次的XML独立重放不应出现定价错误",
            )
        )
        return
    if processing_mode == core.MODE_DAILY:
        core.assign_channels(retained)
    actual_records = payload.get("records")
    if not isinstance(actual_records, list) or len(actual_records) != len(all_records):
        errors.append(
            validation_error(
                "OUTPUT_STRUCTURED_RECORD_COUNT_MISMATCH",
                f"structured records应保留全部 {len(all_records)} 条XML源记录",
            )
        )
        return
    actual_by_sequence = {
        item.get("source_sequence"): item for item in actual_records if isinstance(item, dict)
    }
    for expected_record in all_records:
        sequence = expected_record["_SOURCE_INDEX"]
        actual = actual_by_sequence.get(sequence)
        if actual is None:
            errors.append(
                validation_error(
                    "OUTPUT_STRUCTURED_SOURCE_SEQUENCE_MISSING",
                    f"structured records缺少source_sequence={sequence}",
                )
            )
            continue
        if expected_record["_OUTCOME"] == "pending":
            expected_record["_OUTCOME"] = "retained"
            if processing_mode == core.MODE_DAILY_MONTHLY:
                expected_record["_CHANNEL_KEY"] = actual.get("channel_key")
                expected_record["_KB_AMOUNT"] = (
                    Decimal(expected_record["NO_OF_ROOMS"] * 100)
                    if actual.get("channel_key") == core.KB_SHEET
                    else None
                )
                standard = core.route_standard(
                    expected_record["COMPANY_NAME"], expected_record["RATE_CODE"]
                )
                if standard is not None and actual.get("channel_key") != standard:
                    errors.append(
                        validation_error(
                            "OUTPUT_STRUCTURED_CHANNEL_MISMATCH",
                            f"source_sequence={sequence}的channel_key应为 {standard}",
                        )
                    )
        expected = core.structured_record(expected_record)
        if actual != expected:
            errors.append(
                validation_error(
                    "OUTPUT_STRUCTURED_RECORD_MISMATCH",
                    f"source_sequence={sequence}的结构化字段与XML确定性推导不一致",
                    f"reservation[{sequence}]",
                )
            )


def validate_daily(
    daily_path: Path,
    business_date: date,
    expected_records: Sequence[Dict[str, Any]],
    errors: List[core.ErrorItem],
) -> None:
    if not daily_path.is_file():
        errors.append(validation_error("OUTPUT_DAILY_MISSING", "找不到候选日报", daily_path.name))
        return
    try:
        workbook = load_workbook(daily_path, data_only=False)
    except Exception as exc:
        errors.append(
            validation_error(
                "OUTPUT_DAILY_UNREADABLE", f"候选日报无法读取：{exc}", daily_path.name
            )
        )
        return
    try:
        if len(workbook.worksheets) != 1:
            errors.append(
                validation_error("OUTPUT_DAILY_SHEET_COUNT", "日报必须且只能包含一个工作表")
            )
        sheet = workbook.active
        expected_title = f"{business_date.month}.{business_date.day}"
        if sheet.title != expected_title:
            errors.append(
                validation_error(
                    "OUTPUT_DAILY_SHEET_NAME",
                    f"日报工作表名应为 {expected_title}",
                    f"{daily_path.name}!{sheet.title}",
                )
            )
        if not validate_headers(sheet, core.DAILY_HEADERS, daily_path.name, errors):
            return
        rows = workbook_rows(sheet, core.DAILY_HEADERS, daily_path.name, errors)
        seen: set = set()
        for row_number, row in rows:
            last_column = get_column_letter(len(core.DAILY_HEADERS))
            location = f"{daily_path.name}!{sheet.title}!A{row_number}:{last_column}{row_number}"
            validate_row_types(row, DAILY_NUMERIC_FIELDS, location, errors)
            arrival = actual_date(row.get("ARRIVAL"))
            departure = actual_date(row.get("DEPARTURE"))
            if arrival and departure and is_number(row.get("NIGHTS")):
                if departure < arrival or int(row["NIGHTS"]) != (departure - arrival).days:
                    errors.append(
                        validation_error(
                            "OUTPUT_NIGHTS_MISMATCH", "日报晚数与日期不一致", location, row
                        )
                    )
            if arrival != business_date:
                errors.append(
                    validation_error(
                        "OUTPUT_BUSINESS_DATE_MISMATCH", "日报ARRIVAL与XML业务日期不一致", location, row
                    )
                )
            if core.text_or_blank(row.get("RATE_CODE")).upper() not in core.RATE_WHITELIST:
                errors.append(
                    validation_error(
                        "OUTPUT_RATE_NOT_WHITELISTED", "日报包含费率白名单外的记录", location, row
                    )
                )
            if all(
                is_number(row.get(field))
                for field in ("REAL PRICE", "NO_OF_ROOMS", "NIGHTS", "TOTAL PRICE")
            ):
                expected_total = row["REAL PRICE"] * row["NO_OF_ROOMS"] * row["NIGHTS"]
                if core.parse_decimal(row["TOTAL PRICE"]) != core.parse_decimal(expected_total):
                    errors.append(
                        validation_error(
                            "OUTPUT_DAILY_TOTAL_PRICE_MISMATCH",
                            "日报TOTAL PRICE不等于REAL PRICE×NO_OF_ROOMS×NIGHTS",
                            location,
                            row,
                        )
                    )
            key = (core.text_or_blank(row.get("DISP_ROOM_NO")), arrival)
            if key in seen:
                errors.append(
                    validation_error(
                        "OUTPUT_DUPLICATE_KEY", "日报包含重复的房号+ARRIVAL", location, row
                    )
                )
            seen.add(key)
        compare_rows(
            rows,
            expected_records,
            core.DAILY_HEADERS,
            daily_path.name,
            sheet.title,
            errors,
        )
    finally:
        workbook.close()


def validate_monthly(
    monthly_path: Path,
    business_date: date,
    expected_records: Sequence[Dict[str, Any]],
    result_payload: Dict[str, Any],
    errors: List[core.ErrorItem],
) -> None:
    if not monthly_path.is_file():
        errors.append(
            validation_error("OUTPUT_MONTHLY_MISSING", "找不到候选月报", monthly_path.name)
        )
        return
    try:
        workbook = load_workbook(monthly_path, data_only=False)
    except Exception as exc:
        errors.append(
            validation_error(
                "OUTPUT_MONTHLY_UNREADABLE", f"候选月报无法读取：{exc}", monthly_path.name
            )
        )
        return
    try:
        for sheet_name in core.STANDARD_SHEETS:
            if sheet_name not in workbook.sheetnames:
                errors.append(
                    validation_error(
                        "OUTPUT_MONTHLY_STANDARD_SHEET_MISSING",
                        f"候选月报缺少标准工作表 {sheet_name}",
                    )
                )
        all_arrivals: List[date] = []
        current_by_sheet: Dict[str, List[Tuple[int, Dict[str, Any]]]] = {}
        unknown_company_sheet: Dict[str, str] = {}
        for sheet in workbook.worksheets:
            headers = core.channel_headers_for_sheet(sheet.title)
            column_count = len(headers)
            last_column = get_column_letter(column_count)
            if not validate_headers(sheet, headers, monthly_path.name, errors):
                continue
            rows = workbook_rows(sheet, headers, monthly_path.name, errors)
            previous: Optional[date] = None
            companies: set = set()
            seen_day_room: set = set()
            current: List[Tuple[int, Dict[str, Any]]] = []
            for row_number, row in rows:
                location = (
                    f"{monthly_path.name}!{sheet.title}!A{row_number}:{last_column}{row_number}"
                )
                numeric_fields = set(CHANNEL_NUMERIC_FIELDS)
                if sheet.title == core.KB_SHEET:
                    numeric_fields.add(core.KB_HEADER)
                validate_row_types(row, numeric_fields, location, errors)
                arrival = actual_date(row.get("ARRIVAL"))
                departure = actual_date(row.get("DEPARTURE"))
                if arrival:
                    all_arrivals.append(arrival)
                    if (arrival.year, arrival.month) != (business_date.year, business_date.month):
                        errors.append(
                            validation_error(
                                "OUTPUT_MONTHLY_WRONG_MONTH",
                                "候选月报包含其他月份的数据",
                                location,
                                row,
                            )
                        )
                    if previous and arrival < previous:
                        errors.append(
                            validation_error(
                                "OUTPUT_MONTHLY_DATE_ORDER",
                                "工作表数据块未按ARRIVAL升序排列",
                                location,
                                row,
                            )
                        )
                    previous = arrival
                    if arrival == business_date:
                        current.append((row_number, row))
                if arrival and departure and is_number(row.get("NIGHTS")):
                    if departure < arrival or int(row["NIGHTS"]) != (departure - arrival).days:
                        errors.append(
                            validation_error(
                                "OUTPUT_NIGHTS_MISMATCH", "月报晚数与日期不一致", location, row
                            )
                        )
                if all(
                    is_number(row.get(field))
                    for field in ("REAL PRICE", "NO_OF_ROOMS", "NIGHTS", "TOTAL PRICE")
                ):
                    expected_total = (
                        row["REAL PRICE"] * row["NO_OF_ROOMS"] * row["NIGHTS"]
                    )
                    if core.parse_decimal(row["TOTAL PRICE"]) != core.parse_decimal(expected_total):
                        errors.append(
                            validation_error(
                                "OUTPUT_TOTAL_PRICE_MISMATCH",
                                "月报TOTAL PRICE不等于REAL PRICE×NO_OF_ROOMS×NIGHTS",
                                location,
                                row,
                            )
                        )
                if sheet.title == core.KB_SHEET and all(
                    is_number(row.get(field)) for field in (core.KB_HEADER, "NO_OF_ROOMS")
                ):
                    expected_kb = row["NO_OF_ROOMS"] * 100
                    if core.parse_decimal(row[core.KB_HEADER]) != core.parse_decimal(expected_kb):
                        errors.append(
                            validation_error(
                                "OUTPUT_KB_MISMATCH",
                                f"月报{core.KB_HEADER}不等于NO_OF_ROOMS×100",
                                location,
                                row,
                            )
                        )
                company = core.text_or_blank(row.get("COMPANY_NAME"))
                rate = core.text_or_blank(row.get("RATE_CODE")).upper()
                expected_sheet = core.route_standard(company, rate)
                if expected_sheet and expected_sheet != sheet.title:
                    errors.append(
                        validation_error(
                            "OUTPUT_ROUTE_MISMATCH",
                            f"记录应路由到 {expected_sheet}，实际位于 {sheet.title}",
                            location,
                            row,
                        )
                    )
                if expected_sheet is None and sheet.title in core.STANDARD_SHEETS:
                    errors.append(
                        validation_error(
                            "OUTPUT_ROUTE_MISMATCH",
                            "非标准公司记录不得位于标准工作表",
                            location,
                            row,
                        )
                    )
                companies.add(company)
                key = (arrival, core.text_or_blank(row.get("DISP_ROOM_NO")))
                if key in seen_day_room:
                    errors.append(
                        validation_error(
                            "OUTPUT_MONTHLY_DUPLICATE_KEY",
                            "同一工作表包含重复的ARRIVAL+房号",
                            location,
                            row,
                        )
                    )
                seen_day_room.add(key)
            current_by_sheet[sheet.title] = current
            if sheet.title not in core.STANDARD_SHEETS:
                if len(companies) > 1:
                    errors.append(
                        validation_error(
                            "OUTPUT_MONTHLY_MIXED_COMPANIES",
                            "非标准工作表不得混放多个公司",
                            f"{monthly_path.name}!{sheet.title}",
                        )
                    )
                elif len(companies) == 1:
                    company = next(iter(companies))
                    if not core.valid_unknown_sheet_title(company, sheet.title):
                        errors.append(
                            validation_error(
                                "OUTPUT_ROUTE_MISMATCH",
                                "非标准工作表名不符合实际公司名及碰撞后缀规则",
                                f"{monthly_path.name}!{sheet.title}",
                                {"COMPANY_NAME": company},
                            )
                        )
                    if company in unknown_company_sheet and unknown_company_sheet[company] != sheet.title:
                        errors.append(
                            validation_error(
                                "OUTPUT_ROUTE_MISMATCH",
                                "同一个非标准公司被拆到多个工作表",
                                f"{monthly_path.name}!{sheet.title}",
                            )
                        )
                    unknown_company_sheet[company] = sheet.title

        expected_by_sheet: Dict[str, List[Dict[str, Any]]] = {}
        for record in expected_records:
            standard = core.route_standard(record["COMPANY_NAME"], record["RATE_CODE"])
            sheet_name = standard or unknown_company_sheet.get(record["COMPANY_NAME"])
            if sheet_name is None:
                errors.append(
                    validation_error(
                        "OUTPUT_ROUTE_MISSING",
                        "找不到非标准公司的目标工作表",
                        record=record,
                    )
                )
                continue
            expected_by_sheet.setdefault(sheet_name, []).append(
                core.to_channel_record(record, sheet_name)
            )
        all_sheet_names = set(current_by_sheet) | set(expected_by_sheet)
        for sheet_name in sorted(all_sheet_names):
            headers = core.channel_headers_for_sheet(sheet_name)
            compare_rows(
                current_by_sheet.get(sheet_name, []),
                expected_by_sheet.get(sheet_name, []),
                headers,
                monthly_path.name,
                sheet_name,
                errors,
            )
        expected_counts = {name: len(rows) for name, rows in expected_by_sheet.items() if rows}
        expected_channel_metrics = core.channel_metrics(expected_counts)
        actual_metrics = result_payload.get("metrics", {}).get("channels")
        if actual_metrics != expected_channel_metrics:
            errors.append(
                validation_error(
                    "OUTPUT_RESULT_CHANNEL_METRIC_MISMATCH",
                    f"渠道计数应为 {expected_channel_metrics}，实际为 {actual_metrics}",
                )
            )
        if not all_arrivals:
            errors.append(validation_error("OUTPUT_MONTHLY_EMPTY", "候选月报没有数据行"))
        else:
            max_arrival = max(all_arrivals)
            expected_name = (
                f"各渠道情况-{business_date.year:04d}年{business_date.month:02d}月-"
                f"更新至{max_arrival.month}.{max_arrival.day}.xlsx"
            )
            if monthly_path.name != expected_name:
                errors.append(
                    validation_error(
                        "OUTPUT_MONTHLY_FILENAME_MISMATCH",
                        f"月报文件名应为 {expected_name}",
                        monthly_path.name,
                    )
                )
    finally:
        workbook.close()


def validate(args: argparse.Namespace) -> List[core.ErrorItem]:
    processing_mode = getattr(args, "mode", core.MODE_DAILY_MONTHLY) or core.MODE_DAILY_MONTHLY
    if processing_mode not in core.PROCESSING_MODES:
        return [
            validation_error(
                "OUTPUT_VALIDATOR_INPUT_INVALID",
                f"独立校验器 mode 必须是 {core.MODE_DAILY} 或 {core.MODE_DAILY_MONTHLY}",
                processing_mode,
            )
        ]
    xml_path = Path(args.xml)
    daily_path = Path(args.daily)
    monthly_arg = getattr(args, "monthly", None)
    monthly_path = Path(monthly_arg) if monthly_arg else None
    result_json = Path(args.result_json)
    structured_result_json = Path(args.structured_result_json)
    price_path = Path(args.price_reference)
    required_paths = [
        (xml_path, ".xml", "XML"),
        (daily_path, ".xlsx", "日报"),
        (result_json, ".json", "result.json"),
        (structured_result_json, ".json", "structured-result.json"),
        (price_path, ".xlsx", "价格对照"),
    ]
    if processing_mode == core.MODE_DAILY_MONTHLY:
        if monthly_path is None:
            return [
                validation_error(
                    "OUTPUT_VALIDATOR_INPUT_INVALID",
                    "daily-monthly 独立校验必须提供月报文件",
                )
            ]
        required_paths.append((monthly_path, ".xlsx", "月报"))
    elif monthly_path is not None:
        return [
            validation_error(
                "OUTPUT_VALIDATOR_INPUT_INVALID",
                "daily 独立校验不得提供月报文件",
                str(monthly_path),
            )
        ]
    for path, suffix, label in required_paths:
        if not path.is_absolute() or not path.is_file() or path.suffix.lower() != suffix:
            return [
                validation_error(
                    "OUTPUT_VALIDATOR_INPUT_INVALID",
                    f"独立校验器的{label}路径必须是存在的绝对{suffix}文件",
                    str(path),
                )
            ]
    try:
        payload = json.loads(result_json.read_text(encoding="utf-8"))
    except Exception as exc:
        return [
            validation_error(
                "OUTPUT_RESULT_UNREADABLE", f"result.json无法读取：{exc}", result_json.name
            )
        ]
    if not isinstance(payload, dict):
        return [validation_error("OUTPUT_RESULT_CONTRACT_MISMATCH", "result.json必须是对象")]
    try:
        structured_payload = json.loads(structured_result_json.read_text(encoding="utf-8"))
    except Exception as exc:
        return [
            validation_error(
                "OUTPUT_STRUCTURED_RESULT_UNREADABLE",
                f"structured-result.json无法读取：{exc}",
                structured_result_json.name,
            )
        ]
    if not isinstance(structured_payload, dict):
        return [
            validation_error(
                "OUTPUT_STRUCTURED_CONTRACT_MISMATCH", "structured-result.json必须是对象"
            )
        ]

    business_date, source_rows, removed_rate, removed_duplicates, records = expected_from_xml(
        xml_path, price_path
    )
    expected_channels: Optional[List[Dict[str, Any]]] = None
    if processing_mode == core.MODE_DAILY:
        expected_channels = core.channel_metrics(core.assign_channels(records))
    errors: List[core.ErrorItem] = []
    validate_result_contract(
        payload,
        business_date,
        source_rows,
        removed_rate,
        removed_duplicates,
        records,
        daily_path,
        monthly_path,
        processing_mode,
        expected_channels,
        errors,
    )
    validate_structured_result_contract(
        structured_payload,
        xml_path,
        daily_path,
        monthly_path,
        result_json,
        business_date,
        source_rows,
        removed_rate,
        removed_duplicates,
        records,
        payload,
        processing_mode,
        errors,
    )
    validate_daily(daily_path, business_date, records, errors)
    if processing_mode == core.MODE_DAILY_MONTHLY and monthly_path is not None:
        validate_monthly(monthly_path, business_date, records, payload, errors)
    return errors


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--mode",
        choices=sorted(core.PROCESSING_MODES),
        default=core.MODE_DAILY_MONTHLY,
    )
    parser.add_argument("--xml", required=True)
    parser.add_argument("--daily", required=True)
    parser.add_argument("--monthly")
    parser.add_argument("--result-json", required=True)
    parser.add_argument("--structured-result-json", required=True)
    parser.add_argument("--price-reference", required=True)
    return parser


def main() -> int:
    try:
        errors = validate(build_parser().parse_args())
        if errors:
            payload = {"status": "failed", "errors": [error.to_dict() for error in errors]}
            print(json.dumps(payload, ensure_ascii=False))
            return 2
        print(json.dumps({"status": "success", "errors": []}, ensure_ascii=False))
        return 0
    except core.ProcessingFailure as exc:
        payload = {"status": "failed", "errors": [error.to_dict() for error in exc.errors]}
        print(json.dumps(payload, ensure_ascii=False))
        return exc.exit_code if exc.exit_code in {2, 3} else 2
    except Exception as exc:
        error = validation_error(
            "INTERNAL_ERROR", f"独立校验器内部错误：{type(exc).__name__}: {exc}"
        )
        print(json.dumps({"status": "failed", "errors": [error.to_dict()]}, ensure_ascii=False))
        traceback.print_exc(file=sys.stderr)
        return 4


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