383 lines
16 KiB
Python
383 lines
16 KiB
Python
import asyncio
|
|
import logging
|
|
import os
|
|
import re
|
|
import sqlite3
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
from fastapi import BackgroundTasks, FastAPI, File, Form, HTTPException, UploadFile
|
|
from fastapi.responses import FileResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
DATA_DIR = Path(os.environ.get("DATA_DIR", "./data"))
|
|
AUDIO_DIR = DATA_DIR / "audio"
|
|
DB_PATH = DATA_DIR / "faerro-kb.sqlite3"
|
|
WHISPER_SERVER_URL = os.environ.get("WHISPER_SERVER_URL", "").strip()
|
|
FRONTEND_DIR = Path(__file__).resolve().parent.parent / "frontend"
|
|
WHISPER_TEST_AUDIO_PATH = Path(__file__).resolve().parent.parent / "audio test.wav"
|
|
TRANSCRIPTION_TASKS: set[asyncio.Task[None]] = set()
|
|
logger = logging.getLogger(__name__)
|
|
logging.getLogger("httpx").setLevel(logging.WARNING)
|
|
|
|
DATA_DIR.mkdir(parents=True, exist_ok=True)
|
|
AUDIO_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
app = FastAPI(title="Faerro KB")
|
|
|
|
|
|
def connect_db() -> sqlite3.Connection:
|
|
connection = sqlite3.connect(DB_PATH)
|
|
connection.row_factory = sqlite3.Row
|
|
return connection
|
|
|
|
|
|
def initialize_db() -> None:
|
|
with connect_db() as connection:
|
|
connection.execute(
|
|
"""CREATE TABLE IF NOT EXISTS notes (
|
|
id TEXT PRIMARY KEY,
|
|
created_at TEXT NOT NULL,
|
|
filename TEXT NOT NULL,
|
|
audio_path TEXT NOT NULL,
|
|
transcript TEXT NOT NULL DEFAULT '',
|
|
status TEXT NOT NULL
|
|
)"""
|
|
)
|
|
connection.execute(
|
|
"""CREATE VIRTUAL TABLE IF NOT EXISTS note_search
|
|
USING fts5(note_id UNINDEXED, transcript)"""
|
|
)
|
|
columns = {row[1] for row in connection.execute("PRAGMA table_info(notes)")}
|
|
if "original_audio_path" not in columns:
|
|
connection.execute("ALTER TABLE notes ADD COLUMN original_audio_path TEXT")
|
|
|
|
|
|
initialize_db()
|
|
|
|
def transcode_audio(source_path: Path, output_path: Path) -> None:
|
|
subprocess.run(
|
|
[
|
|
"ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "error", "-y",
|
|
"-i", str(source_path), "-vn", "-ac", "1", "-ar", "22050",
|
|
"-c:a", "pcm_s16le", str(output_path),
|
|
],
|
|
check=True,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.PIPE,
|
|
timeout=300,
|
|
)
|
|
|
|
|
|
def update_transcript(note_id: str, transcript: str, status: str) -> None:
|
|
with connect_db() as connection:
|
|
result = connection.execute(
|
|
"UPDATE notes SET transcript = ?, status = ? WHERE id = ?",
|
|
(transcript, status, note_id),
|
|
)
|
|
connection.execute("DELETE FROM note_search WHERE note_id = ?", (note_id,))
|
|
if result.rowcount and transcript.strip():
|
|
connection.execute(
|
|
"INSERT INTO note_search (note_id, transcript) VALUES (?, ?)",
|
|
(note_id, transcript.strip()),
|
|
)
|
|
|
|
|
|
def remove_transcribed_audio(note_id: str) -> None:
|
|
with connect_db() as connection:
|
|
note = connection.execute(
|
|
"SELECT audio_path, original_audio_path FROM notes WHERE id = ?", (note_id,)
|
|
).fetchone()
|
|
if not note:
|
|
return
|
|
for audio_path in (note["audio_path"], note["original_audio_path"]):
|
|
if audio_path:
|
|
Path(audio_path).unlink(missing_ok=True)
|
|
with connect_db() as connection:
|
|
connection.execute(
|
|
"UPDATE notes SET audio_path = '', original_audio_path = NULL WHERE id = ?",
|
|
(note_id,),
|
|
)
|
|
|
|
|
|
async def transcribe_note(note_id: str, audio_path: Path) -> None:
|
|
if not WHISPER_SERVER_URL:
|
|
return
|
|
try:
|
|
async with httpx.AsyncClient(timeout=900) as client:
|
|
with audio_path.open("rb") as audio_file:
|
|
response = await client.post(
|
|
WHISPER_SERVER_URL,
|
|
params={"task": "transcribe", "output": "json"},
|
|
files={"audio_file": (audio_path.name, audio_file, "application/octet-stream")},
|
|
)
|
|
response.raise_for_status()
|
|
transcript = response.json().get("text", "").strip()
|
|
except Exception:
|
|
update_transcript(note_id, "", "transcription_failed")
|
|
return
|
|
update_transcript(note_id, transcript, "transcribed" if transcript else "empty_transcript")
|
|
if transcript:
|
|
remove_transcribed_audio(note_id)
|
|
|
|
|
|
async def test_whisper_connection() -> bool:
|
|
if not WHISPER_SERVER_URL:
|
|
logger.warning("Whisper startup test skipped: WHISPER_SERVER_URL is not configured")
|
|
return False
|
|
if not WHISPER_TEST_AUDIO_PATH.is_file():
|
|
logger.error("Whisper startup test failed: bundled test audio is missing")
|
|
return False
|
|
|
|
started_at = time.monotonic()
|
|
try:
|
|
async with httpx.AsyncClient(timeout=httpx.Timeout(120, connect=5)) as client:
|
|
with WHISPER_TEST_AUDIO_PATH.open("rb") as audio_file:
|
|
response = await client.post(
|
|
WHISPER_SERVER_URL,
|
|
params={"encode": "true", "task": "transcribe", "output": "json"},
|
|
files={"audio_file": (WHISPER_TEST_AUDIO_PATH.name, audio_file, "audio/wav")},
|
|
)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
transcript = result.get("text") if isinstance(result, dict) else None
|
|
if not isinstance(transcript, str) or not transcript.strip():
|
|
logger.error(
|
|
"Whisper startup test failed: HTTP %s returned no transcript (%.1fs)",
|
|
response.status_code,
|
|
time.monotonic() - started_at,
|
|
)
|
|
return False
|
|
language = result.get("language", "unknown")
|
|
if not isinstance(language, str) or not re.fullmatch(r"[A-Za-z-]{1,12}", language):
|
|
language = "unknown"
|
|
segments = result.get("segments", [])
|
|
segment_count = len(segments) if isinstance(segments, list) else 0
|
|
logger.info(
|
|
"Whisper startup test passed: HTTP %s, language=%s, segments=%d, elapsed=%.1fs",
|
|
response.status_code,
|
|
language,
|
|
segment_count,
|
|
time.monotonic() - started_at,
|
|
)
|
|
return True
|
|
except httpx.HTTPStatusError as error:
|
|
logger.error("Whisper startup test failed: HTTP %s", error.response.status_code)
|
|
except Exception as error:
|
|
logger.error("Whisper startup test failed: %s", type(error).__name__)
|
|
return False
|
|
|
|
|
|
@app.on_event("startup")
|
|
async def resume_pending_transcriptions() -> None:
|
|
with connect_db() as connection:
|
|
connection.execute(
|
|
"UPDATE notes SET status = 'awaiting_transcription' WHERE status = 'transcribing'"
|
|
)
|
|
if not await test_whisper_connection():
|
|
return
|
|
with connect_db() as connection:
|
|
notes = connection.execute(
|
|
"""SELECT id, audio_path FROM notes
|
|
WHERE status IN ('awaiting_transcription', 'transcribing', 'transcription_failed')"""
|
|
).fetchall()
|
|
for note in notes:
|
|
audio_path = Path(note["audio_path"])
|
|
if audio_path.is_file():
|
|
with connect_db() as connection:
|
|
connection.execute(
|
|
"UPDATE notes SET status = 'transcribing' WHERE id = ?",
|
|
(note["id"],),
|
|
)
|
|
task = asyncio.create_task(transcribe_note(note["id"], audio_path))
|
|
TRANSCRIPTION_TASKS.add(task)
|
|
task.add_done_callback(TRANSCRIPTION_TASKS.discard)
|
|
else:
|
|
with connect_db() as connection:
|
|
connection.execute(
|
|
"UPDATE notes SET status = 'transcription_failed' WHERE id = ?",
|
|
(note["id"],),
|
|
)
|
|
|
|
|
|
@app.get("/api/health")
|
|
def health() -> dict[str, str]:
|
|
return {"status": "ok"}
|
|
|
|
|
|
@app.post("/api/notes", status_code=201)
|
|
async def create_note(
|
|
background_tasks: BackgroundTasks,
|
|
recording_id: str = Form(...),
|
|
audio: UploadFile = File(...),
|
|
created_at: str = Form(...),
|
|
) -> dict[str, str]:
|
|
try:
|
|
note_id = str(uuid.UUID(recording_id))
|
|
except ValueError as error:
|
|
raise HTTPException(status_code=400, detail="Invalid recording ID") from error
|
|
with connect_db() as connection:
|
|
existing = connection.execute("SELECT status FROM notes WHERE id = ?", (note_id,)).fetchone()
|
|
if existing:
|
|
return {"id": note_id, "status": existing["status"]}
|
|
suffix = Path(audio.filename or "recording.webm").suffix.lower()
|
|
if not re.fullmatch(r"\.[a-z0-9]{1,9}", suffix):
|
|
suffix = ".audio"
|
|
original_audio_path = AUDIO_DIR / f"{note_id}-original{suffix}"
|
|
audio_path = AUDIO_DIR / f"{note_id}.wav"
|
|
temporary_output = None
|
|
try:
|
|
with original_audio_path.open("wb") as destination:
|
|
while chunk := await audio.read(1024 * 1024):
|
|
destination.write(chunk)
|
|
with tempfile.NamedTemporaryFile(suffix=".wav", dir=AUDIO_DIR, delete=False) as temporary_file:
|
|
temporary_output = Path(temporary_file.name)
|
|
await asyncio.to_thread(transcode_audio, original_audio_path, temporary_output)
|
|
temporary_output.replace(audio_path)
|
|
except FileNotFoundError as error:
|
|
original_audio_path.unlink(missing_ok=True)
|
|
if temporary_output:
|
|
temporary_output.unlink(missing_ok=True)
|
|
raise HTTPException(status_code=500, detail="Audio conversion is unavailable") from error
|
|
except subprocess.CalledProcessError as error:
|
|
original_audio_path.unlink(missing_ok=True)
|
|
if temporary_output:
|
|
temporary_output.unlink(missing_ok=True)
|
|
raise HTTPException(status_code=400, detail="Unsupported or invalid audio file") from error
|
|
except subprocess.TimeoutExpired as error:
|
|
original_audio_path.unlink(missing_ok=True)
|
|
if temporary_output:
|
|
temporary_output.unlink(missing_ok=True)
|
|
raise HTTPException(status_code=422, detail="Audio conversion timed out") from error
|
|
except Exception as error:
|
|
original_audio_path.unlink(missing_ok=True)
|
|
audio_path.unlink(missing_ok=True)
|
|
if temporary_output:
|
|
temporary_output.unlink(missing_ok=True)
|
|
raise HTTPException(status_code=500, detail="Could not store audio") from error
|
|
finally:
|
|
await audio.close()
|
|
|
|
try:
|
|
parsed_date = datetime.fromisoformat(created_at.replace("Z", "+00:00"))
|
|
if parsed_date.tzinfo is None:
|
|
parsed_date = parsed_date.replace(tzinfo=timezone.utc)
|
|
created_at = parsed_date.astimezone(timezone.utc).isoformat()
|
|
except ValueError:
|
|
created_at = datetime.now(timezone.utc).isoformat()
|
|
|
|
status = "awaiting_transcription" if not WHISPER_SERVER_URL else "transcribing"
|
|
with connect_db() as connection:
|
|
connection.execute(
|
|
"""INSERT OR IGNORE INTO notes
|
|
(id, created_at, filename, audio_path, original_audio_path, status)
|
|
VALUES (?, ?, ?, ?, ?, ?)""",
|
|
(note_id, created_at, audio.filename or "recording.audio", str(audio_path), str(original_audio_path), status),
|
|
)
|
|
if WHISPER_SERVER_URL:
|
|
background_tasks.add_task(transcribe_note, note_id, audio_path)
|
|
return {"id": note_id, "status": status}
|
|
|
|
|
|
@app.post("/api/text-notes", status_code=201)
|
|
def create_text_note(payload: dict[str, str]) -> dict[str, str]:
|
|
text = payload.get("text", "").strip()
|
|
if not text:
|
|
raise HTTPException(status_code=400, detail="Note text is required")
|
|
try:
|
|
note_id = str(uuid.UUID(payload.get("id", "")))
|
|
except ValueError as error:
|
|
raise HTTPException(status_code=400, detail="Invalid note ID") from error
|
|
try:
|
|
parsed_date = datetime.fromisoformat(payload.get("created_at", "").replace("Z", "+00:00"))
|
|
if parsed_date.tzinfo is None:
|
|
parsed_date = parsed_date.replace(tzinfo=timezone.utc)
|
|
created_at = parsed_date.astimezone(timezone.utc).isoformat()
|
|
except ValueError:
|
|
created_at = datetime.now(timezone.utc).isoformat()
|
|
|
|
with connect_db() as connection:
|
|
existing = connection.execute("SELECT status FROM notes WHERE id = ?", (note_id,)).fetchone()
|
|
if existing:
|
|
return {"id": note_id, "status": existing["status"]}
|
|
connection.execute(
|
|
"""INSERT INTO notes (id, created_at, filename, audio_path, transcript, status)
|
|
VALUES (?, ?, ?, '', ?, 'written')""",
|
|
(note_id, created_at, "Written note", text),
|
|
)
|
|
connection.execute("INSERT INTO note_search (note_id, transcript) VALUES (?, ?)", (note_id, text))
|
|
return {"id": note_id, "status": "written"}
|
|
|
|
|
|
@app.get("/api/notes")
|
|
def list_notes(q: str = "", status: str = "") -> list[dict[str, str]]:
|
|
with connect_db() as connection:
|
|
search_terms = re.findall(r"[\w]+", q, flags=re.UNICODE)
|
|
if search_terms:
|
|
query = " AND ".join(f'"{term.replace(chr(34), chr(34) * 2)}"' for term in search_terms)
|
|
if status:
|
|
rows = connection.execute(
|
|
"""SELECT n.id, n.created_at, n.filename, n.transcript, n.status
|
|
FROM note_search s JOIN notes n ON n.id = s.note_id
|
|
WHERE note_search MATCH ? AND n.status = ?
|
|
ORDER BY n.created_at DESC LIMIT 100""",
|
|
(query, status),
|
|
).fetchall()
|
|
else:
|
|
rows = connection.execute(
|
|
"""SELECT n.id, n.created_at, n.filename, n.transcript, n.status
|
|
FROM note_search s JOIN notes n ON n.id = s.note_id
|
|
WHERE note_search MATCH ? ORDER BY n.created_at DESC LIMIT 100""",
|
|
(query,),
|
|
).fetchall()
|
|
elif status:
|
|
rows = connection.execute(
|
|
"""SELECT id, created_at, filename, transcript, status
|
|
FROM notes WHERE status = ? ORDER BY created_at DESC LIMIT 100""",
|
|
(status,),
|
|
).fetchall()
|
|
else:
|
|
rows = connection.execute(
|
|
"""SELECT id, created_at, filename, transcript, status
|
|
FROM notes ORDER BY created_at DESC LIMIT 100"""
|
|
).fetchall()
|
|
return [dict(row) for row in rows]
|
|
|
|
|
|
@app.delete("/api/notes/{note_id}")
|
|
def delete_note(note_id: str) -> dict[str, str]:
|
|
with connect_db() as connection:
|
|
note = connection.execute(
|
|
"SELECT audio_path, original_audio_path FROM notes WHERE id = ?", (note_id,)
|
|
).fetchone()
|
|
if not note:
|
|
raise HTTPException(status_code=404, detail="Note not found")
|
|
connection.execute("DELETE FROM note_search WHERE note_id = ?", (note_id,))
|
|
connection.execute("DELETE FROM notes WHERE id = ?", (note_id,))
|
|
if note["audio_path"]:
|
|
Path(note["audio_path"]).unlink(missing_ok=True)
|
|
if note["original_audio_path"]:
|
|
Path(note["original_audio_path"]).unlink(missing_ok=True)
|
|
return {"id": note_id, "status": "deleted"}
|
|
|
|
|
|
@app.put("/api/notes/{note_id}/transcript")
|
|
def save_transcript(note_id: str, payload: dict[str, str]) -> dict[str, str]:
|
|
transcript = payload.get("transcript", "").strip()
|
|
with connect_db() as connection:
|
|
note = connection.execute("SELECT status FROM notes WHERE id = ?", (note_id,)).fetchone()
|
|
if not note:
|
|
raise HTTPException(status_code=404, detail="Note not found")
|
|
status = "written" if note["status"] == "written" else "transcribed" if transcript else "awaiting_transcription"
|
|
update_transcript(note_id, transcript, status)
|
|
if transcript and note["status"] != "written":
|
|
remove_transcribed_audio(note_id)
|
|
return {"id": note_id, "status": status}
|
|
|
|
|
|
app.mount("/", StaticFiles(directory=FRONTEND_DIR, html=True), name="frontend") |