family_home_lab/portal/main.py

"""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 celery_app, 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)
    task_id = ""
    try:
        task_id = transcribe_audio.delay(bucket, key, user.username).id
    except Exception:
        logger.warning("could not enqueue transcription (broker down?)", exc_info=True)
    return RedirectResponse(f"/transcriber?status=started&task={task_id}", 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)
    )
    task_id = ""
    try:
        task_id = transcribe_audio.delay(bucket, key, user.username).id
    except Exception:
        logger.warning("could not enqueue transcription after upload", exc_info=True)
    return RedirectResponse(f"/transcriber?status=started&task={task_id}", status_code=303)


@app.get("/api/transcriber/status")
async def transcriber_status(task: str):
    from celery.result import AsyncResult
    r = AsyncResult(task, app=celery_app)
    return {"state": r.state, "result": r.result if r.state == "SUCCESS" else None}


@app.post("/transcriber/delete")
async def transcriber_delete(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:
        await run_in_threadpool(lambda: s3_client().delete_object(Bucket=bucket, Key=key))
    except Exception:
        logger.warning("transcriber delete failed", exc_info=True)
    return RedirectResponse("/transcriber", status_code=303)


@app.post("/files/delete")
async def files_delete(request: Request, name: Annotated[str, Form()]) -> HTMLResponse:
    user = await _current_user(request)
    if user is None:
        return RedirectResponse("/login", status_code=303)
    p = _resolve_shared(user.username, name)
    if p is not None:
        try:
            p.unlink()
        except Exception:
            logger.warning("files delete failed", exc_info=True)
    return RedirectResponse("/files", 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)
    task_id = ""
    try:
        task_id = transcribe_sheetmusic.delay(bucket, key, user.username).id
    except Exception:
        logger.warning("could not enqueue sheet-music transcription", exc_info=True)
    return RedirectResponse(f"/transcriber?status=started-mus&task={task_id}", 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}