"""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 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 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) @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"} @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(_AUDIO_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}