import asyncio import json import logging import os import re import sqlite3 import subprocess import tempfile import time import uuid import xml.etree.ElementTree as ET from datetime import datetime, timedelta, timezone from pathlib import Path from urllib.parse import unquote, urlsplit 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" BACKUP_SETTINGS_PATH = DATA_DIR / "webdav-backup.json" WHISPER_SERVER_URL = os.environ.get("WHISPER_SERVER_URL", "").strip() BACKUP_WEBDAV_URL = os.environ.get("BACKUP_WEBDAV_URL", "").strip() BACKUP_WEBDAV_USERNAME = os.environ.get("BACKUP_WEBDAV_USERNAME", "") BACKUP_WEBDAV_PASSWORD = os.environ.get("BACKUP_WEBDAV_PASSWORD", "") 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() BACKUP_TASK: asyncio.Task[None] | None = None BACKUP_LOCK = asyncio.Lock() 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)""" ) connection.execute( """CREATE TABLE IF NOT EXISTS backup_state ( id INTEGER PRIMARY KEY CHECK (id = 1), revision INTEGER NOT NULL DEFAULT 0, completed_revision INTEGER NOT NULL DEFAULT 0, last_error TEXT )""" ) connection.execute("INSERT OR IGNORE INTO backup_state (id) VALUES (1)") 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()), ) if result.rowcount and status != "transcription_failed": connection.execute("UPDATE backup_state SET revision = revision + 1 WHERE id = 1") if result.rowcount and status != "transcription_failed": schedule_backup() def get_backup_settings() -> dict[str, str]: if BACKUP_WEBDAV_URL: return { "url": BACKUP_WEBDAV_URL, "username": BACKUP_WEBDAV_USERNAME, "password": BACKUP_WEBDAV_PASSWORD, } try: settings = json.loads(BACKUP_SETTINGS_PATH.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): settings = {} return { "url": settings.get("url", ""), "username": settings.get("username", ""), "password": settings.get("password", ""), } def save_backup_settings(settings: dict[str, str]) -> None: temporary_path = BACKUP_SETTINGS_PATH.with_suffix(".tmp") temporary_path.write_text(json.dumps(settings), encoding="utf-8") temporary_path.chmod(0o600) temporary_path.replace(BACKUP_SETTINGS_PATH) def backup_state() -> dict[str, object]: with connect_db() as connection: row = connection.execute( "SELECT revision, completed_revision, last_error FROM backup_state WHERE id = 1" ).fetchone() settings = get_backup_settings() return { "enabled": bool(settings["url"]), "managed_by_environment": bool(BACKUP_WEBDAV_URL), "url": settings["url"], "username": settings["username"], "password_set": bool(settings["password"]), "revision": row["revision"], "completed_revision": row["completed_revision"], "pending": row["revision"] > row["completed_revision"], "last_error": row["last_error"], } def schedule_backup() -> None: global BACKUP_TASK if BACKUP_TASK and not BACKUP_TASK.done(): return try: loop = asyncio.get_running_loop() except RuntimeError: return BACKUP_TASK = loop.create_task(run_backup()) async def file_chunks(path: Path): with path.open("rb") as backup_file: while chunk := await asyncio.to_thread(backup_file.read, 1024 * 1024): yield chunk async def create_database_snapshot(path: Path) -> None: def copy_database() -> None: source = sqlite3.connect(DB_PATH) destination = sqlite3.connect(path) try: source.backup(destination) finally: destination.close() source.close() await asyncio.to_thread(copy_database) def webdav_file_url(folder_url: str) -> str: parsed = urlsplit(folder_url) if parsed.scheme not in {"http", "https"} or not parsed.netloc or parsed.query or parsed.fragment: raise ValueError("Enter a valid WebDAV folder URL") if parsed.username or parsed.password: raise ValueError("Put WebDAV credentials in the username and password fields") return folder_url.rstrip("/") + "/faerro-kb.sqlite3" def retained_backup_archives(names: list[str], now: datetime) -> set[str]: archives = [] for name in names: match = re.fullmatch(r"faerro-kb-(\d{8}T\d{6}Z)-r\d+\.sqlite3", name) if not match: continue timestamp = datetime.strptime(match.group(1), "%Y%m%dT%H%M%SZ").replace(tzinfo=timezone.utc) if timestamp <= now: archives.append((name, timestamp)) daily: dict[object, tuple[str, datetime]] = {} weekly: dict[object, tuple[str, datetime]] = {} monthly: dict[object, tuple[str, datetime]] = {} yearly: dict[object, tuple[str, datetime]] = {} today = now.date() current_week = today - timedelta(days=today.weekday()) current_month = now.year * 12 + now.month for name, timestamp in archives: date = timestamp.date() age_days = (today - date).days week = date - timedelta(days=date.weekday()) age_weeks = (current_week - week).days // 7 age_months = current_month - (timestamp.year * 12 + timestamp.month) age_years = now.year - timestamp.year for buckets, key, age, limit in ( (daily, date, age_days, 7), (weekly, (week.isocalendar().year, week.isocalendar().week), age_weeks, 4), (monthly, (timestamp.year, timestamp.month), age_months, 12), (yearly, timestamp.year, age_years, 5), ): if 0 <= age < limit and (key not in buckets or timestamp > buckets[key][1]): buckets[key] = (name, timestamp) return { name for buckets in (daily, weekly, monthly, yearly) for name, _timestamp in buckets.values() } async def rotate_remote_backups(client: httpx.AsyncClient, folder_url: str) -> None: request_body = b""" """ response = await client.request( "PROPFIND", folder_url, headers={"Depth": "1", "Content-Type": "application/xml"}, content=request_body, ) response.raise_for_status() root = ET.fromstring(response.content) folder_path = urlsplit(folder_url).path.rstrip("/") remote_archives = [] for item in root.findall("{DAV:}response"): href = item.findtext("{DAV:}href") if not href: continue item_url = httpx.URL(folder_url).join(href) item_path = item_url.path if item_path.rsplit("/", 1)[0].rstrip("/") != folder_path: continue name = unquote(item_path.rsplit("/", 1)[-1]) if re.fullmatch(r"faerro-kb-\d{8}T\d{6}Z-r\d+\.sqlite3", name): remote_archives.append((name, str(item_url))) retained = retained_backup_archives([name for name, _url in remote_archives], datetime.now(timezone.utc)) for name, item_url in remote_archives: if name not in retained: response = await client.delete(item_url) response.raise_for_status() async def run_backup() -> None: async with BACKUP_LOCK: while True: settings = get_backup_settings() if not settings["url"]: return with connect_db() as connection: state = connection.execute( "SELECT revision, completed_revision FROM backup_state WHERE id = 1" ).fetchone() if state["revision"] <= state["completed_revision"]: return revision = state["revision"] snapshot_path = None try: target_url = webdav_file_url(settings["url"]) with tempfile.NamedTemporaryFile(suffix=".sqlite3", dir=DATA_DIR, delete=False) as snapshot: snapshot_path = Path(snapshot.name) await create_database_snapshot(snapshot_path) with sqlite3.connect(snapshot_path) as snapshot_connection: snapshot_connection.execute( "UPDATE backup_state SET completed_revision = ?, last_error = NULL WHERE id = 1", (revision,), ) auth = (settings["username"], settings["password"]) if settings["username"] else None archive_name = ( f"faerro-kb-{datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')}" f"-r{revision:010d}.sqlite3" ) async with httpx.AsyncClient(timeout=120, auth=auth) as client: response = await client.put( f"{settings['url'].rstrip('/')}/{archive_name}", content=file_chunks(snapshot_path), ) response.raise_for_status() response = await client.put(target_url, content=file_chunks(snapshot_path)) response.raise_for_status() await rotate_remote_backups(client, settings["url"]) with connect_db() as connection: connection.execute( "UPDATE backup_state SET completed_revision = ?, last_error = NULL WHERE id = 1", (revision,), ) except httpx.HTTPStatusError as error: error_message = f"WebDAV returned HTTP {error.response.status_code}" with connect_db() as connection: connection.execute("UPDATE backup_state SET last_error = ? WHERE id = 1", (error_message,)) logger.warning("WebDAV database backup failed: HTTP %s", error.response.status_code) return except Exception as error: error_message = "WebDAV backup failed; check the folder URL and credentials" if isinstance(error, ValueError): error_message = str(error) with connect_db() as connection: connection.execute("UPDATE backup_state SET last_error = ? WHERE id = 1", (error_message,)) logger.warning("WebDAV database backup failed: %s", type(error).__name__) return finally: if snapshot_path: snapshot_path.unlink(missing_ok=True) 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: if BACKUP_WEBDAV_URL: with connect_db() as connection: connection.execute( "UPDATE backup_state SET revision = 1 WHERE id = 1 AND revision = 0 AND completed_revision = 0" ) schedule_backup() 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.get("/api/backup") def get_backup_status() -> dict[str, object]: return backup_state() @app.put("/api/backup/config") async def configure_backup(payload: dict[str, str]) -> dict[str, object]: if BACKUP_WEBDAV_URL: raise HTTPException(status_code=409, detail="WebDAV backup is managed by environment variables") url = payload.get("url", "").strip() username = payload.get("username", "").strip() current = get_backup_settings() password = "" if payload.get("clear_password") == "true" else current["password"] if payload.get("password"): password = payload["password"] if url: try: webdav_file_url(url) except ValueError as error: raise HTTPException(status_code=400, detail=str(error)) from error save_backup_settings({"url": url, "username": username, "password": password}) if url: with connect_db() as connection: connection.execute("UPDATE backup_state SET revision = revision + 1 WHERE id = 1") schedule_backup() return backup_state() @app.post("/api/backup/test") async def test_backup_connection() -> dict[str, str]: settings = get_backup_settings() if not settings["url"]: raise HTTPException(status_code=400, detail="Configure a WebDAV folder first") try: folder_url = settings["url"].rstrip("/") webdav_file_url(folder_url) test_url = f"{folder_url}/faerro-kb-test-{uuid.uuid4().hex}" auth = (settings["username"], settings["password"]) if settings["username"] else None async with httpx.AsyncClient(timeout=30, auth=auth) as client: response = await client.request( "PROPFIND", folder_url, headers={"Depth": "1", "Content-Type": "application/xml"}, content=b'', ) response.raise_for_status() response = await client.put(test_url, content=b"Faerro KB WebDAV test") response.raise_for_status() response = await client.delete(test_url) response.raise_for_status() return {"status": "ok", "message": "WebDAV listing, write, and delete test passed."} except httpx.HTTPStatusError as error: raise HTTPException( status_code=502, detail=f"WebDAV returned HTTP {error.response.status_code} during the test", ) from error except Exception as error: raise HTTPException(status_code=502, detail="Could not connect to the WebDAV folder") from error @app.post("/api/backup/retry", status_code=202) async def retry_backup() -> dict[str, object]: if not get_backup_settings()["url"]: raise HTTPException(status_code=400, detail="Configure a WebDAV folder first") schedule_backup() return backup_state() @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") async 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")