"""Scheduled job for half-yearly cloud-server owner inventory.

Pair to ``cloud_owner_handlers.py`` (which targets cloud SCOPES) but
operating on individual SERVERS aggregated per owner. One card per
owner per cron tick, listing every server the owner holds that is
due for confirmation.

Cron: ``cmdb.server_owner_semiannual_confirmation`` — weekly Mon 09:00
Asia/Shanghai (default; admin can override via scheduled-jobs config).
The weekly cadence (vs daily for scopes) is deliberate: server-level
churn is slower than scope-level grants, and a weekly tick batches
ownership changes from the previous week into one card per owner
rather than sending the same owner multiple cards within a few days.

Skip rules (enforced here, not in service.list_due_owners — keeps the
service test-friendly while letting the cron decide policy):
  - owner not found / soft-deleted
  - owner.employment_status ∈ ('left', 'leaver_pending') — the leaver
    offboarding flow already attaches these servers to a replacement
"""
from __future__ import annotations

import logging
import uuid
from datetime import datetime
from typing import Any
from zoneinfo import ZoneInfo

from sqlalchemy.ext.asyncio import AsyncSession

from app.scheduled_jobs.registry import register_scheduled_job

logger = logging.getLogger(__name__)


@register_scheduled_job(
    key="cmdb.server_owner_semiannual_confirmation",
    display_name="云服务器 Owner 半年盘点",
    description=(
        "每周一 09:00（Asia/Shanghai）扫描所有有 owner 的云服务器,"
        "找出 last_owner_confirmed_at 距今超过 180 天的(或从未确认且"
        "first_seen_at 已超过 90 天的),按 owner 聚合发送一张飞书卡片。"
        "owner 一键确认全部或前往管理页逐台处理。"
        "跳过规则:无 owner / owner 离职 / 新机器(<90 天)/ 已停机或下线。"
    ),
)
async def run_server_owner_semiannual_confirmation(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    """Send the per-owner server-inventory card to every due owner.

    Returns a summary dict for ``last_result_summary``:
      ``owners_prompted`` — owners that received a fresh card
      ``owners_deduped`` — owners whose row for today already existed
        (cron re-run on the same day)
      ``servers_in_cards`` — total servers referenced across all sent
        cards (sanity-check that the cron actually had work to do)
      ``errors`` — per-owner exception count
      ``trace_id`` — single id correlating all sends + audit entries
    """
    from app.core.database import async_session_factory
    from app.domains.cmdb.server_owner_confirmation_service import (
        EXCLUDED_EMPLOYMENT_STATUSES,
        ServerOwnerConfirmationService,
    )
    from app.identity.person_models import IdentityPerson

    trace_id = uuid.uuid4().hex
    owners_prompted = 0
    owners_deduped = 0
    servers_in_cards = 0
    errors = 0

    # Step 0 — reconcile stale pending rows BEFORE scanning. The
    # re-prompt interlock in ``list_due_owners`` skips any owner with
    # an open pending row; if that row is stuck on dead servers
    # (see ``reconcile_pending``'s docstring) the owner would never be
    # prompted again. Reconciling first lets a fully-dead row close and
    # the owner's live servers mint a fresh card in the same tick.
    # Own session + commit so a reconcile hiccup can't poison the scan.
    reconcile_summary: dict[str, Any] = {}
    try:
        async with async_session_factory() as recon_session:
            recon_service = ServerOwnerConfirmationService(recon_session)
            reconcile_summary = await recon_service.reconcile_pending(
                trace_id=trace_id,
            )
            await recon_session.commit()
    except Exception:
        logger.exception("server owner confirmation pre-scan reconcile failed")
        reconcile_summary = {"error": True}

    # First pass — discover eligible servers grouped by owner. Uses
    # the cron's shared session for a single read.
    async with async_session_factory() as scan_session:
        scan_service = ServerOwnerConfirmationService(scan_session)
        # Use a stable "today" so all owners share the same period_start
        # — important for the UNIQUE(owner_person_id, period_start)
        # idempotency window. Without this, an owner whose card fires at
        # 23:59:59 followed by another at 00:00:01 next day would land
        # on two different period_start rows.
        ref_date = datetime.now(ZoneInfo("Asia/Shanghai")).date()
        grouped = await scan_service.list_due_owners(today=ref_date)

    owner_ids = list(grouped.keys())

    # Second pass — one transaction per owner so a single failure
    # doesn't roll back the whole sweep. Mirrors cloud_owner_handlers.
    for owner_id in owner_ids:
        try:
            async with async_session_factory() as owner_session:
                service = ServerOwnerConfirmationService(owner_session)
                owner = await owner_session.get(
                    IdentityPerson, uuid.UUID(owner_id),
                )
                if owner is None or owner.deleted_at is not None:
                    continue
                if owner.employment_status in EXCLUDED_EMPLOYMENT_STATUSES:
                    continue

                server_ids = grouped[owner_id]
                _row, deduplicated = await service.create_pending_for_owner(
                    owner=owner,
                    server_ids=server_ids,
                    period_start=ref_date,
                    trace_id=trace_id,
                )
                await owner_session.commit()
                if deduplicated:
                    owners_deduped += 1
                else:
                    owners_prompted += 1
                    servers_in_cards += len(server_ids)
        except Exception:
            logger.exception(
                "server owner confirmation failed for owner=%s", owner_id,
            )
            errors += 1

    summary = {
        "owners_prompted": owners_prompted,
        "owners_deduped": owners_deduped,
        "servers_in_cards": servers_in_cards,
        "errors": errors,
        "reconcile": reconcile_summary,
        "trace_id": trace_id,
    }

    # Same systemic-failure gate as the scope cron: raise if everything
    # failed OR a majority of attempts failed. Lets the failure_notifier
    # surface real outages instead of treating "0/100 cards sent" as ok.
    total_attempted = owners_prompted + owners_deduped + errors
    majority_failed = (
        total_attempted >= 4 and errors * 2 >= total_attempted
    )
    if (errors > 0 and owners_prompted == 0) or majority_failed:
        raise RuntimeError(
            f"cmdb.server_owner_semiannual_confirmation: errors={errors} "
            f"owners_prompted={owners_prompted} "
            f"owners_deduped={owners_deduped} — systemic failure "
            "(check Feishu webhook + open_id resolver)."
        )
    return summary


@register_scheduled_job(
    key="cmdb.server_owner_confirmation_reconcile",
    display_name="云服务器半年盘点待办对账",
    description=(
        "每日校正未响应的半年盘点确认单:重新按扫描期的合格规则检查快照里"
        "的每台机器,剔除此后已停机/下线/软删/换 owner/归档租户的机器;"
        "快照清空则自动关单(auto_resolved),不打扰 owner 确认一台已经"
        "不存在或已停机的机器;剩余机器都已被 owner 确认过的,按已响应"
        "收尾。防止卡死的 pending 单永久阻断该 owner 的后续盘点卡片。"
    ),
)
async def run_server_owner_confirmation_reconcile(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    """Daily snapshot-vs-fleet reconciliation for pending confirmations.

    Thin wrapper over ``ServerOwnerConfirmationService.reconcile_pending``
    — see its docstring for the eligibility rules and the 2026-07-14
    incident that motivated it. Mutates on the executor-provided
    session; the executor commits after the handler returns.
    """
    from app.domains.cmdb.server_owner_confirmation_service import (
        ServerOwnerConfirmationService,
    )

    service = ServerOwnerConfirmationService(db)
    return await service.reconcile_pending(trace_id=uuid.uuid4().hex)