"""AI executive summary for the Platform Impact (平台成果) page. Generates a bilingual leadership narrative from the exact payload the page renders (``build_impact_payload``) via the Bailian LLM, and caches it in ``insights_impact_summaries``. Mirrors the AI-spending summary pipeline (the proven implementation in :mod:`app.domains.ai_spending.service`) with two deliberate deltas: * **No PII sanitisation needed** — the impact payload is pure aggregate counts and currency totals; nothing person-shaped ever reaches the third-party LLM. * **Quota defaults instead of a required setting** — manual regenerates are limited per-user-per-day; when the ``insights_impact.regenerate_per_user_per_day`` platform setting is absent we default to 5 rather than failing, because unlike AI-spending this surface has no settings admin page of its own. LLM credentials are the platform-wide Bailian config already used by AI-spending (``ai_spending.llm.api_key`` / ``.model`` / ``.region``) — one LLM configuration for the whole platform, not a second key to provision and rotate. """ 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 InsightsImpactSummary 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__) SETTING_IMPACT_REGEN_LIMIT = "insights_impact.regenerate_per_user_per_day" _REGEN_LIMIT_DEFAULT = 5 _COUNTER_KEY_PREFIX = "insights_impact.regen_counter" _TZ = ZoneInfo("Asia/Shanghai") _SYSTEM_PROMPT = """\ You are the IT operations chief-of-staff for Helix, a unified internal IT platform (people, hardware, software seats, cloud accounts and permissions, API keys, onboarding/offboarding automation, security governance). You write the executive summary shown at the top of the leadership-facing "Platform Impact" page. You are given one JSON object with the page's own aggregates: `scale` (live managed inventory), `reclaim` (cumulative access/asset recycling since launch, plus `recent_30d` momentum), `flows` (workflows served), `security`, `automation`, `savings` (estimated, annualized), and the cumulative growth series. Write for a CEO/CIO audience: concrete, factual, confident, zero filler. Lead with the single most impressive fact. Quote real numbers from the data — never invent, never extrapolate beyond what is given. Treat `savings` as an estimate and say so when citing it. Mention recent momentum (the 30-day numbers) so the story is "still compounding", not a static stock. Respond with ONE JSON object: { "summary_md_zh": "3-5 short paragraphs of simplified Chinese markdown", "summary_md_en": "the same narrative in English markdown", "highlights": [ { "title_zh": "...", "title_en": "...", "detail_zh": "one sentence", "detail_en": "one sentence", "tone": "win" | "governance" | "momentum" } ] } 3 to 5 highlights. `win` = cost/efficiency achievement, `governance` = security/compliance achievement, `momentum` = recent-30d movement. Both languages must carry the same facts. """ def _build_user_prompt(payload: dict[str, Any]) -> str: """Compact, deterministic projection of the page payload. Trend arrays are trimmed to the cumulative tail — the narrative needs "where we are + recent slope", not 12 months of buckets, and a smaller prompt is cheaper and harder for the model to misread. """ cumulative = payload.get("cumulative") or [] slim = { "as_of": payload.get("as_of"), "scale": payload.get("scale"), "reclaim": payload.get("reclaim"), "flows": payload.get("flows"), "security": payload.get("security"), "automation": payload.get("automation"), "savings": payload.get("savings"), "cumulative_latest": cumulative[-1] if cumulative else None, "cumulative_3_buckets_ago": ( cumulative[-4] if len(cumulative) >= 4 else None ), } return json.dumps(slim, ensure_ascii=False, default=str) 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: InsightsImpactSummary | None) -> dict[str, Any]: if row is None: return {"available": False} return { "available": True, "period": row.period, "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_impact_summary(db: AsyncSession) -> dict[str, Any]: """Newest cached summary, regardless of period.""" row = (await db.execute( select(InsightsImpactSummary) .order_by(InsightsImpactSummary.generated_at.desc()) .limit(1) )).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 AI-spending's quota. 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_IMPACT_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="IMPACT_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_impact_summary( db: AsyncSession, *, actor_id: uuid.UUID | None, trace_id: str, enforce_quota: bool, ) -> dict[str, Any]: """Build payload → prompt → Bailian → upsert today's bilingual row. ``actor_id=None`` is the scheduled-job path (no quota). Raises AppException with a stable code on every failure mode so both the router and the cron handler surface real reasons. """ 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="IMPACT_SUMMARY_NO_ACTOR", message="Quota enforcement requires an actor.", status_code=400, ) await _consume_regen_quota(str(actor_id)) from app.dashboard.insights_router import build_impact_payload # noqa: PLC0415 payload = await build_impact_payload(db, "12m") 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.impact_summary") try: llm_json = await client.chat_json( system=_SYSTEM_PROMPT, user=_build_user_prompt(payload), max_tokens=4096, 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 {} period = str(payload.get("as_of") or datetime.now(UTC).date().isoformat()) highlights = llm_json.get("highlights") row = (await db.execute( select(InsightsImpactSummary).where( InsightsImpactSummary.period == period, InsightsImpactSummary.language == "bilingual", ) )).scalar_one_or_none() if row is None: row = InsightsImpactSummary(period=period, language="bilingual") 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.aggregates = { "reclaim": payload.get("reclaim"), "savings": payload.get("savings"), } 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.impact_summary.generate", resource_type="insights_impact_summary", resource_id=period, status="success", detail={"period": period, "model": row.model_used}, ) await db.flush() return _row_to_dict(row)