"""闲置 API Key —— 「这把 Key 还有人用吗」终于有人问了(migration 603)。 Why (2026-09-05 audit). Three apikey crons existed — budget, expiry, rotation retire — and none of them asked whether a live key is *used*. The identical pattern was already built next door for cloud subscriptions (``cmdb.zero_spend_alert``: last calendar month = $0 → one card with keep / clean up / contact IT), and ``compute_assignment_spend`` already produces the per-key monthly number the budget cron reads. This module joins the two. Shape: * ``apikey.idle_check`` (monthly, the 3rd, after billing ingest) evaluates every ``status='active'`` key old enough to have had a full month of use. Last calendar month spend ``== $0`` with authoritative data ⇒ idle. Bailian (no per-key BSS cost) and keys missing their attribution binding are ``skipped_no_data`` — recorded, never guessed. * **IT always gets a digest** (one card per admin per month). **The holder's card is gated** by ``apikey.idle_alert.enabled_for_employees`` — default OFF, per the 2026-09-05 decision "new employee reach ships IT-first, opened per class from the admin console". The card, the actions and the dedup are all built; the switch only decides whether the holder is paged. * ``apikey_idle_alerts_dedup`` (``assignment × billing_month``) is the ledger: it records that the month was evaluated idle, the streak at that time, and whether the holder was told. It is also what the admin "闲置" filter reads — no JSON path queries. * Holder actions: **keep** (a reason, silence for ``keep_months``), **release** (the existing manager-approved release flow — a $0 key may still be wired into a job the holder forgot), **contact IT** (support entry). ``detail.idle`` on the assignment carries the live state the UI renders: ``{last_idle_month, streak, last_checked_at, kept_until, keep_reason, kept_by_person_id, kept_at}``. """ from __future__ import annotations import logging import uuid from dataclasses import dataclass from datetime import UTC, date, datetime, timedelta from decimal import Decimal from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.exceptions import AppException, ForbiddenError from app.core.txn import best_effort from app.domains.apikey.models import ( ApiKeyAssignment, ApiKeyIdleAlertsDedup, ApiKeyProvider, ) from app.domains.apikey.spend import compute_assignment_spend from app.identity.person_models import IdentityPerson from app.integrations.models import PlatformSetting from app.integrations.service import get_platform_setting logger = logging.getLogger(__name__) SETTING_KEY = "apikey.idle_alert" TEMPLATE_EMPLOYEE = "apikey.idle_alert" TEMPLATE_IT_DIGEST = "apikey.idle_it_digest" DIGEST_RESOURCE_TYPE = "apikey_idle_digest" # Ledger look-back the admin "闲置" list reads: a key idle in either of the # two most recent evaluated months is still a live question. _LIST_MONTHS = 2 # Cap per IT digest card — the card links to the filtered console for the rest. _DIGEST_MAX_ROWS = 15 @dataclass(frozen=True) class IdleSettings: enabled_for_employees: bool = False min_age_days: int = 45 keep_months: int = 6 @classmethod def from_setting(cls, value: Any) -> IdleSettings: if not isinstance(value, dict): return cls() try: min_age = max(7, min(365, int(value.get("min_age_days") or 45))) except (TypeError, ValueError): min_age = 45 try: keep = max(1, min(24, int(value.get("keep_months") or 6))) except (TypeError, ValueError): keep = 6 return cls( enabled_for_employees=bool(value.get("enabled_for_employees", False)), min_age_days=min_age, keep_months=keep, ) def to_setting(self) -> dict[str, Any]: return { "enabled_for_employees": self.enabled_for_employees, "min_age_days": self.min_age_days, "keep_months": self.keep_months, } async def load_idle_settings(db: AsyncSession) -> IdleSettings: raw = await get_platform_setting(SETTING_KEY, default=None, session=db) return IdleSettings.from_setting(raw) async def save_idle_settings(db: AsyncSession, settings: IdleSettings) -> IdleSettings: row = await db.get(PlatformSetting, SETTING_KEY) if row is None: row = PlatformSetting( key=SETTING_KEY, value=settings.to_setting(), description="闲置 API Key 提醒(见迁移 603)。", ) db.add(row) else: row.value = settings.to_setting() await db.flush() return settings def previous_month(today: date) -> tuple[date, date, str]: """``(month_start, month_end, 'YYYY-MM')`` of the calendar month before ``today``. Cron fires on the 3rd; a manual run on Jan 2 still means December.""" first_of_this = today.replace(day=1) month_end = first_of_this - timedelta(days=1) month_start = month_end.replace(day=1) return month_start, month_end, month_start.strftime("%Y-%m") def _prev_period(period: str) -> str: y, m = (int(x) for x in period.split("-")) return f"{y - 1}-12" if m == 1 else f"{y}-{m - 1:02d}" def _aware(dt: datetime) -> datetime: return dt if dt.tzinfo is not None else dt.replace(tzinfo=UTC) async def evaluate_assignment( db: AsyncSession, *, assignment: ApiKeyAssignment, month_start: date, month_end: date, period: str, settings: IdleSettings, now: datetime, ) -> dict[str, Any]: """Decide idle / active / skipped for one key for ``period``. Writes ``detail.idle`` and (when idle) the dedup ledger row. Returns a summary dict with ``result`` ∈ {``idle``, ``active``, ``skipped_too_new``, ``skipped_kept``, ``skipped_no_data``} and, for idle, ``streak`` and ``deduped`` (True when this month's row already existed — a re-run). """ detail = dict(assignment.detail or {}) idle_state = dict(detail.get("idle") or {}) started = assignment.valid_from or ( assignment.created_at.date() if assignment.created_at else None ) if started is None or (month_end - started).days < settings.min_age_days: return {"result": "skipped_too_new"} kept_until_raw = idle_state.get("kept_until") if kept_until_raw: try: kept_until = date.fromisoformat(str(kept_until_raw)) except ValueError: kept_until = None if kept_until is not None and kept_until >= month_start: return {"result": "skipped_kept", "kept_until": kept_until.isoformat()} spend = await compute_assignment_spend( db, assignment=assignment, month_start=month_start, month_end=month_end, ) idle_state["last_checked_at"] = now.isoformat() idle_state["last_checked_period"] = period if not spend.get("data_available"): idle_state["last_result"] = "no_data" idle_state["no_data_reason"] = spend.get("data_caveat_code") or spend.get("data_caveat") detail["idle"] = idle_state assignment.detail = detail return { "result": "skipped_no_data", "reason": spend.get("data_caveat_code") or spend.get("data_caveat"), } mtd = spend.get("mtd_usd") spent = mtd if isinstance(mtd, Decimal) else Decimal(str(mtd or 0)) if spent > 0: idle_state["last_result"] = "active" idle_state["streak"] = 0 idle_state["last_active_month"] = period idle_state.pop("last_idle_month", None) detail["idle"] = idle_state assignment.detail = detail return {"result": "active", "mtd_usd": str(spent)} # Idle. Streak continues only when the previous month was also idle. prev_streak = int(idle_state.get("streak") or 0) streak = ( prev_streak + 1 if idle_state.get("last_idle_month") == _prev_period(period) and prev_streak > 0 else 1 ) idle_state["last_result"] = "idle" idle_state["streak"] = streak idle_state["last_idle_month"] = period detail["idle"] = idle_state assignment.detail = detail existing = (await db.execute( select(ApiKeyIdleAlertsDedup).where( ApiKeyIdleAlertsDedup.assignment_id == assignment.id, ApiKeyIdleAlertsDedup.billing_month == period, ) )).scalar_one_or_none() if existing is None: db.add(ApiKeyIdleAlertsDedup( assignment_id=assignment.id, billing_month=period, consecutive_months=streak, employee_notified=False, )) await db.flush() deduped = False else: existing.consecutive_months = streak deduped = True return {"result": "idle", "streak": streak, "deduped": deduped} async def keep_idle_assignment( db: AsyncSession, *, assignment: ApiKeyAssignment, actor: IdentityPerson, reason: str, settings: IdleSettings, trace_id: str, is_member: bool, ) -> dict[str, Any]: """Holder (or a shared-access member) says "still needed".""" if assignment.assigned_to != actor.id and not is_member: raise ForbiddenError("Only the key's owner or a member may keep it.") reason = (reason or "").strip() if len(reason) < 3: raise AppException( code="IDLE_KEEP_REASON_REQUIRED", message="Say in a sentence why the key is still needed.", status_code=400, ) today = datetime.now(UTC).date() kept_until = today + timedelta(days=30 * settings.keep_months) detail = dict(assignment.detail or {}) idle_state = dict(detail.get("idle") or {}) idle_state.update({ "kept_until": kept_until.isoformat(), "keep_reason": reason[:500], "kept_by_person_id": str(actor.id), "kept_at": datetime.now(UTC).isoformat(), }) detail["idle"] = idle_state assignment.detail = detail await db.flush() await AuditService(db).log( trace_id=trace_id, actor_type="user", actor_user_id=actor.id, action="apikey.idle.keep", resource_type="api_key_assignment", resource_id=str(assignment.id), status="success", detail={ "request_number": assignment.request_number, "kept_until": kept_until.isoformat(), "reason": reason[:500], "streak": idle_state.get("streak"), }, ) return {"kept_until": kept_until.isoformat(), "keep_reason": reason[:500]} async def list_idle_assignments( db: AsyncSession, *, months: int = _LIST_MONTHS, ) -> list[dict[str, Any]]: """Admin ledger: keys evaluated idle in the last ``months`` periods, newest month first, longest streak first, with live keep state.""" today = datetime.now(UTC).date() periods: list[str] = [] cursor = today for _ in range(months): _ms, _me, label = previous_month(cursor) periods.append(label) cursor = _ms rows = (await db.execute( select(ApiKeyIdleAlertsDedup, ApiKeyAssignment, ApiKeyProvider, IdentityPerson) .join(ApiKeyAssignment, ApiKeyAssignment.id == ApiKeyIdleAlertsDedup.assignment_id) .join(ApiKeyProvider, ApiKeyProvider.id == ApiKeyAssignment.provider_id) .join(IdentityPerson, IdentityPerson.id == ApiKeyAssignment.assigned_to) .where(ApiKeyIdleAlertsDedup.billing_month.in_(periods)) .order_by( ApiKeyIdleAlertsDedup.billing_month.desc(), ApiKeyIdleAlertsDedup.consecutive_months.desc(), ) )).all() seen: set[uuid.UUID] = set() out: list[dict[str, Any]] = [] for dedup, assignment, provider, holder in rows: if assignment.id in seen: continue seen.add(assignment.id) idle_state = dict((assignment.detail or {}).get("idle") or {}) out.append({ "assignment_id": str(assignment.id), "request_number": assignment.request_number, "project_name": assignment.project_name, "provider_name": provider.name, "status": assignment.status, "holder_person_id": str(holder.id), "holder_label": holder.display_name or holder.work_email, "holder_email": holder.work_email, "holder_employment_status": holder.employment_status, "billing_month": dedup.billing_month, "streak": int(dedup.consecutive_months), "employee_notified": bool(dedup.employee_notified), "kept_until": idle_state.get("kept_until"), "keep_reason": idle_state.get("keep_reason"), "last_result": idle_state.get("last_result"), "monthly_budget_usd": ( float(assignment.monthly_budget_usd) if assignment.monthly_budget_usd is not None else None ), "valid_until": assignment.valid_until.isoformat() if assignment.valid_until else None, }) return out async def _lang(person: IdentityPerson | None) -> str: raw = ((getattr(person, "preferred_language", None) or "") if person else "").lower() if raw.startswith("en"): return "en" if raw.startswith("ja"): return "ja" return "zh" async def notify_holder( db: AsyncSession, *, assignment: ApiKeyAssignment, provider_name: str, period: str, streak: int, settings: IdleSettings, trace_id: str, ) -> bool: """One card to the holder with keep / release / contact-IT actions. Returns True when a card was dispatched.""" from app.notifications.service import ( # noqa: PLC0415 CardAction, CardStat, LocalizedText, NotificationEnvelope, NotificationService, ) holder = await db.get(IdentityPerson, assignment.assigned_to) if ( holder is None or holder.deleted_at is not None or (holder.employment_status or "").lower() != "active" or not holder.work_email ): return False lang = await _lang(holder) detail_url = f"/me/api-keys/{assignment.id}" title = { "zh": f"💤 这把 API Key 上月没有花费:{assignment.request_number}({provider_name})", "en": f"💤 This API key had no spend last month: {assignment.request_number} ({provider_name})", "ja": f"💤 この API キーは先月使われていません:{assignment.request_number}({provider_name})", }[lang] body = { "zh": ( f"项目「{assignment.project_name}」的这把 Key 在 {period} 花费为 $0," f"已连续 {streak} 个月。还在用就点「保留」写一句理由;不需要了就「释放」" "(需主管确认);拿不准联系 IT。" ), "en": ( f"The key for \"{assignment.project_name}\" had $0 spend in {period} — " f"{streak} month(s) in a row. Still using it? Keep it with a one-line " "reason. Done with it? Release it (your manager confirms). Unsure — contact IT." ), "ja": ( f"プロジェクト「{assignment.project_name}」のこのキーは {period} の費用が $0 で、" f"{streak} か月連続です。まだ使うなら「保持」で理由を一言、不要なら「解放」" "(上長が確認)、判断できなければ IT へ。" ), }[lang] await NotificationService(db).dispatch( NotificationEnvelope( recipient=holder, title=title, body_text=body, channels=["feishu", "email"], template_key=TEMPLATE_EMPLOYEE, context={ "request_number": assignment.request_number, "provider_name": provider_name, "project_name": assignment.project_name, "period": period, "streak": str(streak), "keep_months": str(settings.keep_months), }, url=detail_url, card_subtitle=LocalizedText( zh=f"{period} 花费 $0 · 连续 {streak} 个月", en=f"$0 spend in {period} · {streak} month(s) running", ja=f"{period} の費用 $0 · {streak} か月連続", ), card_stats=[ CardStat( label=LocalizedText(zh="项目", en="Project", ja="プロジェクト"), value=assignment.project_name, ), CardStat( label=LocalizedText(zh="提供方", en="Provider", ja="プロバイダ"), value=provider_name, ), ], card_actions=[ CardAction( label=LocalizedText(zh="保留(写理由)", en="Keep (with reason)", ja="保持(理由を記入)"), url=f"{detail_url}?idle=keep", style="primary", ), CardAction( label=LocalizedText(zh="释放", en="Release", ja="解放"), url=f"{detail_url}?idle=release", style="danger", ), CardAction( label=LocalizedText(zh="联系 IT", en="Contact IT", ja="IT に連絡"), url="/me/support", style="default", ), ], related_resource_type="api_key_assignment", related_resource_id=assignment.id, ), trace_id=trace_id, actor_user_id=None, ) return True async def send_it_digest( db: AsyncSession, *, period: str, items: list[dict[str, Any]], settings: IdleSettings, trace_id: str, ) -> int: """One digest card per active IT admin per period (outbox-deduped). Returns the number of admins reached.""" from app.auth.models import AuthRole, AuthUserRole # noqa: PLC0415 from app.notifications.models import NotificationOutbox # noqa: PLC0415 from app.notifications.service import ( # noqa: PLC0415 CardAction, CardHighlight, CardStat, LocalizedText, NotificationEnvelope, NotificationService, ) if not items: return 0 digest_id = uuid.uuid5(uuid.NAMESPACE_URL, f"helix:apikey-idle-digest:{period}") already = set((await db.execute( select(NotificationOutbox.recipient_person_id).where( NotificationOutbox.related_resource_type == DIGEST_RESOURCE_TYPE, # CHAR(36) column: bind the string form, a UUID object never matches # (PR2 review, 2026-09-05 — the dedupe read was silently empty). NotificationOutbox.related_resource_id == str(digest_id), ) )).scalars().all()) admin_ids = list((await db.execute( select(AuthUserRole.user_id) .join(AuthRole, AuthRole.id == AuthUserRole.role_id) .where(AuthRole.name.in_(("global_admin", "it_admin"))) )).scalars().all()) streak2 = [i for i in items if int(i.get("streak") or 0) >= 2] ranked = sorted(items, key=lambda i: (-int(i.get("streak") or 0), i.get("request_number") or "")) lines: list[str | LocalizedText] = [ f"[b]{i['request_number']}[/b] · {i['provider_name']} · {i['holder_label']} · " f"{i['streak']}m" for i in ranked[:_DIGEST_MAX_ROWS] ] if len(ranked) > _DIGEST_MAX_ROWS: lines.append(f"… +{len(ranked) - _DIGEST_MAX_ROWS}") delivery = LocalizedText( zh="已开启" if settings.enabled_for_employees else "未开启(仅 IT 汇总)", en="ON" if settings.enabled_for_employees else "OFF (IT digest only)", ja="オン" if settings.enabled_for_employees else "オフ(IT 向け集計のみ)", ) svc = NotificationService(db) sent = 0 for aid in admin_ids: if aid in already: continue admin = await db.get(IdentityPerson, aid) if ( admin is None or admin.deleted_at is not None or (admin.employment_status or "").lower() != "active" or not admin.work_email ): continue lang = await _lang(admin) delivery_txt = {"zh": delivery.zh, "en": delivery.en, "ja": delivery.ja}[lang] title = { "zh": f"API Key 闲置月报:{len(items)} 把 Key 在 {period} 花费为 $0", "en": f"Idle API keys — {len(items)} key(s) had $0 spend in {period}", "ja": f"アイドル API キー月報:{period} に費用 $0 のキーが {len(items)} 件", }[lang] body = { "zh": f"{period} 有 {len(items)} 把 active 的 API Key 花费为 $0(其中 {len(streak2)} 把已连续 ≥2 个月)。员工提醒当前{delivery_txt}。", "en": f"{len(items)} active API key(s) had $0 spend in {period} ({len(streak2)} idle for 2+ months). Employee reminders are currently {delivery_txt}.", "ja": f"{period} に費用 $0 だった有効な API キーは {len(items)} 件(うち {len(streak2)} 件は 2 か月以上連続)。従業員へのリマインドは現在{delivery_txt}。", }[lang] # SAVEPOINT + ``step.reason()``, replacing ``except Exception`` + # ``str(exc)``. ``dispatch`` already self-isolates its own writes # (``enqueue`` and both of ``_deliver``'s flushes sit inside # ``begin_nested``), so this site was not poisoning the session — # the savepoint is defence in depth. The DEFECT was the error text: # ``str(exc)`` renders an ``AppException`` as an empty sentinel, so # the warning ended in a colon and nothing. async with best_effort( db, label=f"api-key idle digest to admin {admin.id}", log=logger, log_level=logging.DEBUG, ) as step: await svc.dispatch( NotificationEnvelope( recipient=admin, title=title, body_text=body, channels=["feishu", "email"], template_key=TEMPLATE_IT_DIGEST, context={ "period": period, "idle_count": str(len(items)), "streak2_count": str(len(streak2)), "employee_delivery": delivery_txt, }, url="/admin/assets/api-keys?idle=1", card_stats=[ CardStat( label=LocalizedText(zh="闲置 Key", en="Idle keys", ja="アイドルキー"), value=str(len(items)), ), CardStat( label=LocalizedText(zh="连续 ≥2 月", en="2+ months", ja="2 か月以上"), value=str(len(streak2)), ), ], card_highlights=[CardHighlight( label=LocalizedText(zh="最久未用", en="Longest idle", ja="最長アイドル"), items=lines, )], card_actions=[CardAction( label=LocalizedText(zh="打开 API Key 管理台", en="Open the API key console", ja="API キー管理台を開く"), url="/admin/assets/api-keys?idle=1", style="primary", )], related_resource_type=DIGEST_RESOURCE_TYPE, related_resource_id=digest_id, ), trace_id=trace_id, actor_user_id=None, ) if step.ok: sent += 1 else: logger.warning( "apikey idle digest to %s failed: %s", admin.work_email, step.reason(), ) return sent async def sweep_idle( db: AsyncSession, *, today: date, settings: IdleSettings, trace_id: str, ) -> dict[str, Any]: """The cron body: evaluate every active key for last month, page holders (when enabled), and return the idle roster for the IT digest.""" now = datetime.now(UTC) month_start, month_end, period = previous_month(today) summary: dict[str, Any] = { "period": period, "evaluated": 0, "idle": 0, "active": 0, "skipped_too_new": 0, "skipped_kept": 0, "skipped_no_data": 0, "employee_cards": 0, "errors": [], "idle_items": [], "employee_delivery": settings.enabled_for_employees, } rows = (await db.execute( select(ApiKeyAssignment, ApiKeyProvider) .join(ApiKeyProvider, ApiKeyProvider.id == ApiKeyAssignment.provider_id) .where(ApiKeyAssignment.status == "active") )).all() for assignment, provider in rows: summary["evaluated"] += 1 # Read the id BEFORE the savepoint: a rollback expires every object # the body dirtied, and an expired attribute read outside an await # raises ``MissingGreenlet``. assignment_id = assignment.id # SAVEPOINT, not a bare ``except``: ``evaluate_assignment`` and # ``notify_holder`` both write. A swallowed flush failure deactivates # the whole transaction, so "one key must not stop the sweep" was a # lie — every REMAINING key then failed with ``PendingRollbackError`` # and the sweep silently produced a truncated idle roster. async with best_effort( db, label=f"idle check for api key assignment {assignment_id}", log=logger, log_level=logging.DEBUG, # the call site logs it with a traceback ) as step: result = await evaluate_assignment( db, assignment=assignment, month_start=month_start, month_end=month_end, period=period, settings=settings, now=now, ) res = result["result"] if res in ("skipped_too_new", "skipped_kept", "skipped_no_data", "active"): summary[res] += 1 continue summary["idle"] += 1 holder = await db.get(IdentityPerson, assignment.assigned_to) item = { "assignment_id": str(assignment.id), "request_number": assignment.request_number, "provider_name": provider.name, "project_name": assignment.project_name, "holder_label": ( (holder.display_name or holder.work_email) if holder else str(assignment.assigned_to) ), "streak": result["streak"], } summary["idle_items"].append(item) if settings.enabled_for_employees: dedup = (await db.execute( select(ApiKeyIdleAlertsDedup).where( ApiKeyIdleAlertsDedup.assignment_id == assignment.id, ApiKeyIdleAlertsDedup.billing_month == period, ) )).scalar_one_or_none() if dedup is not None and not dedup.employee_notified: sent = await notify_holder( db, assignment=assignment, provider_name=provider.name, period=period, streak=result["streak"], settings=settings, trace_id=trace_id, ) if sent: dedup.employee_notified = True summary["employee_cards"] += 1 await db.flush() if not step.ok: logger.warning( "apikey.idle_check: assignment %s failed: %s", assignment_id, step.reason(), exc_info=step.error, ) summary["errors"].append({ "assignment_id": str(assignment_id), "error": step.reason(max_len=200) or "", }) return summary __all__ = [ "DIGEST_RESOURCE_TYPE", "IdleSettings", "SETTING_KEY", "TEMPLATE_EMPLOYEE", "TEMPLATE_IT_DIGEST", "evaluate_assignment", "keep_idle_assignment", "list_idle_assignments", "load_idle_settings", "notify_holder", "previous_month", "save_idle_settings", "send_it_digest", "sweep_idle", ]