server-26/drb-c2-core/app/routers/calls.py

from datetime import datetime, timezone, timedelta
from fastapi import APIRouter, BackgroundTasks, HTTPException, Query, Depends
from pydantic import BaseModel
from typing import Optional
from app.internal import firestore as fstore
from app.internal.auth import (
    require_admin_token,
    require_service_or_firebase_token,
    resolve_caller_org_id,
    reprocess_limiter,
)
from app.internal.storage import gcs_uri_for_call, with_playback_url


class TranscriptUpdate(BaseModel):
    transcript: str

router = APIRouter(prefix="/calls", tags=["calls"])


@router.get("")
async def list_calls(
    node_id: Optional[str] = Query(None),
    status: Optional[str] = Query(None),
    system_id: Optional[str] = Query(None),
    decoded: dict = Depends(require_service_or_firebase_token),
):
    filters = {}
    if node_id:
        filters["node_id"] = node_id
    if status:
        filters["status"] = status
    if system_id:
        filters["system_id"] = system_id
    org_id = await resolve_caller_org_id(decoded)
    if org_id is not None:  # service key / platform admin stay unrestricted
        filters["org_id"] = org_id
    calls = await fstore.collection_list("calls", **filters)
    # audio_url is not stored — it's a short-lived signed link minted per read.
    return [with_playback_url(c) for c in calls]


@router.get("/{call_id}")
async def get_call(call_id: str, decoded: dict = Depends(require_service_or_firebase_token)):
    call = await fstore.doc_get("calls", call_id)
    if not call:
        raise HTTPException(404, f"Call '{call_id}' not found.")
    org_id = await resolve_caller_org_id(decoded)
    if org_id is not None and call.get("org_id") != org_id:
        raise HTTPException(404, f"Call '{call_id}' not found.")
    return with_playback_url(call)


@router.post("/{call_id}/reprocess")
async def reprocess_call(
    call_id: str,
    background_tasks: BackgroundTasks,
    _: dict = Depends(require_admin_token),
):
    """
    Re-run the full intelligence pipeline (transcription -> extraction ->
    correlation) for a call. Admin-only (SAAS_PLAN.md B2c) — this was
    previously gated only by "any valid Firebase token", which meant any
    signed-in viewer could loop it and burn the owner's OpenAI/Gemini
    credits (DEFERRED.md, calls.py:42). The rate limiter below is a second
    guard against the same thing happening from a compromised/careless
    admin session, not the primary fix.
    """
    call = await fstore.doc_get("calls", call_id)
    if not call:
        raise HTTPException(404, f"Call '{call_id}' not found.")
    reprocess_limiter.check(call_id)

    from app.routers.upload import _run_intelligence_pipeline

    gcs_uri = gcs_uri_for_call(call)

    background_tasks.add_task(
        _run_intelligence_pipeline,
        call_id=call_id,
        node_id=call.get("node_id"),
        system_id=call.get("system_id"),
        talkgroup_id=call.get("talkgroup_id"),
        talkgroup_name=call.get("talkgroup_name"),
        gcs_uri=gcs_uri,
    )
    return {"ok": True, "call_id": call_id}


@router.post("/close-stale")
async def close_stale_calls(
    older_than_minutes: int = Query(30, ge=1, le=1440, description="Close active calls started more than this many minutes ago."),
    dry_run: bool = Query(False, description="If true, return what would be closed without writing."),
    _: dict = Depends(require_admin_token),
):
    """
    Find and close calls stuck in 'active' status — e.g. because a node rebooted
    before sending an end-call event.  Returns the list of affected call IDs.
    """
    cutoff = datetime.now(timezone.utc) - timedelta(minutes=older_than_minutes)
    active_calls = await fstore.collection_list("calls", status="active")

    stale = []
    for call in active_calls:
        started_raw = call.get("started_at")
        if not started_raw:
            continue
        if isinstance(started_raw, datetime):
            started = started_raw if started_raw.tzinfo else started_raw.replace(tzinfo=timezone.utc)
        else:
            try:
                started = datetime.fromisoformat(str(started_raw).replace("Z", "+00:00"))
            except Exception:
                continue
        if started < cutoff:
            stale.append(call)

    if not dry_run:
        now_iso = datetime.now(timezone.utc).isoformat()
        for call in stale:
            await fstore.doc_set("calls", call["call_id"], {
                "status": "ended",
                "ended_at": now_iso,
            })

    return {
        "dry_run": dry_run,
        "older_than_minutes": older_than_minutes,
        "count": len(stale),
        "call_ids": [c["call_id"] for c in stale],
    }


@router.patch("/{call_id}/transcript")
async def patch_transcript(
    call_id: str,
    body: TranscriptUpdate,
    background_tasks: BackgroundTasks,
    _: dict = Depends(require_admin_token),
):
    """Overwrite a call's transcript and re-run intelligence extraction."""
    call = await fstore.doc_get("calls", call_id)
    if not call:
        raise HTTPException(404, f"Call '{call_id}' not found.")

    # Save user correction as transcript_corrected; leave original transcript intact.
    # Clear stale intelligence fields so re-extraction runs fresh.
    await fstore.doc_set("calls", call_id, {
        "transcript_corrected": body.transcript,
        "tags": [],
        "severity": "unknown",
        "location": None,
        "units": [],
        "vehicles": [],
        "embedding": None,
    })

    # Unlink from ALL current incidents so re-correlation starts clean.
    # Handles both old single incident_id and new incident_ids list.
    old_ids: list[str] = call.get("incident_ids") or (
        [call["incident_id"]] if call.get("incident_id") else []
    )
    for old_incident_id in old_ids:
        old_incident = await fstore.doc_get("incidents", old_incident_id)
        if old_incident:
            remaining = [c for c in (old_incident.get("call_ids") or []) if c != call_id]
            if remaining:
                await fstore.doc_set("incidents", old_incident_id, {
                    "call_ids": remaining,
                    "summary_stale": True,
                })
            else:
                await fstore.doc_set("incidents", old_incident_id, {
                    "call_ids": [],
                    "status": "resolved",
                    "summary_stale": True,
                })
    await fstore.doc_set("calls", call_id, {"incident_ids": [], "incident_id": None})

    # Learn from the correction: diff original → corrected and add new tokens to vocabulary
    system_id = call.get("system_id")
    original_text = call.get("transcript_corrected") or call.get("transcript") or ""
    if system_id and original_text:
        from app.internal.vocabulary_learner import learn_from_correction
        await learn_from_correction(system_id, original_text, body.transcript)

    from app.routers.upload import _run_extraction_pipeline
    background_tasks.add_task(
        _run_extraction_pipeline,
        call_id=call_id,
        node_id=call.get("node_id"),
        system_id=call.get("system_id"),
        talkgroup_id=call.get("talkgroup_id"),
        talkgroup_name=call.get("talkgroup_name"),
        transcript=body.transcript,
        segments=call.get("segments"),
        preserve_transcript_correction=True,
    )
    return {"ok": True, "call_id": call_id}