from __future__ import annotations

import hashlib
import json
import sys
import unittest
import zipfile
from decimal import Decimal
from pathlib import Path

from openpyxl import load_workbook


PROJECT_ROOT = Path(__file__).resolve().parents[1]
SKILL_ROOT = PROJECT_ROOT / "opera-daily-channel-report"
SCRIPTS = SKILL_ROOT / "scripts"
PRICE_REFERENCE = SKILL_ROOT / "references" / "价格对照.xlsx"
ARCHIVE = PROJECT_ROOT / "opera-daily-channel-report.zip"

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


def sha256_bytes(value: bytes) -> str:
    return hashlib.sha256(value).hexdigest()


class PackageContractTests(unittest.TestCase):
    def test_required_skill_layout(self):
        required = {
            "SKILL.md",
            "scripts/process_reports.py",
            "scripts/validate_outputs.py",
            "references/business-rules.md",
            "references/field-contracts.md",
            "references/error-contract.md",
            "references/codex-result.schema.json",
            "references/structured-output.md",
            "references/structured-result.schema.json",
            "references/价格对照.xlsx",
            "assets/daily-template.xlsx",
            "assets/channel-report-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, required)

    def test_frozen_constants_and_company_identities(self):
        self.assertEqual(core.VERSION, "1.0")
        self.assertEqual(core.DAILY_RESULT_VERSION, "2.0")
        self.assertEqual(core.PROCESSOR_VERSION, "2.2.0")
        self.assertEqual(core.STRUCTURED_RESULT_SCHEMA_VERSION, "1.0")
        self.assertEqual(core.DAILY_STRUCTURED_RESULT_SCHEMA_VERSION, "2.0")
        self.assertEqual(core.PROCESSING_MODES, {"daily", "daily-monthly"})
        self.assertEqual(
            core.STANDARD_SHEETS,
            ["LIANTAI-GROUP", "LIANTAI-FIT", "QBD", "DY-AI-Easy-KB", "FENGRUN"],
        )
        self.assertEqual(len(core.DAILY_HEADERS), 19)
        self.assertEqual(core.DAILY_HEADERS[-2:], ["REAL PRICE", "TOTAL PRICE"])
        self.assertEqual(len(core.CHANNEL_HEADERS), 19)
        self.assertEqual(core.CHANNEL_HEADERS[-2:], ["REAL PRICE", "TOTAL PRICE"])
        self.assertEqual(len(core.KB_CHANNEL_HEADERS), 20)
        self.assertEqual(core.KB_CHANNEL_HEADERS[-1], "KB（100/晚/间）")
        self.assertEqual(len(core.RATE_WHITELIST), 20)
        self.assertEqual(core.ZERO_TOTAL_COMPANIES, {"RAINBOW/AI", "GUANGZHOU GO EASY"})
        self.assertEqual(core.ZERO_TOTAL_RATE_CODES, {"LBMS", "LBSM"})
        self.assertEqual(core.price_company_key("T- Rainbow Holiday Se"), "RAINBOW/AI")
        self.assertEqual(core.price_company_key("T- FENGRUN TRAVEL TRA"), "FENGRUN")
        self.assertEqual(core.price_company_key("T- HANATOUR TD CO., L"), "HANA TOUR")
        self.assertEqual(core.price_company_key("T- HONGTAI TRAVEL (TH"), "HONGTAI")
        self.assertEqual(core.price_company_key("T- Guangzhou Go Easy"), "GUANGZHOU GO EASY")

    def test_price_reference_is_unique_and_contains_alias_targets(self):
        price_map = core.load_price_map(PRICE_REFERENCE.resolve())
        self.assertEqual(len(price_map), 73)
        cases = {
            ("HANA TOUR", "WHKR2100B", Decimal("2100")): Decimal("2100"),
            ("HANA TOUR", "WHKR2100B", Decimal("3300")): Decimal("3300"),
            ("FENGRUN", "GLSPCB", Decimal("1800")): Decimal("1800"),
            ("HONGTAI", "LBKB", Decimal("900")): Decimal("1800"),
            ("RAINBOW/AI", "LBSM", Decimal("900")): Decimal("0"),
        }
        for key, expected in cases.items():
            with self.subTest(key=key):
                self.assertEqual(price_map[key], expected)

    def test_templates_have_exact_headers_and_no_data_or_formulas(self):
        daily_path = SKILL_ROOT / "assets" / "daily-template.xlsx"
        daily = load_workbook(daily_path, data_only=False)
        try:
            for sheet in daily.worksheets:
                column_count = len(core.DAILY_HEADERS)
                headers = [
                    sheet.cell(1, column).value for column in range(1, column_count + 1)
                ]
                self.assertEqual(headers, core.DAILY_HEADERS)
                self.assertIsNone(sheet.cell(1, column_count + 1).value)
                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:
                        self.assertIn(cell.value, (None, ""))
                        self.assertNotEqual(cell.data_type, "f")
        finally:
            daily.close()

        monthly_path = SKILL_ROOT / "assets" / "channel-report-template.xlsx"
        monthly = load_workbook(monthly_path, data_only=False)
        try:
            self.assertEqual(monthly.sheetnames, core.STANDARD_SHEETS)
            for sheet in monthly.worksheets:
                expected_headers = core.channel_headers_for_sheet(sheet.title)
                column_count = len(expected_headers)
                headers = [
                    sheet.cell(1, column).value for column in range(1, column_count + 1)
                ]
                self.assertEqual(headers, expected_headers)
                self.assertIsNone(sheet.cell(1, column_count + 1).value)
                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:
                        self.assertIn(cell.value, (None, ""))
                        self.assertNotEqual(cell.data_type, "f")
        finally:
            monthly.close()

    def test_result_schema_strictly_supports_legacy_and_daily_only_versions(self):
        schema_path = SKILL_ROOT / "references" / "codex-result.schema.json"
        schema = json.loads(schema_path.read_text(encoding="utf-8"))
        self.assertFalse(schema["additionalProperties"])
        self.assertEqual(
            schema["properties"]["version"],
            {"type": "string", "enum": ["1.0", "2.0"]},
        )
        self.assertEqual(schema["properties"]["processing_mode"], {"type": "string", "const": "daily"})
        self.assertEqual(schema["properties"]["status"]["enum"], ["success", "failed"])
        self.assertEqual(
            schema["required"],
            [
                "version",
                "status",
                "business_date",
                "month_key",
                "message",
                "metrics",
                "outputs",
                "errors",
            ],
        )

    def test_zip_payload_matches_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
        }
        with zipfile.ZipFile(ARCHIVE) as archive:
            archived_files = {
                name: archive.read(name)
                for name in archive.namelist()
                if not name.endswith("/")
            }
        self.assertEqual(set(archived_files), set(source_files))
        for name, path in source_files.items():
            with self.subTest(path=name):
                self.assertEqual(sha256_bytes(archived_files[name]), sha256_bytes(path.read_bytes()))


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