"""Minimal access-gated static server for the Sync Pilot showcase (HF Docker Space). Serves the pre-built Vite ``dist/`` behind an ``ACCESS_KEY`` login page, mirroring the Streamlit dashboard's gate: * ACCESS_KEY unset → served directly (rely on the Space being private). * ACCESS_KEY set → a login page is shown until the correct key is entered; a signed cookie then keeps the session for 30 days. All routes (incl. assets + catalog.json) are gated, so nothing leaks pre-auth. """ from __future__ import annotations import hashlib import io import json import logging import os import threading import time from datetime import datetime, timezone import requests from flask import Flask, Response, g, jsonify, make_response, redirect, request, send_from_directory # Timestamped usage logging lives here (not the model-heavy backend) so it can be # shipped without a slow backend restart. The proxy sees every /api/live/* call, # so it's the full feature-usage trail. UTC ISO timestamps. logging.Formatter.converter = time.gmtime logging.basicConfig(level=logging.INFO, format="%(asctime)sZ | %(levelname)s | %(message)s", datefmt="%Y-%m-%dT%H:%M:%S") log = logging.getLogger("showcase") ACCESS_KEY = os.getenv("ACCESS_KEY", "") PORT = int(os.getenv("PORT", "7860")) DIST = os.path.join(os.path.dirname(os.path.abspath(__file__)), "dist") # Live-tagging proxy: the showcase server forwards /api/live/* to the *private* # backend Space, holding both the HF token (passes the platform's private-Space # gate) and the backend's ACCESS_KEY (passes the app's compute gate). The browser # thus needs no keys and makes only same-origin calls. Backend stays private, so # its source + model artifacts are never exposed. LIVE_BACKEND_URL = os.getenv("LIVE_BACKEND_URL", "").rstrip("/") LIVE_HF_TOKEN = os.getenv("HF_TOKEN", "") LIVE_ACCESS_KEY = os.getenv("LIVE_ACCESS_KEY", "") # the backend's ACCESS_KEY app = Flask(__name__) # Owner-only admin key for the usage map (set ADMIN_KEY as a Space secret to keep # it private; if unset it falls back to the normal login gate). ADMIN_KEY = os.getenv("ADMIN_KEY", "") # Visitor usage, accumulated from requests (the proxy sees every /api/live/* + # page load). Geo is looked up lazily + cached. Persisted to a HF *dataset* so it # survives Space restarts/rebuilds (the container filesystem and memory both reset # on restart, so a plain file wouldn't last). Loaded on startup, flushed on a # timer. Reuses HF_TOKEN (the same secret the proxy already holds). _USAGE: dict[str, dict] = {} _USAGE_LOCK = threading.Lock() _USAGE_DIRTY = {"v": False} # set on each new hit; the flush thread clears it USAGE_DATASET = os.getenv("USAGE_DATASET", "emresar/sync-pilot-usage") USAGE_FILE = "usage_store.json" USAGE_FLUSH_SEC = int(os.getenv("USAGE_FLUSH_SEC", "60")) def _now() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def _load_usage() -> None: """Load the persisted usage store from the HF dataset. Best-effort: a missing file, missing token, or read-only token just starts the map empty.""" if not LIVE_HF_TOKEN: log.info("usage persistence off: no HF_TOKEN") return try: from huggingface_hub import hf_hub_download path = hf_hub_download(USAGE_DATASET, USAGE_FILE, repo_type="dataset", token=LIVE_HF_TOKEN) with open(path, encoding="utf-8") as fh: data = json.load(fh) ips = data.get("ips", {}) with _USAGE_LOCK: for ip, rec in ips.items(): _USAGE[ip] = rec log.info("usage store loaded from %s: %d IPs", USAGE_DATASET, len(ips)) except Exception as exc: # noqa: BLE001 log.info("usage store not loaded (%s) — starting empty", str(exc)[:120]) def _save_usage() -> None: """Flush the in-memory store to the HF dataset. Non-fatal on failure.""" if not LIVE_HF_TOKEN: return with _USAGE_LOCK: payload = json.dumps({"ips": _USAGE, "updated": _now()}, ensure_ascii=False).encode("utf-8") try: from huggingface_hub import HfApi HfApi(token=LIVE_HF_TOKEN).upload_file( path_or_fileobj=io.BytesIO(payload), path_in_repo=USAGE_FILE, repo_id=USAGE_DATASET, repo_type="dataset", commit_message="usage update") except Exception as exc: # noqa: BLE001 log.warning("usage store save failed: %s", str(exc)[:120]) def _usage_flush_loop() -> None: while True: time.sleep(USAGE_FLUSH_SEC) if _USAGE_DIRTY["v"]: _USAGE_DIRTY["v"] = False # clear before save; a hit mid-save re-dirties _save_usage() _USAGE_STARTED = {"v": False} def _start_usage_persistence() -> None: if _USAGE_STARTED["v"]: return _USAGE_STARTED["v"] = True _load_usage() threading.Thread(target=_usage_flush_loop, daemon=True).start() def _client_ip() -> str: return request.headers.get("X-Forwarded-For", request.remote_addr or "?").split(",")[0].strip() def _geo(ip: str) -> dict: """IP → {lat, lon, city, country} via ip-api.com (free, no key, HTTP/port 80 which HF egress allows). Returns {} on failure/private IP.""" try: r = requests.get(f"http://ip-api.com/json/{ip}", params={"fields": "status,lat,lon,city,country"}, timeout=4) d = r.json() if d.get("status") == "success": return {"lat": d["lat"], "lon": d["lon"], "city": d.get("city"), "country": d.get("country")} except Exception: # noqa: BLE001 pass return {} def _admin_token() -> str: return hashlib.sha256(f"sync_pilot_admin_{ADMIN_KEY}".encode()).hexdigest()[:24] def _admin_ok() -> bool: """Owner-only gate for /admin/*. Prefers a signed cookie set via /admin/login (no key in the URL); still accepts ?key= as a fallback for scripts/curl. With ADMIN_KEY unset it degrades to the normal showcase login cookie.""" if ADMIN_KEY: return (request.cookies.get("sp_admin") == _admin_token() or request.args.get("key") == ADMIN_KEY) return _authed() @app.before_request def _usage_start() -> None: g._t0 = time.perf_counter() @app.after_request def _usage_log(resp: Response) -> Response: """One timestamped line per meaningful request + record the visitor IP for the usage map. Static-asset chatter (.js/.css/.woff) and admin polls are skipped.""" try: p = request.path leaf = p.rsplit("/", 1)[-1] if (p.startswith("/api/live/") or p == "/" or "." not in leaf) and not p.startswith("/admin"): dt = time.perf_counter() - getattr(g, "_t0", time.perf_counter()) ip = _client_ip() log.info("%s %s -> %s %.2fs ip=%s", request.method, p, resp.status_code, dt, ip) if ip and ip != "?" and not ip.startswith(("10.", "192.168.", "127.")): with _USAGE_LOCK: rec = _USAGE.setdefault(ip, {"ip": ip, "count": 0, "first": _now(), "geo": None}) rec["count"] += 1 rec["last"] = _now() _USAGE_DIRTY["v"] = True except Exception: # noqa: BLE001 - logging must never break a response pass return resp def _live_headers() -> dict[str, str]: h = {"X-Access-Key": LIVE_ACCESS_KEY} if LIVE_HF_TOKEN: h["Authorization"] = f"Bearer {LIVE_HF_TOKEN}" return h def _proxy(method: str, path: str, **kw) -> Response: if not _authed(): return jsonify({"detail": "unauthorized"}), 401 # type: ignore[return-value] if not LIVE_BACKEND_URL: return jsonify({"detail": "live backend not configured"}), 503 # type: ignore[return-value] try: r = requests.request(method, f"{LIVE_BACKEND_URL}{path}", headers=_live_headers(), timeout=35, **kw) except requests.RequestException as exc: return jsonify({"detail": f"backend unreachable: {exc}"}), 502 # type: ignore[return-value] return Response(r.content, status=r.status_code, content_type=r.headers.get("content-type", "application/json")) @app.post("/api/live/tag") def live_tag() -> Response: return _proxy("POST", "/tag", json=request.get_json(force=True, silent=True) or {}) @app.get("/api/live/status/") def live_status(job_id: str) -> Response: return _proxy("GET", f"/status/{job_id}") @app.post("/api/live/sync") def live_sync() -> Response: return _proxy("POST", "/sync", json=request.get_json(force=True, silent=True) or {}) @app.post("/api/live/tag-file") def live_tag_file() -> Response: """Proxy a multipart audio upload to the private backend's /tag-file.""" if not _authed(): return jsonify({"detail": "unauthorized"}), 401 # type: ignore[return-value] if not LIVE_BACKEND_URL: return jsonify({"detail": "live backend not configured"}), 503 # type: ignore[return-value] f = request.files.get("file") if f is None: return jsonify({"detail": "no file uploaded"}), 422 # type: ignore[return-value] try: r = requests.post(f"{LIVE_BACKEND_URL}/tag-file", headers=_live_headers(), files={"file": (f.filename, f.stream, f.mimetype)}, timeout=120) except requests.RequestException as exc: return jsonify({"detail": f"backend unreachable: {exc}"}), 502 # type: ignore[return-value] return Response(r.content, status=r.status_code, content_type=r.headers.get("content-type", "application/json")) def _token() -> str: return hashlib.sha256(f"sync_pilot_showcase_{ACCESS_KEY}".encode()).hexdigest()[:24] def _authed() -> bool: return not ACCESS_KEY or request.cookies.get("sp_auth") == _token() _LOGIN = """ Sync Pilot — access

Sync Pilot

Catalog Intelligence · access key required

{err}
""" @app.post("/login") def login(): if ACCESS_KEY and request.form.get("key") == ACCESS_KEY: resp = make_response(redirect("/")) # SameSite=None; Secure so the cookie survives the cross-origin iframe # that huggingface.co embeds the *.hf.space app in (Lax is dropped there). resp.set_cookie("sp_auth", _token(), max_age=60 * 60 * 24 * 30, httponly=True, samesite="None", secure=True) return resp return redirect("/?error=1") _MAP_HTML = """ Sync Pilot — kullanım haritası
Kullanım haritası · 0 IP · 0 istek · —
""" @app.route("/admin/usage") def admin_usage(): if not _admin_ok(): return jsonify({"detail": "unauthorized"}), 401 # type: ignore[return-value] with _USAGE_LOCK: items = [dict(r) for r in _USAGE.values()] for r in items: # lazily geolocate new IPs, cached back into _USAGE if r.get("geo") is None: geo = _geo(r["ip"]) with _USAGE_LOCK: if r["ip"] in _USAGE: _USAGE[r["ip"]]["geo"] = geo r["geo"] = geo _USAGE_DIRTY["v"] = True # persist the resolved geo on next flush return jsonify({"updated": _now(), "total_ips": len(items), "total_hits": sum(r["count"] for r in items), "points": [r for r in items if r.get("geo")]}) _ADMIN_LOGIN = _LOGIN.replace('action="/login"', 'action="/admin/login"').replace( "Catalog Intelligence · access key required", "Usage map · admin key required") @app.get("/admin/login") def admin_login_form(): if not ADMIN_KEY: return Response("ADMIN_KEY not set — the admin map uses the normal login gate.", mimetype="text/plain") return _ADMIN_LOGIN.format(err="Invalid key" if request.args.get("error") else ""), 200 @app.post("/admin/login") def admin_login(): if ADMIN_KEY and request.form.get("key") == ADMIN_KEY: resp = make_response(redirect("/admin/map")) resp.set_cookie("sp_admin", _admin_token(), max_age=60 * 60 * 24 * 30, httponly=True, samesite="None", secure=True) return resp return redirect("/admin/login?error=1") @app.route("/admin/map") def admin_map(): if not _admin_ok(): return redirect("/admin/login") # one-time key entry → cookie, no key in URL return Response(_MAP_HTML, mimetype="text/html") @app.route("/", defaults={"path": ""}) @app.route("/") def serve(path: str): if not _authed(): err = "Invalid key" if request.args.get("error") else "" return _LOGIN.format(err=err), 200 full = os.path.join(DIST, path) if path and os.path.isfile(full): return send_from_directory(DIST, path) return send_from_directory(DIST, "index.html") # SPA fallback # Start dataset-backed usage persistence at import (runs under gunicorn's worker # and under `python app.py`). Single worker (Dockerfile -w 1) → one consistent store. _start_usage_persistence() if __name__ == "__main__": app.run(host="0.0.0.0", port=PORT)