Source code for lalandre_rag.summaries.models

"""Data models, constants, and helpers for canonical act summaries."""

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime
from typing import Any, Dict, Optional

CANONICAL_SUMMARY_KIND = "canonical"
CANONICAL_SUMMARY_PROMPT_VERSION = "canonical_summary_v1"
SUMMARY_STATUS_PENDING = "pending"
SUMMARY_STATUS_READY = "ready"
SUMMARY_STATUS_FAILED = "failed"




@dataclass(frozen=True)
class CanonicalSummarySnapshot:
    """Canonical summary state returned by the summary service."""

    act_id: int
    celex: str
    language: str
    status: str
    is_stale: bool
    summary: Optional[str]
    generated_at: Optional[datetime]
    prompt_version: Optional[str]
    model_id: Optional[str]
    source_version_id: Optional[int]
    error_text: Optional[str]
    trace: Dict[str, Any]

    @property
    def available(self) -> bool:
        """Return whether the snapshot contains a ready-to-use summary."""
        return bool(self.summary and self.status == SUMMARY_STATUS_READY)





class SummaryTraceRecorder:
    """Build structured trace payloads for summary generation and lookup."""



    @staticmethod
    def lookup(*, status: str, is_stale: bool, reason: Optional[str] = None) -> Dict[str, Any]:
        """Build trace metadata for summary lookup operations."""
        trace: Dict[str, Any] = {
            "lookup_status": status,
            "is_stale": bool(is_stale),
        }
        if reason:
            trace["reason"] = reason
        return trace




    @staticmethod
    def generation(
        *,
        mode: str,
        context_chars: int,
        subdivisions_used: int,
        model_id: str,
        prompt_version: str,
        extra: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        """Build trace metadata for summary generation operations."""
        trace: Dict[str, Any] = {
            "mode": mode,
            "context_chars": int(context_chars),
            "subdivisions_used": int(subdivisions_used),
            "model_id": model_id,
            "prompt_version": prompt_version,
        }
        if extra:
            trace.update(extra)
        return trace






def build_default_summary_questions(celex: str) -> set[str]:
    """Return the default summary prompts recognized for a CELEX identifier."""
    normalized = celex.strip()
    if not normalized:
        return set()
    return {
        f"résumé {normalized}".casefold(),
        f"résumé de {normalized}".casefold(),
        f"resume {normalized}".casefold(),
        f"resume de {normalized}".casefold(),
    }





def is_default_summary_question(question: str, celex: str) -> bool:
    """Return whether *question* matches the default summary prompt family."""
    normalized_question = question.strip()
    if not normalized_question:
        return True
    return normalized_question.casefold() in build_default_summary_questions(celex)