"""One-shot backfill of historical celebration rows + system posts.

For every active employee, INSERT ``feed_celebrations`` + matching
``feed_posts`` (``system_birthday`` / ``system_anniversary``) for:

* Every birthday between hire_date and (target_date - 1 day). Birthdays
  that fell BEFORE the person joined are skipped (the user explicitly
  asked for 入职后的生日庆祝 only).
* Every hire-date anniversary between hire_date+1y and (target_date -
  1 day) — year ≥ 1, so the joining day itself is never marked as an
  "anniversary".

Today's celebrations are NOT written by this script — the hourly cron
``feed.scan_today_celebrations`` already owns that path. We backfill
historical-only so a re-run after a deploy doesn't fight the cron.

**NO notifications are written**. The user is explicit: 不能打扰到员工.
The hourly notifier cron only picks up rows where ``occasion_date ==
today`` so historical rows are naturally immune; and we never write
``notification_outbox`` directly.

**Aggregate banners are NOT backfilled**. A "5 people had their
anniversary today" banner dated to 2023-08-14 clutters the timeline
without adding signal — the per-person system posts are the
load-bearing visible item.

**Privacy preserved**. Birthdays are skipped when the person's
``identity_person_preferences['feed.birthday_visibility']`` is
``'hidden'`` (same rule the live cron uses). Anniversaries are always
backfilled — no opt-out exists for those (matches the feature spec).

Idempotent via the UNIQUE constraint
``(person_id, occasion_date, celebration_type)`` on
``feed_celebrations`` — re-runs insert 0 new rows. Dry-run by default;
pass ``--apply`` to write.

Run::

    # dry-run: print counts, write nothing
    uv run python scripts/backfill_feed_celebrations_2026_06_02.py

    # actually write
    uv run python scripts/backfill_feed_celebrations_2026_06_02.py --apply

    # narrow to a single person for testing
    uv run python scripts/backfill_feed_celebrations_2026_06_02.py \\
        --person-id <uuid> --apply

    # backfill up to a specific historical date (default = today)
    uv run python scripts/backfill_feed_celebrations_2026_06_02.py \\
        --through 2026-05-31 --apply
"""
from __future__ import annotations

import argparse
import asyncio
import hashlib
import logging
import sys
import uuid
from collections.abc import Iterable
from datetime import UTC, date, datetime, timedelta
from zoneinfo import ZoneInfo

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

from app.audit.service import AuditService
from app.core.database import async_session_factory
from app.core.safe_logging import get_safe_logger
from app.domains.feed.celebration_service import (
    ANNIVERSARY_ILLUSTRATION_KEYS,
    BIRTHDAY_ILLUSTRATION_KEYS,
)
from app.domains.feed.models import (
    BIRTHDAY_VISIBILITY_HIDDEN,
    CELEBRATION_TYPE_ANNIVERSARY,
    CELEBRATION_TYPE_BIRTHDAY,
    POST_TYPE_SYSTEM_ANNIVERSARY,
    POST_TYPE_SYSTEM_BIRTHDAY,
    PREF_BIRTHDAY_VISIBILITY,
    FeedCelebration,
    FeedPost,
    IdentityPersonPreference,
)
from app.identity.person_models import IdentityPerson

# Force-register every Base subclass so create_all / metadata scans
# don't surface "table referenced in FK but not declared" warnings.
from app.main import app  # noqa: F401 — load all model imports

logger = get_safe_logger("backfill_feed_celebrations")
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)


# ── Helpers (duplicated from celebration_service to keep this script
#    self-contained — copying ~6 lines is cheaper than depending on a
#    private symbol that might change). ────────────────────────────


def _pick_illustration(
    person_id: str, occasion: date, celebration_type: str,
) -> str:
    pool = (
        BIRTHDAY_ILLUSTRATION_KEYS
        if celebration_type == CELEBRATION_TYPE_BIRTHDAY
        else ANNIVERSARY_ILLUSTRATION_KEYS
    )
    seed = f"{person_id}|{occasion.isoformat()}|{celebration_type}".encode()
    # Must match celebration_service's historical presentation bucket; this
    # is deterministic UI selection, not protection of the person id.
    digest = hashlib.sha1(
        seed,
        usedforsecurity=False,
    ).hexdigest()
    idx = int(digest, 16) % len(pool)
    return pool[idx]


def _format_anniv_body(name: str, years: int) -> str:
    return (
        f"今天是 @{name} 的入职 {years} 周年纪念日,"
        f"大家快来送祝福吧!"
    )


def _format_birthday_body(name: str) -> str:
    return f"今天是 @{name} 的生日,大家快来送祝福吧!"


def _shift_for_leap(anchor: date, target_year: int) -> date | None:
    """Return the date in ``target_year`` matching ``anchor``'s month/day,
    with Feb-29 → Feb-28 fallback in non-leap years. Returns None if
    the date can't be constructed (shouldn't happen with the fallback).
    """
    try:
        return date(target_year, anchor.month, anchor.day)
    except ValueError:
        if anchor.month == 2 and anchor.day == 29:
            return date(target_year, 2, 28)
        return None


def _past_anniversaries(
    hire_date: date, through: date,
) -> Iterable[tuple[date, int]]:
    """Yield ``(occasion_date, years)`` for every hire-anniversary in
    ``(hire_date, through]`` — i.e. year ≥ 1 and strictly past
    ``hire_date``, up to and including ``through``. Today is excluded
    from the caller's range so the live cron isn't shadowed.
    """
    if hire_date >= through:
        return
    for year_n in range(1, through.year - hire_date.year + 1):
        occ = _shift_for_leap(hire_date, hire_date.year + year_n)
        if occ is None or occ > through:
            continue
        yield occ, year_n


def _past_birthdays(
    birthday: date, hire_date: date, through: date,
) -> Iterable[date]:
    """Yield each past birthday in ``[hire_date, through]`` —
    birthdays before the person joined are skipped (per the user's
    "入职后的生日庆祝" rule). Years 0 (birthday on hire-day) is
    included if the birthday's MM-DD coincides with the hire day.
    """
    if hire_date > through:
        return
    for target_year in range(hire_date.year, through.year + 1):
        occ = _shift_for_leap(birthday, target_year)
        if occ is None:
            continue
        if occ < hire_date or occ > through:
            continue
        yield occ


# ── Per-person backfill ───────────────────────────────────────────


async def _is_birthday_hidden(db: AsyncSession, person_id: str) -> bool:
    row = (await db.execute(
        select(IdentityPersonPreference.pref_value).where(
            IdentityPersonPreference.person_id == person_id,
            IdentityPersonPreference.pref_key == PREF_BIRTHDAY_VISIBILITY,
        )
    )).scalar_one_or_none()
    return row == BIRTHDAY_VISIBILITY_HIDDEN


async def _insert_celebration_and_post(
    db: AsyncSession,
    *,
    person: IdentityPerson,
    celebration_type: str,
    occasion_date: date,
    years: int,
) -> tuple[bool, str | None]:
    """Insert a single celebration + system post pair. Returns
    ``(inserted, post_type)``. ``inserted=False`` means a duplicate
    UNIQUE collision (already backfilled), which is the
    idempotent-rerun signal — not an error.
    """
    person_id_s = str(person.id)
    display_name = person.display_name or person.work_email or "员工"

    # Pre-existence check so we can short-circuit without a SAVEPOINT
    # round-trip on the common rerun case.
    existing = (await db.execute(
        select(FeedCelebration).where(
            FeedCelebration.person_id == person_id_s,
            FeedCelebration.occasion_date == occasion_date,
            FeedCelebration.celebration_type == celebration_type,
        )
    )).scalar_one_or_none()
    if existing is not None:
        return False, None

    illustration_key = _pick_illustration(
        person_id_s, occasion_date, celebration_type,
    )
    cel = FeedCelebration(
        person_id=person_id_s,
        celebration_type=celebration_type,
        occasion_date=occasion_date,
        years=years,
        display_name_snapshot=display_name,
        department_snapshot=person.primary_department,
        avatar_url_snapshot=person.avatar_url,
        illustration_key=illustration_key,
        # Mark every notification role as already-skipped so a future
        # cron / human-trigger that walks historical celebrations
        # CAN'T accidentally fire pushes. Belt-and-braces: the live
        # cron filters by ``occasion_date == today`` so historical
        # rows aren't picked up anyway, but a deliberate "re-notify"
        # admin tool added later shouldn't be able to spam either.
        notify_state_json={
            "self": "skipped:backfill",
            "manager": "skipped:backfill",
            "hrbp": "skipped:backfill",
        },
    )
    db.add(cel)
    try:
        async with db.begin_nested():
            await db.flush()
    except IntegrityError:
        # Race / pre-existing row — treat as already-done.
        return False, None

    if celebration_type == CELEBRATION_TYPE_ANNIVERSARY:
        post_type = POST_TYPE_SYSTEM_ANNIVERSARY
        body = _format_anniv_body(display_name, years)
    else:
        post_type = POST_TYPE_SYSTEM_BIRTHDAY
        body = _format_birthday_body(display_name)

    # Anchor created_at to the historical occasion date so the
    # timeline rendering places this post at the right spot in
    # history. Use noon UTC so the row sorts mid-day regardless of
    # the viewer's TZ.
    historical_ts = datetime.combine(
        occasion_date, datetime.min.time(), tzinfo=UTC,
    ).replace(hour=12)

    post = FeedPost(
        author_person_id=None,
        post_type=post_type,
        visibility="public",
        category="general",
        body=body,
        celebration_id=cel.id,
        is_pinned=False,
        created_at=historical_ts,
    )
    db.add(post)
    await db.flush()
    return True, post_type


async def _backfill_one_person(
    db: AsyncSession,
    *,
    person: IdentityPerson,
    through: date,
) -> dict[str, int]:
    stats = {"anniversary": 0, "birthday": 0, "skipped_existing": 0}

    if person.hire_date is None:
        return stats

    # Anniversaries.
    for occ, years in _past_anniversaries(person.hire_date, through):
        inserted, _ = await _insert_celebration_and_post(
            db,
            person=person,
            celebration_type=CELEBRATION_TYPE_ANNIVERSARY,
            occasion_date=occ,
            years=years,
        )
        if inserted:
            stats["anniversary"] += 1
        else:
            stats["skipped_existing"] += 1

    # Birthdays — respect hidden preference.
    if person.birthday is not None:
        person_id_s = str(person.id)
        if not await _is_birthday_hidden(db, person_id_s):
            for occ in _past_birthdays(
                person.birthday, person.hire_date, through,
            ):
                inserted, _ = await _insert_celebration_and_post(
                    db,
                    person=person,
                    celebration_type=CELEBRATION_TYPE_BIRTHDAY,
                    occasion_date=occ,
                    years=0,
                )
                if inserted:
                    stats["birthday"] += 1
                else:
                    stats["skipped_existing"] += 1

    return stats


# ── Driver ────────────────────────────────────────────────────────


async def run(*, apply: bool, through: date, person_id: str | None) -> None:
    totals = {
        "persons_seen": 0,
        "persons_skipped_no_hire_date": 0,
        "anniversary_inserted": 0,
        "birthday_inserted": 0,
        "celebrations_already_present": 0,
        "failures": 0,
    }

    async with async_session_factory() as db:
        q = select(IdentityPerson).where(
            IdentityPerson.employment_status == "active",
            IdentityPerson.deleted_at.is_(None),
        )
        if person_id:
            q = q.where(IdentityPerson.id == uuid.UUID(person_id))
        rows = (await db.execute(q)).scalars().all()
        logger.info("Selected %d active person(s) for backfill", len(rows))

        for person in rows:
            totals["persons_seen"] += 1
            if person.hire_date is None:
                totals["persons_skipped_no_hire_date"] += 1
                continue
            try:
                async with db.begin_nested():
                    stats = await _backfill_one_person(
                        db, person=person, through=through,
                    )
                totals["anniversary_inserted"] += stats["anniversary"]
                totals["birthday_inserted"] += stats["birthday"]
                totals["celebrations_already_present"] += stats[
                    "skipped_existing"
                ]
                if stats["anniversary"] + stats["birthday"] > 0:
                    logger.info(
                        "  %s — +%d anniversaries, +%d birthdays (skipped %d)",
                        person.display_name or person.work_email,
                        stats["anniversary"],
                        stats["birthday"],
                        stats["skipped_existing"],
                    )
            except Exception:
                totals["failures"] += 1
                logger.exception(
                    "person %s (%s) failed — rolled back per-person SAVEPOINT",
                    person.display_name, person.id,
                )

        if apply:
            # One audit row per successful run, NOT per person — the
            # row count + summary is forensically sufficient and a
            # per-person audit row would balloon ``audit_logs``.
            await AuditService(db).log(
                trace_id="backfill_feed_celebrations_2026_06_02",
                actor_type="script",
                action="feed.celebrations.backfill",
                status="success",
                resource_type="feed_celebrations",
                resource_id=None,
                detail={
                    "through": through.isoformat(),
                    "person_filter": person_id,
                    **totals,
                },
            )
            await db.commit()
            logger.info("Committed.")
        else:
            await db.rollback()
            logger.info("Dry run — rolled back; no rows written.")

    logger.info("Summary: %s", totals)


def _parse_date(value: str) -> date:
    try:
        return date.fromisoformat(value)
    except ValueError as exc:
        raise argparse.ArgumentTypeError(
            f"--through must be YYYY-MM-DD (got {value!r})",
        ) from exc


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--apply",
        action="store_true",
        help="Actually write to the DB. Default is dry-run.",
    )
    parser.add_argument(
        "--through",
        type=_parse_date,
        default=None,
        help=(
            "Backfill up to and including this date (YYYY-MM-DD). "
            "Default = today in Asia/Shanghai - 1 day so the cron's "
            "live 'today' work isn't shadowed. Pass a past date to "
            "narrow the window."
        ),
    )
    parser.add_argument(
        "--person-id",
        default=None,
        help="Backfill only this person (UUID). Useful for spot-checking.",
    )
    args = parser.parse_args()

    if args.through is None:
        today_sh = datetime.now(ZoneInfo("Asia/Shanghai")).date()
        # Stop the day before today's date — the live cron owns today.
        args.through = today_sh - timedelta(days=1)
        logger.info(
            "Defaulting --through to yesterday (SHA): %s", args.through,
        )

    asyncio.run(
        run(
            apply=args.apply,
            through=args.through,
            person_id=args.person_id,
        ),
    )
    return 0


if __name__ == "__main__":
    sys.exit(main())