"""Durable records for deterministic intelligence and AI interpretation. Raw domain rows never live here. Snapshots contain aggregate facts plus a local-only reference map; analyses contain validated bilingual model output; recommendations are the audited lifecycle record used by the action layer. """ from __future__ import annotations import uuid from datetime import UTC, datetime from decimal import Decimal from typing import Any from sqlalchemy import ( JSON, Boolean, DateTime, ForeignKey, Index, Integer, Numeric, String, Text, UniqueConstraint, func, text, ) from sqlalchemy.dialects import mysql from sqlalchemy.orm import Mapped, mapped_column from app.core.database import Base from app.core.sql_types import GUID as Uuid class IntelligenceSnapshot(Base): """Immutable versioned score and aggregate fact package.""" __tablename__ = "intelligence_snapshots" __mapper_args__ = {"eager_defaults": True} __table_args__ = ( UniqueConstraint( "subject_type", "subject_id", "audience_key", "score_version", "facts_hash", name="uq_intel_snapshot_identity", ), Index( "ix_intel_snapshot_latest", "subject_type", "subject_id", "audience_key", "created_at", ), Index("ix_intel_snapshot_data_as_of", "data_as_of"), ) id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) subject_type: Mapped[str] = mapped_column(String(24), nullable=False) subject_id: Mapped[str] = mapped_column(String(128), nullable=False) audience_key: Mapped[str] = mapped_column(String(128), nullable=False) score_version: Mapped[str] = mapped_column(String(32), nullable=False) facts_hash: Mapped[str] = mapped_column(String(64), nullable=False) overall_score: Mapped[Decimal] = mapped_column(Numeric(5, 2), nullable=False) confidence_score: Mapped[Decimal] = mapped_column(Numeric(5, 2), nullable=False) grade: Mapped[str] = mapped_column(String(24), nullable=False) dimensions: Mapped[list[dict[str, Any]]] = mapped_column(JSON, nullable=False) aggregate_facts: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False) # Never included in an LLM prompt. It maps anonymous evidence references # back to local objects and is filtered again at response time by RBAC. reference_map: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False) data_as_of: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True).with_variant(mysql.DATETIME(fsp=6), "mysql"), nullable=False, default=lambda: datetime.now(UTC), server_default=text("CURRENT_TIMESTAMP(6)"), ) class IntelligenceAnalysis(Base): """One validated bilingual interpretation of one immutable snapshot.""" __tablename__ = "intelligence_analyses" __mapper_args__ = {"eager_defaults": True} __table_args__ = ( UniqueConstraint("generation_key", name="uq_intel_analysis_generation_key"), Index( "ix_intel_analysis_latest", "subject_type", "subject_id", "audience_key", "generated_at", ), Index("ix_intel_analysis_status_created", "status", "created_at"), Index("ix_intel_analysis_generator_date", "generated_by_user_id", "created_at"), Index("ix_intel_analysis_facts_hash", "facts_hash"), ) id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) snapshot_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("intelligence_snapshots.id", ondelete="RESTRICT"), nullable=False, ) subject_type: Mapped[str] = mapped_column(String(24), nullable=False) subject_id: Mapped[str] = mapped_column(String(128), nullable=False) audience_key: Mapped[str] = mapped_column(String(128), nullable=False) analysis_kind: Mapped[str] = mapped_column(String(32), nullable=False) facts_hash: Mapped[str] = mapped_column(String(64), nullable=False) generation_key: Mapped[str] = mapped_column(String(64), nullable=False) status: Mapped[str] = mapped_column(String(16), nullable=False, default="pending") summary_zh: Mapped[str | None] = mapped_column(Text) summary_en: Mapped[str | None] = mapped_column(Text) summary_fact_refs: Mapped[list[str] | None] = mapped_column(JSON) conclusions: Mapped[list[dict[str, Any]] | None] = mapped_column(JSON) risks: Mapped[list[dict[str, Any]] | None] = mapped_column(JSON) opportunities: Mapped[list[dict[str, Any]] | None] = mapped_column(JSON) forecasts: Mapped[list[dict[str, Any]] | None] = mapped_column(JSON) recommendations: Mapped[list[dict[str, Any]] | None] = mapped_column(JSON) data_gaps: Mapped[list[dict[str, Any]] | None] = mapped_column(JSON) model_used: Mapped[str | None] = mapped_column(String(128)) input_token_count: Mapped[int | None] = mapped_column(Integer) output_token_count: Mapped[int | None] = mapped_column(Integer) generated_by_user_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"), ) data_as_of: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) generated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) failure_code: Mapped[str | None] = mapped_column(String(64)) failure_message: Mapped[str | None] = mapped_column(String(512)) cache_reused: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) forced: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now(), ) class IntelligenceRecommendation(Base): """Deduplicated recommendation and its complete handling lifecycle.""" __tablename__ = "intelligence_recommendations" __mapper_args__ = {"eager_defaults": True} __table_args__ = ( UniqueConstraint("dedupe_key", name="uq_intel_recommendation_dedupe_key"), Index("ix_intel_recommendation_status_severity", "status", "severity"), Index("ix_intel_recommendation_subject", "subject_type", "subject_id"), Index("ix_intel_recommendation_domain_created", "domain", "created_at"), ) id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) snapshot_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("intelligence_snapshots.id", ondelete="RESTRICT"), nullable=False, ) analysis_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("intelligence_analyses.id", ondelete="SET NULL"), ) subject_type: Mapped[str] = mapped_column(String(24), nullable=False) subject_id: Mapped[str] = mapped_column(String(128), nullable=False) audience_key: Mapped[str] = mapped_column(String(128), nullable=False) dedupe_key: Mapped[str] = mapped_column(String(64), nullable=False) domain: Mapped[str] = mapped_column(String(32), nullable=False) severity: Mapped[str] = mapped_column(String(16), nullable=False) priority: Mapped[int] = mapped_column(Integer, nullable=False) title_zh: Mapped[str] = mapped_column(String(256), nullable=False) title_en: Mapped[str] = mapped_column(String(256), nullable=False) summary_zh: Mapped[str] = mapped_column(Text, nullable=False) summary_en: Mapped[str] = mapped_column(Text, nullable=False) evidence_refs: Mapped[list[dict[str, Any]]] = mapped_column(JSON, nullable=False) action_contract: Mapped[dict[str, Any] | None] = mapped_column(JSON) expected_monthly_savings: Mapped[Decimal | None] = mapped_column(Numeric(18, 4)) expected_annual_savings: Mapped[Decimal | None] = mapped_column(Numeric(18, 4)) confirmed_savings: Mapped[Decimal | None] = mapped_column(Numeric(18, 4)) realized_savings: Mapped[Decimal | None] = mapped_column(Numeric(18, 4)) currency: Mapped[str | None] = mapped_column(String(8)) status: Mapped[str] = mapped_column(String(16), nullable=False, default="new") ignored_reason: Mapped[str | None] = mapped_column(String(1000)) assigned_to_person_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"), ) last_changed_by_user_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"), ) first_seen_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) last_seen_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) viewed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) resolved_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) ignored_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now(), ) class IntelligenceRecommendationSavingsLink(Base): """Audited one-to-one attribution from a real cloud saving to a suggestion. The savings event remains the source of truth for the amount because a stopped/restarted machine can self-correct its realized saving. This row only records why that event belongs to one explicitly-started Intelligence recommendation; a unique event link prevents double counting. """ __tablename__ = "intelligence_recommendation_savings_links" __mapper_args__ = {"eager_defaults": True} __table_args__ = ( UniqueConstraint("savings_event_id", name="uq_intel_rec_saving_event"), UniqueConstraint( "recommendation_id", "savings_event_id", name="uq_intel_rec_saving_pair", ), Index("ix_intel_rec_saving_recommendation", "recommendation_id"), ) id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) recommendation_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("intelligence_recommendations.id", ondelete="RESTRICT"), nullable=False, ) # Cross-domain by design: savings events are created by the established # CMDB efficiency lifecycle and should never be reimplemented here. savings_event_id: Mapped[uuid.UUID] = mapped_column(Uuid, nullable=False) server_id: Mapped[str] = mapped_column(String(36), nullable=False) linked_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) class IntelligenceGenerationLease(Base): """Database-portable single-flight lease for costly model generation.""" __tablename__ = "intelligence_generation_leases" __table_args__ = (Index("ix_intel_lease_expires", "expires_at"),) lease_key: Mapped[str] = mapped_column(String(191), primary_key=True) owner_token: Mapped[str] = mapped_column(String(64), nullable=False) expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now(), )