from datetime import datetime from typing import Annotated from fastapi import APIRouter, BackgroundTasks, Depends, File, HTTPException, Request, UploadFile, WebSocket, WebSocketDisconnect from sqlalchemy.orm import Session from app.api.dependencies import context_builder, event_publisher, memory_manager, orchestrator, tool_registry from app.domain.schemas import ( ActionStatus, DashboardRead, EventType, NoteCreate, NoteRead, PendingActionRead, PendingActionUpdate, ReservationCreate, ReservationPatch, ReservationRead, TaskCreate, TaskPatch, TaskRead, TranscriptChunk, TranscriptionJobRead, ) from app.executor.action_executor import ActionExecutor from app.infrastructure.database import Note, PendingAction, Reservation, Task, get_session from app.infrastructure.soniox import SonioxTranscriptionService, jobs router = APIRouter() DbSession = Annotated[Session, Depends(get_session)] @router.post("/tasks", response_model=TaskRead, operation_id="create_task", summary="Create task") def create_task(payload: TaskCreate, session: DbSession) -> Task: task = Task(**payload.model_dump()) session.add(task) session.commit() session.refresh(task) return task @router.patch("/tasks/{task_id}", response_model=TaskRead, operation_id="update_task", summary="Update task") def update_task(task_id: int, payload: TaskPatch, session: DbSession) -> Task: task = session.get(Task, task_id) if not task: raise HTTPException(status_code=404, detail="Task not found") for key, value in payload.model_dump(exclude_unset=True).items(): setattr(task, key, value) session.commit() session.refresh(task) return task @router.post("/notes", response_model=NoteRead, operation_id="create_note", summary="Create note") def create_note(payload: NoteCreate, session: DbSession) -> Note: note = Note(**payload.model_dump()) session.add(note) session.commit() session.refresh(note) return note @router.post("/reservations", response_model=ReservationRead, operation_id="create_reservation", summary="Create reservation") def create_reservation(payload: ReservationCreate, session: DbSession) -> Reservation: reservation = Reservation(**payload.model_dump()) session.add(reservation) session.commit() session.refresh(reservation) return reservation @router.patch( "/reservations/{reservation_id}", response_model=ReservationRead, operation_id="update_reservation", summary="Update reservation", ) def update_reservation(reservation_id: int, payload: ReservationPatch, session: DbSession) -> Reservation: reservation = session.get(Reservation, reservation_id) if not reservation: raise HTTPException(status_code=404, detail="Reservation not found") for key, value in payload.model_dump(exclude_unset=True).items(): setattr(reservation, key, value) session.commit() session.refresh(reservation) return reservation @router.get("/pending-actions", response_model=list[PendingActionRead]) def list_pending_actions(session: DbSession) -> list[PendingAction]: return session.query(PendingAction).order_by(PendingAction.created_at.desc()).all() @router.patch("/pending-actions/{action_id}", response_model=PendingActionRead) def edit_pending_action(action_id: int, payload: PendingActionUpdate, session: DbSession) -> PendingAction: action = session.get(PendingAction, action_id) if not action: raise HTTPException(status_code=404, detail="Pending action not found") if payload.payload is not None: action.payload = payload.payload session.commit() session.refresh(action) return action @router.post("/pending-actions/{action_id}/approve", response_model=PendingActionRead) async def approve_action(action_id: int, session: DbSession, request: Request) -> PendingAction: action = session.get(PendingAction, action_id) if not action: raise HTTPException(status_code=404, detail="Pending action not found") action.status = ActionStatus.APPROVED session.commit() session.refresh(action) generated = await ActionExecutor(tool_registry(request)).execute(session, action) event_publisher().publish( EventType.API_CALL_GENERATED, f"approve-{action_id}", {"action_id": action_id, "status": generated.status, "api_call": generated.execution_result}, ) return generated @router.post("/pending-actions/{action_id}/reject", response_model=PendingActionRead) def reject_action(action_id: int, session: DbSession) -> PendingAction: action = session.get(PendingAction, action_id) if not action: raise HTTPException(status_code=404, detail="Pending action not found") action.status = ActionStatus.REJECTED session.commit() session.refresh(action) event_publisher().publish(EventType.ACTION_REJECTED, f"reject-{action_id}", {"action_id": action_id}) return action @router.get("/dashboard", response_model=DashboardRead) def dashboard(session: DbSession): return DashboardRead( transcript=context_builder().transcript(), context=context_builder().snapshot( { "tasks": session.query(Task).count(), "notes": session.query(Note).count(), "reservations": session.query(Reservation).count(), } ), memories=memory_manager().retrieve("wedding venue flowers tasks", limit=6), pending_actions=session.query(PendingAction).order_by(PendingAction.created_at.desc()).all(), tasks=session.query(Task).order_by(Task.created_at.desc()).all(), notes=session.query(Note).order_by(Note.created_at.desc()).all(), reservations=session.query(Reservation).order_by(Reservation.created_at.desc()).all(), events=event_publisher().recent(), ) @router.get("/tools") def tools(request: Request): return [tool.model_dump() for tool in tool_registry(request).list()] @router.post("/transcriptions", response_model=TranscriptionJobRead, status_code=202) async def create_transcription_job( request: Request, background_tasks: BackgroundTasks, audio: UploadFile = File(...), ) -> TranscriptionJobRead: payload = await audio.read() if not payload: raise HTTPException(status_code=400, detail="No audio file provided.") job = jobs.create() background_tasks.add_task( SonioxTranscriptionService().process_upload, job.job_id, payload, audio.filename or "recording.webm", request, ) return job @router.get("/transcriptions/{job_id}", response_model=TranscriptionJobRead) def get_transcription_job(job_id: str) -> TranscriptionJobRead: job = jobs.get(job_id) if not job: raise HTTPException(status_code=404, detail="Transcription job not found.") return job @router.websocket("/ws/transcript") async def transcript_socket(websocket: WebSocket): await websocket.accept() db = next(get_session()) try: while True: raw = await websocket.receive_json() if "timestamp" not in raw: raw["timestamp"] = datetime.utcnow().isoformat() chunk = TranscriptChunk.model_validate(raw) action = await orchestrator(websocket).ingest(db, chunk) # type: ignore[arg-type] await websocket.send_json({"ok": True, "pending_action_id": action.id if action else None}) except WebSocketDisconnect: return finally: db.close()