"""FastAPI web frontend: browse jobs, view variants/intermediates, remix.""" from __future__ import annotations import logging import re from pathlib import Path from typing import Any from urllib.parse import quote from fastapi import FastAPI, Form, HTTPException, Query, Request from fastapi.responses import FileResponse, HTMLResponse, RedirectResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from . import config, pipeline, remix from .logging_config import configure_logging configure_logging() logger = logging.getLogger("livef12.web") app = FastAPI(title=config.SITE_TITLE) BASE_DIR = Path(__file__).resolve().parent templates = Jinja2Templates(directory=str(BASE_DIR / "templates")) app.mount("/static", StaticFiles(directory=str(BASE_DIR / "static")), name="static") # Job id = sanitized source stem (may include dots). _JOB_ID_RE = re.compile(r"^[A-Za-z0-9._-]+$") def _validate_job_id(job_id: str) -> str: raw = job_id or "" if not raw.strip() or "/" in raw or "\\" in raw or ".." in raw: raise HTTPException(status_code=400, detail="Ungueltige Job-ID") stem = pipeline.sanitize_stem(raw) if not _JOB_ID_RE.match(stem): raise HTTPException(status_code=400, detail="Ungueltige Job-ID") return stem def _job_exists(job_id: str) -> str: stem = _validate_job_id(job_id) if pipeline.read_status(stem) is None and not pipeline.job_paths(stem).original.exists(): raise HTTPException(status_code=404, detail="Job nicht gefunden") return stem def _safe_job_file(job_id: str, filename: str) -> Path: """Resolve a basename under variants/ or intermediates/ for this stem.""" stem = _validate_job_id(job_id) name = Path(filename).name if name != filename or "/" in filename or "\\" in filename or name.startswith("."): raise HTTPException(status_code=400, detail="Ungueltiger Pfad") if not name.startswith(f"{stem}_"): raise HTTPException(status_code=400, detail="Ungueltiger Pfad") for directory in (config.VARIANTS_DIR, config.INTERMEDIATES_DIR): candidate = (directory / name).resolve() if directory.resolve() not in candidate.parents and candidate != directory.resolve(): continue if candidate.is_file(): return candidate raise HTTPException(status_code=404, detail="Datei nicht gefunden") def list_jobs() -> list[dict[str, Any]]: jobs = [] for job_id in pipeline.list_job_ids(): data = pipeline.read_meta(job_id) or {} variants = data.get("variants") or [] jobs.append( { "job_id": job_id, "status": data.get("status", "unknown"), "created_at": data.get("created_at") or "", "variant_count": len(variants), "thumbnail": variants[-1].get("file") if variants else None, } ) jobs.sort(key=lambda j: j["created_at"] or j["job_id"], reverse=True) return jobs @app.get("/", response_class=HTMLResponse) def index(request: Request) -> HTMLResponse: return templates.TemplateResponse( request, "index.html", {"jobs": list_jobs(), "site_title": config.SITE_TITLE}, ) @app.get("/jobs/{job_id}", response_class=HTMLResponse) def job_detail(request: Request, job_id: str) -> HTMLResponse: stem = _job_exists(job_id) paths = pipeline.job_paths(stem) status = pipeline.read_status(stem) or {} manifest = pipeline.read_manifest(stem) or {"variants": []} intermediates = ( sorted(p.name for p in paths.intermediates.glob(f"{stem}_*.png") if p.is_file()) if paths.intermediates.exists() else [] ) return templates.TemplateResponse( request, "job.html", { "site_title": config.SITE_TITLE, "job_id": stem, "status": status, "manifest": manifest, "original_name": paths.original.name if paths.original.exists() else None, "rembg_name": paths.rembg.name if paths.rembg.exists() else None, "has_rembg": paths.rembg.exists(), "intermediates": intermediates, }, ) @app.get("/jobs/{job_id}/files/{filename:path}") def job_file(job_id: str, filename: str) -> FileResponse: path = _safe_job_file(job_id, filename) return FileResponse(path) @app.get("/jobs/{job_id}/remix", response_class=HTMLResponse) def remix_form( request: Request, job_id: str, error: str | None = None, from_variant: str | None = Query(None, alias="from"), ) -> HTMLResponse: stem = _job_exists(job_id) paths = pipeline.job_paths(stem) if not paths.rembg.exists(): raise HTTPException(status_code=409, detail="Job hat noch kein Rembg-Ergebnis, Remix noch nicht moeglich.") assets = pipeline.load_assets() options = remix.build_remix_options(assets) prefill: dict[str, str] = { "bg_filter": "", "bg_blend": "", "fg_filter": "", "fg_blend": "", "opacity": config.BLEND_OPACITY if config.BLEND_OPACITY in config.OPACITY_CHOICES else "30%", } if from_variant: manifest = pipeline.read_manifest(stem) or {} match = next((v for v in manifest.get("variants", []) if v.get("id") == from_variant), None) if match: opacity = match.get("blend_opacity") or prefill["opacity"] if opacity not in config.OPACITY_CHOICES: try: pct = int(str(opacity).rstrip("%")) nearest = min( config.OPACITY_CHOICES, key=lambda c: abs(int(c.rstrip("%")) - pct), ) opacity = nearest except ValueError: opacity = prefill["opacity"] prefill = { "bg_filter": match.get("background_filter") or "", "bg_blend": match.get("background_blend") or "", "fg_filter": match.get("foreground_filter") or "", "fg_blend": match.get("foreground_blend") or "", "opacity": opacity, } return templates.TemplateResponse( request, "remix.html", { "site_title": config.SITE_TITLE, "job_id": stem, "options": options, "opacity_choices": config.OPACITY_CHOICES, "prefill": prefill, "from_variant": from_variant, "error": error, }, ) @app.post("/jobs/{job_id}/remix") def remix_submit( job_id: str, bg_filter: str = Form(...), bg_blend: str = Form(...), fg_filter: str = Form(...), fg_blend: str = Form(...), opacity: str = Form(...), ) -> RedirectResponse: stem = _job_exists(job_id) choice = remix.RemixChoice( bg_filter=bg_filter, bg_blend=bg_blend, fg_filter=fg_filter, fg_blend=fg_blend, opacity=opacity, ) try: entry = remix.create_remix_variant(stem, choice) logger.info("[%s] remix created variant %s", stem, entry["id"]) except pipeline.PipelineError as exc: logger.warning("[%s] remix failed: %s", stem, exc) return RedirectResponse(url=f"/jobs/{stem}/remix?error={quote(str(exc))}", status_code=303) return RedirectResponse(url=f"/jobs/{stem}", status_code=303)