"""Parse an uploaded holiday spreadsheet into reviewable rows.

Upload safety — size caps, column/row caps, MIME allow-list, zip-bomb
preflight — is entirely delegated to :func:`app.domains.asset.bulk_import.read_bulk_rows`,
the platform's one hardened spreadsheet contract. This module only does the
domain mapping on top of it.

The flow is deliberately two-step: **preview** parses and classifies every row
without writing anything, and **commit** applies the rows HR approved. So a
spreadsheet with a bad date on line 40 shows HR exactly which line is wrong
instead of half-importing, and a re-import of an unchanged file is visibly a
no-op rather than a silent 300-row rewrite.

Header matching is alias-based and case-insensitive in both English and Chinese,
because the file HR uploads is as often a hand-made sheet as it is one Helix
exported. The English aliases are exactly the export's column labels, so an
export → edit → import round-trip needs no renaming.
"""
from __future__ import annotations

import re
from dataclasses import dataclass, field
from datetime import date, datetime
from typing import Any

from fastapi import UploadFile

from app.core.geo import normalize_country_code
from app.domains.asset.bulk_import import read_bulk_rows
from app.domains.hr.holiday_models import MAKEUP_WORKDAY, PUBLIC_HOLIDAY
from app.domains.hr.holiday_service import (
    MAX_BULK_ROWS,
    BulkRow,
    validate_year,
)

# Canonical field -> accepted header spellings (lower-cased, punctuation-stripped).
_HEADER_ALIASES: dict[str, tuple[str, ...]] = {
    "country": ("country", "country code", "countryregion", "region",
                "国家", "国家地区", "地区", "国家/地区"),
    "day": ("date", "day", "holiday date", "日期", "假期日期"),
    "name_en": ("holiday", "holiday name", "name", "name en", "english name",
                "假期", "假期名称", "名称", "英文名称"),
    "name_local": ("holiday local", "local name", "name local", "local",
                   "本地名称", "中文名称", "当地名称"),
    "day_type": ("type", "day type", "category", "类型", "假期类型"),
    "note": ("note", "notes", "remark", "remarks", "comment",
             "备注", "说明"),
    "group_key": ("group", "group key", "分组", "假期分组"),
}

_TYPE_ALIASES: dict[str, str] = {
    "public holiday": PUBLIC_HOLIDAY,
    "public_holiday": PUBLIC_HOLIDAY,
    "holiday": PUBLIC_HOLIDAY,
    "off": PUBLIC_HOLIDAY,
    "假期": PUBLIC_HOLIDAY,
    "法定假日": PUBLIC_HOLIDAY,
    "法定节假日": PUBLIC_HOLIDAY,
    "放假": PUBLIC_HOLIDAY,
    "makeup workday": MAKEUP_WORKDAY,
    "makeup_workday": MAKEUP_WORKDAY,
    "makeup": MAKEUP_WORKDAY,
    "workday": MAKEUP_WORKDAY,
    "补班": MAKEUP_WORKDAY,
    "调休": MAKEUP_WORKDAY,
    "调休上班": MAKEUP_WORKDAY,
    "上班": MAKEUP_WORKDAY,
}

_MONTH_NAMES = (
    "January", "February", "March", "April", "May", "June", "July",
    "August", "September", "October", "November", "December",
)
_EN_MONTHS = {m.lower(): i for i, m in enumerate(_MONTH_NAMES, start=1)}
_EN_MONTHS.update({m[:3]: i for m, i in list(_EN_MONTHS.items())})

# Longest-first so "June" is preferred over the "Jun" abbreviation.
_MONTH_ALT = "|".join(
    sorted({*_MONTH_NAMES, *(m[:3] for m in _MONTH_NAMES)}, key=len, reverse=True)
)

# "Thursday, January 1, 2026" / "January 1 2026" — the source document's format.
#
# The month token is an alternation of the actual month names, and every
# whitespace run is bounded. An earlier version opened with an optional
# ``(?:[A-Za-z]+,?\s+)?`` weekday prefix before a generic ``([A-Za-z]+)`` month:
# both groups could consume the same letters, so a line of 8,000 letters took
# ~650 ms of backtracking (CodeQL py/polynomial-redos). Since HR pastes up to
# 20,000 characters of untrusted text through this, that is a real denial of
# service, not a theoretical one. The weekday prefix needs no pattern at all —
# ``search`` simply starts at the month.
_LONG_DATE = re.compile(
    rf"\b({_MONTH_ALT})\b\.?[ \t]{{0,3}}(\d{{1,2}})[ \t]{{0,3}},?[ \t]{{0,3}}(\d{{4}})",
    re.IGNORECASE,
)
# "2026年5月1日" — literal anchors between bounded runs, so no ambiguity.
_CJK_DATE = re.compile(
    r"(\d{4})\s{0,3}年\s{0,3}(\d{1,2})\s{0,3}月\s{0,3}(\d{1,2})\s{0,3}日?"
)

_ROW_STATUS_NEW = "new"
_ROW_STATUS_DUPLICATE = "duplicate"
_ROW_STATUS_ERROR = "error"


@dataclass
class ParsedRow:
    """One spreadsheet line, classified for the preview table."""

    line: int
    status: str
    country: str | None = None
    day: date | None = None
    name_en: str | None = None
    name_local: str | None = None
    day_type: str = PUBLIC_HOLIDAY
    note: str | None = None
    group_key: str | None = None
    message: str | None = None

    def to_payload(self) -> dict[str, Any]:
        return {
            "line": self.line,
            "status": self.status,
            "country": self.country,
            "day": self.day.isoformat() if self.day else None,
            "name_en": self.name_en,
            "name_local": self.name_local,
            "day_type": self.day_type,
            "note": self.note,
            "group_key": self.group_key,
            "message": self.message,
        }

    def to_bulk_row(self) -> BulkRow:
        assert self.country and self.day and self.name_en
        return BulkRow(
            country=self.country,
            day=self.day,
            name_en=self.name_en,
            name_local=self.name_local,
            day_type=self.day_type,
            note=self.note,
            group_key=self.group_key,
        )


@dataclass
class ParseReport:
    rows: list[ParsedRow] = field(default_factory=list)
    unmapped_headers: list[str] = field(default_factory=list)
    missing_required: list[str] = field(default_factory=list)

    @property
    def importable(self) -> list[ParsedRow]:
        return [r for r in self.rows if r.status == _ROW_STATUS_NEW]

    def counts(self) -> dict[str, int]:
        out = {_ROW_STATUS_NEW: 0, _ROW_STATUS_DUPLICATE: 0, _ROW_STATUS_ERROR: 0}
        for row in self.rows:
            out[row.status] = out.get(row.status, 0) + 1
        return out


def _normalize_header(raw: Any) -> str:
    text = str(raw or "").strip().lower()
    # Drop punctuation and collapse whitespace: "Holiday (Local)" -> "holiday local".
    text = re.sub(r"[()\[\]_./\\-]+", " ", text)
    return re.sub(r"\s+", " ", text).strip()


def _map_headers(sample: dict[str, Any]) -> tuple[dict[str, str], list[str]]:
    """Map spreadsheet headers onto canonical field names."""
    mapping: dict[str, str] = {}
    unmapped: list[str] = []
    for raw in sample:
        norm = _normalize_header(raw)
        # Compare against aliases with spaces removed too, so "countrycode",
        # "country code" and "Country_Code" all land on the same field.
        squashed = norm.replace(" ", "")
        for field_name, aliases in _HEADER_ALIASES.items():
            if field_name in mapping.values():
                continue
            if norm in aliases or squashed in {a.replace(" ", "") for a in aliases}:
                mapping[raw] = field_name
                break
        else:
            if norm:
                unmapped.append(str(raw))
    return mapping, unmapped


def parse_date(value: Any) -> date | None:
    """Accept every date shape a real HR spreadsheet contains."""
    if value is None:
        return None
    if isinstance(value, datetime):
        return value.date()
    if isinstance(value, date):
        return value
    text = str(value).strip()
    if not text:
        return None
    # ISO first — it is what the Helix export writes.
    try:
        return date.fromisoformat(text[:10])
    except ValueError:
        pass
    cjk = _CJK_DATE.search(text)
    if cjk:
        try:
            return date(int(cjk.group(1)), int(cjk.group(2)), int(cjk.group(3)))
        except ValueError:
            return None
    long_match = _LONG_DATE.search(text)
    if long_match:
        month = _EN_MONTHS.get(long_match.group(1).lower())
        if month:
            try:
                return date(
                    int(long_match.group(3)), month, int(long_match.group(2)),
                )
            except ValueError:
                return None
    # Year-first slash/dot forms are unambiguous.
    for fmt in ("%Y/%m/%d", "%Y.%m.%d", "%Y-%m-%d"):
        try:
            return datetime.strptime(text, fmt).date()
        except ValueError:
            continue
    # ``5/9/2026`` is May 9 to an American and 5 September to everyone else, and
    # guessing is worse than refusing: the earlier version tried %d/%m/%Y first,
    # so a US-formatted sheet silently became a date four months away with no
    # error row for HR to notice. Only the unambiguous half is accepted; the rest
    # comes back None and surfaces as "unreadable date" in the preview.
    ambiguous = re.fullmatch(r"(\d{1,2})[/.](\d{1,2})[/.](\d{4})", text)
    if ambiguous:
        a, b, year = (int(g) for g in ambiguous.groups())
        if a > 12 and b <= 12:
            try:
                return date(year, b, a)   # only day-first can be meant
            except ValueError:
                return None
        if b > 12 and a <= 12:
            try:
                return date(year, a, b)   # only month-first can be meant
            except ValueError:
                return None
        return None
    return None


def parse_day_type(value: Any) -> str | None:
    """``None`` signals an unrecognised value, which is reported not defaulted."""
    text = str(value or "").strip().lower()
    if not text:
        return PUBLIC_HOLIDAY
    return _TYPE_ALIASES.get(text)


def _clean(value: Any, *, limit: int) -> str | None:
    text = str(value or "").strip()
    if not text:
        return None
    return text[:limit]


def parse_rows(
    raw_rows: list[dict[str, Any]],
    *,
    existing_keys: set[tuple[str, str, str]] | None = None,
) -> ParseReport:
    """Classify every spreadsheet line as importable, duplicate, or in error.

    ``existing_keys`` holds ``(country, iso date, name_key)`` triples already in
    the database; matching lines are marked ``duplicate`` so HR sees at a glance
    how much of the upload is actually new. Duplicates *within the file* are
    flagged the same way — a sheet listing 2026-05-01 twice under one name is a
    mistake worth showing, not something to silently dedupe.
    """
    from app.domains.hr.holiday_models import name_key_for

    report = ParseReport()
    if not raw_rows:
        return report

    mapping, unmapped = _map_headers(raw_rows[0])
    report.unmapped_headers = unmapped
    mapped_fields = set(mapping.values())
    report.missing_required = [
        f for f in ("country", "day", "name_en") if f not in mapped_fields
    ]
    if report.missing_required:
        return report

    seen: set[tuple[str, str, str]] = set(existing_keys or set())
    for offset, raw in enumerate(raw_rows):
        # Line 1 is the header row in the user's spreadsheet.
        line = offset + 2
        values: dict[str, Any] = {}
        for raw_key, field_name in mapping.items():
            values[field_name] = raw.get(raw_key)

        if all(
            str(v or "").strip() == "" for v in values.values()
        ):
            continue  # blank line

        country = normalize_country_code(_clean(values.get("country"), limit=64))
        day = parse_date(values.get("day"))
        name_en = _clean(values.get("name_en"), limit=255)
        day_type = parse_day_type(values.get("day_type"))

        problems: list[str] = []
        if not country:
            problems.append(
                f"unrecognised country {str(values.get('country') or '').strip()!r}"
            )
        if day is None:
            problems.append(
                f"unreadable date {str(values.get('day') or '').strip()!r}"
            )
        elif not 2000 <= day.year <= 2100:
            problems.append(f"year {day.year} is outside 2000-2100")
        if not name_en:
            problems.append("holiday name is empty")
        if day_type is None:
            problems.append(
                f"unrecognised type {str(values.get('day_type') or '').strip()!r}"
            )

        if problems:
            report.rows.append(ParsedRow(
                line=line,
                status=_ROW_STATUS_ERROR,
                country=country,
                day=day,
                name_en=name_en,
                name_local=_clean(values.get("name_local"), limit=255),
                day_type=day_type or PUBLIC_HOLIDAY,
                note=_clean(values.get("note"), limit=500),
                group_key=_clean(values.get("group_key"), limit=96),
                message="; ".join(problems),
            ))
            continue

        assert country and day and name_en and day_type
        key = (country, day.isoformat(), name_key_for(name_en))
        duplicate = key in seen
        seen.add(key)
        report.rows.append(ParsedRow(
            line=line,
            status=_ROW_STATUS_DUPLICATE if duplicate else _ROW_STATUS_NEW,
            country=country,
            day=day,
            name_en=name_en,
            name_local=_clean(values.get("name_local"), limit=255),
            day_type=day_type,
            note=_clean(values.get("note"), limit=500),
            group_key=_clean(values.get("group_key"), limit=96),
            message="already in the calendar" if duplicate else None,
        ))
    return report


async def parse_upload(
    upload: UploadFile,
    *,
    existing_keys: set[tuple[str, str, str]] | None = None,
) -> ParseReport:
    """Read + classify an uploaded ``.xlsx``/``.csv``.

    ``read_bulk_rows`` enforces the shared upload contract (5 MB, 2000 rows, 100
    columns, MIME allow-list, zip-bomb preflight) and raises the platform's
    standard errors, so nothing here needs to re-check those.
    """
    raw_rows = await read_bulk_rows(upload)
    if len(raw_rows) > MAX_BULK_ROWS:
        from app.core.exceptions import ValidationError

        raise ValidationError(
            f"a holiday import may not exceed {MAX_BULK_ROWS} rows "
            f"(got {len(raw_rows)})",
        )
    report = parse_rows(raw_rows, existing_keys=existing_keys)
    for row in report.rows:
        if row.day is not None and row.status != _ROW_STATUS_ERROR:
            validate_year(row.day.year)
    return report