Spaces:
Running
Running
| """ | |
| 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) |