"""AI assistance for maintaining the public-holiday calendar — via 百炼 (Qwen).

Two jobs, both of which turn a chore into a review:

* :func:`parse_text` — HR pastes whatever they were sent (a government notice, an
  EOR platform email, a list copied out of a PDF) and gets back structured
  ``(country, date, holiday name)`` rows to check and confirm. Typing 33 China
  dates by hand is exactly the work a model should do.
* :func:`suggest_closing` — rewrites the one line of the 放假通知 whose whole job
  is to *not* read like a template, so a Mid-Autumn notice does not close on the
  same sentence as a Labour Day one.

**The model never writes to the database.** ``parse_text`` returns candidate rows;
HR confirms them in the console, which posts them through the ordinary
``/bulk`` endpoint with its ordinary validation, conflict detection and audit.
That keeps a hallucinated date a visible suggestion rather than a silent row, and
means the AI path has no privilege the manual path lacks.

Both degrade gracefully. With no LLM key configured, ``parse_text`` falls back to
:mod:`app.domains.hr.holiday_ai_service._deterministic_parse` — a real regex
extractor that handles the formats these notices actually use — and
``suggest_closing`` falls back to the seasonal library in
:mod:`app.domains.hr.holiday_notice`. Neither returns a dead end.
"""
from __future__ import annotations

import json
import re
from dataclasses import dataclass, field
from datetime import date

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.geo import normalize_country_code
from app.core.safe_logging import get_safe_logger
from app.domains.hr.holiday_import import parse_date
from app.domains.hr.holiday_models import (
    MAKEUP_WORKDAY,
    PUBLIC_HOLIDAY,
    name_key_for,
)
from app.domains.hr.holiday_notice import EN, ZH, default_closing
from app.domains.hr.holiday_service import HolidayGroup
from app.integrations.models import PlatformSetting

logger = get_safe_logger(__name__)

# The platform LLM credential the rest of the AI features already use.
_KEY_API = "ai_spending.llm.api_key"
_KEY_MODEL = "ai_spending.llm.model"
_KEY_REGION = "ai_spending.llm.region"

#: Cap on pasted text. Long enough for a full multi-country year (HR's own 2026
#: document is ~12k characters of table text), short enough to bound the prompt.
MAX_INPUT_CHARS = 20_000
#: Cap on rows returned from one parse. Half the bulk-write ceiling on purpose:
#: a parse HR has to review row by row should not be able to fill a whole batch.
MAX_PARSED_ROWS = 500

_SYSTEM_PARSE = (
    "You extract public-holiday data from text an HR team was sent (government "
    "notices, EOR platform emails, spreadsheets pasted as text, in any "
    "language).\n"
    "\n"
    "Return ONLY a JSON object of the form:\n"
    '{"rows": [{"country": "CN", "date": "2026-05-01", '
    '"name_en": "Labor Day Holiday", "name_local": "劳动节", '
    '"day_type": "public_holiday"}], "notes": "..."}\n'
    "\n"
    "Rules:\n"
    "- One object PER CALENDAR DATE. A holiday spanning 2026-05-01 to 2026-05-05 "
    "is five objects with the same name, not one object with a range.\n"
    "- country: ISO-3166 alpha-2 (CN, JP, HK, TW, US, GB, AE...). If the text "
    "never says which country, use the country hint given by the user; if there "
    "is no hint either, omit the row.\n"
    "- date: strict YYYY-MM-DD. Resolve relative or partial dates using the year "
    "stated in the text. NEVER guess a date you cannot read — omit the row and "
    "say so in notes.\n"
    "- name_en: the holiday's English name. Translate if the source is not "
    "English. Keep an '(observed)' marker if the source has one.\n"
    "- name_local: the name in the local language when the source gives one, "
    "else null.\n"
    '- day_type: "public_holiday" for a day off. Use "makeup_workday" for a '
    "weekend day designated a WORKING day to offset a long holiday (Chinese "
    "调休/补班). These are workdays, not holidays — classifying one as a holiday "
    "is a serious error.\n"
    "- notes: anything you could not parse, or an assumption you had to make. "
    "Keep it under 300 characters. Empty string if nothing to report.\n"
    "\n"
    "Do not invent holidays that are not in the text. Fewer correct rows beats "
    "more guessed ones."
)

_SYSTEM_CLOSING = (
    "You write the single closing sentence of a company holiday announcement — "
    "the warm sign-off after the dates and the safety reminders.\n"
    "\n"
    "Constraints:\n"
    "- ONE or TWO short sentences. No greeting, no bullet points, no emoji, no "
    "quotation marks around your answer.\n"
    "- Fit the specific holiday and the season it falls in. A Mid-Autumn line "
    "should not read like a Labour Day line.\n"
    "- Warm and human, but restrained. This goes to the whole company from HR: "
    "no exclamation-stacking, no corporate cheer, no 'team' clichés, no "
    "flattery about how hard everyone works beyond a brief acknowledgement.\n"
    "- End by looking ahead to being back at work.\n"
    "- Output the sentence and nothing else."
)


# ─────────────────────────── LLM plumbing ───────────────────────────


async def _setting(db: AsyncSession, key: str) -> str | None:
    return (
        await db.execute(
            select(PlatformSetting.value).where(PlatformSetting.key == key)
        )
    ).scalar_one_or_none()


async def llm_client(db: AsyncSession):
    """The shared 百炼 client, or ``None`` when no key is configured."""
    api_key = (await _setting(db, _KEY_API) or "").strip()
    if not api_key:
        return None
    from app.domains.bailian.llm_client import BailianLLMClient

    return BailianLLMClient(
        api_key=api_key,
        model=await _setting(db, _KEY_MODEL),
        region=await _setting(db, _KEY_REGION), purpose="hr.holiday_notice")


# ─────────────────────────── parse ───────────────────────────


@dataclass
class ParsedCandidate:
    """One row the model (or the fallback) believes it found."""

    country: str
    day: date
    name_en: str
    name_local: str | None = None
    day_type: str = PUBLIC_HOLIDAY
    # Set when the row duplicates something already in the calendar.
    duplicate: bool = False

    def to_payload(self) -> dict[str, object]:
        return {
            "country": self.country,
            "day": self.day.isoformat(),
            "name_en": self.name_en,
            "name_local": self.name_local,
            "day_type": self.day_type,
            "status": "duplicate" if self.duplicate else "new",
        }


@dataclass
class ParseOutcome:
    rows: list[ParsedCandidate] = field(default_factory=list)
    notes: str = ""
    # "llm" or "deterministic" — shown in the console so HR knows which ran.
    engine: str = "llm"
    dropped: list[str] = field(default_factory=list)

    def payload(self) -> dict[str, object]:
        return {
            "engine": self.engine,
            "notes": self.notes,
            "rows": [r.to_payload() for r in self.rows],
            "dropped": self.dropped[:50],
            "counts": {
                "new": sum(1 for r in self.rows if not r.duplicate),
                "duplicate": sum(1 for r in self.rows if r.duplicate),
                "dropped": len(self.dropped),
            },
        }


def _coerce_rows(
    raw_rows: object, *, country_hint: str | None, dropped: list[str],
) -> list[ParsedCandidate]:
    """Validate the model's output row by row, dropping what cannot be trusted.

    Everything the model returns is treated as a claim to check, not data:
    a country that does not resolve, a date that will not parse, a missing name
    or an unknown ``day_type`` drops the row with a reason HR can read. A model
    that hallucinated a date should cost HR one visible dropped line, not a wrong
    holiday in the calendar.
    """
    if not isinstance(raw_rows, list):
        dropped.append("model did not return a row list")
        return []

    out: list[ParsedCandidate] = []
    for index, item in enumerate(raw_rows[:MAX_PARSED_ROWS], start=1):
        if not isinstance(item, dict):
            dropped.append(f"row {index}: not an object")
            continue
        country = normalize_country_code(
            str(item.get("country") or "").strip()
        ) or country_hint
        day = parse_date(item.get("date") or item.get("day"))
        name_en = str(item.get("name_en") or item.get("name") or "").strip()[:255]
        raw_type = str(item.get("day_type") or PUBLIC_HOLIDAY).strip()
        day_type = raw_type if raw_type in (PUBLIC_HOLIDAY, MAKEUP_WORKDAY) else None

        label = f"row {index} ({item.get('date') or '?'} {name_en or '?'})"
        if not country:
            dropped.append(f"{label}: no resolvable country")
            continue
        if day is None:
            dropped.append(f"{label}: unreadable date")
            continue
        if not 2000 <= day.year <= 2100:
            dropped.append(f"{label}: year {day.year} out of range")
            continue
        if not name_en:
            dropped.append(f"{label}: no holiday name")
            continue
        if day_type is None:
            dropped.append(f"{label}: unknown day_type {raw_type!r}")
            continue

        local_raw = item.get("name_local")
        name_local = str(local_raw).strip()[:255] if local_raw else None
        out.append(ParsedCandidate(
            country=country,
            day=day,
            name_en=name_en,
            name_local=name_local or None,
            day_type=day_type,
        ))
    if isinstance(raw_rows, list) and len(raw_rows) > MAX_PARSED_ROWS:
        dropped.append(
            f"only the first {MAX_PARSED_ROWS} rows were kept "
            f"({len(raw_rows)} returned)"
        )
    return out


# ── deterministic fallback ──

_MONTH_NAMES = (
    "January", "February", "March", "April", "May", "June", "July",
    "August", "September", "October", "November", "December",
)
_MONTHS = {m.lower(): i for i, m in enumerate(_MONTH_NAMES, start=1)}
_MONTHS.update({m[:3]: i for m, i in list(_MONTHS.items())})
_MONTH_ALT = "|".join(
    sorted({*_MONTH_NAMES, *(m[:3] for m in _MONTH_NAMES)}, key=len, reverse=True)
)

# "2026-05-01 Labor Day" / "May 1, 2026 — Labor Day" / "2026年5月1日 劳动节"
#
# All three are anchored on literals with bounded whitespace runs. The long form
# matches on the actual month names rather than a generic letter run followed by
# an optional letter-run prefix: that earlier shape let both groups consume the
# same letters, and a line of 8,000 letters cost ~400 ms of backtracking
# (CodeQL py/polynomial-redos). Text here is pasted by a human, up to
# MAX_INPUT_CHARS, so quadratic behaviour is a real denial of service.
_LINE_ISO = re.compile(r"(\d{4})[-/.](\d{1,2})[-/.](\d{1,2})")
_LINE_CJK = 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}日?"
)
_LINE_LONG = re.compile(
    rf"\b({_MONTH_ALT})\b\.?[ \t]{{0,3}}(\d{{1,2}})[ \t]{{0,3}},?[ \t]{{0,3}}(\d{{4}})",
    re.IGNORECASE,
)
_MAKEUP_HINT = re.compile(
    r"补班|调休|makeup|working day|work day", re.IGNORECASE,
)


def _deterministic_parse(
    text: str, *, country_hint: str | None, year_hint: int | None,
) -> ParseOutcome:
    """Line-oriented regex extraction, for when no LLM key is configured.

    Deliberately conservative: one date per line, the rest of the line as the
    name. It handles the shapes these notices actually arrive in (ISO, "May 1,
    2026", "2026年5月1日") and reports what it skipped, so HR gets a working
    importer rather than an empty panel and a shrug.
    """
    outcome = ParseOutcome(engine="deterministic")
    if not country_hint:
        outcome.notes = (
            "No LLM key is configured, so a plain text scan ran instead. It "
            "needs a country — pick one and try again."
        )
        return outcome

    seen: set[tuple[str, str]] = set()
    for raw_line in text.splitlines():
        line = " ".join(raw_line.split())
        if not line:
            continue
        day: date | None = None
        matched = ""
        for pattern in (_LINE_ISO, _LINE_CJK):
            m = pattern.search(line)
            if m:
                try:
                    day = date(int(m.group(1)), int(m.group(2)), int(m.group(3)))
                except ValueError:
                    day = None
                matched = m.group(0)
                break
        if day is None:
            m = _LINE_LONG.search(line)
            if m and m.group(1).lower() in _MONTHS:
                try:
                    day = date(
                        int(m.group(3)), _MONTHS[m.group(1).lower()], int(m.group(2)),
                    )
                except ValueError:
                    day = None
                matched = m.group(0) if day else ""
        if day is None:
            if matched:
                # A line that LOOKED like a date but is not one (2026-02-30) was
                # dropped in silence, contradicting the promise to report skips.
                outcome.dropped.append(f"{line[:60]}: not a real date")
            continue
        if year_hint and day.year != year_hint:
            outcome.dropped.append(f"{line[:60]}: not in {year_hint}")
            continue

        name = line.replace(matched, " ")
        # Strip the leftover separators and weekday words a table line carries.
        name = re.sub(
            r"\b(mon|tues|wednes|thurs|fri|satur|sun)day\b", " ", name,
            flags=re.IGNORECASE,
        )
        name = re.sub(r"[|\t]+", " ", name)
        name = " ".join(name.strip(" -–—:·,;()").split())
        if not name:
            outcome.dropped.append(f"{line[:60]}: date with no holiday name")
            continue

        day_type = MAKEUP_WORKDAY if _MAKEUP_HINT.search(line) else PUBLIC_HOLIDAY
        key = (day.isoformat(), name_key_for(name))
        if key in seen:
            continue
        if len(outcome.rows) >= MAX_PARSED_ROWS:
            # Same cap as the LLM path. Without it a 20,000-character paste
            # yielded ~1600 candidates that HR would review and select, only for
            # /bulk to reject the batch on its own 1000-row limit — wasted work
            # ending in a dead end.
            outcome.dropped.append(
                f"only the first {MAX_PARSED_ROWS} rows were kept "
                f"(the text contains more)"
            )
            break
        seen.add(key)
        has_cjk = any("一" <= ch <= "鿿" for ch in name)
        outcome.rows.append(ParsedCandidate(
            country=country_hint,
            day=day,
            name_en=name[:255],
            name_local=name[:255] if has_cjk else None,
            day_type=day_type,
        ))

    outcome.notes = (
        "No LLM key is configured, so a plain text scan ran instead: one date "
        "per line, the rest of the line as the holiday name. Check the names "
        "before importing."
    )
    return outcome


async def parse_text(
    db: AsyncSession,
    *,
    text: str,
    country_hint: str | None = None,
    year_hint: int | None = None,
    existing_keys: set[tuple[str, str, str]] | None = None,
) -> ParseOutcome:
    """Turn pasted holiday text into candidate rows for HR to confirm.

    ``existing_keys`` holds ``(country, iso date, name_key)`` triples already
    stored; matches are flagged ``duplicate`` rather than dropped, so HR can see
    that a pasted notice is mostly already in the calendar.
    """
    body = (text or "").strip()
    if not body:
        return ParseOutcome(engine="none", notes="Nothing to parse.")
    body = body[:MAX_INPUT_CHARS]
    hint = normalize_country_code(country_hint) if country_hint else None

    client = await llm_client(db)
    if client is None:
        outcome = _deterministic_parse(
            body, country_hint=hint, year_hint=year_hint,
        )
    else:
        prompt_parts = [
            "Extract every public holiday and makeup workday from the text below.",
        ]
        if hint:
            prompt_parts.append(
                f"Country hint (use when the text does not say): {hint}"
            )
        if year_hint:
            prompt_parts.append(f"Year hint: {year_hint}")
        prompt_parts.append("--- TEXT ---")
        prompt_parts.append(body)
        try:
            raw = await client.chat_json(
                system=_SYSTEM_PARSE,
                user="\n".join(prompt_parts),
                max_tokens=8000,
                temperature=0.0,
            )
        except Exception as exc:  # noqa: BLE001 — any LLM failure degrades
            logger.warning("holiday AI parse failed, falling back: %s", exc)
            outcome = _deterministic_parse(
                body, country_hint=hint, year_hint=year_hint,
            )
            outcome.notes = (
                "The AI parse failed, so a plain text scan ran instead. "
                + outcome.notes
            )
        else:
            payload = raw if isinstance(raw, dict) else _loads(raw)
            outcome = ParseOutcome(engine="llm")
            outcome.rows = _coerce_rows(
                payload.get("rows") if isinstance(payload, dict) else None,
                country_hint=hint,
                dropped=outcome.dropped,
            )
            note = payload.get("notes") if isinstance(payload, dict) else None
            outcome.notes = str(note or "")[:300]
            if not outcome.rows:
                # An empty LLM result is not proof the text has no holidays;
                # try the regex scan before telling HR there is nothing here.
                fallback = _deterministic_parse(
                    body, country_hint=hint, year_hint=year_hint,
                )
                if fallback.rows:
                    fallback.notes = (
                        "The AI parse found nothing usable, so a plain text scan "
                        "ran instead. " + fallback.notes
                    )
                    outcome = fallback

    if year_hint:
        kept = [r for r in outcome.rows if r.day.year == year_hint]
        for row in outcome.rows:
            if row.day.year != year_hint:
                outcome.dropped.append(
                    f"{row.day.isoformat()} {row.name_en}: not in {year_hint}"
                )
        outcome.rows = kept

    if existing_keys:
        for row in outcome.rows:
            key = (row.country, row.day.isoformat(), name_key_for(row.name_en))
            row.duplicate = key in existing_keys

    # Same-date-same-name rows within one parse are a model slip, not two
    # holidays; keep the first.
    unique: list[ParsedCandidate] = []
    seen: set[tuple[str, str, str]] = set()
    for row in outcome.rows:
        key = (row.country, row.day.isoformat(), name_key_for(row.name_en))
        if key in seen:
            continue
        seen.add(key)
        unique.append(row)
    outcome.rows = sorted(unique, key=lambda r: (r.country, r.day, r.name_en))
    return outcome


def _loads(raw: object) -> dict:
    if isinstance(raw, dict):
        return raw
    try:
        parsed = json.loads(str(raw))
    except (TypeError, ValueError):
        return {}
    return parsed if isinstance(parsed, dict) else {}


# ─────────────────────────── closing line ───────────────────────────


async def suggest_closing(
    db: AsyncSession, *, group: HolidayGroup, language: str = ZH,
) -> dict[str, object]:
    """Propose the notice's closing sentence, falling back to the seasonal one."""
    lang = EN if language == EN else ZH
    fallback = default_closing(group, lang)

    client = await llm_client(db)
    if client is None:
        return {
            "closing": fallback,
            "engine": "deterministic",
            "note": "No LLM key is configured; this is the seasonal default.",
        }

    name = group.name_local or group.name_en if lang == ZH else group.name_en
    user = "\n".join([
        f"Holiday: {name}"
        + (f" ({group.name_en})" if lang == ZH and group.name_local else ""),
        f"Country/region: {group.country}",
        f"Dates: {group.start.isoformat()} to {group.end.isoformat()} "
        f"({group.total_days} calendar days off)",
        (
            f"Makeup workdays: "
            f"{', '.join(d.isoformat() for d in group.makeup_days)}"
            if group.makeup_days else "Makeup workdays: none"
        ),
        f"Write the sentence in {'Simplified Chinese' if lang == ZH else 'English'}.",
    ])
    try:
        text = await client.chat_text(
            system=_SYSTEM_CLOSING, user=user, max_tokens=300, temperature=0.8,
        )
    except Exception as exc:  # noqa: BLE001
        logger.warning("holiday closing suggestion failed: %s", exc)
        return {
            "closing": fallback,
            "engine": "deterministic",
            "note": "The AI call failed; this is the seasonal default.",
        }

    cleaned = _clean_closing(text)
    if not cleaned:
        return {
            "closing": fallback,
            "engine": "deterministic",
            "note": "The AI returned nothing usable; this is the seasonal default.",
        }
    return {"closing": cleaned, "engine": "llm", "note": ""}


#: A closing line is one or two sentences. Anything longer is the model
#: ignoring the brief, and truncating mid-sentence would look worse than
#: falling back, so the caller treats an over-long answer as unusable.
_MAX_CLOSING_CHARS = 220


def _clean_closing(raw: object) -> str:
    """Strip the wrappers a model adds, and reject an answer that ignored the brief."""
    text = str(raw or "").strip()
    if not text:
        return ""
    # Truncate BEFORE the strip patterns run. An over-long reply is rejected
    # below anyway, and scanning an unbounded one for a trailing fence is
    # needless work on input the model controls.
    text = text[: _MAX_CLOSING_CHARS * 4]
    # Models like to fence or quote a single-sentence answer.
    text = re.sub(r"^```[a-zA-Z]*[ \t]{0,4}", "", text)
    text = re.sub(r"[ \t]{0,4}```$", "", text).strip()
    if len(text) >= 2 and text[0] in "\"'“‘「『" and text[-1] in "\"'”’」』":
        text = text[1:-1].strip()
    # Collapse to a single line; a multi-paragraph answer is not a closing line.
    text = " ".join(text.split())
    if len(text) > _MAX_CLOSING_CHARS:
        return ""
    return text