"""AI diagnosis for the HR 洞察 tabs (人员洞察 / 入离职洞察). Turns the exact payload each tab renders (``build_people_payload`` / ``build_lifecycle_payload``) into a bilingual HR-analyst read via the platform Bailian LLM, cached per ``(section, range_key)`` in ``insights_hr_ai_summaries``. Mirrors the proven Platform-Impact summary pipeline (:mod:`app.dashboard.insights_impact_summary`) with two deltas: * **Per-window cache** — the HR tabs are range-parameterised, so a row is keyed on ``(section, range)`` rather than a single all-time period. * **No cron** — generation is on-demand (a global_admin / hr clicks 生成 on the AI panel); a missing row renders as an inviting empty state. LLM credentials are the platform-wide Bailian config already used by AI-spending / the impact summary (``ai_spending.llm.api_key`` / ``.model`` / ``.region``). The digests fed to the model are pure aggregate counts — never a person row, email, or name reaches the third-party LLM (the international-assignment / upcoming-leaver rosters are reduced to counts before the prompt is built). """ from __future__ import annotations import json import logging import uuid from datetime import UTC, datetime from typing import Any from zoneinfo import ZoneInfo from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.exceptions import AppException from app.dashboard.insights_models import InsightsHrAiSummary from app.domains.ai_spending.service import ( DEFAULT_MODEL, DEFAULT_REGION, SETTING_LLM_API_KEY, SETTING_LLM_MODEL, SETTING_LLM_REGION, ) from app.domains.bailian.llm_client import BailianLLMClient from app.integrations.models import PlatformSetting logger = logging.getLogger(__name__) SECTIONS = ("people", "lifecycle") SETTING_HR_AI_REGEN_LIMIT = "insights_hr.regenerate_per_user_per_day" _REGEN_LIMIT_DEFAULT = 20 # 2 sections × up to 5 ranges × a couple retries _COUNTER_KEY_PREFIX = "insights_hr.regen_counter" _TZ = ZoneInfo("Asia/Shanghai") _PEOPLE_SYSTEM_PROMPT = """\ You are the People Analytics partner for Helix, the company's unified IT platform. You write the diagnosis shown at the top of the HR "People insights" (人员洞察) tab, for an HR / People-Ops audience. You are given ONE JSON object of aggregate workforce statistics for a chosen time window: headcount and status split, hires/leavers/net over the window, an approximate attrition rate, tenure distribution, top locations / countries / job titles, the manager span-of-control distribution, department sizes, and international-assignment counts. Everything is an aggregate — there are no individual people in the data. Write like a sharp HR analyst briefing a People lead: concrete, specific, decision-useful, no filler. Lead with the single most important read on the workforce. Quote real numbers from the data — never invent, never extrapolate beyond what is given; call the attrition figure approximate. Look for things HR actually acts on: growth vs. churn balance, tenure concentration (lots of very-new hires = onboarding load + flight risk; lots of long-tenure with no juniors = succession risk), over-wide or over-narrow manager spans, geographic / departmental concentration. Respond with ONE JSON object: { "summary_md_zh": "2-4 short paragraphs of simplified-Chinese markdown", "summary_md_en": "the same narrative in English markdown", "highlights": [ { "title_zh": "...", "title_en": "...", "detail_zh": "one sentence citing a number", "detail_en": "one sentence citing a number", "tone": "strength" | "watch" | "risk" } ] } 3 to 5 highlights. `strength` = a healthy signal, `watch` = worth monitoring, `risk` = needs action now. Both languages carry the same facts. """ _LIFECYCLE_SYSTEM_PROMPT = """\ You are the People-Operations partner for Helix, the company's unified IT platform. You write the diagnosis shown at the top of the HR "Joiner/Mover/Leaver insights" (入离职洞察) tab, for an HR / IT-Ops audience. You are given ONE JSON object of aggregate onboarding & offboarding operations for a chosen time window: onboarding volume and how long provisioning takes (bucketed), offboarding process states, count of overdue blocked leavers, average offboarding completion days, the breakdown and aging of pending offboarding blockers (by resource type), recovered hardware value, and upcoming-leaver counts. Everything is an aggregate — no individual people. Write like an operations analyst who owns the joiner/leaver SLA: crisp, specific, action-oriented, no filler. Lead with the biggest operational risk or win. Quote real numbers. Focus on what Ops acts on: onboarding speed (are new hires ready on day one?), aging offboarding blockers (security exposure — access still live after someone left), overdue blocked leavers, which resource type jams offboarding most, and recovered-asset value. Respond with ONE JSON object: { "summary_md_zh": "2-4 short paragraphs of simplified-Chinese markdown", "summary_md_en": "the same narrative in English markdown", "highlights": [ { "title_zh": "...", "title_en": "...", "detail_zh": "one sentence citing a number", "detail_en": "one sentence citing a number", "tone": "strength" | "watch" | "risk" } ] } 3 to 5 highlights. `strength` = SLA / process running well, `watch` = worth monitoring, `risk` = needs action now (e.g. aging access blockers). Both languages carry the same facts. """ def _sum(d: dict[str, Any] | None, *keys: str) -> int: if not isinstance(d, dict): return 0 return sum(int(d.get(k, 0) or 0) for k in keys) def _slim_people(payload: dict[str, Any], rng: str) -> dict[str, Any]: """Compact, PII-free projection of the People payload for the LLM.""" trend = payload.get("trend") or [] hires_window = sum(int(b.get("hires", 0)) for b in trend) leavers_window = sum(int(b.get("leavers", 0)) for b in trend) headcount = int(payload.get("headcount", 0) or 0) tenure = payload.get("tenure_distribution") or {} span = payload.get("manager_span") or {} span_buckets = span.get("buckets") or {} def _top(d: dict[str, Any] | None, n: int) -> dict[str, int]: if not isinstance(d, dict): return {} return dict( sorted(d.items(), key=lambda kv: kv[1], reverse=True)[:n] ) depts = payload.get("by_department_tree") or [] top_departments = sorted( ( {"name": r.get("department"), "total": int(r.get("total_count", 0))} for r in depts if r.get("depth", 0) >= 1 ), key=lambda r: r["total"], reverse=True, )[:6] return { "range": rng, "headcount": headcount, "by_status": payload.get("by_status") or {}, "window": { "hires": hires_window, "leavers": leavers_window, "net": hires_window - leavers_window, }, "approx_attrition_rate": ( round(leavers_window / headcount, 3) if headcount else None ), "tenure_distribution": tenure, "new_hire_share_lt_3m": ( round(_sum(tenure, "<1m", "1-3m") / headcount, 3) if headcount else None ), # Bands split at the badge-wall milestones (5-7y / 7-10y / 10y+) since # 2026-08-19; the sum keeps this metric's meaning ("满 5 年及以上"). "tenure_5y_plus": int(_sum(tenure, "5-7y", "7-10y", "10y+")), "top_locations": _top(payload.get("by_location"), 6), "top_countries": _top(payload.get("by_country"), 6), "top_job_titles": _top(payload.get("by_job_title"), 8), "manager_span": { "managers_count": span.get("managers_count", 0), "p50": span.get("p50", 0), "p90": span.get("p90", 0), "max": span.get("max", 0), "wide_span_managers": _sum(span_buckets, "10-15", "16+"), "buckets": span_buckets, }, "top_departments": top_departments, "international_assignment_count": ( (payload.get("international_assignment") or {}).get("count", 0) ), } def _slim_lifecycle(payload: dict[str, Any], rng: str) -> dict[str, Any]: """Compact, PII-free projection of the Lifecycle payload.""" onb_trend = payload.get("onboarding_trend") or [] hires_window = sum(int(b.get("hires", 0)) for b in onb_trend) onb_buckets = payload.get("onboarding_buckets") or {} onb_total = sum(int(v or 0) for v in onb_buckets.values()) age_buckets = payload.get("blocker_age_buckets") or {} upcoming = payload.get("upcoming_leavers") or [] return { "range": rng, "onboarding": { "hires_window": hires_window, "avg_onboarding_days": payload.get("avg_onboarding_days", 0), "buckets": onb_buckets, "slow_share_gt_7d": ( round(_sum(onb_buckets, "7-14d", "14d+") / onb_total, 3) if onb_total else None ), }, "offboarding": { "status_counts": payload.get("offboarding_status_counts") or {}, "blocked_past_due": payload.get("blocked_past_due", 0), "avg_offboarding_days": payload.get("avg_offboarding_days", 0), }, "blockers": { "by_resource": payload.get("blockers_by_resource") or {}, "age_buckets": age_buckets, "aging_gt_7d": _sum(age_buckets, "7-14d", "14d+"), }, "recovered_asset_value": payload.get("recovered_asset_value", 0), "recovered_asset_count": payload.get("recovered_asset_count", 0), "upcoming_leavers": { "count": len(upcoming), "within_7d": sum( 1 for r in upcoming if int(r.get("days_left", 99)) <= 7 ), "with_pending_blockers": sum( 1 for r in upcoming if int(r.get("pending_blockers", 0)) > 0 ), }, } async def _build_digest( db: AsyncSession, section: str, rng: str, ) -> dict[str, Any]: """Fetch the tab payload + reduce it to the compact LLM digest.""" from app.dashboard.insights_router import ( # noqa: PLC0415 build_lifecycle_payload, build_people_payload, ) if section == "people": payload = await build_people_payload(db, rng) return _slim_people(payload, rng) payload = await build_lifecycle_payload(db, rng) return _slim_lifecycle(payload, rng) def _system_prompt(section: str) -> str: return ( _PEOPLE_SYSTEM_PROMPT if section == "people" else _LIFECYCLE_SYSTEM_PROMPT ) async def _get_setting(db: AsyncSession, key: str) -> Any: row = await db.get(PlatformSetting, key) return row.value if row else None def _row_to_dict(row: InsightsHrAiSummary | None) -> dict[str, Any]: if row is None: return {"available": False} return { "available": True, "section": row.section, "range": row.range_key, "summary_md_en": row.summary_md_en, "summary_md_zh": row.summary_md_zh, "highlights": row.highlights or [], "model_used": row.model_used, "generated_at": ( row.generated_at.isoformat() if row.generated_at else None ), } async def get_hr_ai_summary( db: AsyncSession, section: str, range_key: str, ) -> dict[str, Any]: """Cached diagnosis for one ``(section, range)`` — never generates.""" row = (await db.execute( select(InsightsHrAiSummary).where( InsightsHrAiSummary.section == section, InsightsHrAiSummary.range_key == range_key, ) )).scalar_one_or_none() return _row_to_dict(row) async def _consume_regen_quota(user_id: str) -> None: """Debit one regenerate for ``user_id`` today (Asia/Shanghai day). Committed in its OWN session so the debit survives a subsequent LLM failure (same anti-retry-storm property as the impact summary). Raises 429 once the daily limit is reached. """ from app.core.database import async_session_factory # noqa: PLC0415 today = datetime.now(_TZ).date().isoformat() counter_key = f"{_COUNTER_KEY_PREFIX}.{user_id}" async with async_session_factory() as session: limit_v = await _get_setting(session, SETTING_HR_AI_REGEN_LIMIT) try: limit = int(limit_v) if limit_v is not None else _REGEN_LIMIT_DEFAULT except (TypeError, ValueError): limit = _REGEN_LIMIT_DEFAULT row = await session.get( PlatformSetting, counter_key, with_for_update=True, populate_existing=True, ) state = row.value if row and isinstance(row.value, dict) else {} used = int(state.get("count", 0)) if state.get("date") == today else 0 if used >= limit: raise AppException( code="HR_AI_SUMMARY_REGEN_QUOTA", message=( f"Daily regenerate limit reached ({limit}/day). " "Try again tomorrow." ), status_code=429, ) new_value = {"date": today, "count": used + 1} if row is None: session.add(PlatformSetting(key=counter_key, value=new_value)) else: row.value = new_value await session.commit() async def generate_hr_ai_summary( db: AsyncSession, *, section: str, range_key: str, actor_id: uuid.UUID | None, trace_id: str, enforce_quota: bool, ) -> dict[str, Any]: """Build digest → prompt → Bailian → upsert the ``(section, range)`` row. Raises AppException with a stable code on every failure mode so the router surfaces a real reason. """ if section not in SECTIONS: raise AppException( code="HR_AI_SUMMARY_BAD_SECTION", message=f"Unknown insights section {section!r}.", status_code=400, ) api_key = (await _get_setting(db, SETTING_LLM_API_KEY) or "") if not isinstance(api_key, str) or not api_key.strip(): raise AppException( code="BAILIAN_LLM_NO_KEY", message=( "No Bailian API key configured " "(platform setting ai_spending.llm.api_key)." ), status_code=400, ) if enforce_quota: if actor_id is None: # defensive: quota implies a human actor raise AppException( code="HR_AI_SUMMARY_NO_ACTOR", message="Quota enforcement requires an actor.", status_code=400, ) await _consume_regen_quota(str(actor_id)) digest = await _build_digest(db, section, range_key) model_v = await _get_setting(db, SETTING_LLM_MODEL) model = model_v if isinstance(model_v, str) and model_v else DEFAULT_MODEL region_v = await _get_setting(db, SETTING_LLM_REGION) region = ( region_v if isinstance(region_v, str) and region_v else DEFAULT_REGION ) client = BailianLLMClient( api_key=api_key, model=model, region=region, timeout=120.0, purpose="dashboard.hr_insights") try: llm_json = await client.chat_json( system=_system_prompt(section), user=json.dumps(digest, ensure_ascii=False, default=str), max_tokens=3000, temperature=0.2, ) except AppException: raise except Exception as exc: # noqa: BLE001 raise AppException( code="BAILIAN_LLM_FAILED", message=f"Bailian LLM call failed: {exc}", status_code=502, ) from exc meta = llm_json.get("_meta") or {} highlights = llm_json.get("highlights") row = (await db.execute( select(InsightsHrAiSummary).where( InsightsHrAiSummary.section == section, InsightsHrAiSummary.range_key == range_key, ) )).scalar_one_or_none() if row is None: row = InsightsHrAiSummary(section=section, range_key=range_key) db.add(row) row.summary_md_en = str(llm_json.get("summary_md_en") or "") or None row.summary_md_zh = str(llm_json.get("summary_md_zh") or "") or None row.highlights = highlights if isinstance(highlights, list) else [] row.metrics = digest row.model_used = str(meta.get("model_used") or model) row.input_token_count = meta.get("input_tokens") row.output_token_count = meta.get("output_tokens") row.generated_by_user_id = actor_id row.generated_at = datetime.now(UTC) audit = AuditService(db) await audit.log( trace_id=trace_id, actor_type="user" if actor_id else "system", actor_user_id=actor_id, action="insights.hr_ai_summary.generate", resource_type="insights_hr_ai_summary", resource_id=f"{section}:{range_key}", status="success", detail={"section": section, "range": range_key, "model": row.model_used}, ) await db.flush() return _row_to_dict(row)