#!/usr/bin/env python3 """Convert one Opera RES_DETAIL XML into a daily XLSX and all-source 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, Mapping, 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 RESULT_VERSION = "4.0" PROCESSOR_VERSION = "4.0.0" STRUCTURED_RESULT_SCHEMA_VERSION = "4.0" # Retired direct-MCP/callback compatibility. This is deliberately an internal, # opt-in success-only projection; the active processor always emits the v4 # contract and is the only path that can enter price review. LEGACY_DIRECT_RESULT_VERSION = "3.0" LEGACY_DIRECT_PROCESSOR_VERSION = "3.0.0" LEGACY_DIRECT_RULE_SET_SHA256 = ( "c41257208324a43e711de13ec9776a5e6486db334757f152531bf8292a2018eb" ) REVIEW_VERSION = "1.0" SKILL_ROOT = Path(__file__).resolve().parent.parent PRICE_REFERENCE = SKILL_ROOT / "references" / "价格对照.xlsx" DAILY_TEMPLATE = SKILL_ROOT / "assets" / "daily-template.xlsx" STRUCTURED_RESULT_SCHEMA = SKILL_ROOT / "references" / "structured-result.schema.json" MANUAL_OVERRIDE_SCHEMA = SKILL_ROOT / "references" / "manual-override.schema.json" RULE_SET_PATHS = ( SKILL_ROOT / "SKILL.md", SKILL_ROOT / "scripts" / "process_daily.py", SKILL_ROOT / "scripts" / "validate_daily.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, MANUAL_OVERRIDE_SCHEMA, PRICE_REFERENCE, DAILY_TEMPLATE, ) FINAL_OUTCOMES = { "retained", "excluded_rate_code", "duplicate", "validation_failed", "price_unmatched", "candidate", } LEGACY_DIRECT_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", ] KB_SHEET = "DY-AI-Easy-KB" RATE_WHITELIST = { "GRPA1", "GRPA2", "GRPA3", "GRPA4", "GRP1", "WHO1", "WHO2", "WHO3", "WHO4", "LTLT", "LBLT", "LBSM", "LBMS", "LBW1", "LBKB", "LBLS", "WHKR2100B", "GL2100B", "GL2200KR", "GLSPCB", } STANDARD_CHANNELS = ["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)}") @dataclass(frozen=True) class ManualOverrideManifest: review_case_id: str sha256: str prices: Dict[Tuple[str, str, Decimal], Decimal] 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() MANUAL_REAL_PRICE_RE = re.compile(r"^(?:0|[1-9][0-9]{0,15})\.[0-9]{2}$") REVIEW_CASE_ID_RE = re.compile(r"^dailyreview-[0-9a-f]{32}$") REVIEW_JOB_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$") SHA256_RE = re.compile(r"^[0-9a-f]{64}$") def canonical_decimal_text(value: Decimal) -> str: """Render a non-exponent decimal deterministically for immutable key material.""" normalized = value.normalize() if normalized == Decimal("-0"): normalized = Decimal(0) return format(normalized, "f") def canonical_json_bytes(value: Any) -> bytes: return json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False, ).encode("utf-8") def strict_json_object(path: Path) -> Tuple[Dict[str, Any], bytes]: """Read only canonical JSON and reject duplicate keys/non-standard constants.""" def unique_pairs(pairs: Sequence[Tuple[str, Any]]) -> Dict[str, Any]: result: Dict[str, Any] = {} for key, value in pairs: if key in result: raise ValueError(f"duplicate JSON key: {key}") result[key] = value return result try: raw = path.read_bytes() decoded = raw.decode("utf-8") value = json.loads( decoded, object_pairs_hook=unique_pairs, parse_constant=lambda token: (_ for _ in ()).throw( ValueError(f"non-standard JSON constant: {token}") ), ) except Exception as exc: raise ValueError(f"人工价格清单不是有效UTF-8 JSON：{exc}") from exc if not isinstance(value, dict): raise ValueError("人工价格清单顶层必须是对象") try: canonical = canonical_json_bytes(value) except (TypeError, ValueError) as exc: raise ValueError(f"人工价格清单无法规范化：{exc}") from exc if raw != canonical: raise ValueError("人工价格清单必须是严格canonical JSON") return value, raw def _manual_manifest_error(message: str, path: Path) -> ProcessingFailure: return ProcessingFailure( [ErrorItem("MANUAL_OVERRIDE_INVALID", "manual_review", message, str(path))] ) def _manifest_key( item: Any, *, override: bool, path: Path, ) -> Tuple[Tuple[str, str, Decimal], Optional[Decimal]]: required = {"company_key", "rate_code", "effective_rate_amount"} if override: required.add("real_price") if not isinstance(item, dict) or set(item) != required: raise _manual_manifest_error("人工价格清单问题键字段不符合契约", path) company_key = item.get("company_key") rate_code = item.get("rate_code") amount_text = item.get("effective_rate_amount") if ( not isinstance(company_key, str) or not company_key or company_key != company_key.strip() or not isinstance(rate_code, str) or not rate_code or rate_code != rate_code.strip().upper() or not isinstance(amount_text, str) ): raise _manual_manifest_error("人工价格清单问题键格式非法", path) try: amount = parse_decimal(amount_text) except ValueError as exc: raise _manual_manifest_error("人工价格清单Opera价格不是有效数字", path) from exc if amount < 0 or amount_text != canonical_decimal_text(amount): raise _manual_manifest_error("人工价格清单Opera价格不是canonical非负金额", path) price: Optional[Decimal] = None if override: price_text = item.get("real_price") if not isinstance(price_text, str) or not MANUAL_REAL_PRICE_RE.fullmatch(price_text): raise _manual_manifest_error("人工价格必须是0至16位整数加两位小数的非负字符串", path) try: price = parse_decimal(price_text) except ValueError as exc: raise _manual_manifest_error("人工价格不是有效金额", path) from exc if price < 0: raise _manual_manifest_error("人工价格不得为负数", path) return (company_key, rate_code, amount), price def load_manual_override_manifest( path: Path, *, job_id: str, review_case_id: str, expected_sha256: str, xml_path: Path, business_date: date, ) -> ManualOverrideManifest: """Load the frozen staff decision list and bind it to this exact replay.""" try: ensure_absolute_file(path, ".json", "MANUAL_OVERRIDE_INVALID", "人工价格清单") except ProcessingFailure: raise if not REVIEW_JOB_ID_RE.fullmatch(job_id): raise _manual_manifest_error("人工价格清单任务标识非法", path) if not REVIEW_CASE_ID_RE.fullmatch(review_case_id): raise _manual_manifest_error("人工价格清单复核标识非法", path) if not SHA256_RE.fullmatch(expected_sha256): raise _manual_manifest_error("人工价格清单哈希格式非法", path) try: payload, raw = strict_json_object(path) except ValueError as exc: raise _manual_manifest_error(str(exc), path) from exc expected_fields = { "review_version", "review_case_id", "job_id", "source_sha256", "business_date", "processor_version", "rule_set_sha256", "issues", "overrides", } if set(payload) != expected_fields: raise _manual_manifest_error("人工价格清单顶层字段不符合契约", path) if payload.get("review_version") != REVIEW_VERSION: raise _manual_manifest_error("人工价格清单复核版本不受支持", path) if payload.get("review_case_id") != review_case_id or payload.get("job_id") != job_id: raise _manual_manifest_error("人工价格清单未绑定当前任务或复核case", path) if payload.get("source_sha256") != sha256_file(xml_path): raise _manual_manifest_error("人工价格清单源XML哈希不一致", path) if payload.get("business_date") != business_date.isoformat(): raise _manual_manifest_error("人工价格清单业务日期不一致", path) if payload.get("processor_version") != PROCESSOR_VERSION: raise _manual_manifest_error("人工价格清单处理器版本不一致", path) if payload.get("rule_set_sha256") != rule_set_sha256(): raise _manual_manifest_error("人工价格清单规则哈希不一致", path) actual_sha256 = hashlib.sha256(raw).hexdigest() if actual_sha256 != expected_sha256: raise _manual_manifest_error("人工价格清单哈希不一致", path) issues = payload.get("issues") overrides = payload.get("overrides") if not isinstance(issues, list) or not issues or not isinstance(overrides, list): raise _manual_manifest_error("人工价格清单问题或价格项必须是非空数组", path) issue_keys: List[Tuple[str, str, Decimal]] = [] override_prices: Dict[Tuple[str, str, Decimal], Decimal] = {} for item in issues: key, _price = _manifest_key(item, override=False, path=path) issue_keys.append(key) for item in overrides: key, price = _manifest_key(item, override=True, path=path) assert price is not None if key in override_prices: raise _manual_manifest_error("人工价格清单存在重复问题键", path) override_prices[key] = price ordered_issue_keys = sorted(issue_keys) ordered_override_keys = sorted(override_prices) if issue_keys != ordered_issue_keys or len(issue_keys) != len(set(issue_keys)): raise _manual_manifest_error("人工价格清单问题键必须唯一且按canonical顺序排列", path) if list(override_prices) != ordered_override_keys: raise _manual_manifest_error("人工价格清单人工价格必须按canonical顺序排列", path) if set(issue_keys) != set(override_prices): raise _manual_manifest_error("人工价格清单必须恰好覆盖完整问题键集合", path) return ManualOverrideManifest( review_case_id=review_case_id, sha256=actual_sha256, prices=override_prices, ) 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, output_dir: Path, result_json: Path, structured_result_json: Path, ) -> None: ensure_absolute_file(xml_path, ".xml", "INPUT_XML_INVALID", "XML文件") 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" 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], manual_prices: Optional[Dict[Tuple[str, str, Decimal], Decimal]] = None, ) -> 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: manual_price = manual_prices.get(key) if manual_prices is not None else None if manual_price is not None: real_price = manual_price record["_PRICING_METHOD"] = "manual_review" append_decision(record, "MANUAL_PRICE_APPLIED") else: 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 else: 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 missing_price_keys( records: Sequence[Dict[str, Any]], price_map: Dict[Tuple[str, str, Decimal], Decimal] ) -> set[Tuple[str, str, Decimal]]: """Derive the exact mutable-key set without trusting a submitted manifest.""" keys: set[Tuple[str, str, Decimal]] = set() for record in records: rate_code = record["RATE_CODE"].strip().upper() company_key = price_company_key(record["COMPANY_NAME"]) if company_key in ZERO_TOTAL_COMPANIES and rate_code in ZERO_TOTAL_RATE_CODES: continue key = (company_key, rate_code, record["EFFECTIVE_RATE_AMOUNT"]) if key not in price_map: keys.add(key) return keys def review_issues( records: Sequence[Dict[str, Any]], price_map: Dict[Tuple[str, str, Decimal], Decimal] ) -> List[Dict[str, Any]]: """Aggregate only price-safe review metadata by normalized pricing key.""" grouped: Dict[Tuple[str, str, Decimal], List[Dict[str, Any]]] = {} for record in records: if record.get("_OUTCOME") != "price_unmatched": continue company_key = record.get("_COMPANY_KEY") if not isinstance(company_key, str) or not company_key: raise ProcessingFailure( [ ErrorItem( "PRICE_REVIEW_KEY_INVALID", "manual_review", "待人工定价记录缺少标准化公司键", f"reservation[{record['_SOURCE_INDEX']}]", ) ] ) key = ( company_key, record["RATE_CODE"].strip().upper(), record["EFFECTIVE_RATE_AMOUNT"], ) grouped.setdefault(key, []).append(record) issues: List[Dict[str, Any]] = [] for key in sorted(grouped): company_key, rate_code, opera_amount = key affected = grouped[key] comparable_prices = [ { "effective_rate_amount": decimal_to_number(candidate_amount), "real_price": decimal_to_number(candidate_price), } for (candidate_company, candidate_rate, candidate_amount), candidate_price in sorted( price_map.items(), key=lambda item: (item[0][2], item[1]) ) if candidate_company == company_key and candidate_rate == rate_code ] issues.append( { "company_key": company_key, "rate_code": rate_code, "effective_rate_amount": decimal_to_number(opera_amount), "candidate_prices": comparable_prices, "affected_records": len(affected), "affected_rooms": sum(int(record["NO_OF_ROOMS"]) for record in affected), "affected_room_nights": sum( int(record["NO_OF_ROOMS"]) * int(record["NIGHTS"]) for record in affected ), } ) return issues 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( "ROUTING_COMPANY_AMBIGUOUS", "routing", 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 sanitized_sheet_base(company: str) -> str: cleaned = INVALID_SHEET_CHARS.sub("", company.strip()) return cleaned[:31] 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( "ROUTING_CHANNEL_KEY_EMPTY", "routing", "公司名清除Excel禁用字符后为空，无法生成channel_key", 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 creating a channel workbook.""" used = set(used_sheet_names or STANDARD_CHANNELS) 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 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, "candidate_rows": 0, "review_required_rows": 0, "review_issue_count": 0, "channels": [], } def channel_metrics(channel_counts: Dict[str, int]) -> List[Dict[str, Any]]: ordered_names = [name for name in STANDARD_CHANNELS if channel_counts.get(name, 0) > 0] ordered_names.extend( sorted( name for name, count in channel_counts.items() if name not in STANDARD_CHANNELS 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, structured: Optional[Path] = None, exception: Optional[Path] = None, errors: Sequence[ErrorItem] = (), ) -> Dict[str, Any]: return { "version": RESULT_VERSION, "status": status, "business_date": business_date.isoformat() if business_date else None, "message": message, "metrics": metrics, "outputs": { "daily_report": daily.name if daily else None, "structured_result": structured.name if structured else None, "exception_report": exception.name if exception else None, }, "errors": [error.to_dict() for error in errors], } 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 legacy_direct_metrics(metrics: Mapping[str, Any]) -> Dict[str, Any]: """Project a successful active run into the frozen v3 metric field set.""" return { "source_rows": metrics["source_rows"], "removed_by_rate_code": metrics["removed_by_rate_code"], "removed_as_duplicates": metrics["removed_as_duplicates"], "output_rows": metrics["output_rows"], "channels": metrics["channels"], } def legacy_direct_result_object( business_date: date, message: str, metrics: Mapping[str, Any], daily: Path, structured: Path, ) -> Dict[str, Any]: """Build only the immutable v3 direct-success result shape. The compatibility route has no review state and never accepts a manual override. It is intentionally unavailable to normal processor callers. """ return { "version": LEGACY_DIRECT_RESULT_VERSION, "status": "success", "business_date": business_date.isoformat(), "message": message, "metrics": legacy_direct_metrics(metrics), "outputs": { "daily_report": daily.name, "structured_result": structured.name, "exception_report": None, }, "errors": [], } def build_legacy_direct_structured_result( business_date: date, records: Sequence[Dict[str, Any]], channels: Sequence[Dict[str, Any]], xml_path: Path, result_json: Path, daily_path: Path, ) -> Dict[str, Any]: """Build the v3 direct-MCP projection after v4 has been independently checked.""" legacy_counts = { outcome: sum(1 for record in records if record.get("_OUTCOME") == outcome) for outcome in sorted(LEGACY_DIRECT_FINAL_OUTCOMES) } if ( any(record.get("_OUTCOME") == "candidate" for record in records) or any(record.get("_PRICING_METHOD") == "manual_review" for record in records) or legacy_counts["validation_failed"] or legacy_counts["price_unmatched"] ): raise ProcessingFailure( [ ErrorItem( "LEGACY_DIRECT_OUTPUT_INVALID", "legacy_direct", "旧direct_mcp兼容输出只能重放无人工价格的完整成功结果", ) ] ) return { "result_schema_version": LEGACY_DIRECT_RESULT_VERSION, "status": "success", "activation_eligible": True, "ingestion_mode": "opera_xml", "business_date": business_date.isoformat(), "processor_version": LEGACY_DIRECT_PROCESSOR_VERSION, "rule_set_sha256": LEGACY_DIRECT_RULE_SET_SHA256, "source_rows": len(records), "removed_by_rate_code": legacy_counts["excluded_rate_code"], "removed_as_duplicates": legacy_counts["duplicate"], "output_rows": legacy_counts["retained"], "outcome_counts": legacy_counts, "channels": list(channels), "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", ), "result_json": artifact_object(result_json, "result_json", "application/json"), "exception_report": None, }, "records": [structured_record(record) for record in records], "errors": [], } def legacy_direct_failed_result_object( business_date: Optional[date], message: str, metrics: Mapping[str, Any], structured: Path, exception: Path, errors: Sequence[ErrorItem], ) -> Dict[str, Any]: """Build the frozen v3 failed shape for old callback consumers.""" return { "version": LEGACY_DIRECT_RESULT_VERSION, "status": "failed", "business_date": business_date.isoformat() if business_date else None, "message": message, "metrics": legacy_direct_metrics(metrics), "outputs": { "daily_report": None, "structured_result": structured.name, "exception_report": exception.name, }, "errors": [error.to_dict() for error in errors], } def build_legacy_direct_failed_structured_result( business_date: Optional[date], records: Sequence[Dict[str, Any]], xml_path: Path, result_json: Optional[Path], exception_path: Path, errors: Sequence[ErrorItem], ) -> Dict[str, Any]: """Build the frozen v3 failed payload without exposing v4 review fields.""" legacy_counts = { outcome: sum(1 for record in records if record.get("_OUTCOME") == outcome) for outcome in sorted(LEGACY_DIRECT_FINAL_OUTCOMES) } if any(record.get("_OUTCOME") == "candidate" for record in records): raise ProcessingFailure( [ ErrorItem( "LEGACY_DIRECT_OUTPUT_INVALID", "legacy_direct", "旧direct_mcp失败输出不得包含候选复核行", ) ] ) return { "result_schema_version": LEGACY_DIRECT_RESULT_VERSION, "status": "failed", "activation_eligible": False, "ingestion_mode": "opera_xml", "business_date": business_date.isoformat() if business_date else None, "processor_version": LEGACY_DIRECT_PROCESSOR_VERSION, "rule_set_sha256": LEGACY_DIRECT_RULE_SET_SHA256, "source_rows": len(records), "removed_by_rate_code": legacy_counts["excluded_rate_code"], "removed_as_duplicates": legacy_counts["duplicate"], "output_rows": 0, "outcome_counts": legacy_counts, "channels": [], "artifacts": { "source_xml": artifact_object(xml_path, "opera_xml", "application/xml"), "daily_report": None, "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], } def write_legacy_direct_structured_result(path: Path, payload: Dict[str, Any]) -> None: """Write the frozen v3 payload without routing it through the v4 checker.""" path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, 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, "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 finalize_review_outcomes(records: Sequence[Dict[str, Any]]) -> None: for record in records: if record.get("_OUTCOME") == "pending": record["_OUTCOME"] = "candidate" 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, exception_path: Optional[Path] = None, manual_override_path: Optional[Path] = None, review_case_id: Optional[str] = None, manual_override_sha256: Optional[str] = None, review_issue_items: Sequence[Dict[str, Any]] = (), errors: Sequence[ErrorItem] = (), ) -> Dict[str, Any]: counts = structured_counts(records) manual_priced_rows = sum( 1 for record in records if record.get("_PRICING_METHOD") == "manual_review" ) return { "result_schema_version": 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, "candidate_rows": counts["candidate"], "review_required_rows": counts["price_unmatched"], "review_issue_count": len(review_issue_items), "review_issues": list(review_issue_items), "review_case_id": review_case_id, "manual_override_sha256": manual_override_sha256, "manually_priced_rows": manual_priced_rows, "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", ), "result_json": artifact_object(result_json, "result_json", "application/json"), "exception_report": artifact_object( exception_path, "exception_xlsx", "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ), "manual_override_json": artifact_object( manual_override_path, "manual_override_json", "application/json", ), }, "records": [structured_record(record) for record in records], "errors": [error.to_dict() for error in errors], } def validate_structured_completeness(payload: Dict[str, Any]) -> None: failures: List[str] = [] if payload.get("result_schema_version") != STRUCTURED_RESULT_SCHEMA_VERSION: 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计数不一致") if payload.get("candidate_rows") != counts.get("candidate"): failures.append("candidate_rows与outcome计数不一致") if payload.get("review_required_rows") != counts.get("price_unmatched"): failures.append("review_required_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") or counts.get("candidate"): 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的源行分解不平衡") if payload.get("review_required_rows") or payload.get("review_issue_count"): failures.append("成功payload不得包含待复核问题") if payload.get("review_issues") != []: failures.append("成功payload不得包含待复核问题明细") elif status == "review_required": if payload.get("activation_eligible"): failures.append("待复核payload不得允许激活") if payload.get("output_rows") != 0: failures.append("待复核payload不得包含正式输出行") if counts.get("validation_failed"): failures.append("待复核payload不得含其他校验失败行") if not counts.get("price_unmatched"): failures.append("待复核payload必须包含缺价行") if payload.get("candidate_rows", 0) + payload.get("review_required_rows", 0) + counts.get( "excluded_rate_code", 0 ) + counts.get("duplicate", 0) != source_rows: failures.append("待复核payload的源行分解不平衡") issues = payload.get("review_issues") if not isinstance(issues, list) or payload.get("review_issue_count") != len(issues): failures.append("待复核问题计数与明细不一致") if payload.get("review_issue_count", 0) <= 0: failures.append("待复核payload必须包含问题键") if payload.get("review_case_id") is not None or payload.get("manual_override_sha256") is not None: failures.append("初始待复核payload不得绑定人工价格清单") if payload.get("manually_priced_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不得包含正式输出行") if payload.get("candidate_rows"): failures.append("失败payload不得包含候选行") else: failures.append("status必须为success、review_required或failed") artifacts = payload.get("artifacts", {}) if not isinstance(artifacts, dict): failures.append("artifacts必须是对象") artifacts = {} expected_artifact_names = { "source_xml", "daily_report", "result_json", "exception_report", "manual_override_json", } if set(artifacts) != expected_artifact_names: failures.append("artifacts字段集不符合契约") if status == "success": required_success_artifacts = ["source_xml", "daily_report", "result_json"] for name in required_success_artifacts: if not artifacts.get(name): failures.append(f"成功payload缺少{name}产物哈希") if artifacts.get("exception_report") is not None: failures.append("成功payload不得包含异常清单") manual_artifact = artifacts.get("manual_override_json") if payload.get("review_case_id") is None: if manual_artifact is not None or payload.get("manual_override_sha256") is not None: failures.append("普通成功payload不得包含人工价格清单") if payload.get("manually_priced_rows") != 0: failures.append("普通成功payload不得包含人工定价行") else: if not REVIEW_CASE_ID_RE.fullmatch(str(payload.get("review_case_id"))): failures.append("人工成功payload复核标识非法") if not isinstance(payload.get("manual_override_sha256"), str) or not SHA256_RE.fullmatch( payload.get("manual_override_sha256") ): failures.append("人工成功payload缺少人工价格清单哈希") if manual_artifact is None or payload.get("manually_priced_rows", 0) <= 0: failures.append("人工成功payload缺少人工价格清单或人工定价行") if status == "review_required": if artifacts.get("daily_report") is not None or artifacts.get("exception_report") is not None: failures.append("待复核payload不得引用日报或异常清单") if artifacts.get("source_xml") is None or artifacts.get("result_json") is None: failures.append("待复核payload缺少源XML或结果产物哈希") if artifacts.get("manual_override_json") is not None: failures.append("待复核payload不得引用人工价格清单") if status == "failed": if artifacts.get("daily_report") is not None: failures.append("失败payload不得引用日报") if artifacts.get("exception_report") is None: failures.append("失败payload必须引用异常清单") if artifacts.get("manual_override_json") is not 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") in {"retained", "candidate"}: 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 record.get("pricing_method") == "manual_review" and "MANUAL_PRICE_APPLIED" not in decisions: failures.append("人工定价行必须包含MANUAL_PRICE_APPLIED决策码") if record.get("outcome") == "price_unmatched": if record.get("real_price") is not None or record.get("total_price") is not None: 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, result_json: Path, structured_result_json: Path, daily_path: Optional[Path] = None, manual_override_path: Optional[Path] = None, review_job_id: Optional[str] = None, review_case_id: Optional[str] = None, manual_override_sha256: Optional[str] = None, review_only: bool = False, ) -> None: validator = Path(__file__).resolve().parent / "validate_daily.py" command = [ sys.executable, str(validator), "--xml", str(xml_path), "--result-json", str(result_json), "--structured-result-json", str(structured_result_json), "--price-reference", str(PRICE_REFERENCE), ] if daily_path is not None: command.extend(["--daily", str(daily_path)]) if review_only: command.append("--review-only") if manual_override_path is not None: command.extend( [ "--manual-override-json", str(manual_override_path), "--review-job-id", str(review_job_id), "--review-case-id", str(review_case_id), "--manual-override-sha256", str(manual_override_sha256), ] ) 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) 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 all_records: List[Dict[str, Any]] = [] manual_override_arg = getattr(args, "manual_override_json", None) review_job_id = getattr(args, "review_job_id", None) review_case_id = getattr(args, "review_case_id", None) manual_override_sha256 = getattr(args, "manual_override_sha256", None) legacy_v3_output = getattr(args, "legacy_v3_output", False) manual_values = (manual_override_arg, review_job_id, review_case_id, manual_override_sha256) try: if not isinstance(legacy_v3_output, bool): raise ProcessingFailure( [ ErrorItem( "LEGACY_DIRECT_INVOCATION_INVALID", "legacy_direct", "旧direct_mcp兼容开关必须是布尔值", ) ], exit_code=3, ) if legacy_v3_output and any(value is not None for value in manual_values): raise ProcessingFailure( [ ErrorItem( "LEGACY_DIRECT_INVOCATION_INVALID", "legacy_direct", "旧direct_mcp兼容重放不允许人工价格清单", ) ], exit_code=3, ) if any(value is not None for value in manual_values) and not all( isinstance(value, str) and value for value in manual_values ): raise ProcessingFailure( [ ErrorItem( "MANUAL_OVERRIDE_INVOCATION_INVALID", "manual_review", "人工价格重放必须同时提供清单、任务、case和哈希", ) ], exit_code=3, ) validate_invocation( xml_path, output_dir, result_json, structured_result_json, ) 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) manifest: Optional[ManualOverrideManifest] = None manual_override_path = Path(manual_override_arg) if manual_override_arg else None if manual_override_path is not None: manifest = load_manual_override_manifest( manual_override_path, job_id=str(review_job_id), review_case_id=str(review_case_id), expected_sha256=str(manual_override_sha256), xml_path=xml_path, business_date=business_date, ) try: expected_keys = missing_price_keys(records, price_map) except CompanyKeywordAmbiguity: expected_keys = set() if expected_keys != set(manifest.prices): raise _manual_manifest_error("人工价格清单问题键集合与原始XML重放不一致", manual_override_path) pricing_errors = apply_prices_classified( records, price_map, manifest.prices if manifest is not None else None, ) if pricing_errors: if ( not legacy_v3_output and all(error.code == "PRICE_UNMATCHED" for error in pricing_errors) ): channel_counts = assign_channels(records) issues = review_issues(records, price_map) finalize_review_outcomes(all_records) metrics["channels"] = [] metrics["output_rows"] = 0 metrics["candidate_rows"] = sum( 1 for record in all_records if record.get("_OUTCOME") == "candidate" ) metrics["review_required_rows"] = sum( 1 for record in all_records if record.get("_OUTCOME") == "price_unmatched" ) metrics["review_issue_count"] = len(issues) review = result_object( "review_required", business_date, "固定价格表存在待人工确认的缺价键", metrics, structured=structured_result_json, errors=pricing_errors, ) write_result(result_json, review) structured_review = build_structured_result( "review_required", business_date, all_records, channel_metrics(channel_counts), xml_path, result_json, review_issue_items=issues, errors=pricing_errors, ) write_structured_result(structured_result_json, structured_review) run_independent_validator( xml_path, result_json, structured_result_json, review_only=True, ) print(json.dumps(review, ensure_ascii=False)) return 0 raise ProcessingFailure(pricing_errors) channel_counts = assign_channels(records) daily_path = output_dir / f"{business_date.month}.{business_date.day}.xlsx" write_daily(records, business_date, daily_path) metrics["channels"] = channel_metrics(channel_counts) metrics["output_rows"] = len(records) finalize_success_outcomes(all_records) success = result_object( "success", business_date, "日报与结构化结果处理成功", metrics, daily=daily_path, structured=structured_result_json, ) 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, manual_override_path=manual_override_path, review_case_id=manifest.review_case_id if manifest is not None else None, manual_override_sha256=manifest.sha256 if manifest is not None else None, ) write_structured_result(structured_result_json, structured_success) run_independent_validator( xml_path, result_json, structured_result_json, daily_path=daily_path, manual_override_path=manual_override_path, review_job_id=str(review_job_id) if manifest is not None else None, review_case_id=manifest.review_case_id if manifest is not None else None, manual_override_sha256=manifest.sha256 if manifest is not None else None, ) if legacy_v3_output: legacy_success = legacy_direct_result_object( business_date, "日报与结构化结果处理成功", metrics, daily_path, structured_result_json, ) write_result(result_json, legacy_success) legacy_structured = build_legacy_direct_structured_result( business_date, all_records, metrics["channels"], xml_path, result_json, daily_path, ) write_legacy_direct_structured_result(structured_result_json, legacy_structured) print(json.dumps(legacy_success, ensure_ascii=False)) return 0 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["candidate_rows"] = 0 metrics["review_required_rows"] = sum( 1 for record in all_records if record.get("_OUTCOME") == "price_unmatched" ) metrics["review_issue_count"] = 0 metrics["channels"] = [] remove_candidates((daily_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, structured=structured_result_json, exception=exception_path, errors=exc.errors, ) 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, ) write_structured_result(structured_result_json, structured_failed) if ( legacy_v3_output and result_json.is_absolute() and is_within(result_json, output_dir) and structured_result_json.is_absolute() and is_within(structured_result_json, output_dir) and structured_result_json.resolve() != result_json.resolve() ): legacy_failed = legacy_direct_failed_result_object( business_date, exc.errors[0].message if exc.errors else "处理失败", metrics, structured_result_json, exception_path, exc.errors, ) write_result(result_json, legacy_failed) write_legacy_direct_structured_result( structured_result_json, build_legacy_direct_failed_structured_result( business_date, all_records, xml_path, result_json, exception_path, exc.errors, ), ) failed = legacy_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["candidate_rows"] = 0 metrics["review_required_rows"] = sum( 1 for record in all_records if record.get("_OUTCOME") == "price_unmatched" ) metrics["review_issue_count"] = 0 metrics["channels"] = [] remove_candidates((daily_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, structured=structured_result_json, exception=exception_path, errors=[error], ) 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], ) write_structured_result(structured_result_json, structured_failed) if ( legacy_v3_output and result_json.is_absolute() and is_within(result_json, output_dir) and structured_result_json.is_absolute() and is_within(structured_result_json, output_dir) and structured_result_json.resolve() != result_json.resolve() ): legacy_failed = legacy_direct_failed_result_object( business_date, error.message, metrics, structured_result_json, exception_path, [error], ) write_result(result_json, legacy_failed) write_legacy_direct_structured_result( structured_result_json, build_legacy_direct_failed_structured_result( business_date, all_records, xml_path, result_json, exception_path, [error], ), ) failed = legacy_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("--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", ) parser.add_argument("--manual-override-json", help="Frozen canonical manual-price JSON manifest") parser.add_argument("--review-job-id", help="Bound job identifier for manual-price replay") parser.add_argument("--review-case-id", help="Bound review case identifier for manual-price replay") parser.add_argument("--manual-override-sha256", help="Expected SHA-256 of the frozen manifest") parser.add_argument("--legacy-v3-output", action="store_true", help=argparse.SUPPRESS) return parser def main() -> int: return process(build_parser().parse_args()) if __name__ == "__main__": raise SystemExit(main())