Implement WebDAV backup feature: add configuration options, API endpoints, and UI for managing database backups
This commit is contained in:
+224
-1
@@ -1,4 +1,5 @@
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
@@ -9,6 +10,7 @@ import time
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from urllib.parse import urlsplit
|
||||
|
||||
import httpx
|
||||
from fastapi import BackgroundTasks, FastAPI, File, Form, HTTPException, UploadFile
|
||||
@@ -18,10 +20,16 @@ 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)
|
||||
|
||||
@@ -53,6 +61,15 @@ def initialize_db() -> None:
|
||||
"""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")
|
||||
@@ -86,6 +103,145 @@ def update_transcript(note_id: str, transcript: str, status: str) -> None:
|
||||
"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"
|
||||
|
||||
|
||||
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
|
||||
async with httpx.AsyncClient(timeout=120, auth=auth) as client:
|
||||
response = await client.put(target_url, content=file_chunks(snapshot_path))
|
||||
response.raise_for_status()
|
||||
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:
|
||||
@@ -175,6 +331,12 @@ async def test_whisper_connection() -> bool:
|
||||
|
||||
@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'"
|
||||
@@ -210,6 +372,67 @@ 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")
|
||||
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.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 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)
|
||||
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,
|
||||
@@ -367,7 +590,7 @@ def delete_note(note_id: str) -> dict[str, str]:
|
||||
|
||||
|
||||
@app.put("/api/notes/{note_id}/transcript")
|
||||
def save_transcript(note_id: str, payload: dict[str, str]) -> dict[str, str]:
|
||||
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()
|
||||
|
||||
Reference in New Issue
Block a user