app/api/canonical_rollout.py

"""Canonical paragraph rollout: scan indexed chunks, optionally apply replacement to .docx."""

import logging

from fastapi import APIRouter, Depends, HTTPException, Request
from sqlalchemy.ext.asyncio import AsyncSession

from app.api.rate_limit import check_generate, check_read
from app.db.database import get_db
from app.models.schemas import (
    CanonicalApplyRequest,
    CanonicalApplyResponse,
    CanonicalRolloutRequest,
    CanonicalRolloutResponse,
)
from app.services.canonical_rollout import (
    apply_canonical_replacement,
    scan_canonical_rollout,
)

logger = logging.getLogger(__name__)

router = APIRouter()


@router.post(
    "/content/canonical-scan",
    response_model=CanonicalRolloutResponse,
    summary="Find chunks semantically similar to a canonical paragraph",
)
async def canonical_scan(
    request: Request,
    body: CanonicalRolloutRequest,
    db: AsyncSession = Depends(get_db),
    _: None = Depends(check_read),
) -> CanonicalRolloutResponse:
    """Discover candidate chunks for upgrading to an improved standard paragraph.

    Uses the same embedding search as the rest of the RAG stack, plus token Jaccard
    versus the canonical text to reduce false positives. Proposed text is either
    the canonical paragraph or an LLM merge (``adapt_with_llm``) when an API key is set.

    Review ``compatibility`` (high / review / low) before any automatic apply policy.
    """
    tenant_id: str = request.state.tenant_id
    try:
        return await scan_canonical_rollout(db, tenant_id, body)
    except Exception as exc:  # noqa: BLE001
        logger.exception("canonical-scan failed: %s", exc)
        raise HTTPException(status_code=500, detail="Canonical scan failed") from exc


@router.post(
    "/content/canonical-apply",
    response_model=CanonicalApplyResponse,
    summary="Replace one chunk in a .docx and re-index (destructive layout)",
)
async def canonical_apply(
    request: Request,
    body: CanonicalApplyRequest,
    db: AsyncSession = Depends(get_db),
    _: None = Depends(check_generate),
) -> CanonicalApplyResponse:
    """Apply a replacement string for one indexed chunk.

    Rebuilds the Word file as plain paragraphs (tables/complex layout are not preserved),
    deletes old vector rows for the document, and re-runs ingestion.

    **Requires** ``confirm_destructive_docx: true``.
    """
    tenant_id: str = request.state.tenant_id
    try:
        return await apply_canonical_replacement(db, tenant_id, body)
    except HTTPException:
        raise
    except Exception as exc:  # noqa: BLE001
        logger.exception("canonical-apply failed: %s", exc)
        raise HTTPException(status_code=500, detail="Canonical apply failed") from exc