from __future__ import annotations

import base64
import hashlib
import json
import tempfile
import threading
import time
import unittest
from datetime import date, datetime
from pathlib import Path
from typing import Any, Dict, Mapping, Optional
from zoneinfo import ZoneInfo

from arr_web.app import PortalApplication, RuntimeHealth, SessionLedger
from arr_web.auth import LoginCredentials
from arr_web.company_jobs import (
    COMPANY_NAMES,
    PersistentCompanyReportCoordinator,
    period_complete_at,
    period_release_at,
    period_to_as_of,
)
from arr_web.contracts import PortalError
from arr_web.downloads import ArtifactDescriptor, ControlledProjectArtifactReader
from company_reports.contracts import ENGLISH_MONTH_NAMES, RESULT_SCHEMA_VERSION


BANGKOK = ZoneInfo("Asia/Bangkok")
TEST_CREDENTIALS = LoginCredentials(
    username="arr-test-operator",
    password="arr-test-password",
)


def decoded(response: Any) -> Dict[str, Any]:
    return json.loads(response.body.decode("utf-8"))


def login_headers(app: PortalApplication) -> Dict[str, str]:
    response = app.handle(
        "POST",
        "/api/login",
        {"Content-Type": "application/json"},
        json.dumps(
            {
                "username": TEST_CREDENTIALS.username,
                "password": TEST_CREDENTIALS.password,
            }
        ).encode("utf-8"),
    )
    if response.status != 200:
        raise AssertionError(f"test login failed with HTTP {response.status}")
    payload = decoded(response)
    cookie = response.headers["Set-Cookie"].split(";", 1)[0]
    return {
        "Cookie": cookie,
        "X-ARR-CSRF": payload["data"]["csrf_token"],
    }


class FixtureExecutor:
    def __init__(self, project_root: Path, *, failed_company: str = "") -> None:
        self.project_root = project_root
        self.output_root = project_root / "outputs" / "company_reports"
        self.failed_company = failed_company
        self.calls: list[tuple[int, int, object]] = []

    def run(self, year: int, month: int, as_of_date: object) -> Mapping[str, Any]:
        self.calls.append((year, month, as_of_date))
        final_day = 29 if year == 2028 and month == 2 else 31
        if month in {4, 6, 9, 11}:
            final_day = 30
        elif month == 2 and year != 2028:
            final_day = 28
        companies = []
        for index, company in enumerate(COMPANY_NAMES, 1):
            counts = {"01-10": index, "11-20": index + 1, f"21-{final_day:02d}": 0}
            if company == self.failed_company:
                companies.append(
                    {
                        "company": company,
                        "status": "failed",
                        "row_count": 0,
                        "period_row_counts": counts,
                        "warnings": [],
                        "errors": [
                            {
                                "code": "COMPANY_REPORT_GROUP_CODE_NOT_FOUND",
                                "stage": "source",
                                "period": "01-10",
                                "record_ids": [100 + index],
                            }
                        ],
                    }
                )
                continue
            filename = f"{company}-{ENGLISH_MONTH_NAMES[month]}-{year}.xlsx"
            path = self.output_root / f"{year:04d}" / f"{month:02d}" / "test" / filename
            path.parent.mkdir(parents=True, exist_ok=True)
            payload = f"safe workbook fixture {company}".encode("utf-8")
            path.write_bytes(payload)
            companies.append(
                {
                    "company": company,
                    "status": "success",
                    "row_count": sum(counts.values()),
                    "period_row_counts": counts,
                    "version_no": index,
                    "warnings": (
                        [
                            {
                                "code": "COMPANY_REPORT_MULTI_PRICE_REVIEW",
                                "stage": "pricing",
                                "period": "01-10",
                                "record_ids": [index],
                            }
                        ]
                        if company == "QBD"
                        else []
                    ),
                    "errors": [],
                    "artifact": {
                        "filename": filename,
                        "storage_key": path.relative_to(self.project_root).as_posix(),
                        "sha256": hashlib.sha256(payload).hexdigest(),
                        "semantic_sha256": "a" * 64,
                    },
                }
            )
        successful = len(COMPANY_NAMES) - (1 if self.failed_company else 0)
        status = "success" if successful == len(COMPANY_NAMES) else "partial_failure"
        return {
            "schema_version": RESULT_SCHEMA_VERSION,
            "status": status,
            "report_year": year,
            "report_month": month,
            "as_of_date": str(as_of_date),
            "requested_companies": list(COMPANY_NAMES),
            "companies": companies,
        }


class BlockingFailureExecutor:
    def __init__(self) -> None:
        self.started = threading.Event()
        self.release = threading.Event()

    def run(self, year: int, month: int, as_of_date: object) -> Mapping[str, Any]:
        self.started.set()
        self.release.wait(timeout=3)
        counts = {"01-10": 0, "11-20": 0, "21-31": 0}
        return {
            "schema_version": RESULT_SCHEMA_VERSION,
            "status": "failed",
            "report_year": year,
            "report_month": month,
            "as_of_date": str(as_of_date),
            "requested_companies": list(COMPANY_NAMES),
            "companies": [
                {
                    "company": company,
                    "status": "failed",
                    "row_count": 0,
                    "period_row_counts": counts,
                    "warnings": [],
                    "errors": [
                        {
                            "code": "COMPANY_REPORT_SOURCE_VERSION_MISSING",
                            "stage": "source",
                            "period": "2026-07",
                            "record_ids": [],
                        }
                    ],
                }
                for company in COMPANY_NAMES
            ],
        }


def wait_terminal(
    coordinator: PersistentCompanyReportCoordinator,
    job_id: str,
    timeout: float = 3.0,
) -> Dict[str, Any]:
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        value = coordinator.get_job(job_id)
        if value["state"] in {"succeeded", "partial_failure", "failed"}:
            return value
        time.sleep(0.01)
    raise AssertionError("company report job did not finish")


class CompanyReportCoordinatorTests(unittest.TestCase):
    def test_period_mapping_and_bangkok_completion_instants(self) -> None:
        self.assertEqual(str(period_to_as_of("2028-02", "21-month-end")), "2028-02-29")
        self.assertEqual(
            period_complete_at("2026-07", "01-10").isoformat(),
            "2026-07-11T00:00:00+07:00",
        )
        self.assertEqual(
            period_complete_at("2026-12", "21-month-end").isoformat(),
            "2027-01-01T00:00:00+07:00",
        )
        self.assertEqual(
            period_release_at("2026-07", "01-10"),
            period_complete_at("2026-07", "01-10"),
        )

    def test_success_is_private_and_download_is_reverified(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            executor = FixtureExecutor(root)
            coordinator = PersistentCompanyReportCoordinator(
                root,
                root / "outputs" / "company_reports",
                executor,
                now=lambda: datetime(2026, 8, 1, tzinfo=BANGKOK),
            )
            try:
                source = {
                    "source_batch_id": 7,
                    "source_type": "excel",
                    "filename": "Booking 报表.xlsx",
                    "source_rows": 12,
                    "worksheet_count": 2,
                    "distinct_group_codes": 8,
                    "room_quantity": 14,
                    "activated_at": "2026-07-31T08:00:00+07:00",
                    "disposition": "current",
                }
                created = coordinator.create("2026-07", "11-20", source=source)
                self.assertEqual(created["source"], source)
                job = wait_terminal(coordinator, created["job_id"])
                self.assertEqual(job["state"], "succeeded")
                self.assertEqual(job["source"], source)
                self.assertEqual(job["requested_companies"], list(COMPANY_NAMES))
                self.assertEqual(len(job["company_results"]), 5)
                self.assertEqual(job["company_results"][1]["warnings"][0]["record_ids"], [2])
                public_text = json.dumps(job, ensure_ascii=False)
                self.assertNotIn("storage_key", public_text)
                self.assertNotIn("sha256", public_text)
                descriptor = coordinator.resolve_download(created["job_id"], "QBD")
                self.assertEqual(descriptor.file_kind, "company_ten_day_xlsx")
                body = ControlledProjectArtifactReader(root).read(descriptor)
                self.assertIn(b"safe workbook fixture", body)
            finally:
                coordinator.close()

    def test_partial_failure_keeps_successful_company_downloads(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            coordinator = PersistentCompanyReportCoordinator(
                root,
                root / "outputs" / "company_reports",
                FixtureExecutor(root, failed_company="HanaTour"),
                now=lambda: datetime(2026, 8, 1, tzinfo=BANGKOK),
            )
            try:
                created = coordinator.create("2026-07", "01-10")
                job = wait_terminal(coordinator, created["job_id"])
                self.assertEqual(job["state"], "partial_failure")
                self.assertEqual(set(job["downloads"]), set(COMPANY_NAMES) - {"HanaTour"})
                with self.assertRaises(PortalError) as caught:
                    coordinator.resolve_download(created["job_id"], "HanaTour")
                self.assertEqual(caught.exception.status, 404)
            finally:
                coordinator.close()

    def test_incomplete_period_can_run_and_future_month_is_rejected(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            executor = FixtureExecutor(root)
            coordinator = PersistentCompanyReportCoordinator(
                root,
                root / "outputs" / "company_reports",
                executor,
                now=lambda: datetime(2026, 7, 10, 23, 59, tzinfo=BANGKOK),
            )
            try:
                created = coordinator.create("2026-07", "01-10")
                job = wait_terminal(coordinator, created["job_id"])
                self.assertEqual(job["state"], "succeeded")
                self.assertEqual(executor.calls, [(2026, 7, date(2026, 7, 10))])
            finally:
                coordinator.close()

        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            coordinator = PersistentCompanyReportCoordinator(
                root,
                root / "outputs" / "company_reports",
                FixtureExecutor(root),
                now=lambda: datetime(2026, 7, 10, 23, 59, tzinfo=BANGKOK),
            )
            try:
                with self.assertRaises(PortalError) as caught:
                    coordinator.create("2026-08", "01-10")
                self.assertEqual(caught.exception.code, "COMPANY_REPORT_MONTH_IN_FUTURE")
            finally:
                coordinator.close()

        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            executor = BlockingFailureExecutor()
            coordinator = PersistentCompanyReportCoordinator(
                root,
                root / "outputs" / "company_reports",
                executor,
                now=lambda: datetime(2026, 8, 1, tzinfo=BANGKOK),
            )
            try:
                first = coordinator.create("2026-07", "01-10")
                self.assertTrue(executor.started.wait(timeout=1))
                with self.assertRaises(PortalError) as caught:
                    coordinator.create("2026-07", "01-10")
                self.assertEqual(caught.exception.code, "COMPANY_REPORT_ALREADY_RUNNING")
                executor.release.set()
                failed_job = wait_terminal(coordinator, first["job_id"])
                self.assertEqual(failed_job["state"], "failed")
                self.assertIsNone(failed_job["message"])
            finally:
                executor.release.set()
                coordinator.close()


class FakeCompanyCoordinator:
    def __init__(self) -> None:
        self.created: list[tuple[str, str, Optional[Mapping[str, Any]]]] = []
        self.listed: list[tuple[object, int, int]] = []
        self.descriptor = ArtifactDescriptor(
            "company_ten_day_xlsx",
            "QBD-July-2026.xlsx",
            "outputs/company_reports/QBD-July-2026.xlsx",
            "a" * 64,
            4,
            "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        )

    def create(
        self,
        report_month: str,
        period: str,
        *,
        source: Optional[Mapping[str, Any]] = None,
    ) -> Dict[str, Any]:
        self.created.append((report_month, period, source))
        return {"job_id": "a" * 32, "state": "queued", "source": source}

    def list_jobs(
        self,
        report_month: object = None,
        limit: int = 50,
        offset: int = 0,
    ) -> tuple[list[Dict[str, Any]], int]:
        self.listed.append((report_month, limit, offset))
        return [
            {
                "job_id": f"{index:032x}",
                "report_month": report_month,
                "limit": limit,
                "offset": offset,
            }
            for index in range(offset, min(offset + limit, 72))
        ], 72

    def list_month_counts(self) -> list[Dict[str, Any]]:
        return [
            {"month_key": "2026-07", "company_count": 4},
            {"month_key": "2026-06", "company_count": 1},
        ]

    def get_job(self, job_id: str) -> Dict[str, Any]:
        return {"job_id": job_id, "state": "succeeded"}

    def resolve_download(self, job_id: str, company: str) -> ArtifactDescriptor:
        return self.descriptor


class FakeReader:
    def read(self, descriptor: ArtifactDescriptor) -> bytes:
        return b"xlsx"


class FakeBookingSources:
    def __init__(self) -> None:
        self.uploads: list[tuple[str, bytes]] = []
        self.draft_page: Dict[str, Any] | None = None
        self.review_updates: list[tuple[Any, int, Any, Any]] = []
        self.review_deletes: list[tuple[Any, int]] = []
        self.review_bulk_deletes: list[tuple[Any, tuple[int, ...]]] = []
        self.review_discards: list[Any] = []
        self.review_activations: list[Any] = []
        self.source: Dict[str, Any] | None = {
            "source_batch_id": 7,
            "source_type": "excel",
            "filename": "Booking 报表.xlsx",
            "source_rows": 12,
            "worksheet_count": 2,
            "distinct_group_codes": 8,
            "room_quantity": 14,
            "activated_at": "2026-07-31T08:00:00+07:00",
            "disposition": "current",
        }

    def current(self) -> Dict[str, Any] | None:
        return self.source

    def draft(self, limit: int = 50, offset: int = 0) -> Dict[str, Any] | None:
        _ = (limit, offset)
        return self.draft_page

    def submit(self, filename: str, payload: bytes) -> Dict[str, Any]:
        self.uploads.append((filename, payload))
        self.draft_page = {
            "summary": {
                "draft_id": "bookingdraft-" + "a" * 32,
                "filename": filename,
                "pending_items": 1,
            },
            "items": [],
            "pagination": {"total": 0, "limit": 50, "offset": 0},
        }
        return self.draft_page

    def update_item(
        self,
        draft_id: Any,
        item_id: int,
        room_type: Any,
        quantity: Any,
    ) -> Dict[str, Any]:
        self.review_updates.append((draft_id, item_id, room_type, quantity))
        return {"item_id": item_id, "room_type": room_type, "quantity": quantity}

    def delete_item(self, draft_id: Any, item_id: int) -> Dict[str, Any]:
        self.review_deletes.append((draft_id, item_id))
        return {"item_id": item_id, "deleted": True}

    def delete_items(self, draft_id: Any, item_ids: object) -> Dict[str, Any]:
        assert isinstance(item_ids, list)
        values = tuple(item_ids)
        self.review_bulk_deletes.append((draft_id, values))
        return {
            "draft_id": draft_id,
            "item_ids": list(values),
            "deleted_count": len(values),
            "deleted": True,
        }

    def discard(self, draft_id: Any) -> Dict[str, Any]:
        self.review_discards.append(draft_id)
        self.draft_page = None
        return {"draft_id": draft_id, "discarded": True}

    def activate(self, draft_id: Any) -> Dict[str, Any]:
        self.review_activations.append(draft_id)
        self.draft_page = None
        assert self.source is not None
        return self.source


class CompanyReportRouteTests(unittest.TestCase):
    def setUp(self) -> None:
        self.coordinator = FakeCompanyCoordinator()
        self.booking_sources = FakeBookingSources()
        self.app = PortalApplication(
            company_reports=self.coordinator,
            booking_sources=self.booking_sources,
            artifact_reader=FakeReader(),
            health=RuntimeHealth(False, False, False, True, True, True),
            sessions=SessionLedger(),
            credentials=TEST_CREDENTIALS,
        )
        self.auth_headers = login_headers(self.app)

    def session_headers(self) -> Dict[str, str]:
        return dict(self.auth_headers)

    def test_create_is_csrf_protected_and_strict(self) -> None:
        body = b'{"report_month":"2026-07","period":"01-10"}'
        self.assertEqual(
            self.app.handle(
                "POST",
                "/api/company-reports/jobs",
                {"Cookie": self.auth_headers["Cookie"]},
                body,
            ).status,
            403,
        )
        response = self.app.handle(
            "POST",
            "/api/company-reports/jobs",
            {**self.session_headers(), "Content-Type": "application/json"},
            body,
        )
        self.assertEqual(response.status, 202)
        self.assertEqual(
            self.coordinator.created,
            [("2026-07", "01-10", self.booking_sources.source)],
        )
        self.assertEqual(decoded(response)["data"]["source"]["source_batch_id"], 7)
        invalid = self.app.handle(
            "POST",
            "/api/company-reports/jobs",
            {**self.session_headers(), "Content-Type": "application/json"},
            b'{"report_month":"2026-07","period":"01-10","companies":[]}',
        )
        self.assertEqual(invalid.status, 400)
        self.assertEqual(decoded(invalid)["error"]["code"], "COMPANY_REPORT_REQUEST_INVALID")

    def test_source_read_and_upload_are_authenticated_and_csrf_protected(self) -> None:
        current = self.app.handle(
            "GET",
            "/api/company-reports/source",
            self.auth_headers,
        )
        self.assertEqual(decoded(current)["data"]["filename"], "Booking 报表.xlsx")

        encoded = base64.urlsafe_b64encode("七月预订.xlsx".encode("utf-8")).decode("ascii").rstrip("=")
        without_csrf = self.app.handle(
            "POST",
            "/api/company-reports/source",
            {
                "Cookie": self.auth_headers["Cookie"],
                "X-ARR-Filename-B64": encoded,
            },
            b"xlsx",
        )
        self.assertEqual(without_csrf.status, 403)
        uploaded = self.app.handle(
            "POST",
            "/api/company-reports/source",
            {**self.session_headers(), "X-ARR-Filename-B64": encoded},
            b"xlsx",
        )
        self.assertEqual(uploaded.status, 200)
        self.assertEqual(self.booking_sources.uploads, [("七月预订.xlsx", b"xlsx")])
        self.assertEqual(decoded(uploaded)["data"]["summary"]["pending_items"], 1)

    def test_review_routes_are_authenticated_csrf_protected_and_strict(self) -> None:
        draft_id = "bookingdraft-" + "a" * 32
        self.booking_sources.draft_page = {
            "summary": {"draft_id": draft_id, "pending_items": 1},
            "items": [],
            "pagination": {"total": 0, "limit": 50, "offset": 0},
        }
        page = self.app.handle(
            "GET",
            "/api/company-reports/source/draft?limit=50&offset=0",
            self.auth_headers,
        )
        self.assertEqual(decoded(page)["data"]["summary"]["draft_id"], draft_id)

        update_body = json.dumps(
            {"draft_id": draft_id, "room_type": "EXTRA BED", "quantity": 2}
        ).encode("utf-8")
        without_csrf = self.app.handle(
            "PATCH",
            "/api/company-reports/source/draft/items/9",
            {"Cookie": self.auth_headers["Cookie"]},
            update_body,
        )
        self.assertEqual(without_csrf.status, 403)
        updated = self.app.handle(
            "PATCH",
            "/api/company-reports/source/draft/items/9",
            {**self.session_headers(), "Content-Type": "application/json"},
            update_body,
        )
        self.assertEqual(updated.status, 200)
        self.assertEqual(
            self.booking_sources.review_updates,
            [(draft_id, 9, "EXTRA BED", 2)],
        )

        deleted = self.app.handle(
            "DELETE",
            "/api/company-reports/source/draft/items/9",
            {**self.session_headers(), "Content-Type": "application/json"},
            json.dumps({"draft_id": draft_id}).encode("utf-8"),
        )
        self.assertEqual(deleted.status, 200)
        self.assertEqual(self.booking_sources.review_deletes, [(draft_id, 9)])

        bulk_body = json.dumps({"draft_id": draft_id, "item_ids": [9, 10]}).encode("utf-8")
        without_bulk_csrf = self.app.handle(
            "DELETE",
            "/api/company-reports/source/draft/items",
            {"Cookie": self.auth_headers["Cookie"]},
            bulk_body,
        )
        self.assertEqual(without_bulk_csrf.status, 403)
        bulk_deleted = self.app.handle(
            "DELETE",
            "/api/company-reports/source/draft/items",
            {**self.session_headers(), "Content-Type": "application/json"},
            bulk_body,
        )
        self.assertEqual(bulk_deleted.status, 200)
        self.assertEqual(decoded(bulk_deleted)["data"]["deleted_count"], 2)
        self.assertEqual(
            self.booking_sources.review_bulk_deletes,
            [(draft_id, (9, 10))],
        )
        invalid_bulk = self.app.handle(
            "DELETE",
            "/api/company-reports/source/draft/items",
            {**self.session_headers(), "Content-Type": "application/json"},
            json.dumps({"draft_id": draft_id, "item_ids": [9], "extra": True}).encode(
                "utf-8"
            ),
        )
        self.assertEqual(invalid_bulk.status, 400)
        self.assertEqual(
            decoded(invalid_bulk)["error"]["code"],
            "BOOKING_EXCEL_REVIEW_REQUEST_INVALID",
        )

        activated = self.app.handle(
            "POST",
            "/api/company-reports/source/draft/activate",
            {**self.session_headers(), "Content-Type": "application/json"},
            json.dumps({"draft_id": draft_id}).encode("utf-8"),
        )
        self.assertEqual(activated.status, 200)
        self.assertEqual(self.booking_sources.review_activations, [draft_id])

    def test_open_review_draft_blocks_company_generation(self) -> None:
        self.booking_sources.draft_page = {
            "summary": {"draft_id": "bookingdraft-" + "a" * 32},
            "items": [],
            "pagination": {"total": 0, "limit": 1, "offset": 0},
        }
        response = self.app.handle(
            "POST",
            "/api/company-reports/jobs",
            {**self.session_headers(), "Content-Type": "application/json"},
            b'{"report_month":"2026-07","period":"01-10"}',
        )
        self.assertEqual(response.status, 409)
        self.assertEqual(decoded(response)["error"]["code"], "BOOKING_EXCEL_REVIEW_OPEN")

    def test_generation_requires_an_accepted_source(self) -> None:
        self.booking_sources.source = None
        response = self.app.handle(
            "POST",
            "/api/company-reports/jobs",
            {**self.session_headers(), "Content-Type": "application/json"},
            b'{"report_month":"2026-07","period":"01-10"}',
        )
        self.assertEqual(response.status, 409)
        self.assertEqual(
            decoded(response)["error"]["code"],
            "BOOKING_EXCEL_SOURCE_REQUIRED",
        )

    def test_list_detail_and_download_routes(self) -> None:
        listing = self.app.handle(
            "GET",
            "/api/company-reports/jobs?month=2026-07&limit=50&offset=50",
            self.auth_headers,
        )
        listing_payload = decoded(listing)
        self.assertEqual(listing_payload["data"][0]["report_month"], "2026-07")
        self.assertEqual(self.coordinator.listed, [("2026-07", 50, 50)])
        self.assertEqual(listing_payload["pagination"]["total"], 72)
        self.assertTrue(listing_payload["pagination"]["has_previous"])
        self.assertFalse(listing_payload["pagination"]["has_next"])
        detail = self.app.handle(
            "GET",
            f"/api/company-reports/jobs/{'a' * 32}",
            self.auth_headers,
        )
        self.assertEqual(decoded(detail)["data"]["state"], "succeeded")
        download = self.app.handle(
            "GET",
            f"/api/company-reports/jobs/{'a' * 32}/downloads/QBD",
            self.auth_headers,
        )
        self.assertEqual(download.status, 200)
        self.assertEqual(download.body, b"xlsx")
        self.assertNotIn("outputs/company_reports", str(download.headers))


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