Logan
138 lines
4.6 KiB
Python
138 lines
4.6 KiB
Python
from typing import Optional
|
|
from fastapi import APIRouter, BackgroundTasks, UploadFile, File, Form, HTTPException, Security
|
|
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
|
from app.internal.storage import upload_audio
|
|
from app.internal import firestore as fstore
|
|
from app.internal.logger import logger
|
|
|
|
router = APIRouter(tags=["upload"])
|
|
|
|
_bearer = HTTPBearer(auto_error=False)
|
|
|
|
|
|
@router.post("/upload")
|
|
async def upload_call_audio(
|
|
background_tasks: BackgroundTasks,
|
|
file: UploadFile = File(...),
|
|
call_id: str = Form(...),
|
|
node_id: str = Form(...),
|
|
talkgroup_id: Optional[int] = Form(None),
|
|
talkgroup_name: Optional[str] = Form(None),
|
|
system_id: Optional[str] = Form(None),
|
|
credentials: Optional[HTTPAuthorizationCredentials] = Security(_bearer),
|
|
):
|
|
"""
|
|
Receive an audio recording from an edge node.
|
|
Upload to GCS, update the call document in Firestore with the audio URL,
|
|
then kick off the intelligence pipeline as a background task.
|
|
"""
|
|
# Verify the per-node API key
|
|
if not credentials:
|
|
raise HTTPException(401, "Missing authorization")
|
|
key_doc = await fstore.doc_get("node_keys", node_id)
|
|
if not key_doc:
|
|
logger.warning(f"Upload 401: no key_doc in Firestore for node_id={node_id!r}")
|
|
raise HTTPException(401, "Invalid node API key")
|
|
if key_doc.get("api_key") != credentials.credentials:
|
|
logger.warning(
|
|
f"Upload 401: key mismatch for node_id={node_id!r} "
|
|
f"(received prefix: {credentials.credentials[:8]}...)"
|
|
)
|
|
raise HTTPException(401, "Invalid node API key")
|
|
|
|
data = await file.read()
|
|
if not data:
|
|
raise HTTPException(400, "Empty file.")
|
|
|
|
filename = file.filename
|
|
audio_url = await upload_audio(data, filename)
|
|
|
|
if audio_url:
|
|
try:
|
|
await fstore.doc_set("calls", call_id, {"audio_url": audio_url})
|
|
except Exception as e:
|
|
logger.warning(f"Could not update call {call_id} with audio_url: {e}")
|
|
|
|
# Convert public GCS URL to gs:// URI for Speech-to-Text
|
|
gcs_uri = _public_url_to_gcs_uri(audio_url)
|
|
|
|
background_tasks.add_task(
|
|
_run_intelligence_pipeline,
|
|
call_id=call_id,
|
|
node_id=node_id,
|
|
system_id=system_id,
|
|
talkgroup_id=talkgroup_id,
|
|
talkgroup_name=talkgroup_name,
|
|
gcs_uri=gcs_uri,
|
|
)
|
|
|
|
return {"url": audio_url}
|
|
|
|
|
|
def _public_url_to_gcs_uri(url: str) -> Optional[str]:
|
|
"""
|
|
Convert a public GCS URL (possibly signed) like
|
|
https://storage.googleapis.com/bucket/calls/file.mp3?Expires=...
|
|
to a gs:// URI usable by Speech-to-Text.
|
|
Returns None if the URL doesn't look like a GCS URL.
|
|
"""
|
|
prefix = "https://storage.googleapis.com/"
|
|
if url and url.startswith(prefix):
|
|
path = url[len(prefix):].split("?")[0] # strip signed-URL query params
|
|
return "gs://" + path
|
|
return None
|
|
|
|
|
|
async def _run_intelligence_pipeline(
|
|
call_id: str,
|
|
node_id: str,
|
|
system_id: Optional[str],
|
|
talkgroup_id: Optional[int],
|
|
talkgroup_name: Optional[str],
|
|
gcs_uri: Optional[str],
|
|
) -> None:
|
|
"""
|
|
Post-upload intelligence pipeline (runs as a background task):
|
|
1. Transcribe audio via Google STT
|
|
2. Extract tags/incident type from transcript
|
|
3. Correlate with existing incidents (or create new one)
|
|
4. Check alert rules and dispatch notifications
|
|
"""
|
|
from app.internal import transcription, intelligence, incident_correlator, alerter
|
|
|
|
transcript: Optional[str] = None
|
|
|
|
# Step 1: Transcription
|
|
if gcs_uri:
|
|
transcript = await transcription.transcribe_call(call_id, gcs_uri, talkgroup_name)
|
|
|
|
# Step 2: Intelligence extraction
|
|
tags: list[str] = []
|
|
incident_type: Optional[str] = None
|
|
location: Optional[str] = None
|
|
if transcript:
|
|
tags, incident_type, location = await intelligence.extract_tags(call_id, transcript, talkgroup_name)
|
|
|
|
# Step 3: Incident correlation
|
|
if incident_type:
|
|
await incident_correlator.correlate_call(
|
|
call_id=call_id,
|
|
node_id=node_id,
|
|
system_id=system_id,
|
|
talkgroup_id=talkgroup_id,
|
|
talkgroup_name=talkgroup_name,
|
|
tags=tags,
|
|
incident_type=incident_type,
|
|
location=location,
|
|
)
|
|
|
|
# Step 4: Alert dispatch (always runs — talkgroup ID rules don't need a transcript)
|
|
await alerter.check_and_dispatch(
|
|
call_id=call_id,
|
|
node_id=node_id,
|
|
talkgroup_id=talkgroup_id,
|
|
talkgroup_name=talkgroup_name,
|
|
tags=tags,
|
|
transcript=transcript,
|
|
)
|