645 lines
22 KiB
Python
645 lines
22 KiB
Python
"""Family Home Lab console — FastAPI + Jinja2 + HTMX.
|
|
|
|
Screens:
|
|
/login POST form -> sets session cookie, redirects
|
|
/ dashboard (tool grid)
|
|
/tool/{id} embed view (iframe well) or link-out
|
|
/admin user + service management (Sam only)
|
|
/api/status HTMX-polled status endpoint (30s)
|
|
|
|
Every page renders against the design tokens in static/tokens.css.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
from zoneinfo import ZoneInfo
|
|
|
|
MELB = ZoneInfo("Australia/Melbourne")
|
|
import json
|
|
import os
|
|
import httpx
|
|
from pathlib import Path
|
|
from typing import Annotated
|
|
|
|
from fastapi import Depends, FastAPI, File, Form, HTTPException, Request, UploadFile
|
|
from fastapi.concurrency import run_in_threadpool
|
|
from fastapi.responses import HTMLResponse, RedirectResponse, Response, Response
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.templating import Jinja2Templates
|
|
from sqlalchemy import select
|
|
|
|
from . import auth
|
|
from .config import settings
|
|
from .database import (
|
|
session_scope,
|
|
User,
|
|
get_session,
|
|
get_user_by_username,
|
|
init_db,
|
|
)
|
|
from .tasks import ensure_user_bucket, transcribe_audio, transcribe_sheetmusic
|
|
from .s3 import list_prefix, list_recent, s3_bucket_name, s3_client
|
|
from .tools import section_color, section_label, tools_for_user
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent
|
|
templates = Jinja2Templates(directory=str(BASE_DIR / "templates"))
|
|
templates.env.globals["settings"] = settings
|
|
|
|
app = FastAPI(title=settings.APP_NAME, docs_url=None if not settings.DEBUG else "/docs")
|
|
app.mount(
|
|
"/static", StaticFiles(directory=str(BASE_DIR / "static")), name="static"
|
|
)
|
|
|
|
|
|
# Per-user tool URL for the dsh chat instance placeholder; actual contract is
|
|
# defined in plan.md §6 to be handed to the dsh agent.
|
|
# (dsh link specialization lives in portal/tools.py::_dsh_for_user)
|
|
|
|
|
|
@app.on_event("startup")
|
|
async def on_startup() -> None:
|
|
await init_db()
|
|
await _seed_admin_if_empty()
|
|
|
|
|
|
async def _seed_admin_if_empty() -> None:
|
|
"""First-run: if the users table is empty, create the admin from env."""
|
|
from sqlalchemy import func
|
|
|
|
admin_user = settings.ADMIN_USERNAME
|
|
admin_pass = settings.ADMIN_PASSWORD
|
|
admin_name = settings.ADMIN_FULLNAME
|
|
if not admin_user or not admin_pass:
|
|
logger.warning("ADMIN_USERNAME/ADMIN_PASSWORD not set — skipping seed")
|
|
return
|
|
|
|
async with session_scope() as session:
|
|
count = await session.scalar(select(func.count()).select_from(User))
|
|
if count:
|
|
return
|
|
admin = User(
|
|
username=admin_user,
|
|
full_name=admin_name or admin_user,
|
|
password_hash=auth.hash_password(admin_pass),
|
|
is_admin=True,
|
|
can_chat=True,
|
|
can_image=True,
|
|
can_video=True,
|
|
can_audio=True,
|
|
is_active=True,
|
|
)
|
|
session.add(admin)
|
|
await session.commit()
|
|
logger.info("Seeded initial admin account %r", admin_user)
|
|
|
|
|
|
def _greeting(now: datetime) -> str:
|
|
h = now.astimezone(MELB).hour
|
|
if h < 5:
|
|
return "Up late"
|
|
if h < 12:
|
|
return "Good morning"
|
|
if h < 18:
|
|
return "Good afternoon"
|
|
return "Good evening"
|
|
|
|
|
|
async def _current_user(request: Request) -> User | None:
|
|
"""Resolve the logged-in user from the session cookie (async-safe)."""
|
|
user = await auth.current_user_dep(request)
|
|
return user
|
|
|
|
|
|
@app.get("/login", response_class=HTMLResponse)
|
|
async def login_page(
|
|
request: Request,
|
|
next: str = "/",
|
|
) -> HTMLResponse:
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"login.html",
|
|
{"greeting": _greeting(datetime.now(timezone.utc)), "next": next},
|
|
)
|
|
|
|
|
|
@app.post("/login")
|
|
async def login_submit(
|
|
request: Request,
|
|
username: Annotated[str, Form()],
|
|
password: Annotated[str, Form()],
|
|
next: str = Form("/"),
|
|
) -> HTMLResponse:
|
|
async with session_scope() as session:
|
|
user = await get_user_by_username(session, username.strip())
|
|
|
|
if user is None or not auth.verify_password(password, user.password_hash):
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"login.html",
|
|
{
|
|
"greeting": _greeting(datetime.now(timezone.utc)),
|
|
"error": "Incorrect username or password.",
|
|
"next": next,
|
|
},
|
|
status_code=401,
|
|
)
|
|
if not user.is_active:
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"login.html",
|
|
{
|
|
"greeting": _greeting(datetime.now(timezone.utc)),
|
|
"error": "This account is disabled.",
|
|
"next": next,
|
|
},
|
|
status_code=403,
|
|
)
|
|
|
|
response = RedirectResponse(next or "/", status_code=303)
|
|
auth.set_session_cookie(request, response, user.username)
|
|
return response
|
|
|
|
|
|
@app.get("/logout")
|
|
async def logout(request: Request):
|
|
response = RedirectResponse("/login", status_code=303)
|
|
auth.clear_session_cookie(request, response)
|
|
return response
|
|
|
|
|
|
@app.get("/", response_class=HTMLResponse)
|
|
async def dashboard(request: Request) -> HTMLResponse:
|
|
user = await _current_user(request)
|
|
if user is None:
|
|
return RedirectResponse("/login", status_code=303)
|
|
groups = tools_for_user(user)
|
|
now = datetime.now(timezone.utc)
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"dashboard.html",
|
|
{
|
|
"user": user,
|
|
"greeting": _greeting(now),
|
|
"sections": [
|
|
{
|
|
"label": section_label(cat),
|
|
"color": section_color(cat),
|
|
"tools": group,
|
|
}
|
|
for cat, group in groups
|
|
],
|
|
},
|
|
)
|
|
|
|
|
|
@app.get("/tool/{tool_id}", response_class=HTMLResponse)
|
|
async def tool_embed(request: Request, tool_id: str) -> HTMLResponse:
|
|
user = await _current_user(request)
|
|
if user is None:
|
|
return RedirectResponse("/login", status_code=303)
|
|
groups = tools_for_user(user)
|
|
tool = next((t for _, g in groups for t in g if t.id == tool_id), None)
|
|
if tool is None:
|
|
return templates.TemplateResponse(
|
|
request, "not_found.html", {"user": user}, status_code=404
|
|
)
|
|
# Workspace files: the user's bucket + the shared pool (does not block the
|
|
# event loop, degrades to empty list on any S3 error).
|
|
user_bucket = s3_bucket_name(user.username)
|
|
my_files = await run_in_threadpool(list_recent, user_bucket, 20)
|
|
shared_files = await run_in_threadpool(list_recent, "shared-media", 20)
|
|
ctx = {
|
|
"user": user,
|
|
"tool": tool,
|
|
"files": my_files,
|
|
"shared_files": shared_files,
|
|
"user_bucket": user_bucket,
|
|
}
|
|
if tool.mode == "window":
|
|
# Link-out fallback for tools that forbid framing.
|
|
return templates.TemplateResponse(request, "linkout.html", ctx)
|
|
return templates.TemplateResponse(request, "embed.html", ctx)
|
|
|
|
|
|
|
|
SHARED_DIR = Path(os.getenv("SHARED_DIR", "/shared-media"))
|
|
|
|
|
|
def _shared_files(username: str) -> list:
|
|
"""List this user's uploaded files in the shared-media folder (created on demand)."""
|
|
if not SHARED_DIR.is_dir():
|
|
return []
|
|
user_dir = SHARED_DIR / username
|
|
if not user_dir.is_dir():
|
|
return []
|
|
out = []
|
|
for f in sorted(user_dir.iterdir()):
|
|
if f.is_file():
|
|
out.append({"name": f.name, "size": f.stat().st_size,
|
|
"img": f.suffix.lower() in _IMAGE_MIME})
|
|
return out
|
|
|
|
|
|
|
|
_IMAGE_MIME = {".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg",
|
|
".gif": "image/gif", ".webp": "image/webp", ".bmp": "image/bmp",
|
|
".svg": "image/svg+xml"}
|
|
|
|
|
|
def _resolve_shared(username: str, name: str):
|
|
"""Return a validated path inside this user's shared folder, or None."""
|
|
if Path(name).name != name: # block path traversal
|
|
return None
|
|
p = (SHARED_DIR / username / name).resolve()
|
|
base = (SHARED_DIR / username).resolve()
|
|
if not str(p).startswith(str(base)):
|
|
return None
|
|
return p if p.is_file() else None
|
|
|
|
|
|
@app.get("/files/raw/{name}")
|
|
async def files_raw(request: Request, name: str) -> Response:
|
|
"""Inline the file (images render as a preview)."""
|
|
user = await _current_user(request)
|
|
if user is None:
|
|
return RedirectResponse("/login", status_code=303)
|
|
p = _resolve_shared(user.username, name)
|
|
if p is None:
|
|
return Response(b"Not found", status_code=404)
|
|
media = _IMAGE_MIME.get(p.suffix.lower(), "application/octet-stream")
|
|
return Response(content=p.read_bytes(), media_type=media)
|
|
|
|
|
|
@app.get("/files/dl/{name}")
|
|
async def files_dl(request: Request, name: str) -> Response:
|
|
"""Download (forced attachment) a file so it can be saved/emailed."""
|
|
user = await _current_user(request)
|
|
if user is None:
|
|
return RedirectResponse("/login", status_code=303)
|
|
p = _resolve_shared(user.username, name)
|
|
if p is None:
|
|
return Response(b"Not found", status_code=404)
|
|
media = _IMAGE_MIME.get(p.suffix.lower(), "application/octet-stream")
|
|
return Response(content=p.read_bytes(), media_type=media,
|
|
headers={"Content-Disposition": f'attachment; filename="{p.name}"'})
|
|
|
|
|
|
@app.get("/files", response_class=HTMLResponse)
|
|
async def files_page(request: Request) -> HTMLResponse:
|
|
user = await _current_user(request)
|
|
if user is None:
|
|
return RedirectResponse("/login", status_code=303)
|
|
return templates.TemplateResponse(
|
|
request, "files.html", {"user": user, "files": _shared_files(user.username)}
|
|
)
|
|
|
|
|
|
@app.post("/files/upload")
|
|
async def files_upload(request: Request, file: Annotated[UploadFile, File()]) -> HTMLResponse:
|
|
user = await _current_user(request)
|
|
if user is None:
|
|
return RedirectResponse("/login", status_code=303)
|
|
error = None
|
|
if not file or not file.filename:
|
|
error = "No file chosen."
|
|
else:
|
|
try:
|
|
name = Path(file.filename).name.lower()
|
|
# allow images + common docs (photos are the main use case)
|
|
if not name.endswith((".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".svg", ".psd", ".xcf", ".pdf", ".txt", ".md")):
|
|
error = f"File type not allowed: {file.filename}"
|
|
else:
|
|
SHARED_DIR.mkdir(exist_ok=True)
|
|
user_dir = SHARED_DIR / user.username
|
|
user_dir.mkdir(exist_ok=True)
|
|
data = await file.read()
|
|
if len(data) > 50 * 1024 * 1024:
|
|
error = "File too large (max 50 MB)."
|
|
else:
|
|
(user_dir / file.filename).write_bytes(data)
|
|
except Exception as exc: # noqa: BLE001
|
|
logger.warning("files upload failed", exc_info=True)
|
|
error = f"Upload failed: {exc}"
|
|
return templates.TemplateResponse(
|
|
request, "files.html",
|
|
{"user": user, "files": _shared_files(user.username),
|
|
"error": error, "ok": error is None and file and file.filename},
|
|
)
|
|
|
|
|
|
|
|
VOICE_AGENT = os.getenv("VOICE_AGENT_URL", "http://192.168.20.13:8501")
|
|
|
|
|
|
@app.get("/voice", response_class=HTMLResponse)
|
|
async def voice_page(request: Request) -> HTMLResponse:
|
|
user = await _current_user(request)
|
|
if user is None:
|
|
return RedirectResponse("/login", status_code=303)
|
|
return templates.TemplateResponse(request, "voice.html", {"user": user})
|
|
|
|
|
|
@app.post("/voice", response_class=HTMLResponse)
|
|
async def voice_send(request: Request, message: Annotated[str, Form()]) -> HTMLResponse:
|
|
user = await _current_user(request)
|
|
if user is None:
|
|
return RedirectResponse("/login", status_code=303)
|
|
command = message.strip()
|
|
error = reply = None
|
|
if command:
|
|
try:
|
|
async with httpx.AsyncClient(timeout=60) as client:
|
|
r = await client.post(f"{VOICE_AGENT}/voice", json={"text": command})
|
|
r.raise_for_status()
|
|
reply = r.json().get("reply") or r.json().get("text")
|
|
except Exception as exc: # noqa: BLE001
|
|
logger.warning("voice agent call failed", exc_info=True)
|
|
error = f"Voice assistant unavailable: {type(exc).__name__}"
|
|
return templates.TemplateResponse(
|
|
request, "voice.html", {"user": user, "command": command, "reply": reply, "error": error}
|
|
)
|
|
|
|
|
|
@app.get("/admin", response_class=HTMLResponse)
|
|
async def admin_panel(request: Request) -> HTMLResponse:
|
|
user = await _current_user(request)
|
|
if user is None:
|
|
return RedirectResponse("/login", status_code=303)
|
|
if not user.is_admin:
|
|
return templates.TemplateResponse(
|
|
request, "forbidden.html", {"user": user}, status_code=403
|
|
)
|
|
async with session_scope() as session:
|
|
result = await session.scalars(select(User).order_by(User.username))
|
|
users = list(result)
|
|
return templates.TemplateResponse(
|
|
request, "admin.html", {"user": user, "users": users}
|
|
)
|
|
|
|
|
|
|
|
@app.get("/account/password", response_class=HTMLResponse)
|
|
async def change_password_page(request: Request) -> HTMLResponse:
|
|
user = await _current_user(request)
|
|
if user is None:
|
|
return RedirectResponse("/login", status_code=303)
|
|
return templates.TemplateResponse(
|
|
request, "change_password.html", {"user": user}
|
|
)
|
|
|
|
|
|
@app.post("/account/password")
|
|
async def change_password(
|
|
request: Request,
|
|
current: Annotated[str, Form()],
|
|
password: Annotated[str, Form()],
|
|
confirm: Annotated[str, Form()],
|
|
) -> HTMLResponse:
|
|
user = await _current_user(request)
|
|
if user is None:
|
|
return RedirectResponse("/login", status_code=303)
|
|
error = None
|
|
try:
|
|
ok_current = auth.verify_password(current, user.password_hash)
|
|
except Exception:
|
|
ok_current = False
|
|
if password != confirm:
|
|
error = "New passwords do not match."
|
|
elif not ok_current:
|
|
error = "Current password is incorrect."
|
|
elif len(password) < 8:
|
|
error = "New password must be at least 8 characters."
|
|
else:
|
|
async with session_scope() as session:
|
|
u = await get_user_by_username(session, user.username)
|
|
if u is not None:
|
|
u.password_hash = auth.hash_password(password)
|
|
await session.commit()
|
|
return RedirectResponse("/?pw=1", status_code=303)
|
|
return templates.TemplateResponse(
|
|
request, "change_password.html", {"user": user, "error": error},
|
|
status_code=400,
|
|
)
|
|
|
|
|
|
@app.post("/admin/users")
|
|
async def admin_add_user(
|
|
request: Request,
|
|
username: Annotated[str, Form()],
|
|
full_name: Annotated[str, Form()],
|
|
password: Annotated[str, Form()],
|
|
) -> HTMLResponse:
|
|
"""Admin: create a user and enqueue their S3 bucket provisioning."""
|
|
admin = await _current_user(request)
|
|
if admin is None:
|
|
return RedirectResponse("/login", status_code=303)
|
|
if not admin.is_admin:
|
|
raise HTTPException(status_code=403, detail="Admins only")
|
|
|
|
async with session_scope() as session:
|
|
existing = await get_user_by_username(session, username.strip())
|
|
if existing is not None:
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"admin.html",
|
|
{"user": admin, "error": "Username already exists."},
|
|
status_code=400,
|
|
)
|
|
user = User(
|
|
username=username.strip(),
|
|
full_name=full_name.strip() or username.strip(),
|
|
password_hash=auth.hash_password(password),
|
|
)
|
|
session.add(user)
|
|
await session.commit()
|
|
new_username = user.username
|
|
# Provision bucket in background. User creation must never fail because a
|
|
# background broker hiccup ; log instead so provisioning can be retried.
|
|
try:
|
|
ensure_user_bucket.delay(new_username)
|
|
except Exception:
|
|
logger.warning("Could not enqueue bucket provisioning for %r (broker down?)", new_username)
|
|
return RedirectResponse("/admin", status_code=303)
|
|
|
|
|
|
|
|
@app.get("/admin/pi", response_class=HTMLResponse)
|
|
async def admin_pi(request: Request) -> HTMLResponse:
|
|
"""Admin: Pi Dashboard — agent session cards sourced from ~/.pi/agent/dashboard.
|
|
|
|
Files are synced from the pi host into /pi-dashboard (mounted ro). Shows
|
|
session name, machine, status, last tool, last-seen, cost estimate.
|
|
"""
|
|
user = await _current_user(request)
|
|
if user is None:
|
|
return RedirectResponse("/login", status_code=303)
|
|
if not user.is_admin:
|
|
raise HTTPException(status_code=403, detail="Admins only")
|
|
pi_dir = Path(os.getenv("PI_DASHBOARD_DIR", "/pi-dashboard"))
|
|
sessions = []
|
|
if pi_dir.is_dir():
|
|
for f in sorted(pi_dir.glob("*.json")):
|
|
try:
|
|
sessions.append(json.loads(f.read_text(encoding="utf-8")))
|
|
except Exception:
|
|
continue
|
|
sessions.sort(key=lambda s: (s.get("last_seen_at") or ""), reverse=True)
|
|
return templates.TemplateResponse(
|
|
request, "admin_pi.html", {"user": user, "sessions": sessions, "pi_dir": str(pi_dir)}
|
|
)
|
|
|
|
@app.post("/admin/users/rename")
|
|
async def admin_rename_user(
|
|
request: Request,
|
|
old_username: Annotated[str, Form()],
|
|
new_username: Annotated[str, Form()],
|
|
) -> HTMLResponse:
|
|
"""Admin: rename a user's login username (keeps password, perms, bucket name)."""
|
|
admin = await _current_user(request)
|
|
if admin is None:
|
|
return RedirectResponse("/login", status_code=303)
|
|
if not admin.is_admin:
|
|
raise HTTPException(status_code=403, detail="Admins only")
|
|
old_username = old_username.strip()
|
|
new_username = new_username.strip()
|
|
error = None
|
|
async with session_scope() as session:
|
|
target = await get_user_by_username(session, old_username)
|
|
if target is None:
|
|
error = f"User {old_username!r} not found."
|
|
elif new_username == old_username:
|
|
error = "No change."
|
|
elif await get_user_by_username(session, new_username) is not None:
|
|
error = f"Username {new_username!r} is already taken."
|
|
else:
|
|
target.username = new_username
|
|
await session.commit()
|
|
logger.info("Admin %s renamed user %s -> %s", admin.username, old_username, new_username)
|
|
async with session_scope() as session:
|
|
result = await session.scalars(select(User).order_by(User.username))
|
|
users = list(result)
|
|
return templates.TemplateResponse(
|
|
request, "admin.html", {"user": admin, "users": users, "error": error},
|
|
status_code=400 if error else 200,
|
|
)
|
|
|
|
|
|
_AUDIO_EXT = {".wav", ".mp3", ".flac", ".ogg", ".m4a", ".aiff", ".aif", ".opus"}
|
|
_VIDEO_EXT = {".mp4", ".mov", ".webm", ".mkv", ".m4v", ".avi", ".mpg", ".mpeg"}
|
|
_TRANS_EXT = _AUDIO_EXT | _VIDEO_EXT
|
|
|
|
|
|
@app.get("/transcriber", response_class=HTMLResponse)
|
|
async def transcriber_page(request: Request) -> HTMLResponse:
|
|
user = await _current_user(request)
|
|
if user is None:
|
|
return RedirectResponse("/login", status_code=303)
|
|
bucket = s3_bucket_name(user.username)
|
|
all_keys = await run_in_threadpool(list_prefix, bucket, "", 400)
|
|
audio_files = [k for k in all_keys
|
|
if k.lower().endswith(tuple(_TRANS_EXT))]
|
|
transcriptions = [k for k in all_keys
|
|
if k.startswith("transcriptions/") or k.startswith("sheetmusic/")]
|
|
return templates.TemplateResponse(
|
|
request, "transcriber.html",
|
|
{"user": user, "bucket": bucket, "audio_files": audio_files,
|
|
"transcriptions": transcriptions, "status": request.query_params.get("status", "")},
|
|
)
|
|
|
|
|
|
@app.post("/transcriber/transcribe")
|
|
async def transcriber_transcribe(
|
|
request: Request,
|
|
key: Annotated[str, Form()],
|
|
) -> HTMLResponse:
|
|
user = await _current_user(request)
|
|
if user is None:
|
|
return RedirectResponse("/login", status_code=303)
|
|
bucket = s3_bucket_name(user.username)
|
|
try:
|
|
transcribe_audio.delay(bucket, key, user.username)
|
|
except Exception:
|
|
logger.warning("could not enqueue transcription (broker down?)", exc_info=True)
|
|
return RedirectResponse("/transcriber?status=started", status_code=303)
|
|
|
|
|
|
@app.post("/transcriber/upload")
|
|
async def transcriber_upload(
|
|
request: Request,
|
|
file: Annotated[UploadFile, File()],
|
|
) -> HTMLResponse:
|
|
user = await _current_user(request)
|
|
if user is None:
|
|
return RedirectResponse("/login", status_code=303)
|
|
import uuid
|
|
bucket = s3_bucket_name(user.username)
|
|
safe = (file.filename or "audio").replace(" ", "_")
|
|
key = f"audio/{uuid.uuid4().hex[:8]}-{safe}"
|
|
content = await file.read()
|
|
await run_in_threadpool(
|
|
lambda: s3_client().put_object(Bucket=bucket, Key=key, Body=content)
|
|
)
|
|
try:
|
|
transcribe_audio.delay(bucket, key, user.username)
|
|
except Exception:
|
|
logger.warning("could not enqueue transcription after upload", exc_info=True)
|
|
return RedirectResponse("/transcriber?status=started", status_code=303)
|
|
|
|
|
|
@app.get("/transcriber/download")
|
|
async def transcriber_download(request: Request, key: str):
|
|
user = await _current_user(request)
|
|
if user is None:
|
|
return RedirectResponse("/login", status_code=303)
|
|
bucket = s3_bucket_name(user.username)
|
|
try:
|
|
data = await run_in_threadpool(
|
|
lambda: (
|
|
s3_client().get_object(Bucket=bucket, Key=key)["Body"]
|
|
.read()
|
|
)
|
|
)
|
|
except Exception:
|
|
return HTMLResponse("not found", status_code=404)
|
|
fname = key.rsplit("/", 1)[-1]
|
|
return Response(content=data, media_type="audio/midi")
|
|
|
|
|
|
|
|
@app.post("/transcriber/sheetmusic")
|
|
async def transcriber_sheetmusic(
|
|
request: Request,
|
|
key: Annotated[str, Form()],
|
|
) -> HTMLResponse:
|
|
user = await _current_user(request)
|
|
if user is None:
|
|
return RedirectResponse("/login", status_code=303)
|
|
bucket = s3_bucket_name(user.username)
|
|
try:
|
|
transcribe_sheetmusic.delay(bucket, key, user.username)
|
|
except Exception:
|
|
logger.warning("could not enqueue sheet-music transcription", exc_info=True)
|
|
return RedirectResponse("/transcriber?status=started-mus", status_code=303)
|
|
|
|
|
|
@app.get("/api/status")
|
|
async def api_status(request: Request) -> HTMLResponse:
|
|
"""HTMX-polled snippet: updates each tool card's status dot (not full page)."""
|
|
user = await _current_user(request)
|
|
if user is None:
|
|
return HTMLResponse("", status_code=401)
|
|
groups = tools_for_user(user)
|
|
# Status is static for now; swap `status` for a live probe later.
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"partials/_status.html",
|
|
{"sections": [{"tools": g} for _, g in groups]},
|
|
)
|
|
|
|
|
|
@app.get("/healthz")
|
|
async def healthz() -> dict:
|
|
return {"ok": True, "app": settings.APP_NAME} |