"""工号系统 service — allocation, CSV import/export, report aggregation.

Allocation invariant: the next 工号 is ``MAX(seq) + 1`` over
``hr_employee_numbers`` only (this table is the sole allocation authority
after migration 249 seeds it from ``identity_persons``). ``employee_no`` is
always the canonical ``P`` + zero-padded-5 string for that ``seq`` so a
caller-supplied ``P2388`` and an existing ``P02388`` collapse to the same
key and dedupe correctly.
"""

from __future__ import annotations

import re
from datetime import UTC, date, datetime
from uuid import UUID

from sqlalchemy import String, and_, cast, func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.exceptions import AppException
from app.domains.hr.models import HrEmployeeNumber, HrEmployeeNumberOverride
from app.identity.person_models import IdentityPerson

_P_NUMERIC = re.compile(r"^P(\d+)$")
_PAD = 5
_MAX_ALLOCATE_RETRIES = 8


def format_employee_no(seq: int) -> str:
    """``2388`` -> ``"P02388"``. Naturally grows past 99999 -> ``P100000``."""
    return f"P{seq:0{_PAD}d}"


def parse_seq(employee_no: str) -> int | None:
    """``"P02388"`` -> ``2388``; non ``P<digits>`` -> ``None``."""
    m = _P_NUMERIC.match(employee_no.strip())
    return int(m.group(1)) if m else None


def canonicalize(employee_no: str) -> tuple[str, int] | None:
    """Return (canonical_no, seq) for a ``P<digits>`` value, else ``None``."""
    seq = parse_seq(employee_no)
    if seq is None:
        return None
    return format_employee_no(seq), seq


# ── Allocation ────────────────────────────────────────────────────────


async def _current_max_seq(db: AsyncSession) -> int:
    """Largest allocated ``seq`` ever (0 if the ledger has no numeric rows).

    Deliberately spans ALL rows, including soft-deleted ones: the
    ``employee_no`` unique index is global (MySQL/TiDB can't do partial
    indexes), so a soft-deleted number still occupies its key. Computing the
    max over all rows means allocation never re-mints a number whose row was
    soft-deleted — which would otherwise collide on the unique index and
    wedge the retry loop.
    """
    val = await db.scalar(select(func.max(HrEmployeeNumber.seq)))
    return int(val) if val is not None else 0


async def _override_floor(db: AsyncSession) -> int:
    """The manually-set floor for the next ``seq`` (0 if HR never set one).

    This is the ``next_seq`` of the most-recent ``hr_employee_number_overrides``
    row. It only ever raises the next number ABOVE ``MAX(seq)+1`` — once
    allocations catch up to it, ``MAX(seq)+1`` wins and the floor is inert.
    Set-time validation guarantees every stored floor was ``> MAX(seq)`` at
    the time it was set, so it can never make allocation re-mint a number.
    """
    val = await db.scalar(
        select(HrEmployeeNumberOverride.next_seq)
        # ``created_at`` is DATETIME(0) (second precision); two overrides
        # set in the same second would order nondeterministically. Add
        # ``id`` as a stable tiebreak so the picked floor is deterministic
        # (both same-second rows are valid floors by set-time validation,
        # so any consistent choice is correct — we just need it stable).
        .order_by(
            HrEmployeeNumberOverride.created_at.desc(),
            HrEmployeeNumberOverride.id.desc(),
        )
        .limit(1)
    )
    return int(val) if val is not None else 0


async def _next_seq(db: AsyncSession) -> tuple[int, int]:
    """Return ``(next_seq, current_max_seq)`` honouring the manual floor."""
    max_seq = await _current_max_seq(db)
    return max(max_seq + 1, await _override_floor(db)), max_seq


async def get_next(db: AsyncSession) -> tuple[str, int, str | None]:
    """Preview the next number without allocating.

    Returns ``(next_employee_no, next_seq, current_max_no_or_None)``. The
    next number honours any manually-set floor (工号维护).
    """
    next_seq, max_seq = await _next_seq(db)
    current_max_no = format_employee_no(max_seq) if max_seq > 0 else None
    return format_employee_no(next_seq), next_seq, current_max_no


async def allocate(
    db: AsyncSession,
    *,
    name: str | None,
    hire_date: date | None,
    remark: str | None,
    created_by: UUID | None,
    bu: str | None = None,
    source: str = "manual",
) -> HrEmployeeNumber:
    """Allocate the next 工号 and persist it.

    Concurrency-safe via the ``employee_no`` unique index + retry: two
    simultaneous allocations that compute the same ``seq`` collide on
    INSERT; the loser re-reads ``MAX(seq)`` and retries (same pattern as
    the asset-tag generator). Caller must NOT have an open nested txn.

    The number honours any manually-set floor (工号维护); the floor is
    re-read each retry so a concurrent set-next is respected too.
    """
    last_err: Exception | None = None
    for _ in range(_MAX_ALLOCATE_RETRIES):
        next_seq, _max_seq = await _next_seq(db)
        row = HrEmployeeNumber(
            employee_no=format_employee_no(next_seq),
            seq=next_seq,
            name=name,
            hire_date=hire_date,
            remark=remark,
            bu=bu,
            source=source,
            created_by=created_by,
        )
        try:
            # Savepoint, NOT a full session rollback: a concurrent
            # allocation that grabbed the same seq collides on the unique
            # index here; only THIS INSERT rolls back, leaving any prior
            # pending work in the caller's transaction (e.g. an MCP
            # tool_runtime txn) intact. We then re-read MAX(seq) and retry.
            async with db.begin_nested():
                db.add(row)
                await db.flush()
            return row
        except IntegrityError as exc:
            last_err = exc
            continue
    raise AppException(
        code="EMPLOYEE_NO_ALLOCATION_FAILED",
        message="工号分配失败,请重试。",
        status_code=409,
    ) from last_err


# ── Per-row maintenance / delete (创建者自助维护) ──────────────────────


async def _get_editable(
    db: AsyncSession,
    *,
    number_id: UUID,
    actor_id: UUID,
    is_admin: bool,
) -> HrEmployeeNumber:
    """Load a live ledger row the actor is allowed to mutate, else raise.

    Authority model: the row's creator may maintain/delete their own
    allocations; ``global_admin`` may touch any row (platform invariant —
    full write everywhere). Seed rows (``created_by IS NULL``) therefore
    have no self-service owner and are admin-only.
    """
    row = await db.get(HrEmployeeNumber, number_id)
    if row is None or row.deleted_at is not None:
        raise AppException(
            code="EMPLOYEE_NO_NOT_FOUND",
            message="工号记录不存在或已删除。",
            status_code=404,
        )
    if not is_admin and (row.created_by is None or row.created_by != actor_id):
        raise AppException(
            code="EMPLOYEE_NO_NOT_OWNER",
            message="只能维护或删除本人创建的工号。",
            status_code=403,
        )
    return row


def _editable_snapshot(row: HrEmployeeNumber) -> dict:
    """The mutable fields of a row, JSON-ready, for an audit before/after diff.

    ``employee_no`` / ``seq`` are intentionally excluded — they are immutable
    and never change on an update."""
    return {
        "name": row.name,
        "hire_date": row.hire_date.isoformat() if row.hire_date else None,
        "remark": row.remark,
        "bu": row.bu,
    }


async def update_number(
    db: AsyncSession,
    *,
    number_id: UUID,
    actor_id: UUID,
    is_admin: bool,
    name: str,
    hire_date: date | None,
    remark: str | None,
    bu: str | None,
) -> tuple[HrEmployeeNumber, dict, dict]:
    """Update the descriptive fields of one ledger row.

    The 工号 itself (``employee_no`` / ``seq``) is immutable — only
    ``name`` / ``hire_date`` / ``remark`` / ``bu`` change. Returns
    ``(row, before, after)`` so the caller can audit the diff. Does NOT
    commit; the caller commits the row + its audit entry atomically.
    """
    row = await _get_editable(
        db, number_id=number_id, actor_id=actor_id, is_admin=is_admin
    )
    before = _editable_snapshot(row)
    row.name = name
    row.hire_date = hire_date
    row.remark = remark
    row.bu = bu
    await db.flush()
    return row, before, _editable_snapshot(row)


async def delete_number(
    db: AsyncSession,
    *,
    number_id: UUID,
    actor_id: UUID,
    is_admin: bool,
) -> dict:
    """Permanently delete one ledger row and return its details for audit.

    HARD delete (not soft): the ``employee_no`` unique index is global, so a
    soft-deleted row would keep occupying its number and block re-allocation.
    Deleting the row physically frees the key, which is what makes a number
    reclaimable — if the deleted ``seq`` was the current maximum, the next
    allocation (``MAX(seq)+1`` over the surviving rows) hands the same number
    out again; if a higher number still exists, the gap is simply left behind
    and allocation continues from the surviving max. The full who/what/when
    survives in ``audit_logs`` (this table has no inbound FKs). Does NOT
    commit; the caller commits the delete + its audit entry atomically.
    """
    row = await _get_editable(
        db, number_id=number_id, actor_id=actor_id, is_admin=is_admin
    )
    info = {
        "employee_no": row.employee_no,
        "seq": row.seq,
        "name": row.name,
        "bu": row.bu,
        "source": row.source,
        "created_by": str(row.created_by) if row.created_by else None,
    }
    await db.delete(row)
    await db.flush()
    return info


# ── Manual "next 工号" maintenance (工号维护) ──────────────────────────


async def set_next(
    db: AsyncSession,
    *,
    next_employee_no: str,
    reason: str,
    created_by: UUID | None,
) -> HrEmployeeNumberOverride:
    """Manually set the next 工号 the allocator will hand out.

    The target must be a canonical ``P<digits>`` value strictly greater than
    the current maximum allocated number — a smaller (or equal-to-max) target
    would collide with an already-allocated 工号. Appends one maintenance
    record carrying the operator, reason, and the before→after next number.
    Does NOT commit; the caller commits the record + its audit row atomically.
    """
    canon = canonicalize(next_employee_no)
    if canon is None:
        raise AppException(
            code="BAD_EMPLOYEE_NO",
            message="工号格式错误,应为 P + 数字(例如 P05000)。",
            status_code=400,
        )
    target_no, target_seq = canon

    max_seq = await _current_max_seq(db)
    if target_seq <= max_seq:
        raise AppException(
            code="NEXT_NO_NOT_GREATER",
            message=(
                f"下一个工号必须大于当前最大工号 {format_employee_no(max_seq)}。"
            ),
            status_code=400,
        )

    # before→after diff for the maintenance log.
    prev_no, prev_seq, _ = await get_next(db)

    row = HrEmployeeNumberOverride(
        next_seq=target_seq,
        next_employee_no=target_no,
        previous_next_seq=prev_seq,
        previous_next_employee_no=prev_no,
        reason=reason.strip(),
        created_by=created_by,
    )
    db.add(row)
    await db.flush()
    return row


async def list_overrides(
    db: AsyncSession, *, page: int = 1, page_size: int = 20,
) -> tuple[list[dict], int]:
    """工号维护记录, newest first, with ``created_by_name`` resolved."""
    total = await db.scalar(
        select(func.count()).select_from(HrEmployeeNumberOverride)
    ) or 0

    operator = IdentityPerson.__table__.alias("operator")
    stmt = (
        select(HrEmployeeNumberOverride, operator.c.display_name)
        .outerjoin(operator, HrEmployeeNumberOverride.created_by == operator.c.id)
        # id tiebreak: stable pagination for same-second rows (DATETIME(0)).
        .order_by(
            HrEmployeeNumberOverride.created_at.desc(),
            HrEmployeeNumberOverride.id.desc(),
        )
        .offset((page - 1) * page_size)
        .limit(page_size)
    )
    result = await db.execute(stmt)
    rows: list[dict] = []
    for row, operator_name in result.all():
        rows.append({
            "id": row.id,
            "next_employee_no": row.next_employee_no,
            "next_seq": row.next_seq,
            "previous_next_employee_no": row.previous_next_employee_no,
            "previous_next_seq": row.previous_next_seq,
            "reason": row.reason,
            "created_by": row.created_by,
            "created_by_name": operator_name,
            "created_at": row.created_at,
        })
    return rows, int(total)


# ── BU (一级部门) resolution ───────────────────────────────────────────


async def known_bu_map(db: AsyncSession) -> dict[str, str]:
    """``lower(name) -> canonical name`` for every live BU.

    Reuses the platform-wide BU derivation (``app/identity/bu_helpers``) so a
    BU set on a 工号 always matches the same set the cost / AI-spending
    dashboards use. Empty dict if the org tree has no BUs yet."""
    from app.identity.bu_helpers import list_bus

    return {
        str(b["name"]).strip().lower(): str(b["name"])
        for b in await list_bus(db)
        if b.get("name")
    }


def resolve_bu(raw: str | None, bu_map: dict[str, str]) -> tuple[str | None, bool]:
    """``(canonical_name | None, ok)``.

    Blank input -> ``(None, True)``. A non-blank value is matched
    case-insensitively against ``bu_map``; on a hit the canonical name is
    returned, on a miss ``(raw, False)`` so the caller can surface a clean
    "unknown BU" error instead of silently storing a typo."""
    s = (raw or "").strip()
    if not s:
        return None, True
    canon = bu_map.get(s.lower())
    if canon is None:
        return s, False
    return canon, True


# ── List / search ─────────────────────────────────────────────────────


def _like_escape(s: str) -> str:
    """Escape LIKE wildcards so a user's literal ``%`` / ``_`` / ``\\`` in a
    substring filter matches literally instead of as a pattern."""
    return s.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")


def _apply_filters(stmt, *, no, name, hire_from, hire_to, source, created_from, created_to):
    conds = [HrEmployeeNumber.deleted_at.is_(None)]
    if no:
        conds.append(HrEmployeeNumber.employee_no.ilike(f"%{_like_escape(no.strip())}%", escape="\\"))
    if name:
        conds.append(HrEmployeeNumber.name.ilike(f"%{_like_escape(name.strip())}%", escape="\\"))
    if hire_from:
        conds.append(HrEmployeeNumber.hire_date >= hire_from)
    if hire_to:
        conds.append(HrEmployeeNumber.hire_date <= hire_to)
    if source:
        conds.append(HrEmployeeNumber.source == source)
    if created_from:
        conds.append(HrEmployeeNumber.created_at >= created_from)
    if created_to:
        conds.append(HrEmployeeNumber.created_at <= created_to)
    return stmt.where(and_(*conds))


async def list_numbers(
    db: AsyncSession,
    *,
    no: str | None = None,
    name: str | None = None,
    hire_from: date | None = None,
    hire_to: date | None = None,
    source: str | None = None,
    created_from: datetime | None = None,
    created_to: datetime | None = None,
    page: int = 1,
    page_size: int = 20,
) -> tuple[list[dict], int]:
    """Filtered + paginated ledger rows, newest number first.

    Each dict carries ``created_by_name`` resolved from the creator's
    ``identity_persons.display_name`` (NULL for seed rows)."""
    filters = dict(
        no=no, name=name, hire_from=hire_from, hire_to=hire_to,
        source=source, created_from=created_from, created_to=created_to,
    )

    total = await db.scalar(
        _apply_filters(
            select(func.count()).select_from(HrEmployeeNumber), **filters
        )
    ) or 0

    creator = IdentityPerson.__table__.alias("creator")
    stmt = (
        _apply_filters(
            select(HrEmployeeNumber, creator.c.display_name), **filters
        )
        .outerjoin(creator, HrEmployeeNumber.created_by == creator.c.id)
        # Numeric numbers first (seq desc), then any seq-less legacy rows.
        # MySQL/TiDB has no NULLS LAST — emulate with ``seq IS NULL`` as the
        # leading sort key (False=0 sorts before True=1, so non-null first).
        .order_by(
            HrEmployeeNumber.seq.is_(None),
            HrEmployeeNumber.seq.desc(),
            HrEmployeeNumber.created_at.desc(),
        )
        .offset((page - 1) * page_size)
        .limit(page_size)
    )
    result = await db.execute(stmt)
    rows: list[dict] = []
    for row, creator_name in result.all():
        rows.append({
            "id": row.id,
            "employee_no": row.employee_no,
            "seq": row.seq,
            "name": row.name,
            "hire_date": row.hire_date,
            "remark": row.remark,
            "bu": row.bu,
            "source": row.source,
            "created_by": row.created_by,
            "created_by_name": creator_name,
            "created_at": row.created_at,
        })
    return rows, int(total)


# ── CSV import ─────────────────────────────────────────────────────────

# Canonical CSV column order (also the export header). Chinese headers
# matching what HR types; the importer is tolerant to surrounding spaces.
CSV_COLUMNS = ["工号", "姓名", "入职日期", "备注", "一级部门"]


def _get(row: dict, key: str) -> str:
    v = row.get(key)
    return str(v).strip() if v is not None else ""


async def import_rows(
    db: AsyncSession,
    rows: list[dict],
    *,
    created_by: UUID | None,
) -> dict:
    """Import parsed CSV rows. Skip-bad-rows-and-report semantics: good rows
    are queued for the caller's commit, bad rows (missing name / bad date /
    bad 工号 format / duplicate) are skipped and returned in ``errors``.

    Empty 工号 -> auto-allocated sequentially from the current max (a running
    cursor so multiple new rows in one file get consecutive numbers).
    Explicit 工号 -> canonicalized + used as given; if it raises the max,
    later auto-allocations continue from it.

    Duplicate detection is DB-authoritative: each insert runs in its own
    savepoint, so a collision against ANY existing row — including a
    soft-deleted one still occupying the global unique index, or a row
    inserted concurrently — surfaces as a clean per-row ``duplicate`` error
    instead of aborting the whole batch on the final commit. The in-memory
    ``seen`` set short-circuits intra-file duplicates without a round-trip.
    This function does NOT commit; the caller (router) commits rows + audit
    atomically.
    """
    cursor = await _current_max_seq(db) + 1
    bu_map = await known_bu_map(db)
    seen: set[str] = set()
    errors: list[dict] = []
    created = 0

    for idx, raw in enumerate(rows, start=1):
        no_raw = _get(raw, "工号")
        name = _get(raw, "姓名")
        hire_raw = _get(raw, "入职日期")
        remark = _get(raw, "备注") or None
        bu, bu_ok = resolve_bu(_get(raw, "一级部门"), bu_map)

        if not name:
            errors.append({"row": idx, "employee_no": no_raw or None, "reason": "missing_name"})
            continue

        hire_date: date | None = None
        if hire_raw:
            try:
                hire_date = date.fromisoformat(hire_raw)
            except ValueError:
                errors.append({"row": idx, "employee_no": no_raw or None, "reason": "bad_hire_date"})
                continue

        if not bu_ok:
            errors.append({"row": idx, "employee_no": no_raw or None, "reason": "unknown_bu"})
            continue

        if no_raw:
            canon = canonicalize(no_raw)
            if canon is None:
                errors.append({"row": idx, "employee_no": no_raw, "reason": "bad_format"})
                continue
            employee_no, seq = canon
        else:
            seq = cursor
            employee_no = format_employee_no(seq)

        is_dup = employee_no in seen
        if not is_dup:
            try:
                async with db.begin_nested():
                    db.add(HrEmployeeNumber(
                        employee_no=employee_no,
                        seq=seq,
                        name=name,
                        hire_date=hire_date,
                        remark=remark,
                        bu=bu,
                        source="import",
                        created_by=created_by,
                    ))
                    await db.flush()
            except IntegrityError:
                is_dup = True

        if is_dup:
            errors.append({"row": idx, "employee_no": employee_no, "reason": "duplicate"})
            # Auto rows must still advance the cursor past the collision so
            # the next new row doesn't re-pick the same number. (Explicit
            # duplicates are always <= max < cursor, so leaving the cursor
            # alone is correct.)
            if not no_raw:
                cursor += 1
            continue

        seen.add(employee_no)
        created += 1
        if no_raw:
            # Explicit number: advance the cursor past it so a later auto
            # row doesn't re-pick this seq.
            if seq >= cursor:
                cursor = seq + 1
        else:
            cursor += 1

    return {
        "total": len(rows),
        "created": created,
        "skipped": len(errors),
        "errors": errors,
    }


# ── Report aggregation ────────────────────────────────────────────────


async def build_report(db: AsyncSession) -> dict:
    """KPI cards + monthly-allocation trend + source distribution + hire-year
    distribution. The monthly trend and the "本月新增" KPI exclude
    ``source='seed'`` rows — those are the one-time migration backfill (all
    stamped at deploy time) and would otherwise drown the genuine
    month-over-month allocation signal."""
    live = HrEmployeeNumber.deleted_at.is_(None)

    total = await db.scalar(
        select(func.count()).select_from(HrEmployeeNumber).where(live)
    ) or 0

    next_no, _next_seq, max_no = await get_next(db)

    # 本月新增 (genuine allocations, seed excluded). Month boundary is UTC,
    # matching the UTC-based ``date_format(created_at)`` bucketing in the
    # trend below so the KPI and the chart agree. This is an internal stat,
    # not an action-triggering cutoff, so we deliberately keep it UTC rather
    # than pulling in convert_tz (which depends on MySQL TZ tables being
    # loaded on prod) — a sub-day mis-bucket at a month edge is acceptable.
    now = datetime.now(UTC)
    month_start = datetime(now.year, now.month, 1, tzinfo=UTC)
    created_this_month = await db.scalar(
        select(func.count()).select_from(HrEmployeeNumber).where(
            live,
            HrEmployeeNumber.source != "seed",
            HrEmployeeNumber.created_at >= month_start,
        )
    ) or 0

    # Monthly allocation trend (seed excluded).
    month_expr = func.date_format(HrEmployeeNumber.created_at, "%Y-%m")
    trend_rows = (await db.execute(
        select(month_expr.label("m"), func.count().label("c"))
        .where(live, HrEmployeeNumber.source != "seed")
        .group_by("m")
        .order_by("m")
    )).all()
    monthly_trend = [{"month": m, "count": int(c)} for m, c in trend_rows]

    # Source distribution (all rows).
    src_rows = (await db.execute(
        select(HrEmployeeNumber.source, func.count())
        .where(live)
        .group_by(HrEmployeeNumber.source)
    )).all()
    source_distribution = [
        {"source": s, "count": int(c)} for s, c in src_rows
    ]

    # Hire-year distribution (all rows; NULL hire_date -> 'unknown').
    year_expr = func.coalesce(
        cast(func.year(HrEmployeeNumber.hire_date), String), "unknown"
    )
    year_rows = (await db.execute(
        select(year_expr.label("y"), func.count().label("c"))
        .where(live)
        .group_by("y")
        .order_by("y")
    )).all()
    hire_year_distribution = [
        {"year": y or "unknown", "count": int(c)} for y, c in year_rows
    ]

    return {
        "kpi": {
            "total": int(total),
            "next_employee_no": next_no,
            "max_employee_no": max_no,
            "created_this_month": int(created_this_month),
        },
        "monthly_trend": monthly_trend,
        "source_distribution": source_distribution,
        "hire_year_distribution": hire_year_distribution,
    }