Laurence
19c3e156a0
A runnable, self-hosted web interface so the whole workflow can be tested end to end: record or upload audio in the browser, watch it transcribe and summarise, browse a library, play back, and export. What changed: - server/app/store.py: SQLite + on-disk audio storage for recordings (stdlib only). - server/app/pipeline.py: background task audio -> transcribe -> summarise via the existing provider layer, updating status (queued/transcribing/summarising/done/error). - server/app/main.py: web API - POST upload, list, detail, audio (Range), delete, and export (txt/md/srt/vtt/json) - and serves the SPA at /. - server/app/web/: Plaud-style single-page UI (index.html, styles.css, app.js). Sidebar library, in-browser recording (MediaRecorder) + file upload, live status polling, audio player, summary (overview/key points/actions), timestamped transcript, exports. - server/Dockerfile + README: two-minute run instructions (default provider: Groq free tier for both Whisper + LLM), and a Docker option. - config: env prefix switched OPENSCRIBE_ -> NIGHTJAR_ to match the brand and the site tutorials; .env.example rewritten with a ready Groq quick-start. - state/TODO: web app recorded as done. Why: - User asked for a Plaud-like web interface to test how it all works. Nothing testable existed before (marketing site is a brochure; pipeline was unwired). This delivers a real, runnable product demo and effectively lands M5/M6 for the HTTP providers. Notes: - Slim by design: AI is offloaded to the configured provider, so no local ML deps needed for the demo. Byte-compiles clean; JS passes node --check. Local faster-whisper still needs its model (M5) for the fully-offline path. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
107 lines
3.2 KiB
Python
107 lines
3.2 KiB
Python
# SPDX-License-Identifier: GPL-3.0-only
|
|
"""Tiny persistence for the web app: recordings in SQLite, audio on disk.
|
|
|
|
Deliberately dependency-free (sqlite3 is stdlib) so the demo runs anywhere. Not meant for
|
|
high concurrency; the hosted service uses Postgres + object storage (see docs/).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import time
|
|
import uuid
|
|
from typing import Any
|
|
|
|
from .config import settings
|
|
|
|
_DB = os.path.join(settings.local_media_dir, "nightjar.db")
|
|
|
|
|
|
def _conn() -> sqlite3.Connection:
|
|
os.makedirs(settings.local_media_dir, exist_ok=True)
|
|
c = sqlite3.connect(_DB)
|
|
c.row_factory = sqlite3.Row
|
|
return c
|
|
|
|
|
|
def init() -> None:
|
|
with _conn() as c:
|
|
c.execute(
|
|
"""CREATE TABLE IF NOT EXISTS recordings (
|
|
id TEXT PRIMARY KEY,
|
|
title TEXT,
|
|
filename TEXT,
|
|
mime TEXT,
|
|
created_at REAL,
|
|
status TEXT, -- queued|transcribing|summarising|done|error
|
|
error TEXT,
|
|
language TEXT,
|
|
transcript TEXT, -- JSON: {text, segments}
|
|
summary TEXT -- JSON: {overview, key_points, action_items}
|
|
)"""
|
|
)
|
|
|
|
|
|
def audio_path(rec_id: str) -> str:
|
|
return os.path.join(settings.local_media_dir, f"{rec_id}.audio")
|
|
|
|
|
|
def create(title: str, filename: str, mime: str) -> str:
|
|
rec_id = "rec_" + uuid.uuid4().hex[:12]
|
|
with _conn() as c:
|
|
c.execute(
|
|
"INSERT INTO recordings (id,title,filename,mime,created_at,status) "
|
|
"VALUES (?,?,?,?,?,?)",
|
|
(rec_id, title, filename, mime, time.time(), "queued"),
|
|
)
|
|
return rec_id
|
|
|
|
|
|
def set_status(rec_id: str, status: str, error: str | None = None) -> None:
|
|
with _conn() as c:
|
|
c.execute("UPDATE recordings SET status=?, error=? WHERE id=?", (status, error, rec_id))
|
|
|
|
|
|
def set_result(rec_id: str, language: str, transcript: dict, summary: dict) -> None:
|
|
with _conn() as c:
|
|
c.execute(
|
|
"UPDATE recordings SET status='done', language=?, transcript=?, summary=? WHERE id=?",
|
|
(language, json.dumps(transcript), json.dumps(summary), rec_id),
|
|
)
|
|
|
|
|
|
def _row(r: sqlite3.Row) -> dict[str, Any]:
|
|
return {
|
|
"id": r["id"],
|
|
"title": r["title"],
|
|
"created_at": r["created_at"],
|
|
"status": r["status"],
|
|
"error": r["error"],
|
|
"language": r["language"],
|
|
"mime": r["mime"],
|
|
"transcript": json.loads(r["transcript"]) if r["transcript"] else None,
|
|
"summary": json.loads(r["summary"]) if r["summary"] else None,
|
|
}
|
|
|
|
|
|
def get(rec_id: str) -> dict | None:
|
|
with _conn() as c:
|
|
r = c.execute("SELECT * FROM recordings WHERE id=?", (rec_id,)).fetchone()
|
|
return _row(r) if r else None
|
|
|
|
|
|
def list_all() -> list[dict]:
|
|
with _conn() as c:
|
|
rows = c.execute("SELECT * FROM recordings ORDER BY created_at DESC").fetchall()
|
|
return [_row(r) for r in rows]
|
|
|
|
|
|
def delete(rec_id: str) -> bool:
|
|
with _conn() as c:
|
|
cur = c.execute("DELETE FROM recordings WHERE id=?", (rec_id,))
|
|
try:
|
|
os.remove(audio_path(rec_id))
|
|
except OSError:
|
|
pass
|
|
return cur.rowcount > 0
|