#!/usr/bin/env python3
"""Prepare a deterministic, read-only legacy import batch from the latest workbook.

This script never connects to PostgreSQL. It keeps source Use/Balance and raw
room-type text, filters rows without Confirmation, and writes auditable JSON
artifacts for the later database importer.
"""

from __future__ import annotations

import hashlib
import json
import re
import sys
from collections import Counter, defaultdict
from datetime import date, datetime
from pathlib import Path
from typing import Any

from openpyxl import load_workbook


PROJECT_ROOT = Path(__file__).resolve().parents[2]
DEFAULT_WORKBOOK = Path(
    "/Users/chillishark/Desktop/Condo公寓/Confirmation Report 2026  สิทธิ์ 15 วันคอนโดเท่านั้น(2).xlsx"
)
DEFAULT_OUTPUT = PROJECT_ROOT / ".planning" / "data_import_audit" / "import_batch_2026_07_31"
MASTER_SHEET = "Total2024-2026"
ROOM_TYPES = {
    "RM1",
    "RM2",
    "RM3",
    "RM4",
    "UG1",
    "UG2",
    "SU1",
    "SU2",
    "SU6",
    "SU3",
    "AC2",
}


def scalar(value: Any) -> Any:
    if isinstance(value, datetime):
        return value.date().isoformat()
    if isinstance(value, date):
        return value.isoformat()
    if isinstance(value, float) and value.is_integer():
        return int(value)
    return value


def text(value: Any) -> str:
    value = scalar(value)
    return "" if value is None else str(value).strip()


def number(value: Any) -> int | None:
    value = scalar(value)
    if isinstance(value, int):
        return value
    if isinstance(value, str) and re.fullmatch(r"-?\d+", value.strip()):
        return int(value.strip())
    return None


def parsed_date(value: Any) -> date | None:
    value = scalar(value)
    if isinstance(value, date):
        return value
    if isinstance(value, str):
        for fmt in ("%Y-%m-%d", "%d-%b-%y", "%d-%b-%Y"):
            try:
                return datetime.strptime(value.strip(), fmt).date()
            except ValueError:
                pass
    return None


def normalize_header(value: Any) -> str:
    return re.sub(r"[^a-z0-9]+", "", text(value).lower())


def normalize_room_type(value: Any) -> str:
    return re.sub(r"\r?\n", "<br>", text(value))


def room_numbers(value: Any) -> list[str]:
    return list(dict.fromkeys(re.findall(r"\b\d{4}\b", text(value))))


def canonical_room_type(raw: str) -> str | None:
    """Only exact standard codes are canonicalized; legacy text is not guessed."""
    return raw if raw in ROOM_TYPES else None


def write_json(path: Path, value: Any) -> None:
    path.write_text(
        json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )


def prepare(workbook_path: Path, output_dir: Path) -> dict[str, Any]:
    workbook_bytes = workbook_path.read_bytes()
    source_sha256 = hashlib.sha256(workbook_bytes).hexdigest()
    workbook = load_workbook(workbook_path, data_only=True, read_only=False)
    if MASTER_SHEET not in workbook.sheetnames:
        raise ValueError(f"Missing master sheet: {MASTER_SHEET}")

    master = workbook[MASTER_SHEET]
    header_map: dict[str, int] = {}
    for column in range(1, master.max_column + 1):
        header = normalize_header(master.cell(2, column).value)
        if header:
            header_map.setdefault(header, column)

    def column(header: str, fallback: int) -> int:
        return header_map.get(normalize_header(header), fallback)

    columns = {
        "no": column("No", 1),
        "transfer_date": column("Transfer Date", 2),
        "name": column("Name", 3),
        "room_no": column("Room No.", 4),
        "room_type": column("Room Type", 5),
        "unit_no": column("Unit No.", 6),
        "member_no": column("Member No.", 7),
        "remaining": column("Remaining stay privileges", 8),
    }

    owners: list[dict[str, Any]] = []
    owner_by_room: dict[str, dict[str, Any]] = {}
    owner_issues: list[dict[str, Any]] = []
    for row in range(4, master.max_row + 1):
        room_no = text(master.cell(row, columns["room_no"]).value)
        if not re.fullmatch(r"\d{4}", room_no):
            continue
        owner = {
            "account_no": number(master.cell(row, columns["no"]).value),
            "transfer_date": scalar(master.cell(row, columns["transfer_date"]).value),
            "name": text(master.cell(row, columns["name"]).value),
            "room_no": room_no,
            "purchased_room_type_code": text(master.cell(row, columns["room_type"]).value),
            "unit_no": text(master.cell(row, columns["unit_no"]).value),
            "member_no": text(master.cell(row, columns["member_no"]).value),
            "reported_remaining": number(master.cell(row, columns["remaining"]).value),
            "source_sheet": MASTER_SHEET,
            "source_row": row,
        }
        missing = [
            key
            for key in (
                "account_no",
                "transfer_date",
                "name",
                "room_no",
                "purchased_room_type_code",
                "unit_no",
                "member_no",
                "reported_remaining",
            )
            if owner[key] in (None, "")
        ]
        if missing:
            owner_issues.append({"source_row": row, "room_no": room_no, "missing": missing})
        if owner["purchased_room_type_code"] not in ROOM_TYPES:
            owner_issues.append(
                {
                    "source_row": row,
                    "room_no": room_no,
                    "invalid_purchased_room_type": owner["purchased_room_type_code"],
                }
            )
        if owner["reported_remaining"] is not None and not 0 <= owner["reported_remaining"] <= 15:
            owner_issues.append(
                {
                    "source_row": row,
                    "room_no": room_no,
                    "remaining_out_of_range": owner["reported_remaining"],
                }
            )
        owners.append(owner)
        owner_by_room[room_no] = owner

    usage: list[dict[str, Any]] = []
    rejected: list[dict[str, Any]] = []
    sheet_diagnostics: list[dict[str, Any]] = []
    arithmetic_issues: list[dict[str, Any]] = []
    chain_issues: list[dict[str, Any]] = []
    required_issues: list[dict[str, Any]] = []
    date_issues: list[dict[str, Any]] = []
    identity_issues: list[dict[str, Any]] = []
    previous_by_sheet: dict[str, dict[str, Any]] = {}
    duplicate_confirmation_locations: dict[str, list[dict[str, Any]]] = defaultdict(list)

    expected_headers = [
        "No.",
        "Confirmation No.",
        "Check -IN",
        "Check-OUT",
        "Night",
        "Room",
        "Total",
        "Use",
        "Balance",
        "Room Type",
        "Remark",
    ]

    for title in workbook.sheetnames:
        if title == MASTER_SHEET:
            continue
        sheet = workbook[title]
        title_rooms = room_numbers(title)
        if not title_rooms:
            sheet_diagnostics.append({"sheet": title, "kind": "template", "usage_count": 0})
            continue
        header_values = [text(sheet.cell(3, column).value).rstrip(" ") for column in range(1, 12)]
        normalized_headers = [header.replace("Check-OUT ", "Check-OUT").strip() for header in header_values]
        header_issue = normalized_headers != expected_headers
        header_rooms = room_numbers(sheet.cell(2, 4).value)
        if set(title_rooms) != set(header_rooms):
            identity_issues.append(
                {
                    "sheet": title,
                    "title_rooms": title_rooms,
                    "header_rooms": header_rooms,
                    "resolution": "sheet title is authoritative",
                }
            )
        accepted_count = 0
        sheet_rows: list[dict[str, Any]] = []
        for row in range(4, sheet.max_row + 1):
            raw = {
                "no": number(sheet.cell(row, 1).value),
                "confirmation_no": text(sheet.cell(row, 2).value),
                "check_in": scalar(sheet.cell(row, 3).value),
                "check_out": scalar(sheet.cell(row, 4).value),
                "night": number(sheet.cell(row, 5).value),
                "room": number(sheet.cell(row, 6).value),
                "total": number(sheet.cell(row, 7).value),
                "use": number(sheet.cell(row, 8).value),
                "balance": number(sheet.cell(row, 9).value),
                "raw_used_room_type": normalize_room_type(sheet.cell(row, 10).value),
                "remark": text(sheet.cell(row, 11).value),
            }
            if not raw["confirmation_no"]:
                populated_columns = [
                    column
                    for column in (3, 4, 6, 10, 11)
                    if text(sheet.cell(row, column).value)
                ]
                if populated_columns:
                    rejected.append(
                        {
                            "reason_code": "missing_confirmation",
                            "reason": "Confirmation No. is empty; clear/exclude this row",
                            "source_sheet": title,
                            "source_row": row,
                            "populated_columns": populated_columns,
                            **raw,
                        }
                    )
                continue

            missing = [
                key
                for key in (
                    "confirmation_no",
                    "check_in",
                    "check_out",
                    "night",
                    "room",
                    "total",
                    "use",
                    "balance",
                    "raw_used_room_type",
                    "remark",
                )
                if raw[key] in (None, "")
            ]
            if missing:
                required_issues.append({"source_sheet": title, "source_row": row, "missing": missing})

            check_in = parsed_date(raw["check_in"])
            check_out = parsed_date(raw["check_out"])
            if (
                check_in is None
                or check_out is None
                or check_out <= check_in
                or raw["night"] != (check_out - check_in).days
            ):
                date_issues.append({"source_sheet": title, "source_row": row, **raw})
            if raw["total"] is None or raw["use"] is None or raw["balance"] is None:
                arithmetic_issues.append({"source_sheet": title, "source_row": row, **raw})
            elif raw["total"] - raw["use"] != raw["balance"]:
                arithmetic_issues.append({"source_sheet": title, "source_row": row, **raw})
            previous = previous_by_sheet.get(title)
            if previous is not None and raw["total"] != previous["balance"]:
                chain_issues.append(
                    {
                        "source_sheet": title,
                        "source_row": row,
                        "expected_total": previous["balance"],
                        "actual_total": raw["total"],
                    }
                )
            previous_by_sheet[title] = raw
            period_year = check_in.year if check_in else None
            record = {
                "owner_room_no": title_rooms[0] if len(title_rooms) == 1 else None,
                "confirmation_no": raw["confirmation_no"],
                "check_in": raw["check_in"],
                "check_out": raw["check_out"],
                "night": raw["night"],
                "room": raw["room"],
                "total": raw["total"],
                "use": raw["use"],
                "balance": raw["balance"],
                "raw_used_room_type": raw["raw_used_room_type"],
                "canonical_used_room_type_code": canonical_room_type(raw["raw_used_room_type"]),
                "remark": raw["remark"],
                "source_sheet": title,
                "source_row": row,
                "source_sequence": raw["no"],
                "period_year": period_year,
                "applied_multiplier": None,
                "rule_version": "legacy-source",
            }
            usage.append(record)
            sheet_rows.append(record)
            accepted_count += 1
            duplicate_confirmation_locations[raw["confirmation_no"]].append(
                {"source_sheet": title, "source_row": row}
            )
        sheet_diagnostics.append(
            {
                "sheet": title,
                "kind": "room_usage",
                "title_rooms": title_rooms,
                "header_rooms": header_rooms,
                "header_issue": header_issue,
                "usage_count": accepted_count,
                "rejected_count": sum(
                    1 for row in rejected if row["source_sheet"] == title
                ),
                "final_balance": sheet_rows[-1]["balance"] if sheet_rows else None,
            }
        )

    duplicate_confirmations = {
        confirmation: locations
        for confirmation, locations in duplicate_confirmation_locations.items()
        if len(locations) > 1
    }
    master_remaining_sum = sum(owner["reported_remaining"] or 0 for owner in owners)
    usage_sum = sum(record["use"] or 0 for record in usage)
    empty_usage_sheets = [
        item["sheet"]
        for item in sheet_diagnostics
        if item["kind"] == "room_usage" and item["usage_count"] == 0
    ]
    noncanonical = [
        record
        for record in usage
        if record["canonical_used_room_type_code"] is None
    ]
    manifest = {
        "mode": "local-import-preflight",
        "source_workbook": str(workbook_path),
        "source_sha256": source_sha256,
        "master_sheet": MASTER_SHEET,
        "rules": {
            "latest_workbook_only": True,
            "missing_confirmation": "clear_or_exclude",
            "usage_owner": "sheet_title",
            "owner_metadata": "master_sheet",
            "history_values": "preserve_source_use_balance_total_room",
            "legacy_applied_multiplier": None,
            "legacy_rule_version": "legacy-source",
            "new_usage_room_type": "standard_code_only",
            "database_write_performed": False,
        },
        "counts": {
            "owner_count": len(owners),
            "unique_owner_room_count": len(owner_by_room),
            "accepted_usage_count": len(usage),
            "rejected_row_count": len(rejected),
            "rejected_missing_confirmation_count": sum(
                item["reason_code"] == "missing_confirmation" for item in rejected
            ),
            "usage_use_sum": usage_sum,
            "master_remaining_sum": master_remaining_sum,
            "implied_used": len(owners) * 15 - master_remaining_sum,
            "canonical_used_room_type_count": len(usage) - len(noncanonical),
            "noncanonical_used_room_type_count": len(noncanonical),
            "duplicate_confirmation_group_count": len(duplicate_confirmations),
        },
        "checks": {
            "owner_issues": owner_issues,
            "required_usage_issues": required_issues,
            "date_night_issues": date_issues,
            "arithmetic_issues": arithmetic_issues,
            "balance_chain_issues": chain_issues,
            "room_gt_one_count": sum((record["room"] or 0) > 1 for record in usage),
            "missing_owner_rooms": sorted(
                {
                    record["owner_room_no"]
                    for record in usage
                    if record["owner_room_no"] not in owner_by_room
                }
            ),
            "empty_usage_sheets": empty_usage_sheets,
        },
        "duplicate_confirmations": duplicate_confirmations,
        "year_counts": dict(sorted(Counter(record["period_year"] for record in usage).items())),
        "raw_used_room_type_counts": dict(
            sorted(Counter(record["raw_used_room_type"] for record in usage).items())
        ),
    }

    output_dir.mkdir(parents=True, exist_ok=True)
    write_json(output_dir / "owners.json", owners)
    write_json(output_dir / "usage_legacy.json", usage)
    write_json(output_dir / "rejected_rows.json", rejected)
    write_json(output_dir / "sheet_diagnostics.json", sheet_diagnostics)
    write_json(output_dir / "identity_issues.json", identity_issues)
    write_json(output_dir / "manifest.json", manifest)
    return manifest


def main() -> None:
    workbook_path = Path(sys.argv[1]).expanduser() if len(sys.argv) > 1 else DEFAULT_WORKBOOK
    output_dir = Path(sys.argv[2]).expanduser() if len(sys.argv) > 2 else DEFAULT_OUTPUT
    if not workbook_path.is_file():
        raise SystemExit(f"Workbook not found: {workbook_path}")
    manifest = prepare(workbook_path, output_dir)
    print(json.dumps(manifest, ensure_ascii=False, indent=2, sort_keys=True))


if __name__ == "__main__":
    main()