Blame for hf_launch.py

"""
hf_launch.py — starts any of the three apps on Hugging Face behind the admin.gamify.pt gate,
WITHOUT changing the app's own code.

    Video Processor   → APP_MODULE=app      (app.py builds a FastAPI "app")
    Thumbnail Editor  → APP_MODULE=server   (server.py builds a FastAPI "app")
    Upload Center     → APP_MODULE=app      (app.py builds Gradio Blocks and calls .launch())

How:
  • If the module calls Blocks.launch(), that call is caught (Gradio's own server is NOT started)
    and the Blocks are served here instead, with the same theme / css the app passed to launch().
  • If the module already builds a FastAPI/Starlette app, that app is used as it is.
  • The GamifyGate (gamify_gate.py) is put in front — unless the app already added it itself.
  • GAMIFY_GATE=off opens the app to everyone; if gamify_gate.py can't be loaded, everything is
    blocked (the app is never opened by accident).

Dockerfile:   CMD ["python", "hf_launch.py"]
"""
import importlib
import inspect
import logging
import os

import gradio as gr

logging.basicConfig(level=logging.INFO)
PORT = int(os.environ.get("PORT", 7860))
MODULE = os.environ.get("APP_MODULE", "app")


def _params(fn):
    try:
        return inspect.signature(fn).parameters
    except (TypeError, ValueError):
        return {}


# ---- 0) clean pasted Secrets: stray spaces / line breaks / surrounding quotes break API keys ----
def _clean_env():
    fixed = []
    for name, value in list(os.environ.items()):
        clean = value.strip()
        if len(clean) >= 2 and clean[0] == clean[-1] and clean[0] in "\"'":
            clean = clean[1:-1].strip()
        if clean != value:
            os.environ[name] = clean
            fixed.append(name)
    if fixed:
        logging.warning(f"hf_launch: removed extra spaces/quotes from: {', '.join(sorted(fixed))}")


_clean_env()

# ---- 1) load the app; catch Blocks.launch() so it doesn't start its own (ungated) server ----
_captured = {}
_original_launch = gr.Blocks.launch


def _capture_launch(self, *args, **kwargs):
    _captured["blocks"], _captured["kwargs"] = self, kwargs
    logging.info("hf_launch: the app called launch() — it will be served behind the gate instead")
    return None


def _import_app_module(name):
    """Imports the app file: <name>.py next to this script — or, if it was uploaded inside a
    folder (e.g. upload_center/app.py), finds it there. Explains clearly when it's missing."""
    import sys
    from pathlib import Path
    here = Path(__file__).resolve().parent
    name = name.replace("\\", "/").removesuffix(".py")
    if "/" in name:                                   # APP_MODULE=upload_center/app
        folder, name = name.rsplit("/", 1)
        sys.path.insert(0, str(here / folder))
    sys.path.insert(0, str(here))
    if not (here / f"{name}.py").exists():
        found = sorted(p for p in here.rglob(f"{name}.py")
                       if ".git" not in p.parts and p.resolve() != Path(__file__).resolve())
        if found:
            logging.warning(f"hf_launch: {name}.py is not next to hf_launch.py — using {found[0].relative_to(here)}")
            sys.path.insert(0, str(found[0].parent))
        else:
            files = sorted(str(p.relative_to(here)) for p in here.rglob("*.py") if ".git" not in p.parts)
            raise SystemExit(
                f"hf_launch: could not find {name}.py in this Space.\n"
                f"  Python files found: {', '.join(files) or '(none)'}\n"
                f"  → Put the app file at the top level of the Space (next to hf_launch.py),\n"
                f"    or set the variable APP_MODULE to its name (e.g. APP_MODULE=server).")
    return importlib.import_module(name)


gr.Blocks.launch = _capture_launch
try:
    module = _import_app_module(MODULE)
finally:
    gr.Blocks.launch = _original_launch


def _is_asgi_app(obj):
    """A FastAPI / Starlette application (has add_middleware and is callable)."""
    return obj is not None and not isinstance(obj, gr.Blocks) and callable(obj) and hasattr(obj, "add_middleware")


# ---- 2) the web app to serve ----
asgi = next((getattr(module, n) for n in ("app", "server", "application") if _is_asgi_app(getattr(module, n, None))),
            None)
if asgi is None:
    from fastapi import FastAPI
    blocks = _captured.get("blocks") or next(
        (getattr(module, n) for n in ("app", "demo", "interface") if isinstance(getattr(module, n, None), gr.Blocks)),
        None)
    if blocks is None:
        raise RuntimeError(f"hf_launch: no Gradio Blocks or FastAPI app found in {MODULE}.py")
    launch_kwargs = _captured.get("kwargs", {})
    mount_params = _params(gr.mount_gradio_app)
    extra = {k: v for k, v in launch_kwargs.items() if k in mount_params and k not in ("app", "blocks", "path")}
    if "css" in launch_kwargs and "css" not in mount_params and hasattr(blocks, "css"):
        blocks.css = launch_kwargs["css"]          # older Gradio: css lives on the Blocks
    asgi = gr.mount_gradio_app(FastAPI(), blocks, path="/", **extra)


# ---- 3) the access gate ----
def _has_gate(app_):
    for m in getattr(app_, "user_middleware", []) or []:
        cls = getattr(m, "cls", None)
        if getattr(cls, "__name__", "") == "GamifyGate":
            return True
    return False


class _BlockEverything:
    """Gate wanted but gamify_gate.py missing: nobody gets in."""
    def __init__(self, app_):
        self.app = app_

    async def __call__(self, scope, receive, send):
        if scope.get("type") != "http" or scope.get("path") == "/healthz":
            return await self.app(scope, receive, send)
        await send({"type": "http.response.start", "status": 503,
                    "headers": [(b"content-type", b"text/plain; charset=utf-8")]})
        await send({"type": "http.response.body",
                    "body": b"Access blocked: gamify_gate.py is missing from this Space."})


if os.environ.get("GAMIFY_GATE", "on").strip().lower() == "off":
    logging.warning("hf_launch: GAMIFY_GATE=off — the app is OPEN to everyone")
elif _has_gate(asgi):
    logging.info("hf_launch: the app already has the access gate")
else:
    try:
        from gamify_gate import GamifyGate
        asgi.add_middleware(GamifyGate)
        logging.info("hf_launch: access gate ON — only through admin.gamify.pt")
    except Exception as e:
        asgi.add_middleware(_BlockEverything)
        logging.error(f"hf_launch: gamify_gate could not be loaded ({e}) — blocking all access")

app = asgi

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=PORT)