"""Report subscriptions, frozen report history, CSV export and AI-assisted interpretation.""" from __future__ import annotations import csv import io import uuid from datetime import UTC, datetime from typing import Any, Literal from fastapi import APIRouter, Depends, Query, Request from fastapi.encoders import jsonable_encoder from fastapi.responses import StreamingResponse from pydantic import BaseModel, Field, field_validator from sqlalchemy import select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.auth.models import AuthUser from app.core.database import get_db from app.core.dependencies import get_current_user from app.core.dependencies import require_active_employee_for_mutations as _leaver_gate from app.core.exceptions import ConflictError, ForbiddenError, NotFoundError, ValidationError from app.core.idempotent_route import idempotent from app.core.rbac import can_view_cloud_resources from app.core.response import paginated_response, success_response from app.domains.cmdb.models import ( CmdbCloudResource, CmdbResourceGovernance, CmdbResourceReportRun, CmdbResourceReportSubscription, ) from app.domains.cmdb.resource_directory_visibility import ( directory_visibility_clause, resolve_directory_visibility, ) from app.domains.cmdb.resource_reporting import ( generate_report_run, next_delivery_at, validate_scope_config, ) router = APIRouter( prefix="/api/v1/cmdb/resource-directory/reports", tags=["cmdb-resource-directory-reports"], dependencies=[Depends(_leaver_gate)], ) class SubscriptionUpdate(BaseModel): version: int | None = Field(default=None, ge=1) frequency: Literal["off", "weekly", "monthly"] channels: list[Literal["feishu", "email"]] = Field(default=["feishu"]) timezone: str = Field(default="Asia/Shanghai", min_length=1, max_length=64) delivery_hour: int = Field(default=9, ge=0, le=23) weekly_day: int = Field(default=0, ge=0, le=6) monthly_day: int = Field(default=1, ge=1, le=28) include_ai: bool = False scope_config: dict[str, Any] | None = None @field_validator("channels") @classmethod def unique_channels(cls, value: list[str]) -> list[str]: if len(value) > 2: raise ValueError("At most two channels are supported") return list(dict.fromkeys(value)) class GenerateReport(BaseModel): days: int = Field(default=30, ge=1, le=366) include_ai: bool = False scope_config: dict[str, Any] | None = None def _subscription_dict(row: CmdbResourceReportSubscription | None) -> dict[str, Any]: if row is None: return { "version": None, "frequency": "off", "channels": ["feishu"], "timezone": "Asia/Shanghai", "delivery_hour": 9, "weekly_day": 0, "monthly_day": 1, "include_ai": False, "scope_config": {}, "next_run_at": None, "last_sent_at": None, } return { "id": str(row.id), "version": row.version, "frequency": row.frequency, "channels": row.channels or ["feishu"], "timezone": row.timezone, "delivery_hour": row.delivery_hour, "weekly_day": row.weekly_day, "monthly_day": row.monthly_day, "include_ai": row.include_ai, "scope_config": row.scope_config or {}, "next_run_at": row.next_run_at, "last_sent_at": row.last_sent_at, "updated_at": row.updated_at, } def _run_dict(row: CmdbResourceReportRun, *, include_data: bool = True) -> dict[str, Any]: data = { "id": str(row.id), "trigger": row.trigger, "status": row.status, "window_start": row.window_start, "window_end": row.window_end, "scope_snapshot": row.scope_snapshot or {}, "interpretation": row.interpretation, "ai_interpretation": row.ai_interpretation, "ai_status": row.ai_status, "fact_hash": row.fact_hash, "generated_at": row.generated_at, "delivered_at": row.delivered_at, "created_at": row.created_at, } if include_data: data["report_data"] = row.report_data or {} return data def _assert_access(user: AuthUser) -> None: if not can_view_cloud_resources(user.role_names): raise ForbiddenError() @router.get("/subscription") async def get_subscription( request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _assert_access(user) row = await db.scalar(select(CmdbResourceReportSubscription).where( CmdbResourceReportSubscription.person_id == str(user.id), )) return success_response(data=_subscription_dict(row), request=request) @router.put("/subscription") @idempotent("cmdb.resource_report_subscription.update", key_params=("user.id",)) async def put_subscription( body: SubscriptionUpdate, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _assert_access(user) try: scope = validate_scope_config(body.scope_config) next_run = next_delivery_at( frequency=body.frequency, timezone=body.timezone, delivery_hour=body.delivery_hour, weekly_day=body.weekly_day, monthly_day=body.monthly_day, ) except ValueError as exc: raise ValidationError(str(exc)) from exc if body.frequency != "off" and not body.channels: raise ValidationError("An active subscription requires at least one channel") row = await db.scalar(select(CmdbResourceReportSubscription).where( CmdbResourceReportSubscription.person_id == str(user.id), ).with_for_update()) if row is not None and body.version != row.version: raise ConflictError("Report subscription changed; reload before saving") before = _subscription_dict(row) now = datetime.now(UTC).replace(tzinfo=None) if row is None: row = CmdbResourceReportSubscription( person_id=str(user.id), created_at=now, updated_at=now, ) db.add(row) row.frequency = body.frequency row.channels = body.channels row.timezone = body.timezone row.delivery_hour = body.delivery_hour row.weekly_day = body.weekly_day row.monthly_day = body.monthly_day row.include_ai = body.include_ai row.scope_config = scope row.next_run_at = next_run row.updated_at = now row.version = row.version + 1 if before.get("version") else 1 try: await db.flush() except IntegrityError as exc: await db.rollback() raise ConflictError("Report subscription was created concurrently") from exc after = _subscription_dict(row) await AuditService(db).log( trace_id=getattr(request.state, "trace_id", str(uuid.uuid4())), actor_type="user", actor_user_id=user.id, action="cmdb.resource_report_subscription.update", status="success", resource_type="cmdb_resource_report_subscription", resource_id=str(row.id), detail=jsonable_encoder({"before": before, "after": after}), ) await db.commit() return success_response(data=after, request=request) @router.post("/generate", status_code=201) @idempotent("cmdb.resource_report.generate", key_params=("user.id",)) async def generate_report( body: GenerateReport, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _assert_access(user) try: scope = validate_scope_config(body.scope_config) except ValueError as exc: raise ValidationError(str(exc)) from exc run = await generate_report_run( db, user=user, trigger="manual", scope=scope, days=body.days, include_ai=body.include_ai, ) await AuditService(db).log( trace_id=getattr(request.state, "trace_id", str(uuid.uuid4())), actor_type="user", actor_user_id=user.id, action="cmdb.resource_report.generate", status="success", resource_type="cmdb_resource_report_run", resource_id=str(run.id), detail={"days": body.days, "scope": scope, "include_ai": body.include_ai, "ai_status": run.ai_status, "fact_hash": run.fact_hash}, ) await db.commit() return success_response(data=_run_dict(run), request=request) @router.get("/runs") async def list_runs( request: Request, page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=100), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _assert_access(user) where = CmdbResourceReportRun.recipient_person_id == str(user.id) from sqlalchemy import func total = int(await db.scalar(select(func.count(CmdbResourceReportRun.id)).where(where)) or 0) rows = list((await db.scalars(select(CmdbResourceReportRun).where(where).order_by( CmdbResourceReportRun.created_at.desc(), CmdbResourceReportRun.id.desc(), ).offset((page - 1) * page_size).limit(page_size))).all()) return paginated_response( [_run_dict(row, include_data=False) for row in rows], total, page, page_size, request, ) @router.get("/runs/{run_id}") async def get_run( run_id: uuid.UUID, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _assert_access(user) row = await db.scalar(select(CmdbResourceReportRun).where( CmdbResourceReportRun.id == run_id, CmdbResourceReportRun.recipient_person_id == str(user.id), )) if row is None: raise NotFoundError("ResourceReportRun", str(run_id)) return success_response(data=_run_dict(row), request=request) def _csv_safe(value: Any) -> str: text = "" if value is None else str(value) if text.startswith(("=", "+", "-", "@", "\t", "\r")): return "'" + text return text @router.get("/export.csv") async def export_directory_csv( request: Request, provider: list[str] | None = Query(default=None), category: list[str] | None = Query(default=None), resource_type: list[str] | None = Query(default=None), integration_id: str | None = Query(default=None, max_length=36), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _assert_access(user) try: scope = validate_scope_config({ "providers": provider or [], "categories": category or [], "resource_types": resource_type or [], "integration_id": integration_id, }) except ValueError as exc: raise ValidationError(str(exc)) from exc vis = await resolve_directory_visibility(db, user_id=str(user.id), role_names=user.role_names) clauses: list[Any] = [ directory_visibility_clause(vis), CmdbCloudResource.deleted_at.is_(None), ] for key, column in ( ("providers", CmdbCloudResource.provider), ("categories", CmdbCloudResource.category), ("resource_types", CmdbCloudResource.resource_type), ): if scope.get(key): clauses.append(column.in_(scope[key])) if scope.get("integration_id"): clauses.append(CmdbCloudResource.integration_id == scope["integration_id"]) export_columns = ( CmdbCloudResource.provider, CmdbCloudResource.category, CmdbCloudResource.resource_type, CmdbCloudResource.name, CmdbCloudResource.provider_resource_id, CmdbCloudResource.scope_external_id, CmdbCloudResource.region, CmdbCloudResource.zone, CmdbCloudResource.lifecycle_status, CmdbCloudResource.environment, CmdbCloudResource.sensitivity, CmdbCloudResource.owner_person_id, CmdbResourceGovernance.business_owner_person_id, CmdbResourceGovernance.technical_owner_person_id, CmdbResourceGovernance.governance_status, CmdbResourceGovernance.criticality, CmdbResourceGovernance.risk_level, CmdbResourceGovernance.compliance_status, CmdbCloudResource.first_seen_at, CmdbCloudResource.last_seen_at, CmdbCloudResource.last_changed_at, ) # Select only exported scalar columns. Selecting the ORM entities would # also hydrate raw_metadata/normalized_metadata/tags for up to 50k rows, # needlessly recreating the same TiDB/client memory risk the bounded # directory materialiser is designed to avoid. rows = (await db.execute(select(*export_columns).outerjoin( CmdbResourceGovernance, CmdbResourceGovernance.resource_id == CmdbCloudResource.id, ).where(*clauses).order_by( CmdbCloudResource.provider, CmdbCloudResource.category, CmdbCloudResource.name, ).limit(50_001))).all() if len(rows) > 50_000: raise ValidationError("Export exceeds 50,000 rows; narrow the filters") buf = io.StringIO() writer = csv.writer(buf) writer.writerow([ "provider", "category", "resource_type", "name", "provider_resource_id", "scope_external_id", "region", "zone", "status", "environment", "sensitivity", "owner_person_id", "business_owner_person_id", "technical_owner_person_id", "governance_status", "criticality", "risk_level", "compliance_status", "first_seen_at", "last_seen_at", "last_changed_at", ]) for row in rows: values = list(row) values[14] = values[14] or "unreviewed" writer.writerow([_csv_safe(value) for value in values]) await AuditService(db).log( trace_id=getattr(request.state, "trace_id", str(uuid.uuid4())), actor_type="user", actor_user_id=user.id, action="cmdb.resource_directory.export", status="success", resource_type="cmdb_cloud_resource", resource_id=None, detail={"scope": scope, "row_count": len(rows)}, ) await db.commit() payload = ("\ufeff" + buf.getvalue()).encode("utf-8") return StreamingResponse( iter([payload]), media_type="text/csv; charset=utf-8", headers={"Content-Disposition": 'attachment; filename="helix-cloud-resources.csv"'}, )