api_bridge.py

import base64
from io import BytesIO
from pathlib import Path
from typing import Any, Callable

from PIL import Image


DecodeRunner = Callable[[Image.Image], str]
SectionParser = Callable[[str], dict[str, Any]]
DecodeBuilder = Callable[..., dict[str, Any]]


class DecodePayloadError(ValueError):
    pass


MAX_DECODE_IMAGE_EDGE = 640


def _strip_data_url(value: str) -> str:
    if "," in value and value.lstrip().startswith("data:"):
        return value.split(",", 1)[1]
    return value


def _image_from_payload(payload: dict[str, Any]) -> tuple[Image.Image, str, str]:
    image_b64 = payload.get("image")
    if not isinstance(image_b64, str) or not image_b64.strip():
        raise DecodePayloadError("Missing base64 image.")

    normalized_b64 = _strip_data_url(image_b64.strip())
    mime = payload.get("mime")
    if not isinstance(mime, str) or not mime.startswith("image/"):
        mime = "image/jpeg"

    try:
        raw = base64.b64decode(normalized_b64, validate=True)
        image = Image.open(BytesIO(raw)).convert("RGB")
        image.thumbnail((MAX_DECODE_IMAGE_EDGE, MAX_DECODE_IMAGE_EDGE), Image.Resampling.LANCZOS)
    except Exception as exc:
        raise DecodePayloadError("Invalid image payload.") from exc

    return image, normalized_b64, mime


def decode_payload_to_culture_decode(
    payload: dict[str, Any],
    *,
    run_decode: DecodeRunner,
    parse_sections: SectionParser,
    sections_to_decode: DecodeBuilder,
) -> dict[str, Any]:
    image, image_b64, mime = _image_from_payload(payload)
    raw = run_decode(image)
    sections = parse_sections(raw)
    decode = dict(sections_to_decode(sections, uploaded_b64=image_b64))

    decode["image"] = f"{mime};base64,{image_b64}" if mime.startswith("data:") else f"data:{mime};base64,{image_b64}"
    decode.pop("_uploaded_b64", None)
    return decode


def create_hackathon_app(
    gradio_demo: Any,
    decode_payload: Callable[[dict[str, Any]], dict[str, Any]],
    narrate_payload: Callable[[dict[str, Any]], dict[str, Any]] | None = None,
    *,
    static_dir: str = "dist",
) -> Any:
    from fastapi import FastAPI, HTTPException
    from fastapi.middleware.cors import CORSMiddleware
    from fastapi.responses import FileResponse
    from fastapi.staticfiles import StaticFiles
    import gradio as gr

    app = FastAPI(title="Local in 30s Hackathon API")
    app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],
        allow_credentials=False,
        allow_methods=["*"],
        allow_headers=["*"],
    )

    @app.get("/api/health")
    async def health() -> dict[str, str]:
        return {"status": "ok"}

    @app.post("/api/decode")
    async def decode_api(payload: dict[str, Any]) -> dict[str, Any]:
        try:
            return decode_payload(payload)
        except DecodePayloadError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        except Exception as exc:
            raise HTTPException(status_code=502, detail=f"AI decode failed: {exc}") from exc

    @app.post("/api/narrate")
    async def narrate_api(payload: dict[str, Any]) -> dict[str, Any]:
        if narrate_payload is None:
            return {"audio": None, "usedFallback": True}

        try:
            return narrate_payload(payload)
        except ValueError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        except Exception as exc:
            raise HTTPException(status_code=502, detail=f"AI narration failed: {exc}") from exc

    dist = Path(static_dir)
    if dist.exists():
        assets = dist / "assets"
        if assets.exists():
            app.mount("/ui/assets", StaticFiles(directory=str(assets)), name="ui-assets")

        @app.get("/")
        async def react_home() -> FileResponse:
            standalone = dist / "standalone-preview.html"
            if standalone.exists():
                return FileResponse(str(standalone))
            return FileResponse(str(dist / "index.html"))

        @app.get("/ui")
        @app.get("/ui/")
        @app.get("/ui/{path:path}")
        async def react_ui(path: str = "") -> FileResponse:
            target = dist / path
            if path and target.exists() and target.is_file():
                return FileResponse(str(target))
            return FileResponse(str(dist / "index.html"))

    return gr.mount_gradio_app(app, gradio_demo, path="/gradio")