from __future__ import annotations

import argparse
import contextlib
import hashlib
import io
import json
import sys
import tempfile
import unittest
import zipfile
from datetime import date
from pathlib import Path

from openpyxl import load_workbook


PROJECT_ROOT = Path(__file__).resolve().parents[1]
SKILL_ROOT = PROJECT_ROOT / "arr-opera-daily-ingest"
SCRIPTS = SKILL_ROOT / "scripts"

sys.path.insert(0, str(SCRIPTS))
import process_daily as core  # noqa: E402
import validate_daily as validator  # noqa: E402


def reservation(
    sequence: int,
    *,
    company: str = "T / Q.B.D. TRAVEL GROUP",
    rate_code: str = "GRPA1",
    rate_amount: str = "900",
    room: str | None = None,
    confirmation: str | None = None,
    full_name: str = "SYNTHETIC GUEST",
    block_code: str = "",
    res_comment: str | None = None,
    rooms: str = "1",
    departure: str = "2026-07-28",
) -> str:
    room = room or f"SYNTHETIC-ROOM-{sequence}"
    confirmation = confirmation or f"SYNTHETIC-CONF-{sequence}"
    comment_xml = ""
    if res_comment is not None:
        comment_xml = (
            "<LIST_G_COMMENT_RESV_NAME_ID><G_COMMENT_RESV_NAME_ID>"
            f"<RES_COMMENT>{res_comment}</RES_COMMENT>"
            "</G_COMMENT_RESV_NAME_ID></LIST_G_COMMENT_RESV_NAME_ID>"
        )
    return f"""
        <G_RESERVATION>
          <ADULTS>2</ADULTS><BLOCK_CODE>{block_code}</BLOCK_CODE><CF_CHILDREN>0</CF_CHILDREN>
          <COMPANY_NAME>{company}</COMPANY_NAME>
          <CONFIRMATION_NO>{confirmation}</CONFIRMATION_NO><DISP_ROOM_NO>{room}</DISP_ROOM_NO>
          <EFFECTIVE_RATE_AMOUNT>{rate_amount}</EFFECTIVE_RATE_AMOUNT><FULL_NAME>{full_name}</FULL_NAME>
          {comment_xml}
          <NO_OF_ROOMS>{rooms}</NO_OF_ROOMS><PRODUCTS>SYNTHETIC</PRODUCTS><RATE_CODE>{rate_code}</RATE_CODE>
          <ROOM_CATEGORY_LABEL>SYNTHETIC ROOM TYPE</ROOM_CATEGORY_LABEL>
          <TRUNC_BEGIN>2026-07-27</TRUNC_BEGIN><TRUNC_END>{departure}</TRUNC_END>
        </G_RESERVATION>
    """


def xml_document(*rows: str) -> str:
    return f"""<?xml version="1.0" encoding="UTF-8"?>
<RES_DETAIL>
  <LIST_G_GROUP_BY1>
    <G_GROUP_BY1>
      <GROUPBY1_SORT_COL>20260727</GROUPBY1_SORT_COL>
      <GROUPBY1_COL>27-07-26</GROUPBY1_COL>
      <LIST_G_RESERVATION>{''.join(rows)}</LIST_G_RESERVATION>
    </G_GROUP_BY1>
  </LIST_G_GROUP_BY1>
</RES_DETAIL>
"""


def success_xml() -> str:
    return xml_document(
        reservation(1, rate_code="NOT-ALLOWED", rate_amount="INVALID"),
        reservation(
            2,
            rate_code="grpa1",
            res_comment="  ab-123  ",
            rooms="2",
            departure="2026-07-30",
        ),
        reservation(
            3,
            room="SYNTHETIC-ROOM-2",
            confirmation="SYNTHETIC-DUPLICATE",
            res_comment="AB-123",
        ),
        reservation(
            4,
            company="T- Rainbow Holiday Service",
            rate_code="LBMS",
            rate_amount="98765",
            block_code="MUST-NOT-BECOME-GROUP-CODE",
            departure="2026-07-27",
        ),
        reservation(
            5,
            company="T / Guangzhou Go-Easy",
            rate_code="LBSM",
            rate_amount="54321",
            res_comment="GO-EASY-GROUP",
        ),
    )


def run_processor(
    xml_text: str,
    root: Path,
    *,
    manual_override: dict[str, object] | None = None,
    review_job_id: str | None = None,
    review_case_id: str | None = None,
):
    root.mkdir(parents=True, exist_ok=True)
    xml_path = root / "synthetic.xml"
    output_dir = root / "output"
    result_path = output_dir / "result.json"
    structured_path = output_dir / "structured-result.json"
    xml_path.write_text(xml_text, encoding="utf-8")
    manual_override_path: Path | None = None
    if manual_override is not None:
        if review_job_id is None or review_case_id is None:
            raise ValueError("manual override requires job and review case identifiers")
        manual_override_path = root / "manual-override.json"
        manual_override_path.write_bytes(core.canonical_json_bytes(manual_override))
    args = argparse.Namespace(
        xml=str(xml_path.resolve()),
        output_dir=str(output_dir.resolve()),
        result_json=str(result_path.resolve()),
        structured_result_json=str(structured_path.resolve()),
        manual_override_json=(
            str(manual_override_path.resolve()) if manual_override_path is not None else None
        ),
        review_job_id=review_job_id,
        review_case_id=review_case_id,
        manual_override_sha256=(
            core.sha256_file(manual_override_path) if manual_override_path is not None else None
        ),
    )
    with contextlib.redirect_stdout(io.StringIO()):
        exit_code = core.process(args)
    result = json.loads(result_path.read_text(encoding="utf-8"))
    structured = json.loads(structured_path.read_text(encoding="utf-8"))
    return exit_code, xml_path, output_dir, result, structured


def manual_override_manifest(
    xml_path: Path,
    review_issues: list[dict[str, object]],
    *,
    job_id: str,
    review_case_id: str,
    real_price: str,
) -> dict[str, object]:
    business_date, _reservations = core.read_xml(xml_path)

    def issue_key(issue: dict[str, object]) -> dict[str, str]:
        return {
            "company_key": str(issue["company_key"]),
            "rate_code": str(issue["rate_code"]),
            "effective_rate_amount": core.canonical_decimal_text(
                core.parse_decimal(issue["effective_rate_amount"])
            ),
        }

    issues = sorted(
        (issue_key(issue) for issue in review_issues),
        key=lambda item: (
            item["company_key"],
            item["rate_code"],
            item["effective_rate_amount"],
        ),
    )
    return {
        "review_version": core.REVIEW_VERSION,
        "review_case_id": review_case_id,
        "job_id": job_id,
        "source_sha256": core.sha256_file(xml_path),
        "business_date": business_date.isoformat(),
        "processor_version": core.PROCESSOR_VERSION,
        "rule_set_sha256": core.rule_set_sha256(),
        "issues": issues,
        "overrides": [{**issue, "real_price": real_price} for issue in issues],
    }


class ArrOperaDailyIngestTests(unittest.TestCase):
    def test_package_is_daily_only_and_self_contained(self):
        expected = {
            "SKILL.md",
            "agents/openai.yaml",
            "scripts/process_daily.py",
            "scripts/validate_daily.py",
            "references/business-rules.md",
            "references/codex-result.schema.json",
            "references/error-contract.md",
            "references/field-contracts.md",
            "references/manual-override.schema.json",
            "references/structured-output.md",
            "references/structured-result.schema.json",
            "references/价格对照.xlsx",
            "assets/daily-template.xlsx",
        }
        actual = {
            str(path.relative_to(SKILL_ROOT))
            for path in SKILL_ROOT.rglob("*")
            if path.is_file()
            and path.name != ".DS_Store"
            and "__pycache__" not in path.parts
        }
        self.assertEqual(actual, expected)
        scripts = "\n".join(
            path.read_text(encoding="utf-8")
            for path in (SCRIPTS / "process_daily.py", SCRIPTS / "validate_daily.py")
        )
        for forbidden in (
            "monthly_report",
            "monthly_base",
            "daily-monthly",
            "write_monthly",
            "read_existing_monthly",
            "MONTHLY_",
        ):
            self.assertNotIn(forbidden, scripts)
        self.assertEqual(core.RESULT_VERSION, "4.0")
        self.assertEqual(core.STRUCTURED_RESULT_SCHEMA_VERSION, "4.0")
        self.assertEqual(core.PROCESSOR_VERSION, "4.0.0")
        self.assertEqual(len(core.DAILY_HEADERS), 19)
        self.assertEqual(len(core.RATE_WHITELIST), 20)

    def test_archive_matches_daily_only_skill_source(self):
        source_files = {
            str(path.relative_to(PROJECT_ROOT)): path
            for path in SKILL_ROOT.rglob("*")
            if path.is_file()
            and path.name != ".DS_Store"
            and "__pycache__" not in path.parts
        }
        for suffix in (".zip", ".skill"):
            archive_path = PROJECT_ROOT / f"arr-opera-daily-ingest{suffix}"
            with self.subTest(archive=archive_path.name):
                self.assertTrue(archive_path.is_file())
                with zipfile.ZipFile(archive_path) as archive:
                    archived_files = {
                        name: archive.read(name)
                        for name in archive.namelist()
                        if not name.endswith("/")
                    }
                    self.assertIsNone(archive.testzip())
                self.assertEqual(set(archived_files), set(source_files))
                for name, path in source_files.items():
                    self.assertEqual(
                        hashlib.sha256(archived_files[name]).hexdigest(),
                        hashlib.sha256(path.read_bytes()).hexdigest(),
                    )

    def test_success_preserves_all_source_outcomes_and_path_free_artifacts(self):
        with tempfile.TemporaryDirectory() as temp_dir:
            exit_code, xml_path, output_dir, result, payload = run_processor(
                success_xml(), Path(temp_dir)
            )
            self.assertEqual(exit_code, 0)
            self.assertEqual(
                set(result),
                {
                    "version",
                    "status",
                    "business_date",
                    "message",
                    "metrics",
                    "outputs",
                    "errors",
                },
            )
            self.assertEqual(
                set(result["outputs"]),
                {"daily_report", "structured_result", "exception_report"},
            )
            self.assertNotIn("monthly_report", json.dumps(result))
            self.assertEqual(result["metrics"]["source_rows"], 5)
            self.assertEqual(result["metrics"]["removed_by_rate_code"], 1)
            self.assertEqual(result["metrics"]["removed_as_duplicates"], 1)
            self.assertEqual(result["metrics"]["output_rows"], 3)
            self.assertEqual(
                result["metrics"]["channels"],
                [
                    {"worksheet": "QBD", "rows": 1},
                    {"worksheet": core.KB_SHEET, "rows": 2},
                ],
            )

            self.assertEqual(payload["result_schema_version"], "4.0")
            self.assertEqual(payload["processor_version"], "4.0.0")
            self.assertEqual(payload["source_rows"], 5)
            self.assertEqual(
                payload["outcome_counts"],
                {
                    "candidate": 0,
                    "duplicate": 1,
                    "excluded_rate_code": 1,
                    "price_unmatched": 0,
                    "retained": 3,
                    "validation_failed": 0,
                },
            )
            records = payload["records"]
            self.assertEqual(
                [record["outcome"] for record in records],
                [
                    "excluded_rate_code",
                    "retained",
                    "duplicate",
                    "retained",
                    "retained",
                ],
            )
            self.assertEqual(records[2]["duplicate_of_source_sequence"], 2)
            self.assertEqual(records[1]["group_code_key"], "AB-123")
            self.assertEqual(records[1]["real_price"], 1800)
            self.assertEqual(records[1]["total_price"], 10800)
            self.assertEqual(records[3]["nights"], 0)
            self.assertEqual(records[3]["real_price"], 0)
            self.assertEqual(records[3]["total_price"], 0)
            self.assertEqual(records[3]["kb_amount"], 100)
            self.assertIsNone(records[3]["group_code_key"])
            self.assertEqual(
                records[3]["booking_source_match_status"], "missing_group_code"
            )
            self.assertNotEqual(records[3]["block_code"], records[3]["group_code_key"])
            self.assertEqual(records[4]["company_key"], "GUANGZHOU GO EASY")
            self.assertEqual(records[4]["real_price"], 0)
            self.assertEqual(records[4]["pricing_method"], "zero_price_exception")
            self.assertTrue(all(row["source_worksheet"] is None for row in records))
            self.assertTrue(all(row["source_row_no"] is None for row in records))

            self.assertEqual(
                set(payload["artifacts"]),
                {
                    "source_xml",
                    "daily_report",
                    "result_json",
                    "exception_report",
                    "manual_override_json",
                },
            )
            for name, expected_path in (
                ("source_xml", xml_path),
                ("daily_report", output_dir / result["outputs"]["daily_report"]),
                ("result_json", output_dir / "result.json"),
            ):
                with self.subTest(artifact=name):
                    artifact = payload["artifacts"][name]
                    self.assertNotIn("path", artifact)
                    self.assertEqual(artifact["sha256"], core.sha256_file(expected_path))
                    self.assertEqual(artifact["byte_size"], expected_path.stat().st_size)
            self.assertIsNone(payload["artifacts"]["exception_report"])
            self.assertIsNone(payload["artifacts"]["manual_override_json"])
            self.assertFalse(
                any(
                    path.suffix.lower() == ".xlsx"
                    and path.name != result["outputs"]["daily_report"]
                    for path in output_dir.iterdir()
                )
            )
            core.validate_structured_completeness(payload)
            errors = validator.validate(
                argparse.Namespace(
                    xml=str(xml_path.resolve()),
                    daily=str(
                        (output_dir / result["outputs"]["daily_report"]).resolve()
                    ),
                    result_json=str((output_dir / "result.json").resolve()),
                    structured_result_json=str(
                        (output_dir / "structured-result.json").resolve()
                    ),
                    price_reference=str(core.PRICE_REFERENCE.resolve()),
                )
            )
            self.assertEqual(errors, [])

    def test_zero_night_is_valid_and_negative_night_is_not(self):
        zero_night = xml_document(
            reservation(1, departure="2026-07-27", res_comment="ZERO-NIGHT")
        )
        with tempfile.TemporaryDirectory() as temp_dir:
            exit_code, _xml, output_dir, result, payload = run_processor(
                zero_night, Path(temp_dir) / "zero"
            )
            self.assertEqual(exit_code, 0)
            self.assertEqual(payload["records"][0]["nights"], 0)
            self.assertEqual(payload["records"][0]["total_price"], 0)
            workbook = load_workbook(
                output_dir / result["outputs"]["daily_report"], data_only=False
            )
            try:
                nights_col = core.DAILY_HEADERS.index("NIGHTS") + 1
                total_col = core.DAILY_HEADERS.index("TOTAL PRICE") + 1
                self.assertEqual(workbook.active.cell(2, nights_col).value, 0)
                self.assertEqual(workbook.active.cell(2, total_col).value, 0)
            finally:
                workbook.close()

        negative_night = xml_document(
            reservation(1, departure="2026-07-26", res_comment="NEGATIVE")
        )
        with tempfile.TemporaryDirectory() as temp_dir:
            exit_code, _xml, output_dir, result, payload = run_processor(
                negative_night, Path(temp_dir) / "negative"
            )
            self.assertEqual(exit_code, 2)
            self.assertEqual(result["status"], "failed")
            self.assertIsNone(result["outputs"]["daily_report"])
            self.assertEqual(payload["records"][0]["nights"], -1)
            self.assertEqual(payload["records"][0]["outcome"], "validation_failed")
            self.assertIn(
                "XML_NEGATIVE_NIGHTS", payload["records"][0]["decision_codes"]
            )
            self.assertFalse(
                any(
                    path.suffix.lower() == ".xlsx"
                    and path.name != "异常清单.xlsx"
                    for path in output_dir.iterdir()
                )
            )

    def test_price_unmatched_enters_review_and_aggregates_safe_keys(self):
        price_review = xml_document(
            reservation(1, rate_code="GRPA1", rate_amount="900"),
            reservation(2, rate_code="GRPA1", rate_amount="1800", rooms="2"),
            reservation(3, rate_code="GRPA1", rate_amount="1800", rooms="1"),
            reservation(4, rate_code="NOT-ALLOWED", rate_amount="INVALID"),
        )
        with tempfile.TemporaryDirectory() as temp_dir:
            exit_code, _xml, output_dir, result, payload = run_processor(
                price_review, Path(temp_dir)
            )
            self.assertEqual(exit_code, 0)
            self.assertEqual(result["status"], "review_required")
            self.assertIsNone(result["outputs"]["daily_report"])
            self.assertEqual(
                [row["outcome"] for row in payload["records"]],
                ["candidate", "price_unmatched", "price_unmatched", "excluded_rate_code"],
            )
            self.assertFalse(payload["activation_eligible"])
            self.assertEqual(payload["output_rows"], 0)
            self.assertEqual(payload["channels"], [])
            self.assertEqual(payload["candidate_rows"], 1)
            self.assertEqual(payload["review_required_rows"], 2)
            self.assertEqual(payload["review_issue_count"], 1)
            self.assertIsNone(payload["artifacts"]["daily_report"])
            self.assertIsNone(payload["artifacts"]["exception_report"])
            self.assertIsNone(payload["artifacts"]["manual_override_json"])
            self.assertEqual(
                payload["review_issues"],
                [
                    {
                        "company_key": "QBD",
                        "rate_code": "GRPA1",
                        "effective_rate_amount": 1800,
                        "candidate_prices": [
                            {"effective_rate_amount": 900, "real_price": 1800},
                            {"effective_rate_amount": 1150, "real_price": 2300},
                            {"effective_rate_amount": 1300, "real_price": 2600},
                            {"effective_rate_amount": 1400, "real_price": 2600},
                            {"effective_rate_amount": 1725, "real_price": 3450},
                            {"effective_rate_amount": 2300, "real_price": 4600},
                        ],
                        "affected_records": 2,
                        "affected_rooms": 3,
                        "affected_room_nights": 3,
                    }
                ],
            )
            self.assertEqual(
                {path.name for path in output_dir.iterdir()},
                {"result.json", "structured-result.json"},
            )
            errors = validator.validate(
                argparse.Namespace(
                    xml=str(_xml.resolve()),
                    result_json=str((output_dir / "result.json").resolve()),
                    structured_result_json=str(
                        (output_dir / "structured-result.json").resolve()
                    ),
                    price_reference=str(core.PRICE_REFERENCE.resolve()),
                    review_only=True,
                )
            )
            self.assertEqual(errors, [])

    def test_final_manual_zero_price_replays_original_xml_and_manifest(self):
        source = xml_document(
            reservation(1, rate_code="GRPA1", rate_amount="900"),
            reservation(2, rate_code="GRPA1", rate_amount="1800", rooms="2"),
            reservation(3, rate_code="GRPA1", rate_amount="1800", rooms="1"),
        )
        job_id = "job-synthetic-review-001"
        review_case_id = "dailyreview-0123456789abcdef0123456789abcdef"
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            review = run_processor(source, root / "review")
            self.assertEqual(review[3]["status"], "review_required")
            manifest = manual_override_manifest(
                review[1],
                review[4]["review_issues"],
                job_id=job_id,
                review_case_id=review_case_id,
                real_price="0.00",
            )
            exit_code, xml_path, output_dir, result, payload = run_processor(
                source,
                root / "final",
                manual_override=manifest,
                review_job_id=job_id,
                review_case_id=review_case_id,
            )
            self.assertEqual(exit_code, 0)
            self.assertEqual(result["status"], "success")
            self.assertEqual(payload["review_case_id"], review_case_id)
            manifest_path = root / "final" / "manual-override.json"
            self.assertEqual(payload["manual_override_sha256"], core.sha256_file(manifest_path))
            self.assertEqual(payload["manually_priced_rows"], 2)
            self.assertEqual([row["outcome"] for row in payload["records"]], ["retained"] * 3)
            self.assertEqual(
                [row["pricing_method"] for row in payload["records"]],
                ["price_reference_exact", "manual_review", "manual_review"],
            )
            self.assertTrue(
                all(
                    "MANUAL_PRICE_APPLIED" in row["decision_codes"]
                    for row in payload["records"][1:]
                )
            )
            self.assertEqual([row["real_price"] for row in payload["records"][1:]], [0, 0])
            self.assertEqual([row["total_price"] for row in payload["records"][1:]], [0, 0])
            self.assertIsNotNone(payload["artifacts"]["manual_override_json"])
            self.assertEqual(
                payload["artifacts"]["manual_override_json"]["sha256"],
                core.sha256_file(manifest_path),
            )
            errors = validator.validate(
                argparse.Namespace(
                    xml=str(xml_path.resolve()),
                    daily=str((output_dir / result["outputs"]["daily_report"]).resolve()),
                    result_json=str((output_dir / "result.json").resolve()),
                    structured_result_json=str(
                        (output_dir / "structured-result.json").resolve()
                    ),
                    price_reference=str(core.PRICE_REFERENCE.resolve()),
                    manual_override_json=str(manifest_path.resolve()),
                    review_job_id=job_id,
                    review_case_id=review_case_id,
                    manual_override_sha256=core.sha256_file(manifest_path),
                )
            )
            self.assertEqual(errors, [])

    def test_price_unmatched_with_other_validation_error_fails(self):
        mixed = xml_document(
            reservation(1, rate_code="GRPA1", rate_amount="1800"),
            reservation(2, departure="2026-07-26"),
        )
        with tempfile.TemporaryDirectory() as temp_dir:
            exit_code, _xml, output_dir, result, payload = run_processor(mixed, Path(temp_dir))
            self.assertEqual(exit_code, 2)
            self.assertEqual(result["status"], "failed")
            self.assertIsNone(result["outputs"]["daily_report"])
            self.assertIsNotNone(result["outputs"]["exception_report"])
            self.assertEqual(payload["status"], "failed")
            self.assertEqual(payload["review_issues"], [])
            self.assertEqual(payload["review_issue_count"], 0)
            self.assertIsNotNone(payload["artifacts"]["exception_report"])
            self.assertIn("异常清单.xlsx", {path.name for path in output_dir.iterdir()})

    def test_independent_validator_rejects_channel_tampering(self):
        with tempfile.TemporaryDirectory() as temp_dir:
            exit_code, xml_path, output_dir, result, payload = run_processor(
                success_xml(), Path(temp_dir)
            )
            self.assertEqual(exit_code, 0)
            structured_path = output_dir / "structured-result.json"
            payload["records"][1]["channel_key"] = "FENGRUN"
            structured_path.write_text(
                json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
                encoding="utf-8",
            )
            errors = validator.validate(
                argparse.Namespace(
                    xml=str(xml_path.resolve()),
                    daily=str(
                        (output_dir / result["outputs"]["daily_report"]).resolve()
                    ),
                    result_json=str((output_dir / "result.json").resolve()),
                    structured_result_json=str(structured_path.resolve()),
                    price_reference=str(core.PRICE_REFERENCE.resolve()),
                )
            )
            self.assertIn(
                "OUTPUT_STRUCTURED_RECORD_MISMATCH",
                {error.code for error in errors},
            )

    def test_contract_schemas_are_strict_and_have_no_monthly_or_local_path(self):
        result_schema = json.loads(
            (SKILL_ROOT / "references" / "codex-result.schema.json").read_text(
                encoding="utf-8"
            )
        )
        structured_schema = json.loads(
            (
                SKILL_ROOT / "references" / "structured-result.schema.json"
            ).read_text(encoding="utf-8")
        )
        self.assertFalse(result_schema["additionalProperties"])
        self.assertEqual(result_schema["properties"]["version"]["const"], "4.0")
        self.assertEqual(
            set(result_schema["properties"]["outputs"]["properties"]),
            {"daily_report", "structured_result", "exception_report"},
        )
        self.assertFalse(structured_schema["additionalProperties"])
        self.assertEqual(
            structured_schema["properties"]["result_schema_version"]["const"], "4.0"
        )
        artifact_properties = structured_schema["$defs"]["artifact"]["properties"]
        self.assertNotIn("path", artifact_properties)
        self.assertNotIn(
            "monthly_xlsx",
            structured_schema["$defs"]["artifact"]["properties"]["file_kind"]["enum"],
        )
        self.assertNotIn(
            "monthly_report",
            structured_schema["properties"]["artifacts"]["properties"],
        )
        retained = structured_schema["$defs"]["record"]["allOf"][1]["then"][
            "properties"
        ]
        self.assertEqual(retained["nights"]["minimum"], 0)

    def test_cli_has_no_legacy_monthly_surface(self):
        parser = core.build_parser()
        with self.assertRaises(SystemExit) as caught, contextlib.redirect_stderr(
            io.StringIO()
        ):
            parser.parse_args(
                [
                    "--xml",
                    "/tmp/input.xml",
                    "--output-dir",
                    "/tmp/output",
                    "--result-json",
                    "/tmp/output/result.json",
                    "--monthly-base",
                    "/tmp/monthly.xlsx",
                ]
            )
        self.assertEqual(caught.exception.code, 2)

    def test_main_prompt_keeps_oss_and_direct_mcp_outside_skill(self):
        prompt_path = (
            PROJECT_ROOT / "prompts" / "arr_opera_daily_main_agent_prompt.md"
        )
        agent_schema_path = (
            PROJECT_ROOT
            / "prompts"
            / "arr_opera_daily_agent_result.schema.json"
        )
        profile_schema_path = (
            PROJECT_ROOT
            / "prompts"
            / "arr_opera_daily_profile_output.schema.json"
        )
        prompt = prompt_path.read_text(encoding="utf-8")
        agent_schema = json.loads(agent_schema_path.read_text(encoding="utf-8"))
        profile_schema = json.loads(profile_schema_path.read_text(encoding="utf-8"))
        self.assertIn("fetch_oss_file", prompt)
        self.assertIn("arr-opera-daily-ingest", prompt)
        self.assertIn("arr_submit_processing_result", prompt)
        self.assertIn("PROGRAM_INPUT.result_submission.submission_grant", prompt)
        self.assertIn("不上传日报", prompt)
        self.assertIn("恰好调用一次", prompt)
        self.assertIn("禁止 `present_files`", prompt)
        self.assertIn("不要读取本机", prompt)
        self.assertIn("不在 Agent 内重复校验", prompt)
        self.assertIn("第二次取件、第二次处理或第二次提交", prompt)
        self.assertIn("唯一允许的工具动作是 `fetch_oss_file`", prompt)
        self.assertIn(
            "object_uri = PROGRAM_INPUT.oss_attachments[0].oss.url",
            prompt,
        )
        self.assertIn("不得拼接或改写 URL/key/文件名", prompt)
        self.assertIn("不得换成 `oss://`", prompt)
        self.assertNotIn('object_uri = "oss://"', prompt)
        self.assertIn("不得再次调用、补参数", prompt)
        self.assertNotIn("ARR_PROGRAM_INPUT_REQUIRED", prompt)
        self.assertNotIn("attachment_fetch_policy", prompt)
        self.assertNotIn("required_before_skill_call", prompt)
        self.assertNotIn("required_after_successful_skill_call", prompt)
        self.assertNotIn("SHA-256", prompt)
        self.assertLess(len(prompt), 1800)
        self.assertNotIn(
            "prompts/arr_opera_daily_program_input.schema.json", prompt
        )
        self.assertNotIn(
            "prompts/arr_opera_daily_agent_result.schema.json", prompt
        )
        self.assertNotIn("POST /api/integrations/super-agent/results", prompt)
        self.assertNotIn("OSS_ACCESS_KEY_ID", prompt)
        self.assertNotIn("OSS_ACCESS_KEY_SECRET", prompt)
        self.assertNotIn("LTAI", prompt)
        self.assertFalse(prompt_path.is_relative_to(SKILL_ROOT))
        self.assertEqual(
            agent_schema["properties"]["contract_version"]["const"],
            "arr-opera-daily-agent-1",
        )
        self.assertEqual(
            agent_schema["$defs"]["processorResult"]["properties"]["version"][
                "const"
            ],
            "3.0",
        )
        self.assertIn("result_json", agent_schema["properties"]["files"]["required"])
        self.assertEqual(
            profile_schema["properties"]["processor_result"]["type"], "object"
        )
        self.assertEqual(profile_schema["properties"]["files"]["type"], "object")
        self.assertEqual(
            set(profile_schema["required"]), set(agent_schema["required"])
        )

    def test_success_artifacts_map_to_current_arr_delivery_envelope(self):
        from arr_ingestion.contracts import DeliveryEnvelope

        with tempfile.TemporaryDirectory() as temp_dir:
            exit_code, xml_path, output_dir, result, payload = run_processor(
                success_xml(), Path(temp_dir)
            )
            self.assertEqual(exit_code, 0)
            daily_path = output_dir / result["outputs"]["daily_report"]
            result_path = output_dir / "result.json"
            structured_path = output_dir / "structured-result.json"

            def reference(
                object_key: str, path: Path, mime_type: str
            ) -> dict[str, object]:
                return {
                    "object_key": object_key,
                    "original_filename": path.name,
                    "sha256": core.sha256_file(path),
                    "byte_size": path.stat().st_size,
                    "mime_type": mime_type,
                }

            delivery = DeliveryEnvelope.from_dict(
                {
                    "delivery_schema_version": "1.0",
                    "delivery_id": "delivery-synthetic-001",
                    "job_id": "job-synthetic-001",
                    "attempt_no": 1,
                    "status": "success",
                    "processor_version": payload["processor_version"],
                    "rule_set_sha256": payload["rule_set_sha256"],
                    "result_schema_version": payload["result_schema_version"],
                    "business_date": payload["business_date"],
                    "artifacts": {
                        "source_xml": reference(
                            "arr/jobs/job-synthetic-001/input/source.xml",
                            xml_path,
                            "application/xml",
                        ),
                        "daily_report": reference(
                            "arr/jobs/job-synthetic-001/output/7.27.xlsx",
                            daily_path,
                            "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
                        ),
                        "result_json": reference(
                            "arr/jobs/job-synthetic-001/output/result.json",
                            result_path,
                            "application/json",
                        ),
                        "structured_result_json": reference(
                            "arr/jobs/job-synthetic-001/output/structured-result.json",
                            structured_path,
                            "application/json",
                        ),
                        "exception_report": None,
                        "manual_override_json": None,
                    },
                }
            )
            self.assertEqual(delivery.processor_version, "4.0.0")
            self.assertEqual(delivery.result_schema_version, "4.0")
            self.assertEqual(delivery.business_date, date(2026, 7, 27))

    def test_same_input_rerun_has_identical_business_records(self):
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            first = run_processor(success_xml(), root / "first")
            second = run_processor(success_xml(), root / "second")
            self.assertEqual(first[0], 0)
            self.assertEqual(second[0], 0)
            self.assertEqual(first[4]["records"], second[4]["records"])
            self.assertEqual(first[4]["channels"], second[4]["channels"])
            self.assertEqual(
                first[4]["rule_set_sha256"], second[4]["rule_set_sha256"]
            )

    def test_company_keyword_and_zero_price_conditions_are_exact(self):
        self.assertEqual(
            core.price_company_key("T- Rainbow Holiday Se"), "RAINBOW/AI"
        )
        self.assertEqual(
            core.price_company_key("T- Guangzhou Go Easy"),
            "GUANGZHOU GO EASY",
        )
        self.assertEqual(core.price_company_key("T / Q.B.D."), "QBD")
        for company, rate_code in (
            ("T- Guangzhou Go Easy", "GRPA1"),
            ("T- Rainbow Holiday Se", "LBW1"),
            ("T- HANATOUR TD CO., L", "LBMS"),
        ):
            record = {
                "_SOURCE_INDEX": 1,
                "_SOURCE_LOCATION": "reservation[1]",
                "_DECISION_CODES": [],
                "_OUTCOME": "pending",
                "_COMPANY_KEY": None,
                "_PRICING_METHOD": None,
                "COMPANY_NAME": company,
                "RATE_CODE": rate_code,
                "EFFECTIVE_RATE_AMOUNT": core.parse_decimal("99999"),
                "CONFIRMATION_NO": "SYNTHETIC",
                "NO_OF_ROOMS": 1,
                "NIGHTS": 1,
                "REAL PRICE": None,
                "TOTAL PRICE": None,
            }
            with self.subTest(company=company, rate_code=rate_code):
                errors = core.apply_prices_classified([record], {})
                self.assertEqual(errors[0].code, "PRICE_UNMATCHED")


if __name__ == "__main__":
    unittest.main(verbosity=2)
