"""HTTP surface for the multi-country public-holiday calendar.

Mounted at ``/api/v1/hr/holidays``. Two audiences on one router:

* **HR maintenance** — reads gated by ``require_hr_read`` (so auditor and
  management can look without gaining write authority), writes by
  ``require_hr_admin`` plus the leaver runtime gate, rate limiting and audit.
* **Employee 公示** — the ``/public/...`` endpoints, open to any active
  employee. Holidays are company-wide published information, so there is no
  narrowing by manager scope; the only personalisation is which country is
  shown *first*.

Every write is audited. Nothing here pushes a notification: the notice
generator returns text for HR to copy, keeping the "no new automated pushes to
employees" rule intact.
"""
from __future__ import annotations

import uuid
from datetime import date, datetime
from typing import Annotated, Any

from fastapi import APIRouter, Depends, File, Query, Request, UploadFile
from sqlalchemy.ext.asyncio import AsyncSession

from app.audit.service import AuditService
from app.auth.models import AuthUser
from app.core.database import get_db
from app.core.dependencies import (
    check_sensitive_endpoint_rate_limit,
    get_current_user,
    require_active_employee,
    require_hr_admin,
    require_hr_read,
)
from app.core.dependencies import (
    require_active_employee_for_mutations as _leaver_runtime_gate,
)
from app.core.exceptions import ConflictError, NotFoundError, ValidationError
from app.core.exporting import xlsx_export_response
from app.core.response import success_response
from app.core.working_hours import resolve_timezone
from app.domains.hr import holiday_export as export
from app.domains.hr import holiday_import as importer
from app.domains.hr import holiday_service as svc
from app.domains.hr.holiday_models import (
    MAKEUP_WORKDAY,
    PUBLIC_HOLIDAY,
    HrPublicHoliday,
    HrPublicHolidayCalendar,
)
from app.domains.hr.holiday_notice import (
    NoticeRequest,
    notice_html,
    poster_from_text,
    render_notice,
)
from app.domains.hr.holiday_schemas import (
    AiClosingBody,
    AiParseBody,
    BulkBody,
    CalendarBody,
    DayPatchBody,
    DayRowBody,
    GroupBody,
    NoticeBody,
    NoticeHtmlBody,
    RollForwardBody,
    SuggestBody,
)

router = APIRouter(prefix="/api/v1/hr/holidays", tags=["hr-holidays"])

_AUDIT_RESOURCE = "hr.public_holiday"


def _trace_id(request: Request) -> str:
    return getattr(request.state, "trace_id", None) or str(uuid.uuid4())


async def _audit(
    request: Request,
    db: AsyncSession,
    user: AuthUser,
    action: str,
    resource_id: str,
    detail: dict[str, Any] | None = None,
) -> None:
    await AuditService(db).log(
        trace_id=_trace_id(request),
        actor_type="user",
        actor_user_id=user.id,
        action=action,
        status="success",
        resource_type=_AUDIT_RESOURCE,
        resource_id=resource_id,
        detail=detail,
    )


# ──────────────────────────── serializers ────────────────────────────


def _day_dict(row: HrPublicHoliday) -> dict[str, Any]:
    return {
        "id": row.id,
        "country": row.country,
        "day": row.day.isoformat(),
        "weekday": row.day.isoweekday(),
        "group_key": row.group_key,
        "name_en": row.name_en,
        "name_local": row.name_local,
        "day_type": row.day_type,
        "note": row.note,
        "source": row.source,
        "updated_at": row.updated_at.isoformat() if row.updated_at else None,
    }


def _group_dict(group: svc.HolidayGroup) -> dict[str, Any]:
    return {
        "country": group.country,
        "group_key": group.group_key,
        "name_en": group.name_en,
        "name_local": group.name_local,
        "note": group.note,
        "year": group.year,
        "start": group.start.isoformat(),
        "end": group.end.isoformat(),
        "days": [d.isoformat() for d in group.days],
        "makeup_days": [d.isoformat() for d in group.makeup_days],
        "total_days": group.total_days,
        "observed_days": group.observed_days,
        "segments": [
            {"start": s.isoformat(), "end": e.isoformat(), "days": (e - s).days + 1}
            for s, e in group.segments
        ],
        "sources": sorted(group.sources),
        "row_ids": group.row_ids,
    }


def _calendar_dict(row: HrPublicHolidayCalendar) -> dict[str, Any]:
    return {
        "country": row.country,
        "year": row.year,
        "label": row.label or row.country,
        "stated_days": row.stated_days,
        "gcal_url": row.gcal_url,
        "region_note": row.region_note,
        "source": row.source,
        "updated_at": row.updated_at.isoformat() if row.updated_at else None,
    }


def _parse_countries(raw: str | None) -> list[str] | None:
    """``"cn,jp"`` → ``["CN", "JP"]``; unrecognised codes are a 422, not silence."""
    if not raw:
        return None
    codes = [c for c in (part.strip() for part in raw.split(",")) if c]
    if not codes:
        return None
    # Deduplicate while keeping the caller's order stable for display.
    seen: set[str] = set()
    unique: list[str] = []
    for code in codes:
        resolved = svc.resolve_country(code, field_name="countries")
        if resolved not in seen:
            seen.add(resolved)
            unique.append(resolved)
    return unique


def _default_year(explicit: int | None) -> int:
    return svc.validate_year(explicit if explicit is not None else date.today().year)


# ──────────────────────────── HR: reads ────────────────────────────


@router.get("/years")
async def get_years(
    request: Request,
    db: AsyncSession = Depends(get_db),
    _gate: None = Depends(require_hr_read()),
):
    """Years the console may switch to.

    Deliberately the union of years that *have* data and the next two calendar
    years. Driving the switcher purely off stored data made next year
    unreachable — HR could not select 2027 to start building it, which is the one
    thing they do every December.
    """
    stored = await svc.available_years(db)
    current = date.today().year
    selectable = sorted({*stored, current, current + 1, current + 2})
    return success_response(
        data={
            "years": selectable,
            "years_with_data": stored,
            "current_year": current,
        },
        request=request,
    )


@router.get("/countries")
async def get_countries(
    request: Request,
    year: Annotated[int | None, Query(ge=svc.MIN_YEAR, le=svc.MAX_YEAR)] = None,
    db: AsyncSession = Depends(get_db),
    _gate: None = Depends(require_hr_read()),
):
    """The country roster, across every year unless one is named.

    The console needs this for its pickers, which must NOT be derived from the
    filtered table: doing that collapsed the filter's own option list to the one
    country already chosen, so a second could never be picked. Omitting ``year``
    also covers a year HR has just opened and not yet filled in.
    """
    # The roster spans every year (so a country stays offerable in a year HR has
    # only just opened), but the counts are for the year being looked at —
    # otherwise a freshly opened 2027 makes every country read "0 days", which is
    # true of 2027 and useless as a hint about the year on screen.
    roster = await svc.list_countries(db)
    counted = {
        str(row["country"]): row
        for row in (await svc.list_countries(db, year=year) if year else [])
    }
    seen: dict[str, dict[str, object]] = {}
    for row in sorted(roster, key=lambda r: -int(r["year"])):
        code = str(row["country"])
        if code in seen:
            continue
        for_year = counted.get(code)
        seen[code] = {
            "country": code,
            # Prefer the label from the year in view; fall back to the newest.
            "label": (for_year or row)["label"],
            "year": year if year is not None else row["year"],
            "day_count": for_year["day_count"] if for_year else 0,
            "stated_days": for_year["stated_days"] if for_year else None,
        }
    return success_response(
        data={"countries": list(seen.values())},
        request=request,
        meta={"total": len(seen)},
    )


@router.get("/calendars")
async def get_calendars(
    request: Request,
    year: Annotated[int | None, Query(ge=svc.MIN_YEAR, le=svc.MAX_YEAR)] = None,
    countries: Annotated[str | None, Query(max_length=512)] = None,
    db: AsyncSession = Depends(get_db),
    _gate: None = Depends(require_hr_read()),
):
    """Per-country roll-up for the console's overview table.

    Carries HR's ``stated_days`` next to the derived ``observed_days`` and an
    explicit match flag, so a divergence is visible in the UI without the
    frontend re-deriving the rule.
    """
    resolved_year = _default_year(year)
    summary = await svc.country_year_summary(
        db, year=resolved_year, countries=_parse_countries(countries),
    )
    return success_response(
        data={"year": resolved_year, "countries": summary},
        request=request,
        meta={"total": len(summary)},
    )


@router.get("/groups")
async def get_groups(
    request: Request,
    year: Annotated[int | None, Query(ge=svc.MIN_YEAR, le=svc.MAX_YEAR)] = None,
    countries: Annotated[str | None, Query(max_length=512)] = None,
    q: Annotated[str | None, Query(max_length=100)] = None,
    db: AsyncSession = Depends(get_db),
    _gate: None = Depends(require_hr_read()),
):
    """Holiday groups for a year — the console's primary table."""
    resolved_year = _default_year(year)
    groups = await svc.search_groups(
        db, year=resolved_year, countries=_parse_countries(countries), query=q,
    )
    return success_response(
        data={
            "year": resolved_year,
            "groups": [_group_dict(g) for g in groups],
        },
        request=request,
        meta={"total": len(groups)},
    )


@router.get("/days")
async def get_days(
    request: Request,
    year: Annotated[int | None, Query(ge=svc.MIN_YEAR, le=svc.MAX_YEAR)] = None,
    month: Annotated[int | None, Query(ge=1, le=12)] = None,
    countries: Annotated[str | None, Query(max_length=512)] = None,
    day_type: Annotated[str | None, Query(max_length=32)] = None,
    db: AsyncSession = Depends(get_db),
    _gate: None = Depends(require_hr_read()),
):
    """Day rows, plus a date-indexed matrix when ``month`` is given.

    The matrix is what the month grid renders; it is built from the same fetch
    rather than a second query per month.
    """
    resolved_year = _default_year(year)
    rows = await svc.list_days(
        db,
        year=resolved_year,
        countries=_parse_countries(countries),
        day_types=[svc.validate_day_type(day_type)] if day_type else None,
    )
    data: dict[str, Any] = {
        "year": resolved_year,
        "days": [_day_dict(r) for r in rows],
    }
    if month is not None:
        data["month"] = month
        data["matrix"] = svc.month_matrix(rows, year=resolved_year, month=month)
    return success_response(data=data, request=request, meta={"total": len(rows)})


# ──────────────────────────── HR: writes ────────────────────────────


@router.post("/groups", status_code=201)
async def post_group(
    request: Request,
    body: GroupBody,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _gate: None = Depends(require_hr_admin()),
    _leaver: None = Depends(_leaver_runtime_gate),
):
    """Create one holiday — all of its days off and makeup workdays at once."""
    group_key, rows = await svc.create_group(
        db,
        country=body.country,
        name_en=body.name_en,
        name_local=body.name_local,
        days=body.days,
        makeup_days=body.makeup_days,
        note=body.note,
        actor_id=str(user.id),
        group_key=body.group_key,
    )
    await _audit(
        request, db, user, "hr.public_holiday.group_create",
        f"{body.country}:{group_key}",
        {
            "country": body.country,
            "name_en": body.name_en,
            "days": [d.isoformat() for d in body.days],
            "makeup_days": [d.isoformat() for d in body.makeup_days],
        },
    )
    await db.commit()
    fresh = await svc.list_days(db, countries=[body.country], group_key=group_key)
    groups = svc.build_groups(fresh)
    return success_response(
        data=_group_dict(groups[0]) if groups else {"group_key": group_key},
        request=request,
        meta={"created_rows": len(rows)},
    )


@router.put("/groups/{country}/{group_key}")
async def put_group(
    request: Request,
    country: str,
    group_key: str,
    body: GroupBody,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _gate: None = Depends(require_hr_admin()),
    _leaver: None = Depends(_leaver_runtime_gate),
):
    """Replace a holiday's name and exact day set (diff-based, not delete+insert)."""
    resolved = svc.resolve_country(country)
    if body.country != resolved:
        raise ValidationError(
            "country in the path and the body must match "
            f"({resolved} vs {body.country})",
        )
    rows = await svc.replace_group(
        db,
        country=resolved,
        group_key=group_key,
        name_en=body.name_en,
        name_local=body.name_local,
        days=body.days,
        makeup_days=body.makeup_days,
        note=body.note,
        actor_id=str(user.id),
    )
    await _audit(
        request, db, user, "hr.public_holiday.group_update",
        f"{resolved}:{group_key}",
        {
            "name_en": body.name_en,
            "days": [d.isoformat() for d in body.days],
            "makeup_days": [d.isoformat() for d in body.makeup_days],
        },
    )
    await db.commit()
    groups = svc.build_groups(rows)
    return success_response(
        data=_group_dict(groups[0]) if groups else {"group_key": group_key},
        request=request,
    )


@router.delete("/groups/{country}/{group_key}")
async def remove_group(
    request: Request,
    country: str,
    group_key: str,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _gate: None = Depends(require_hr_admin()),
    _leaver: None = Depends(_leaver_runtime_gate),
):
    resolved = svc.resolve_country(country)
    removed = await svc.delete_group(db, country=resolved, group_key=group_key)
    await _audit(
        request, db, user, "hr.public_holiday.group_delete",
        f"{resolved}:{group_key}", {"rows_deleted": removed},
    )
    await db.commit()
    return success_response(
        data={"country": resolved, "group_key": group_key, "rows_deleted": removed},
        request=request,
    )


@router.post("/days", status_code=201)
async def post_day(
    request: Request,
    body: DayRowBody,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _gate: None = Depends(require_hr_admin()),
    _leaver: None = Depends(_leaver_runtime_gate),
):
    """Add a single day — for extending an existing holiday or a one-off."""
    result = await svc.bulk_upsert(
        db,
        rows=[
            svc.BulkRow(
                country=body.country,
                day=body.day,
                name_en=body.name_en,
                name_local=body.name_local,
                day_type=body.day_type,
                note=body.note,
                group_key=body.group_key,
            )
        ],
        source="manual",
        actor_id=str(user.id),
    )
    if result.created == 0:
        # A duplicate is a conflict, not bad input — and ``create_group`` already
        # raises ConflictError for exactly this case, so keep the two write paths
        # returning the same status for the same situation.
        raise ConflictError(
            f"{body.day.isoformat()} already carries a holiday named "
            f"{body.name_en!r} in {body.country}",
        )
    await _audit(
        request, db, user, "hr.public_holiday.day_create",
        f"{body.country}:{body.day.isoformat()}",
        {"name_en": body.name_en, "day_type": body.day_type},
    )
    await db.commit()
    rows = await svc.list_days(
        db, countries=[body.country], start=body.day, end=body.day,
    )
    match = next((r for r in rows if r.name_en == body.name_en), None)
    return success_response(
        data=_day_dict(match) if match else {"created": result.created},
        request=request,
    )


@router.put("/days/{day_id}")
async def patch_day(
    request: Request,
    day_id: int,
    body: DayPatchBody,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _gate: None = Depends(require_hr_admin()),
    _leaver: None = Depends(_leaver_runtime_gate),
):
    before = await svc.get_day(db, day_id=day_id)
    snapshot = {
        "day": before.day.isoformat(),
        "name_en": before.name_en,
        "day_type": before.day_type,
    }
    row = await svc.update_day(
        db,
        day_id=day_id,
        day=body.day,
        name_en=body.name_en,
        name_local=body.name_local,
        day_type=body.day_type,
        note=body.note,
        clear_local=body.clear_local,
        clear_note=body.clear_note,
        actor_id=str(user.id),
    )
    await _audit(
        request, db, user, "hr.public_holiday.day_update", str(day_id),
        {"before": snapshot, "after": {
            "day": row.day.isoformat(),
            "name_en": row.name_en,
            "day_type": row.day_type,
        }},
    )
    await db.commit()
    return success_response(data=_day_dict(row), request=request)


@router.delete("/days/{day_id}")
async def remove_day(
    request: Request,
    day_id: int,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _gate: None = Depends(require_hr_admin()),
    _leaver: None = Depends(_leaver_runtime_gate),
):
    row = await svc.delete_day(db, day_id=day_id)
    await _audit(
        request, db, user, "hr.public_holiday.day_delete", str(day_id),
        {"country": row.country, "day": row.day.isoformat(), "name_en": row.name_en},
    )
    await db.commit()
    return success_response(
        data={"deleted": day_id, "country": row.country, "day": row.day.isoformat()},
        request=request,
    )


@router.post("/bulk")
async def post_bulk(
    request: Request,
    body: BulkBody,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _gate: None = Depends(require_hr_admin()),
    _leaver: None = Depends(_leaver_runtime_gate),
):
    """Batch write — the commit half of the import flow and the AI-parse sink."""
    result = await svc.bulk_upsert(
        db,
        rows=[
            svc.BulkRow(
                country=r.country, day=r.day, name_en=r.name_en,
                name_local=r.name_local, day_type=r.day_type,
                note=r.note, group_key=r.group_key,
            )
            for r in body.rows
        ],
        source=body.source,
        actor_id=str(user.id),
        overwrite=body.overwrite,
    )
    await _audit(
        request, db, user, "hr.public_holiday.bulk_write", "bulk",
        {
            "submitted": len(body.rows),
            "created": result.created,
            "updated": result.updated,
            "skipped": result.skipped,
            "overwrite": body.overwrite,
            "source": body.source,
        },
    )
    await db.commit()
    return success_response(
        data={
            "created": result.created,
            "updated": result.updated,
            "skipped": result.skipped,
            "calendars_created": result.calendars_created,
            # Bounded: a 1000-row import must not return 1000 lines of prose.
            "details": result.details[:200],
            "details_truncated": len(result.details) > 200,
        },
        request=request,
    )


@router.put("/calendars/{country}/{year}")
async def put_calendar(
    request: Request,
    country: str,
    year: int,
    body: CalendarBody,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _gate: None = Depends(require_hr_admin()),
    _leaver: None = Depends(_leaver_runtime_gate),
):
    """Create or update a country-year's metadata (also creates the coverage row)."""
    row = await svc.upsert_calendar(
        db,
        country=country,
        year=year,
        label=body.label,
        stated_days=body.stated_days,
        gcal_url=body.gcal_url,
        region_note=body.region_note,
        clear_gcal=body.clear_gcal,
        clear_note=body.clear_note,
        clear_stated=body.clear_stated,
        actor_id=str(user.id),
    )
    await _audit(
        request, db, user, "hr.public_holiday.calendar_upsert",
        f"{row.country}:{row.year}",
        {"stated_days": row.stated_days, "has_gcal": bool(row.gcal_url)},
    )
    await db.commit()
    return success_response(data=_calendar_dict(row), request=request)


@router.delete("/calendars/{country}/{year}")
async def remove_calendar(
    request: Request,
    country: str,
    year: int,
    cascade: Annotated[bool, Query()] = False,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _gate: None = Depends(require_hr_admin()),
    _leaver: None = Depends(_leaver_runtime_gate),
):
    """Delete a country-year. Refuses to orphan days unless ``cascade`` is set."""
    result = await svc.delete_calendar(
        db, country=country, year=year, cascade=cascade,
    )
    resolved = svc.resolve_country(country)
    await _audit(
        request, db, user, "hr.public_holiday.calendar_delete",
        f"{resolved}:{year}", {"cascade": cascade, **result},
    )
    await db.commit()
    return success_response(
        data={"country": resolved, "year": year, **result}, request=request,
    )


@router.post("/restore-defaults")
async def restore_defaults(
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _gate: None = Depends(require_hr_admin()),
    _leaver: None = Depends(_leaver_runtime_gate),
):
    """Re-apply the shipped 2026 baseline, preserving every HR edit.

    Adds back what is missing and fills blank metadata; never overwrites a value
    HR has set. Idempotent — running it twice changes nothing the second time.
    """
    result = await svc.restore_seed(db, actor_id=str(user.id))
    await _audit(
        request, db, user, "hr.public_holiday.restore_defaults", "seed", result,
    )
    await db.commit()
    return success_response(data=result, request=request)


@router.post("/calendars/roll-forward")
async def roll_forward(
    request: Request,
    body: RollForwardBody,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _gate: None = Depends(require_hr_admin()),
    _leaver: None = Depends(_leaver_runtime_gate),
):
    """Open next year's calendar by copying this year's country roster.

    Copies the countries, their labels and their regional-holiday advisories —
    **not the dates**, and not the subscription links. Holidays move year to
    year, so a copied date would be a wrong date that looks authoritative; HR
    fills the dates in per country with the Excel import or the AI parse.
    Idempotent, so re-running only adds what is missing.
    """
    result = await svc.roll_forward_year(
        db,
        from_year=body.from_year,
        to_year=body.to_year,
        countries=body.countries,
        actor_id=str(user.id),
    )
    await _audit(
        request, db, user, "hr.public_holiday.roll_forward",
        f"{body.from_year}->{body.to_year}", result,
    )
    await db.commit()
    return success_response(data=result, request=request)


@router.post("/suggest")
async def suggest_from_dataset(
    request: Request,
    body: SuggestBody,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _gate: None = Depends(require_hr_admin()),
    _leaver: None = Depends(_leaver_runtime_gate),
):
    """Suggest one country-year's holidays from the offline statutory dataset.

    Writes nothing. Returns candidates on the same review-then-confirm path as
    the AI text parse, flagged against what is already stored, and HR posts the
    ones they accept to ``/bulk``. Statutory dates are a first draft, not the
    answer: PingCAP's calendar carries company choices no dataset knows, and an
    EOR provider's observed set can differ.
    """
    from app.domains.hr import holiday_source

    outcome = holiday_source.suggest(body.country, body.year)
    existing = {
        (row.day.isoformat(), row.name_key)
        for row in await svc.list_days(
            db, countries=[body.country], year=body.year,
        )
    }
    from app.domains.hr.holiday_keys import name_key_for

    rows = []
    for row in outcome.rows:
        duplicate = (row.day.isoformat(), name_key_for(row.name_en)) in existing
        rows.append({
            **row.to_payload(),
            "country": outcome.country,
            "status": "duplicate" if duplicate else "new",
        })
    payload = outcome.payload()
    payload["rows"] = rows
    payload["counts"] = {
        "new": sum(1 for r in rows if r["status"] == "new"),
        "duplicate": sum(1 for r in rows if r["status"] == "duplicate"),
        "holidays": sum(1 for r in rows if r["day_type"] == PUBLIC_HOLIDAY),
        "makeup": sum(1 for r in rows if r["day_type"] == MAKEUP_WORKDAY),
    }
    await _audit(
        request, db, user, "hr.public_holiday.suggest",
        f"{outcome.country}:{outcome.year}",
        {
            "supported": outcome.supported,
            "provider": outcome.provider,
            "provider_version": outcome.provider_version,
            "rows": len(rows),
        },
    )
    await db.commit()
    return success_response(data=payload, request=request)


# ──────────────────────────── import ────────────────────────────


@router.post("/import/preview")
async def preview_import(
    request: Request,
    file: Annotated[UploadFile, File()],
    db: AsyncSession = Depends(get_db),
    _gate: None = Depends(require_hr_admin()),
    _leaver: None = Depends(_leaver_runtime_gate),
):
    """Parse an uploaded sheet and classify every line. Writes nothing.

    HR reviews the result, then posts the importable rows to ``/bulk``. Keeping
    the two halves stateless means no server-side draft to expire, and the rows
    HR actually confirms are the rows that get written.
    """
    # Scoped to the years an import can plausibly target rather than the whole
    # table: this set exists only to mark duplicates, and reading every row of
    # every year to build it grows without bound.
    existing = await svc.duplicate_keys(db)
    report = await importer.parse_upload(file, existing_keys=existing)
    counts = report.counts()
    return success_response(
        data={
            "rows": [r.to_payload() for r in report.rows],
            "counts": counts,
            "unmapped_headers": report.unmapped_headers,
            "missing_required": report.missing_required,
            "importable": len(report.importable),
        },
        request=request,
        meta={"total": len(report.rows)},
    )


# ──────────────────────────── exports ────────────────────────────


async def _export_scope(
    db: AsyncSession, year: int, countries: str | None,
) -> tuple[list[HrPublicHoliday], list[HrPublicHolidayCalendar], list[str] | None]:
    codes = _parse_countries(countries)
    rows = await svc.list_days(db, year=year, countries=codes)
    calendars = await svc.list_calendars(db, year=year, countries=codes)
    return rows, calendars, codes


@router.get("/export.xlsx")
async def export_xlsx(
    request: Request,
    year: Annotated[int | None, Query(ge=svc.MIN_YEAR, le=svc.MAX_YEAR)] = None,
    countries: Annotated[str | None, Query(max_length=512)] = None,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _gate: None = Depends(require_hr_read()),
):
    """Flat one-row-per-day export. Its columns are the import parser's aliases,
    so an export can be edited and imported straight back."""
    resolved_year = _default_year(year)
    rows, calendars, codes = await _export_scope(db, resolved_year, countries)
    labels = {c.country: (c.label or c.country) for c in calendars}
    return await xlsx_export_response(
        db,
        request,
        user,
        columns=list(export.EXPORT_COLUMNS),
        rows=export.excel_rows(rows, labels),
        total_matched=len(rows),
        filename_stem=f"public_holidays_{resolved_year}",
        sheet_title=f"Holidays {resolved_year}",
        audit_action="hr.public_holiday.export_xlsx",
        audit_resource_type=_AUDIT_RESOURCE,
        filters={"year": resolved_year, "countries": codes or "all"},
    )


@router.get("/export.docx")
async def export_docx(
    request: Request,
    year: Annotated[int | None, Query(ge=svc.MIN_YEAR, le=svc.MAX_YEAR)] = None,
    countries: Annotated[str | None, Query(max_length=512)] = None,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _gate: None = Depends(require_hr_read()),
):
    """Rebuild HR's published Word document from the stored calendar."""
    resolved_year = _default_year(year)
    rows, calendars, codes = await _export_scope(db, resolved_year, countries)
    if not rows:
        raise NotFoundError(
            f"no holidays stored for {resolved_year}"
            + (f" in {', '.join(codes)}" if codes else ""),
        )
    return await export.docx_export_response(
        db, request, user,
        year=resolved_year, rows=rows, calendars=calendars, countries=codes,
    )


# ──────────────────────────── notice generator ────────────────────────────


@router.post("/notice")
async def build_notice(
    request: Request,
    body: NoticeBody,
    db: AsyncSession = Depends(get_db),
    _gate: None = Depends(require_hr_admin()),
    _leaver: None = Depends(_leaver_runtime_gate),
):
    """Render a copy-ready 放假通知 for one holiday.

    Returns text only — Helix never sends it. HR copies it into whichever channel
    they choose, which keeps the announcement under human control.

    Gated on ``require_hr_admin`` rather than ``require_hr_read``: this composes
    an all-hands announcement in the company's voice, which is an HR act, not a
    governance read. The read gate would have let auditor and management draft
    one, and it disagreed with the MCP twin, which was already HR-only.
    """
    rows = await svc.list_days(
        db, countries=[body.country], group_key=body.group_key,
    )
    if not rows:
        raise NotFoundError(
            f"holiday group {body.group_key!r} not found for {body.country}",
        )
    groups = svc.build_groups(rows)
    group = groups[0]
    if group.is_makeup_only:
        raise ValidationError(
            "this group has only makeup workdays, so there is no holiday to announce",
        )
    region_label = body.region_label
    if region_label is None:
        region_label = await _default_region_label(db, group)
    # The poster's footer names the region even for China, where the title
    # deliberately carries no qualifier — so it needs the calendar label
    # regardless of what the title uses.
    calendar = await svc.get_calendar(db, country=group.country, year=group.year)
    poster_region = (calendar.label if calendar else None) or group.country
    try:
        payload = render_notice(
            group,
            NoticeRequest(
                language=body.language,
                include_tips=body.include_tips,
                include_emoji=body.include_emoji,
                closing_zh=body.closing_zh,
                closing_en=body.closing_en,
                region_label=region_label,
            ),
            poster_region=poster_region,
        )
    except ValueError as exc:
        raise ValidationError(str(exc)) from exc
    return success_response(data=payload, request=request)


@router.post("/notice/html")
async def build_notice_html(
    request: Request,
    body: NoticeHtmlBody,
    _gate: None = Depends(require_hr_admin()),
    _leaver: None = Depends(_leaver_runtime_gate),
):
    """Convert notice text to the rich-text flavour the copy button pastes.

    The console's textarea is editable — HR routinely tweaks a sentence before
    sending — so the ``html`` returned with the original render goes stale the
    moment they type. The copy button therefore converts whatever is on screen
    at that moment, and it does it HERE rather than in the browser so there is
    exactly one implementation of the conversion (frontend stays presentation
    only). Pure function of the request body: no DB, no audit, nothing to
    make idempotent.
    """
    return success_response(
        data={
            "html": notice_html(body.text),
            # The poster is rebuilt from the SAME text, so the image the console
            # previews always matches what HR has on screen — including their
            # edits. `base` carries only what the text cannot express (the day
            # the calendar glyph shows, the country the skyline comes from, the
            # footer region).
            "poster": poster_from_text(body.text, dict(body.base or {})),
        },
        request=request,
    )


async def _default_region_label(
    db: AsyncSession, group: svc.HolidayGroup,
) -> str | None:
    """Qualify the audience for a non-China holiday.

    A Chinese notice for the China office needs no qualifier ("2026 年劳动节放假
    通知"). One about Japan's Marine Day does, or an all-hands reader cannot tell
    whose holiday it is.
    """
    if group.country == "CN":
        return None
    calendar = await svc.get_calendar(
        db, country=group.country, year=group.year,
    )
    return (calendar.label if calendar else None) or group.country


# ──────────────────────────── employee 公示 ────────────────────────────


@router.get("/public/countries")
async def public_countries(
    request: Request,
    year: Annotated[int | None, Query(ge=svc.MIN_YEAR, le=svc.MAX_YEAR)] = None,
    db: AsyncSession = Depends(get_db),
    user: AuthUser = Depends(require_active_employee),
):
    """Country picker for the employee page, with the reader's own country marked."""
    resolved_year = _default_year(year)
    countries = await svc.list_countries(db, year=resolved_year)
    mine, _person = await svc.resolve_person_country(db, person_id=str(user.id))
    return success_response(
        data={
            "year": resolved_year,
            "countries": countries,
            "my_country": mine,
            # False tells the UI to fall back to "all countries" rather than
            # showing an empty page for someone whose country is unmaintained.
            "my_country_covered": bool(
                mine and any(c["country"] == mine for c in countries)
            ),
        },
        request=request,
    )


@router.get("/public")
async def public_holidays(
    request: Request,
    year: Annotated[int | None, Query(ge=svc.MIN_YEAR, le=svc.MAX_YEAR)] = None,
    month: Annotated[int | None, Query(ge=1, le=12)] = None,
    scope: Annotated[str, Query(pattern="^(mine|all|custom)$")] = "mine",
    countries: Annotated[str | None, Query(max_length=512)] = None,
    db: AsyncSession = Depends(get_db),
    user: AuthUser = Depends(require_active_employee),
):
    """The employee 公示 payload: month grid, year table, and what is coming up.

    ``scope=mine`` resolves the reader's own country and silently widens to all
    countries when that country has no calendar — an employee in an unmaintained
    location gets the company-wide view instead of a blank page.
    """
    resolved_year = _default_year(year)
    mine, person = await svc.resolve_person_country(db, person_id=str(user.id))

    if scope == "custom":
        selected = _parse_countries(countries)
        if not selected:
            raise ValidationError("scope=custom requires at least one country")
    elif scope == "mine" and mine:
        selected = [mine]
    else:
        selected = None  # all maintained countries

    rows = await svc.list_days(db, year=resolved_year, countries=selected)
    if scope == "mine" and mine and not rows:
        # Unmaintained home country: widen rather than show nothing.
        selected = None
        rows = await svc.list_days(db, year=resolved_year, countries=None)

    calendars = await svc.list_calendars(
        db, year=resolved_year, countries=selected,
    )
    today = datetime.now(resolve_timezone(person)).date() if person else date.today()
    upcoming = await svc.upcoming_groups(
        db,
        countries=selected or [c.country for c in calendars],
        today=today,
        limit=3,
    )
    # Whole groups, so a cross-year holiday does not report a truncated day
    # count here while the notice HR sends reports the real one.
    groups = await svc.complete_groups(db, rows)

    data: dict[str, Any] = {
        "year": resolved_year,
        "today": today.isoformat(),
        "scope": scope,
        "my_country": mine,
        "countries": [_calendar_dict(c) for c in calendars],
        "selected_countries": selected or [c.country for c in calendars],
        "groups": [_group_dict(g) for g in groups],
        "upcoming": [
            {**_group_dict(g), "days_until": (g.start - today).days}
            for g in upcoming
        ],
    }
    if month is not None:
        data["month"] = month
        data["matrix"] = svc.month_matrix(rows, year=resolved_year, month=month)
    return success_response(
        data=data, request=request, meta={"total": len(groups)},
    )


@router.get("/public/upcoming")
async def public_upcoming(
    request: Request,
    limit: Annotated[int, Query(ge=1, le=10)] = 3,
    scope: Annotated[str, Query(pattern="^(mine|all)$")] = "mine",
    db: AsyncSession = Depends(get_db),
    user: AuthUser = Depends(require_active_employee),
):
    """Just the "next holiday" card — cheap enough for the feed sidebar.

    ``today`` is the reader's *local* date, so someone in Singapore does not see
    yesterday's holiday because the server is on UTC.
    """
    mine, person = await svc.resolve_person_country(db, person_id=str(user.id))
    today = datetime.now(resolve_timezone(person)).date() if person else date.today()
    if scope == "mine" and mine:
        selected = [mine]
    else:
        selected = [str(c["country"]) for c in await svc.list_countries(db)]
    groups = await svc.upcoming_groups(
        db, countries=selected, today=today, limit=limit,
    )
    if scope == "mine" and mine and not groups:
        selected = [str(c["country"]) for c in await svc.list_countries(db)]
        groups = await svc.upcoming_groups(
            db, countries=selected, today=today, limit=limit,
        )
    return success_response(
        data={
            "today": today.isoformat(),
            "my_country": mine,
            "scope": scope,
            "upcoming": [
                {
                    "country": g.country,
                    "group_key": g.group_key,
                    "name_en": g.name_en,
                    "name_local": g.name_local,
                    "start": g.start.isoformat(),
                    "end": g.end.isoformat(),
                    "total_days": g.total_days,
                    "days_until": (g.start - today).days,
                    "in_progress": g.start <= today <= g.end,
                    "makeup_days": [d.isoformat() for d in g.makeup_days],
                }
                for g in groups
            ],
        },
        request=request,
    )


@router.get("/public/day")
async def public_day(
    request: Request,
    day: Annotated[date, Query()],
    country: Annotated[str | None, Query(max_length=64)] = None,
    db: AsyncSession = Depends(get_db),
    user: AuthUser = Depends(require_active_employee),
):
    """Is this specific date a holiday? Defaults to the reader's own country."""
    resolved: str | None
    if country:
        resolved = svc.resolve_country(country)
    else:
        resolved, _person = await svc.resolve_person_country(
            db, person_id=str(user.id),
        )
    if not resolved:
        raise ValidationError(
            "no country could be resolved for you — pass ?country= explicitly",
        )
    rows = await svc.day_is_holiday(db, country=resolved, day=day)
    cal = await svc.get_calendar(db, country=resolved, year=day.year)
    holidays = [r for r in rows if r.day_type == PUBLIC_HOLIDAY]
    makeups = [r for r in rows if r.day_type == MAKEUP_WORKDAY]
    return success_response(
        data={
            "country": resolved,
            "day": day.isoformat(),
            "coverage": "covered" if cal else "no_data",
            "is_public_holiday": bool(holidays),
            "is_makeup_workday": bool(makeups),
            "entries": [_day_dict(r) for r in rows],
        },
        request=request,
    )

# ──────────────────────────── AI assistance ────────────────────────────


@router.post("/ai/parse")
async def ai_parse(
    request: Request,
    body: AiParseBody,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _gate: None = Depends(require_hr_admin()),
    _leaver: None = Depends(_leaver_runtime_gate),
    # The sensitive-action limiter stays HERE and only here in this router: an
    # LLM call costs real money per request, so 10/hour/user is a sane guard.
    #
    # It used to be on all 16 holiday endpoints, which was wrong twice over.
    # Holiday data is company-published and non-personal, and HR has full
    # authority over it — reading or editing it is not a PII reveal. Worse, the
    # limiter's budget is GLOBAL per user: the notice dialog re-renders on every
    # language / tips / emoji toggle, so clicking through a few holidays burnt
    # the user's entire hourly allowance for PII reveal and exports across the
    # whole product. Reported from prod as "随便点点就 Too many requests".
    _rl: None = Depends(check_sensitive_endpoint_rate_limit),
):
    """Turn pasted holiday text into candidate rows for HR to confirm.

    Writes nothing. HR reviews the candidates and posts the ones they accept to
    ``/bulk``, which applies the same validation, conflict detection and audit as
    a manual entry — so the AI path has no privilege the manual path lacks, and a
    hallucinated date stays a visible suggestion instead of becoming a row.

    Degrades rather than dead-ends: with no LLM key configured (or on any model
    failure) a deterministic text scan runs instead, and the response says which
    engine produced the rows. One model call at the client's 90s default, well
    inside the frontend's ``LLM_SYNC_TIMEOUT_MS`` budget.
    """
    from app.domains.hr import holiday_ai_service as ai

    existing = await svc.duplicate_keys(
        db, years=[body.year] if body.year else None,
    )
    outcome = await ai.parse_text(
        db,
        text=body.text,
        country_hint=body.country,
        year_hint=body.year,
        existing_keys=existing,
    )
    await _audit(
        request, db, user, "hr.public_holiday.ai_parse", "ai",
        {
            "engine": outcome.engine,
            "chars": len(body.text or ""),
            "country_hint": body.country,
            "year_hint": body.year,
            "rows_returned": len(outcome.rows),
            "dropped": len(outcome.dropped),
        },
    )
    await db.commit()
    return success_response(data=outcome.payload(), request=request)


@router.post("/notice/closing")
async def ai_closing(
    request: Request,
    body: AiClosingBody,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _gate: None = Depends(require_hr_admin()),
    _leaver: None = Depends(_leaver_runtime_gate),
    _rl: None = Depends(check_sensitive_endpoint_rate_limit),
):
    """Suggest an alternative closing sentence for one holiday's notice.

    The closing line is the one part of the notice whose job is to not read like
    a template, so it is the one part worth asking a model for. Falls back to the
    seasonal default when no key is configured or the answer ignores the brief.
    """
    from app.domains.hr import holiday_ai_service as ai

    rows = await svc.list_days(
        db, countries=[body.country], group_key=body.group_key,
    )
    if not rows:
        raise NotFoundError(
            f"holiday group {body.group_key!r} not found for {body.country}",
        )
    group = svc.build_groups(rows)[0]
    if group.is_makeup_only:
        raise ValidationError(
            "this group has only makeup workdays, so there is no notice to close",
        )
    result = await ai.suggest_closing(db, group=group, language=body.language)
    # Audited like /ai/parse: this is the one endpoint that hits the model on
    # every click, so AI spend and prompt attribution need a trail.
    await _audit(
        request, db, user, "hr.public_holiday.ai_closing",
        f"{body.country}:{body.group_key}",
        {"engine": result.get("engine"), "language": body.language},
    )
    await db.commit()
    return success_response(data=result, request=request)