"""``software_seat.usage_signal_refresh`` — multi-vendor usage collection (PR4).

Runs before ``software_seat.idle_reclaim`` each day so the reclaim job
reads signals collected hours ago, not days. If this job stops running,
the freshness gate downgrades every row to ``unknown`` within
``STALE_AFTER_HOURS`` and reclaims stop — the safe direction.
"""
from __future__ import annotations

import logging
from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession

from app.scheduled_jobs.registry import register_scheduled_job

logger = logging.getLogger(__name__)

USAGE_SIGNAL_REFRESH_HANDLER = "software_seat.usage_signal_refresh"


@register_scheduled_job(
    key=USAGE_SIGNAL_REFRESH_HANDLER,
    display_name="坐席使用信号 · 多厂商采集",
    description=(
        "每日 02:40 UTC(排在 software_seat.idle_reclaim 之前)。对每个启用且绑定了活跃 "
        "software_connectors 的集成,若其 provider 有采集器(jira / google_workspace / "
        "zoom),读一次厂商名册,按 work_email 匹配到人,写入 asset_seat_usage_signals。"
        "四态:active / never(厂商明说从未登录)/ unknown(厂商没给可用值)/ "
        "unsupported(该厂商答不了)。读不到的厂商保留原有行不覆盖 —— 过期的真证据好过"
        "新鲜的假话,新鲜度闸自会退役它。"
    ),
)
async def run_usage_signal_refresh(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    from app.integrations.usage_signal_service import (  # noqa: PLC0415
        refresh_usage_signals,
    )

    summary = await refresh_usage_signals(db)
    await db.commit()
    if summary.get("errors"):
        # Partial success is still a failure worth paging on: a vendor
        # that cannot be read means its seats quietly stop being
        # reclaimable, and nothing else would say so.
        raise RuntimeError(
            f"{USAGE_SIGNAL_REFRESH_HANDLER}: "
            f"{len(summary['errors'])} provider(s) failed: "
            + " | ".join(str(e) for e in summary["errors"][:5])
        )
    return summary