openscribe/server/app/main.py

# SPDX-License-Identifier: GPL-3.0-only
"""Nightjar web app - a Plaud-style interface plus the JSON API.

Upload or record audio, watch it transcribe and summarise, browse a library, play back and
export. The heavy AI is done by whichever providers you configure (see .env.example);
point it at a free Groq key and it works end to end.
"""
from __future__ import annotations

import json
import os
from contextlib import asynccontextmanager

from fastapi import BackgroundTasks, FastAPI, HTTPException, UploadFile
from fastapi.responses import FileResponse, PlainTextResponse, Response
from fastapi.staticfiles import StaticFiles

from . import store
from .config import settings
from .pipeline import process

WEB_DIR = os.path.join(os.path.dirname(__file__), "web")


@asynccontextmanager
async def lifespan(app: FastAPI):
    os.makedirs(settings.local_media_dir, exist_ok=True)
    store.init()
    yield


app = FastAPI(title="Nightjar", version="0.1.0", lifespan=lifespan)


@app.get("/health")
def health() -> dict:
    from .providers import build_summariser, build_transcriber

    try:
        summariser = build_summariser().name
    except Exception as exc:
        summariser = f"unavailable ({type(exc).__name__})"
    return {
        "status": "ok",
        "transcriber": build_transcriber().name,
        "summariser": summariser,
    }


@app.post("/api/recordings")
async def create_recording(
    file: UploadFile, background: BackgroundTasks, title: str | None = None
) -> dict:
    rec_id = store.create(
        title=title or (file.filename or "Recording"),
        filename=file.filename or "audio",
        mime=file.content_type or "application/octet-stream",
    )
    with open(store.audio_path(rec_id), "wb") as f:
        while chunk := await file.read(1 << 20):
            f.write(chunk)
    background.add_task(process, rec_id)  # transcribe + summarise off the request path
    return store.get(rec_id)


@app.get("/api/recordings")
def list_recordings() -> list[dict]:
    return store.list_all()


@app.get("/api/recordings/{rec_id}")
def get_recording(rec_id: str) -> dict:
    rec = store.get(rec_id)
    if not rec:
        raise HTTPException(404, "No such recording")
    return rec


@app.get("/api/recordings/{rec_id}/audio")
def get_audio(rec_id: str):
    rec = store.get(rec_id)
    if not rec:
        raise HTTPException(404, "No such recording")
    return FileResponse(store.audio_path(rec_id), media_type=rec["mime"] or "audio/webm")


@app.delete("/api/recordings/{rec_id}", status_code=204)
def delete_recording(rec_id: str):
    if not store.delete(rec_id):
        raise HTTPException(404, "No such recording")
    return Response(status_code=204)


def _ts(seconds: float, sep: str) -> str:
    h, rem = divmod(int(seconds), 3600)
    m, s = divmod(rem, 60)
    ms = int((seconds - int(seconds)) * 1000)
    return f"{h:02d}:{m:02d}:{s:02d}{sep}{ms:03d}"


@app.get("/api/recordings/{rec_id}/export")
def export_recording(rec_id: str, format: str = "txt"):
    rec = store.get(rec_id)
    if not rec:
        raise HTTPException(404, "No such recording")
    t = rec.get("transcript") or {}
    s = rec.get("summary") or {}
    segs = t.get("segments") or []

    if format == "json":
        return Response(json.dumps(rec, indent=2), media_type="application/json")
    if format == "txt":
        return PlainTextResponse(t.get("text", ""))
    if format == "md":
        body = [f"# {rec['title']}", "", "## Summary", s.get("overview", ""), ""]
        if s.get("key_points"):
            body += ["### Key points"] + [f"- {p}" for p in s["key_points"]] + [""]
        if s.get("action_items"):
            body += ["### Action items"] + [f"- [ ] {a}" for a in s["action_items"]] + [""]
        body += ["## Transcript", "", t.get("text", "")]
        return PlainTextResponse("\n".join(body), media_type="text/markdown")
    if format in ("srt", "vtt"):
        sep = "," if format == "srt" else "."
        lines = ["WEBVTT", ""] if format == "vtt" else []
        for i, seg in enumerate(segs, 1):
            if format == "srt":
                lines.append(str(i))
            lines.append(f"{_ts(seg['start'], sep)} --> {_ts(seg['end'], sep)}")
            lines.append(seg["text"].strip())
            lines.append("")
        return PlainTextResponse("\n".join(lines))
    raise HTTPException(400, "Unknown format")


# Serve the web UI at the root (must be mounted after the API routes).
app.mount("/", StaticFiles(directory=WEB_DIR, html=True), name="web")