#!/usr/bin/env python3 """ startup.py — Web server with loading screen during initialization. Starts serving immediately with a funny loading page, then initializes the full app in a background thread and hot-swaps when ready. """ import asyncio import contextlib import os import re import threading import time import traceback import uvicorn import uvicorn.logging from fastapi import FastAPI from fastapi.responses import HTMLResponse, JSONResponse from fastapi.staticfiles import StaticFiles import startup_progress as sp LOADING_HTML = """ OpenAjaj — Startuje...
OpenAjajOpenAjaj
poháněný Necyklopedií — jedinou encyklopedií, která se odvážila říct pravdu
Probouzím mozkovou hmotu z necyklopedického spánku...
""" _CS_TOKEN = re.compile(r"(cs=)[^&\s\"]+") class RedactingAccessFormatter(uvicorn.logging.AccessFormatter): """Keep captcha session tokens out of the access log. The captcha redirects to /?cs= and the request line is logged verbatim, so every solve used to leave a replayable 7-day token in logs/access.log — which analytics/report.py then reads. """ def formatMessage(self, record) -> str: return _CS_TOKEN.sub(r"\1***", super().formatMessage(record)) # Phase 1: Start a minimal loading server immediately loading_app = FastAPI() loading_app.mount("/static", StaticFiles(directory="static"), name="static") real_app = None app_ready = threading.Event() @loading_app.get("/api/init-status") async def init_status(): """Return current initialization progress for the loading page.""" return JSONResponse(sp.snapshot()) @loading_app.get("/{path:path}") async def loading_page(path: str = ""): if app_ready.is_set(): return HTMLResponse( '' 'Přesměrovávám...' ) if f"/{path}" in _EARLY_PATHS: # web.py is still importing: a second or two, not the whole boot return HTMLResponse( '' 'Spouštím přepis…' ) return HTMLResponse(LOADING_HTML) # Transcription, the captcha in front of it and /stats need nothing from # web.boot(), so they are served by the real app as soon as web.py is imported. # The boot runs models and the index, which can be slow or fail; they must not # wait on it. _EARLY_PATHS = { "/t", "/transcribe", "/api/stt", "/api/stt/check", "/api/stt/usage", "/api/stt/token", "/captcha", "/api/captcha/challenge", "/api/captcha/verify", "/stats", "/api/visits", "/api/stt/quality", "/robots.txt", } def init_real_app(): """Import the full app (transcription goes live), then boot the rest.""" global real_app import web real_app = web.app print("=== Transcription ready ===", flush=True) try: web.boot() except Exception: # noqa: BLE001 - a failed boot must not take /t down with it traceback.print_exc() return app_ready.set() print("=== Real app ready, swap complete ===", flush=True) # An early path arriving while web.py imports waits for it rather than getting # a stand-in page; past this, the stand-in (a 1 s refresh) answers instead. _IMPORT_WAIT_S = 30 class SwappableApp: """The loading page until the boot is done; transcription from import on.""" async def __call__(self, scope, receive, send): early = scope.get("path") in _EARLY_PATHS for _ in range(_IMPORT_WAIT_S * 10): if not early or real_app is not None: break await asyncio.sleep(0.1) if real_app is not None and (app_ready.is_set() or early): await real_app(scope, receive, send) else: await loading_app(scope, receive, send) def _kill_existing_server(): """Kill any existing server on the same port.""" port = int(os.environ.get("PORT", "8000")) import subprocess # Best effort: no lsof (OSError), a hung lsof (SubprocessError), or a PID # that is gone or not ours (OSError) — a real conflict fails the bind later. with contextlib.suppress(OSError, subprocess.SubprocessError, ValueError): result = subprocess.run( ["lsof", "-ti", f":{port}"], capture_output=True, text=True, timeout=5, check=False) pids = result.stdout.strip().split() for pid in pids: if pid and pid != str(os.getpid()): os.kill(int(pid), 9) print(f"Killed existing server (PID {pid}) on port {port}", flush=True) time.sleep(0.5) def main(): _kill_existing_server() init_thread = threading.Thread(target=init_real_app, daemon=True) init_thread.start() port = int(os.environ.get("PORT", "8000")) # The captcha gate is only installed when SPACE_ID is set, so a local run # has no gate at all — don't offer it to the whole network. Set HOST to # override when running the container somewhere other than HF. host = os.environ.get("HOST") or ("0.0.0.0" if os.environ.get("SPACE_ID") else "127.0.0.1") uvicorn.run(SwappableApp(), host=host, port=port, access_log=True, log_config={ "version": 1, "disable_existing_loggers": False, "formatters": { "access": { "()": RedactingAccessFormatter, # the class, not a path: startup.py runs as __main__ "fmt": '%(asctime)s %(client_addr)s - "%(request_line)s" %(status_code)s', "datefmt": "%Y-%m-%dT%H:%M:%S", "use_colors": False, # colour codes break analytics/parser.py }, "default": { "fmt": "%(levelprefix)s %(message)s", "use_colors": False, }, # "format", not "fmt": the formatters above are # "()"-style callables that take fmt as a kwarg, but a # plain dictConfig formatter reads "format" and # silently falls back to bare %(message)s without it. "app": { "format": "%(asctime)s %(levelname)s %(message)s", "datefmt": "%Y-%m-%dT%H:%M:%S", }, }, "handlers": { "access_file": { "class": "logging.FileHandler", "filename": "logs/access.log", "formatter": "access", }, "default": { "class": "logging.StreamHandler", "formatter": "default", "stream": "ext://sys.stdout", }, # Rotating, not plain: a refusal line is written by # whatever is hammering the limit, so the volume is set # by the caller and not by us. "app_file": { "class": "logging.handlers.RotatingFileHandler", "filename": "logs/app.log", "maxBytes": 5_000_000, "backupCount": 3, "formatter": "app", }, }, "loggers": { "uvicorn.access": {"handlers": ["access_file"], "level": "INFO", "propagate": False}, "uvicorn.error": {"handlers": ["default"], "level": "INFO"}, "ajaj": {"handlers": ["app_file"], "level": "INFO", "propagate": False}, }, }) if __name__ == "__main__": main()