"""Permission-rechecked cloud directory reports and bounded AI interpretation.""" from __future__ import annotations import hashlib import json import uuid from calendar import monthrange from datetime import UTC, date, datetime, timedelta from typing import Any from zoneinfo import ZoneInfo, ZoneInfoNotFoundError from sqlalchemy import and_, case, delete, exists, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.core.dependencies import _load_user_with_roles from app.core.exceptions import format_exception_for_log from app.core.locale import Lang, resolve_lang from app.core.rbac import can_manage_cloud, can_view_cloud_resources from app.core.safe_logging import get_safe_logger from app.domains.cmdb.models import ( CmdbCloudResource, CmdbCloudResourceChange, CmdbResourceFinding, CmdbResourceGovernance, CmdbResourceGovernanceRule, CmdbResourceReportRun, CmdbResourceReportSubscription, ) from app.domains.cmdb.resource_directory_ownership import inherited_scope_owner_exists from app.domains.cmdb.resource_directory_visibility import ( DirectoryVisibility, directory_visibility_clause, resolve_directory_visibility, ) from app.identity.person_models import IdentityPerson from app.integrations.service import get_platform_setting from app.notifications.service import NotificationEnvelope, NotificationService logger = get_safe_logger(__name__) _ACTIVE_FINDINGS = ("open", "acknowledged") _ALLOWED_FILTERS = frozenset({"providers", "categories", "resource_types", "integration_id"}) def next_delivery_at( *, frequency: str, timezone: str, delivery_hour: int, weekly_day: int, monthly_day: int, after: datetime | None = None, ) -> datetime | None: if frequency == "off": return None try: zone = ZoneInfo(timezone) except ZoneInfoNotFoundError as exc: raise ValueError("Unknown IANA timezone") from exc now = (after or datetime.now(UTC)).astimezone(zone) if frequency == "weekly": days = (weekly_day - now.weekday()) % 7 candidate = (now + timedelta(days=days)).replace( hour=delivery_hour, minute=0, second=0, microsecond=0, ) if candidate <= now: candidate += timedelta(days=7) elif frequency == "monthly": day = min(monthly_day, monthrange(now.year, now.month)[1]) candidate = now.replace(day=day, hour=delivery_hour, minute=0, second=0, microsecond=0) if candidate <= now: year = now.year + (1 if now.month == 12 else 0) month = 1 if now.month == 12 else now.month + 1 day = min(monthly_day, monthrange(year, month)[1]) candidate = candidate.replace(year=year, month=month, day=day) else: raise ValueError("frequency must be off, weekly, or monthly") return candidate.astimezone(UTC).replace(tzinfo=None) def validate_scope_config(value: dict[str, Any] | None) -> dict[str, Any]: config = value or {} unknown = set(config) - _ALLOWED_FILTERS if unknown: raise ValueError(f"Unsupported report filters: {', '.join(sorted(unknown))}") clean: dict[str, Any] = {} for key in ("providers", "categories", "resource_types"): values = config.get(key) or [] if not isinstance(values, list) or len(values) > 50: raise ValueError(f"{key} must be a list with at most 50 values") if any(not isinstance(item, str) or not item.strip() or len(item) > 192 for item in values): raise ValueError(f"{key} contains an invalid value") if values: clean[key] = list(dict.fromkeys(item.strip() for item in values)) integration_id = config.get("integration_id") if integration_id: try: clean["integration_id"] = str(uuid.UUID(str(integration_id))) except ValueError as exc: raise ValueError("integration_id must be a UUID") from exc if len(json.dumps(clean, ensure_ascii=False)) > 16_384: raise ValueError("Report scope is too large") return clean def _resource_clauses(vis: DirectoryVisibility, scope: dict[str, Any]) -> list[Any]: clauses: list[Any] = [ directory_visibility_clause(vis), CmdbCloudResource.deleted_at.is_(None), ] mapping = { "providers": CmdbCloudResource.provider, "categories": CmdbCloudResource.category, "resource_types": CmdbCloudResource.resource_type, } for key, column in mapping.items(): if scope.get(key): clauses.append(column.in_(scope[key])) if scope.get("integration_id"): clauses.append(CmdbCloudResource.integration_id == scope["integration_id"]) return clauses async def build_report_data( db: AsyncSession, *, vis: DirectoryVisibility, scope: dict[str, Any], days: int, ) -> dict[str, Any]: days = max(1, min(days, 366)) end = datetime.now(UTC).date() start = end - timedelta(days=days - 1) clauses = _resource_clauses(vis, scope) governed_owner = exists(select(CmdbResourceGovernance.id).where( CmdbResourceGovernance.resource_id == CmdbCloudResource.id, or_( CmdbResourceGovernance.business_owner_person_id.is_not(None), CmdbResourceGovernance.technical_owner_person_id.is_not(None), ), ).correlate(CmdbCloudResource)) totals = (await db.execute(select( func.count(CmdbCloudResource.id), func.sum(case((CmdbCloudResource.sensitivity == "restricted", 1), else_=0)), func.sum(case((and_( CmdbCloudResource.owner_person_id.is_(None), ~governed_owner, ~inherited_scope_owner_exists(), ), 1), else_=0)), ).where(*clauses))).one() async def grouped(column: Any, limit: int = 20) -> list[dict[str, Any]]: rows = (await db.execute(select(column, func.count(CmdbCloudResource.id)).where( *clauses, ).group_by(column).order_by(func.count(CmdbCloudResource.id).desc()).limit(limit))).all() return [{"key": str(key or "unknown"), "count": int(count)} for key, count in rows] finding_rows = (await db.execute(select( CmdbResourceFinding.severity, func.count(CmdbResourceFinding.id), ).join( CmdbCloudResource, CmdbCloudResource.id == CmdbResourceFinding.resource_id, ).where( *clauses, CmdbResourceFinding.status.in_(_ACTIVE_FINDINGS), ).group_by(CmdbResourceFinding.severity))).all() finding_rule_rows = (await db.execute(select( CmdbResourceGovernanceRule.rule_key, func.count(CmdbResourceFinding.id), ).join( CmdbResourceFinding, CmdbResourceFinding.rule_id == CmdbResourceGovernanceRule.id, ).join( CmdbCloudResource, CmdbCloudResource.id == CmdbResourceFinding.resource_id, ).where( *clauses, CmdbResourceFinding.status.in_(_ACTIVE_FINDINGS), ).group_by(CmdbResourceGovernanceRule.rule_key).order_by( func.count(CmdbResourceFinding.id).desc(), ).limit(30))).all() change_rows = (await db.execute(select( CmdbCloudResourceChange.change_type, func.count(CmdbCloudResourceChange.id), ).join( CmdbCloudResource, CmdbCloudResource.id == CmdbCloudResourceChange.resource_id, ).where( *clauses, CmdbCloudResourceChange.observed_at >= datetime.combine(start, datetime.min.time()), ).group_by(CmdbCloudResourceChange.change_type))).all() change_resource_type_rows = (await db.execute(select( CmdbCloudResource.resource_type, func.count(CmdbCloudResourceChange.id), ).join( CmdbCloudResourceChange, CmdbCloudResourceChange.resource_id == CmdbCloudResource.id, ).where( *clauses, CmdbCloudResourceChange.observed_at >= datetime.combine( start, datetime.min.time(), ), ).group_by(CmdbCloudResource.resource_type).order_by( func.count(CmdbCloudResourceChange.id).desc(), ).limit(30))).all() findings_by_rule = { str(key): int(count) for key, count in finding_rule_rows } return { "window": {"start": start.isoformat(), "end": end.isoformat(), "days": days}, "totals": { "resources": int(totals[0] or 0), "restricted": int(totals[1] or 0), "unowned": int(totals[2] or 0), "open_findings": sum(int(count) for _, count in finding_rows), }, "by_provider": await grouped(CmdbCloudResource.provider), "by_category": await grouped(CmdbCloudResource.category), "by_resource_type": await grouped(CmdbCloudResource.resource_type, 30), "by_lifecycle_status": await grouped(CmdbCloudResource.lifecycle_status), "by_environment": await grouped(CmdbCloudResource.environment), "findings_by_severity": [ {"key": str(key), "count": int(count)} for key, count in finding_rows ], "findings_by_rule": [ {"key": str(key), "count": int(count)} for key, count in finding_rule_rows ], "coverage_gaps": { "monitoring_missing": findings_by_rule.get( "resource.monitoring_missing", 0, ), "cost_unavailable": findings_by_rule.get( "resource.cost_unavailable", 0, ), }, "changes": {str(key): int(count) for key, count in change_rows}, "changes_by_resource_type": [ {"key": str(key), "count": int(count)} for key, count in change_resource_type_rows ], "generated_at": datetime.now(UTC).isoformat(), } def deterministic_interpretation(data: dict[str, Any], *, lang: Lang = "zh") -> str: totals = data["totals"] changes = data.get("changes", {}) changed = int(changes.get("changed", 0)) + int(changes.get("discovered", 0)) if not totals["resources"]: return { "zh": "当前权限与筛选范围内没有云资源。请先确认云连接、扫描状态或资源授权范围。", "en": "No cloud resources are visible in the current permission and filter scope. Check the cloud connection, scan status, or resource authorization.", "ja": "現在の権限とフィルター範囲にはクラウドリソースがありません。クラウド接続、スキャン状態、またはリソース権限を確認してください。", }[lang] templates = { "zh": ( "优先处理 {findings} 个未关闭治理发现", "补齐 {unowned} 个无明确责任人的资源", "复核 {restricted} 个敏感资源的分级与最小权限", "核对窗口内 {changed} 次新增或变化", "补齐 {monitoring} 个应监控资源的数据链路", "修复 {cost} 个资源所属连接的成本数据", "当前资源目录没有显著治理缺口；建议继续关注扫描新鲜度和资源变化趋势。", "；", "。", ), "en": ( "Prioritize {findings} open governance findings", "Assign accountable owners to {unowned} resources", "Review classification and least privilege for {restricted} sensitive resources", "Review {changed} discoveries or changes in the window", "Restore monitoring data for {monitoring} resources that should be monitored", "Restore cost data for {cost} resource connections", "No material governance gap is visible; continue monitoring scan freshness and resource-change trends.", "; ", ".", ), "ja": ( "未解決のガバナンス指摘 {findings} 件を優先対応", "責任者未設定のリソース {unowned} 件を割り当て", "機密リソース {restricted} 件の分類と最小権限を確認", "期間内の新規・変更 {changed} 件を確認", "監視対象リソース {monitoring} 件のデータ経路を復旧", "{cost} 件のリソース接続でコストデータを復旧", "重大なガバナンス上の欠落はありません。スキャン鮮度と変更傾向の監視を継続してください。", "；", "。", ), }[lang] priorities: list[str] = [] if totals["open_findings"]: priorities.append(templates[0].format(findings=totals["open_findings"])) if totals["unowned"]: priorities.append(templates[1].format(unowned=totals["unowned"])) if totals["restricted"]: priorities.append(templates[2].format(restricted=totals["restricted"])) if changed: priorities.append(templates[3].format(changed=changed)) coverage_gaps = data.get("coverage_gaps") or {} if int(coverage_gaps.get("monitoring_missing", 0)): priorities.append(templates[4].format( monitoring=int(coverage_gaps["monitoring_missing"]), )) if int(coverage_gaps.get("cost_unavailable", 0)): priorities.append(templates[5].format( cost=int(coverage_gaps["cost_unavailable"]), )) if not priorities: return templates[6] return templates[7].join(priorities[:4]) + templates[8] async def _ai_interpretation( db: AsyncSession, data: dict[str, Any], *, lang: Lang, ) -> tuple[str | None, str]: api_key = await get_platform_setting("ai_spending.llm.api_key", None, session=db) if not isinstance(api_key, str) or not api_key.strip(): return None, "not_configured" model = await get_platform_setting( "ai_spending.llm.model", "qwen-max-latest", session=db, ) region = await get_platform_setting("ai_spending.llm.region", None, session=db) from app.domains.bailian.llm_client import BailianLLMClient client = BailianLLMClient(api_key.strip(), model=str(model), region=region, timeout=30.0, purpose="cloud.resource_report") facts = json.dumps(data, ensure_ascii=False, sort_keys=True, separators=(",", ":")) language_name = {"zh": "Chinese", "en": "English", "ja": "Japanese"}[lang] try: text = await client.chat_text( system=( "You are an enterprise cloud-governance analyst. Use only the supplied aggregate facts. " f"Write in {language_name}, within 300 words: " "lead with the conclusion, then at most three priority actions. Never invent amounts, causes, " "compliance conclusions, or resource names." ), user=facts[:20_000], max_tokens=600, temperature=0.2, ) except Exception as exc: # noqa: BLE001 logger.warning("cloud resource report AI interpretation failed: %s", str(exc)[:200]) return None, "error" return (text or "").strip()[:4_000] or None, "generated" async def generate_report_run( db: AsyncSession, *, user: IdentityPerson, trigger: str, subscription: CmdbResourceReportSubscription | None = None, scope: dict[str, Any] | None = None, days: int = 30, include_ai: bool = False, ) -> CmdbResourceReportRun: role_names = user.role_names if not can_view_cloud_resources(role_names): raise PermissionError("Cloud directory access is no longer available") vis = await resolve_directory_visibility(db, user_id=str(user.id), role_names=role_names) scope = validate_scope_config(scope) data = await build_report_data(db, vis=vis, scope=scope, days=days) lang = resolve_lang(user) interpretation = deterministic_interpretation(data, lang=lang) hash_data = { "facts": {key: value for key, value in data.items() if key != "generated_at"}, # AI output is language-specific. Including the resolved locale keeps a # user who changes language from receiving a 24-hour cached narrative # in the previous language while preserving deterministic fact reuse. "language": lang, } fact_hash = hashlib.sha256( json.dumps(hash_data, sort_keys=True, separators=(",", ":")).encode(), ).hexdigest() ai_text: str | None = None ai_status = "not_requested" # AI consumes the already permission-scoped aggregate facts produced # above. It has no broader retrieval path, so ordinary employees may use # it safely for resources they can already see. if include_ai: cached = await db.scalar(select(CmdbResourceReportRun).where( CmdbResourceReportRun.recipient_person_id == str(user.id), CmdbResourceReportRun.fact_hash == fact_hash, CmdbResourceReportRun.ai_interpretation.is_not(None), CmdbResourceReportRun.generated_at >= ( datetime.now(UTC).replace(tzinfo=None) - timedelta(hours=24) ), ).order_by(CmdbResourceReportRun.generated_at.desc()).limit(1)) if cached is not None: ai_text, ai_status = cached.ai_interpretation, "cached" else: ai_text, ai_status = await _ai_interpretation(db, data, lang=lang) permission_fingerprint = hashlib.sha256(json.dumps({ "user_id": str(user.id), "roles": sorted(role_names), "full": vis.full, "person_ids": sorted(vis.person_ids), "evaluated_at": data["generated_at"], }, sort_keys=True, separators=(",", ":")).encode()).hexdigest() run = CmdbResourceReportRun( subscription_id=str(subscription.id) if subscription else None, recipient_person_id=str(user.id), trigger=trigger, status="generated", window_start=date.fromisoformat(data["window"]["start"]), window_end=date.fromisoformat(data["window"]["end"]), scope_snapshot=scope, report_data=data, interpretation=interpretation, ai_interpretation=ai_text, ai_status=ai_status, fact_hash=fact_hash, permission_fingerprint=permission_fingerprint, generated_at=datetime.now(UTC).replace(tzinfo=None), created_at=datetime.now(UTC).replace(tzinfo=None), ) db.add(run) await db.flush() return run async def dispatch_due_reports(db: AsyncSession, *, limit: int = 100) -> dict[str, int]: now = datetime.now(UTC).replace(tzinfo=None) expired_ids = list((await db.scalars(select(CmdbResourceReportRun.id).where( CmdbResourceReportRun.created_at < now - timedelta(days=730), ).order_by(CmdbResourceReportRun.created_at).limit(1_000))).all()) if expired_ids: await db.execute(delete(CmdbResourceReportRun).where( CmdbResourceReportRun.id.in_(expired_ids), )) await db.commit() summary = { "due": 0, "generated": 0, "queued": 0, "skipped": 0, "failed": 0, "pruned": len(expired_ids), } # Lock and finish one due subscription per transaction. Committing inside # a preloaded multi-row result would release the locks for every remaining # row and let a concurrent scheduler enqueue the same report twice. for _ in range(limit): subscription = await db.scalar(select(CmdbResourceReportSubscription).where( CmdbResourceReportSubscription.frequency.in_(("weekly", "monthly")), CmdbResourceReportSubscription.next_run_at.is_not(None), CmdbResourceReportSubscription.next_run_at <= now, ).order_by(CmdbResourceReportSubscription.next_run_at).limit(1).with_for_update( skip_locked=True, )) if subscription is None: break subscription_id = str(subscription.id) person_id = subscription.person_id scope_snapshot = dict(subscription.scope_config or {}) summary["due"] += 1 try: user = await _load_user_with_roles(db, uuid.UUID(subscription.person_id)) if user is None or user.employment_status == "left": subscription.frequency = "off" subscription.next_run_at = None summary["skipped"] += 1 await db.commit() continue days = 7 if subscription.frequency == "weekly" else 31 run = await generate_report_run( db, user=user, trigger="scheduled", subscription=subscription, scope=subscription.scope_config or {}, days=days, include_ai=subscription.include_ai, ) data = run.report_data totals = data["totals"] path = "/admin/cloud/resources?tab=reports" if can_manage_cloud(user.role_names) else "/my/cloud/resources?tab=reports" lang = resolve_lang(user) notification_copy = { "zh": { "weekly": "云资源目录周报", "monthly": "云资源目录月报", "body": "资源 {resources} 个，期间变化 {changes} 次，未关闭发现 {findings} 个，无责任人 {unowned} 个，应监控但缺数据 {monitoring} 个，成本数据不可用 {cost} 个。{interpretation}", }, "en": { "weekly": "Cloud resource directory weekly report", "monthly": "Cloud resource directory monthly report", "body": "{resources} resources, {changes} changes in the period, {findings} open findings, {unowned} without an owner, {monitoring} expected monitoring gaps, and {cost} unavailable cost-data connections. {interpretation}", }, "ja": { "weekly": "クラウドリソースディレクトリ週次レポート", "monthly": "クラウドリソースディレクトリ月次レポート", "body": "リソース {resources} 件、期間内変更 {changes} 件、未解決の指摘 {findings} 件、責任者未設定 {unowned} 件、監視データ欠損 {monitoring} 件、コストデータ利用不可 {cost} 件。{interpretation}", }, }[lang] await NotificationService(db).enqueue(NotificationEnvelope( recipient=user, title=notification_copy[subscription.frequency], body_text=notification_copy["body"].format( resources=totals["resources"], findings=totals["open_findings"], unowned=totals["unowned"], changes=sum(int(value) for value in data.get("changes", {}).values()), monitoring=int(data.get("coverage_gaps", {}).get("monitoring_missing", 0)), cost=int(data.get("coverage_gaps", {}).get("cost_unavailable", 0)), interpretation=run.ai_interpretation or run.interpretation or "", )[:8_000], channels=list(subscription.channels or ["feishu"]), url=path, related_resource_type="cmdb_resource_report_run", related_resource_id=run.id, )) run.status = "queued" subscription.last_sent_at = now subscription.next_run_at = next_delivery_at( frequency=subscription.frequency, timezone=subscription.timezone, delivery_hour=subscription.delivery_hour, weekly_day=subscription.weekly_day, monthly_day=subscription.monthly_day, after=datetime.now(UTC) + timedelta(minutes=1), ) summary["generated"] += 1 summary["queued"] += 1 except PermissionError: subscription.frequency = "off" subscription.next_run_at = None summary["skipped"] += 1 except Exception as exc: # noqa: BLE001 error_message = format_exception_for_log(exc, max_len=2_000) logger.exception("resource report dispatch failed for subscription %s", subscription_id) # Discard any partially built report/outbox rows before recording # the isolated failure. A failed flush leaves AsyncSession unusable # until rollback, and committing partial notification state would # make the six-hour retry generate a duplicate report. await db.rollback() failed_subscription = await db.scalar(select( CmdbResourceReportSubscription, ).where( CmdbResourceReportSubscription.id == subscription_id, CmdbResourceReportSubscription.frequency.in_(("weekly", "monthly")), CmdbResourceReportSubscription.next_run_at.is_not(None), CmdbResourceReportSubscription.next_run_at <= now, ).with_for_update()) if failed_subscription is None: summary["skipped"] += 1 continue subscription = failed_subscription subscription.next_run_at = now + timedelta(hours=6) summary["failed"] += 1 db.add(CmdbResourceReportRun( subscription_id=subscription_id, recipient_person_id=person_id, trigger="scheduled", status="failed", window_start=now.date(), window_end=now.date(), scope_snapshot=scope_snapshot, report_data={}, fact_hash="0" * 64, permission_fingerprint="0" * 64, error_message=error_message, generated_at=now, created_at=now, )) await db.commit() return summary