""" ls_proxy_hf.py โ€” Reverse proxy for Label Studio on HuggingFace Spaces. Runs on port 7860 (HF exposed port), forwards to Label Studio on port 8080. Features kept from local version: - Custom /projects landing page with flag emojis - Annotation privacy: non-admins only see their own annotations - Auto-advance to next task after Submit - Per-user progress bar at top - goNextTask via label stream (not admin-only DM actions) """ import json import re as _re import sqlite3 import os import smtplib import threading import requests from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from flask import Flask, request, Response, make_response app = Flask(__name__, static_folder=None) LS_URL = "http://localhost:8080" def _find_db_path(): """Find the LS SQLite DB with the most users (handles different LS data dir configs).""" import glob as _g, sqlite3 as _s3 candidates = sorted(_g.glob("/data/**/*.sqlite3", recursive=True) + _g.glob("/label-studio/**/*.sqlite3", recursive=True)) best, best_n = "/data/ls/label_studio.sqlite3", -1 for p in candidates: try: c = _s3.connect(p) n = c.execute("SELECT COUNT(*) FROM htx_user").fetchone()[0] c.close() if n > best_n: best, best_n = p, n except Exception: pass return best DB_PATH = _find_db_path() HF_SPACE_URL = "https://trustllmeu-saga-annotation.hf.space" def _backup_db(): """Trigger an immediate DB backup after an annotation is submitted.""" try: import subprocess subprocess.run( ["python3", "/app/db_sync.py", "backup"], timeout=60, capture_output=True, ) except Exception as e: print(f"[db_sync] Immediate backup failed: {e}", flush=True) # Flag emoji prefix โ†’ (ISO 2-letter code for flagcdn.com, language label) # Keyed by the flag emoji that LS project titles begin with. # This is resilient to LS re-assigning project IDs on DB resets. _EMOJI_FLAGS = { "\U0001F1EE\U0001F1F8": ("is", "รslenska"), # ๐Ÿ‡ฎ๐Ÿ‡ธ IS "\U0001F1E9\U0001F1F0": ("dk", "Dansk"), # ๐Ÿ‡ฉ๐Ÿ‡ฐ DA "\U0001F1F3\U0001F1F4": ("no", "Norsk"), # ๐Ÿ‡ณ๐Ÿ‡ด NB "\U0001F1F8\U0001F1EA": ("se", "Svenska"), # ๐Ÿ‡ธ๐Ÿ‡ช SV "\U0001F1E9\U0001F1EA": ("de", "Deutsch"), # ๐Ÿ‡ฉ๐Ÿ‡ช DE "\U0001F1EB\U0001F1EE": ("fi", "Suomi"), # ๐Ÿ‡ซ๐Ÿ‡ฎ FI } def _flag_for_title(title): """Return (country_code, lang_label) from the leading flag emoji in a project title.""" for emoji, info in _EMOJI_FLAGS.items(): if title.startswith(emoji): return info return ("", "") def _load_manifest_ids(): """Return set of allowed project ID strings from projects/manifest.json.""" # Try reading from file for candidate in [ os.path.join(os.path.dirname(os.path.abspath(__file__)), "projects", "manifest.json"), "/app/projects/manifest.json", ]: try: if os.path.exists(candidate): ids = {str(e["id"]) for e in json.load(open(candidate))} print(f"[manifest] Loaded {len(ids)} IDs from {candidate}: {sorted(ids)}", flush=True) return ids except Exception as e: print(f"[manifest] Failed to load {candidate}: {e}", flush=True) # Hardcoded fallback: the 8 canonical projects always have DB IDs 1-8 on HF Space fallback = {"1", "2", "3", "4", "5", "6", "7", "8"} print(f"[manifest] Using hardcoded fallback: {sorted(fallback)}", flush=True) return fallback _MANIFEST_IDS = _load_manifest_ids() def _load_manifest_tasks(): """Return {project_id_str: task_count} from manifest.json.""" for candidate in [ os.path.join(os.path.dirname(os.path.abspath(__file__)), "projects", "manifest.json"), "/app/projects/manifest.json", ]: try: if os.path.exists(candidate): return {str(e["id"]): int(e.get("tasks", 0)) for e in json.load(open(candidate))} except Exception: pass return {} _MANIFEST_TASKS = _load_manifest_tasks() # --------------------------------------------------------------------------- # Annotation-submission email notifications # --------------------------------------------------------------------------- _NOTIFY_FROM = "hoda.fakharzade@gmail.com" _NOTIFY_TO = "hodfa71@liu.se" # App password: HF Space secret GMAIL_APP_PASS, fallback to file in /data def _gmail_app_pass(): v = os.environ.get("GMAIL_APP_PASS", "").strip() if v: return v try: return open("/data/ls/gmail_app_pass").read().strip() except Exception: return "" def _send_completion_email(project_title: str, annotator_email: str, done: int, total: int): """Send email only when a rater finishes ALL tasks in a project.""" app_pass = _gmail_app_pass() if not app_pass: print("[webhook] No Gmail app password โ€” skipping completion email", flush=True) return subject = f"[SAGA] โœ“ {annotator_email} completed โ€” {project_title}" body = ( f"Annotator : {annotator_email}\n" f"Project : {project_title}\n" f"Status : COMPLETED ({done}/{total} tasks)\n\n" f"View at: {HF_SPACE_URL}\n" ) msg = MIMEMultipart() msg["Subject"] = subject msg["From"] = _NOTIFY_FROM msg["To"] = _NOTIFY_TO msg.attach(MIMEText(body, "plain")) try: with smtplib.SMTP("smtp.gmail.com", 587, timeout=15) as s: s.ehlo(); s.starttls(); s.ehlo() s.login(_NOTIFY_FROM, app_pass) s.sendmail(_NOTIFY_FROM, [_NOTIFY_TO], msg.as_string()) print(f"[webhook] Completion email โ†’ {_NOTIFY_TO} ({annotator_email} finished {project_title})", flush=True) except Exception as exc: print(f"[webhook] Email send failed: {exc}", flush=True) FAVICON_TAG = '' PROLIFIC_CC = { "is": os.environ.get("PROLIFIC_CC_IS", "SAGAIS77"), "da": os.environ.get("PROLIFIC_CC_DA", "SAGADA42"), "nb": os.environ.get("PROLIFIC_CC_NB", "SAGANB15"), "sv": os.environ.get("PROLIFIC_CC_SV", "SAGASV33"), } PROLIFIC_LANG_PROJECTS = { "is": [("2", "Pairwise Comparison")], "da": [("4", "Pairwise Comparison")], "nb": [("5", "Likert Rating"), ("6", "Pairwise Comparison")], "sv": [("8", "Pairwise Comparison")], } # Per-project metadata for single-project Prolific studies (?project=N) PROLIFIC_PROJECT_META = { "1": {"flag": "๐Ÿ‡ฎ๐Ÿ‡ธ", "lang": "Icelandic / รslenska", "task": "Likert Rating", "cc": os.environ.get("PROLIFIC_CC_P1", "ISLIK01"), "est_min": 90}, "2": {"flag": "๐Ÿ‡ฎ๐Ÿ‡ธ", "lang": "Icelandic / รslenska", "task": "Pairwise Comparison", "cc": os.environ.get("PROLIFIC_CC_P2", "ISPAIR02"), "est_min": 45}, "3": {"flag": "๐Ÿ‡ฉ๐Ÿ‡ฐ", "lang": "Danish / Dansk", "task": "Likert Rating", "cc": os.environ.get("PROLIFIC_CC_P3", "DALIK03"), "est_min": 60}, "4": {"flag": "๐Ÿ‡ฉ๐Ÿ‡ฐ", "lang": "Danish / Dansk", "task": "Pairwise Comparison", "cc": os.environ.get("PROLIFIC_CC_P4", "DAPAIR04"), "est_min": 30}, "5": {"flag": "๐Ÿ‡ณ๐Ÿ‡ด", "lang": "Norwegian / Norsk", "task": "Likert Rating", "cc": os.environ.get("PROLIFIC_CC_P5", "NBLIK05"), "est_min": 33}, "6": {"flag": "๐Ÿ‡ณ๐Ÿ‡ด", "lang": "Norwegian / Norsk", "task": "Pairwise Comparison", "cc": os.environ.get("PROLIFIC_CC_P6", "NBPAIR06"), "est_min": 25}, "7": {"flag": "๐Ÿ‡ธ๐Ÿ‡ช", "lang": "Swedish / Svenska", "task": "Likert Rating", "cc": os.environ.get("PROLIFIC_CC_P7", "SVLIK07"), "est_min": 60}, "8": {"flag": "๐Ÿ‡ธ๐Ÿ‡ช", "lang": "Swedish / Svenska", "task": "Pairwise Comparison", "cc": os.environ.get("PROLIFIC_CC_P8", "SVPAIR08"), "est_min": 45}, } PRIVACY_SCRIPT = """ """ # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _fwd_headers(req, extra_exclude=()): # Strip cookie (passed separately via cookies= param) and proxy headers # so LS sees a plain HTTP request and uses simple token-match CSRF rules. skip = {"host", "content-length", "transfer-encoding", "cookie", "x-forwarded-proto", "x-forwarded-for", "x-real-ip", "x-forwarded-host"} | set(extra_exclude) h = {k: v for k, v in req.headers if k.lower() not in skip} h["Host"] = "localhost:8080" if "Referer" in h: h["Referer"] = h["Referer"].replace(req.host_url.rstrip("/"), LS_URL) # Rewrite Origin to LS URL so Django CSRF doesn't reject cross-domain requests # (saga-research.com is not in CSRF_TRUSTED_ORIGINS; rewriting to localhost passes validation) if "Origin" in h: h["Origin"] = LS_URL # X-LS-Token: caller passes LS API token; we swap it in as Authorization for LS. # Also strip any HF Bearer token (it's only for HF proxy auth, LS doesn't understand it). ls_token = req.headers.get("X-LS-Token") or req.args.get("ls_token") if ls_token: h["Authorization"] = f"Token {ls_token}" elif h.get("Authorization", "").startswith("Bearer "): del h["Authorization"] # HF-only header; LS uses sessionid cookie instead return h def _out_headers(resp, extra_exclude=()): skip = {"transfer-encoding"} | {s.lower() for s in extra_exclude} try: items = list(resp.raw.headers.items()) except AttributeError: items = list(resp.headers.items()) return [(k, v) for k, v in items if k.lower() not in skip] _user_cache = {} _USER_CACHE_TTL = 1800 # 30 min โ€” reduces /api/current-user/whoami spam in LS logs def get_current_user(req): import time as _time session_key = req.cookies.get("sessionid") or req.cookies.get("csrftoken", "") # No session cookie at all โ†’ definitely not logged in; skip the LS call entirely # (avoids the 401 /api/current-user/whoami spam in LS logs from unauthenticated browsers) if not req.cookies.get("sessionid"): return None, False now = _time.time() if session_key and session_key in _user_cache: uid, is_admin, exp = _user_cache[session_key] if now < exp: return uid, is_admin try: r = requests.get( f"{LS_URL}/api/current-user/whoami", headers=_fwd_headers(req), cookies=_ls_cookies(req), timeout=5, allow_redirects=False, ) if r.status_code == 200: uid = r.json().get("id") if uid: conn = sqlite3.connect(DB_PATH) row = conn.execute( "SELECT is_staff, is_superuser FROM htx_user WHERE id=?", (uid,) ).fetchone() conn.close() is_admin = bool(row and (row[0] or row[1])) if session_key: _user_cache[session_key] = (uid, is_admin, now + _USER_CACHE_TTL) return uid, is_admin except Exception: pass return None, False _STARTING_PAGE = """ SAGA โ€” Startingโ€ฆ

โณ Label Studio is startingโ€ฆ

This page refreshes automatically every 5 seconds.

""" def _ls_unavailable(): return Response(_STARTING_PAGE, status=503, content_type="text/html; charset=utf-8", headers={"Retry-After": "5"}) def _ls_cookies(req): """Only pass LS-relevant cookies to the backend; strip HF-only cookies like 'token'.""" return {k: v for k, v in req.cookies.items() if k not in ("token",)} def _rewrite_hf_urls(html): """Replace absolute HF Space URLs with relative paths. LS is configured with LABEL_STUDIO_HOST=hf.space URL and injects it into APP_SETTINGS.hostname and all static asset hrefs. When accessed via saga-research.com the browser would load scripts/CSS from hf.space and the React app would POST API calls back to hf.space โ€” bypassing our proxy and sending no cookies (different domain) โ†’ 401. Stripping the host makes every URL relative, so the browser uses whichever domain it's on. """ return html.replace(HF_SPACE_URL, "") def _proxy_stream(path): url = f"{LS_URL}{path}" if request.query_string: url += "?" + request.query_string.decode() try: resp = requests.request( method=request.method, url=url, headers=_fwd_headers(request), data=request.get_data(), cookies=_ls_cookies(request), allow_redirects=False, stream=True, timeout=30, ) except requests.exceptions.ConnectionError: return _ls_unavailable() except requests.exceptions.Timeout: return Response("Request timed out โ€” LS may be overloaded.", status=504, content_type="text/plain") return Response( resp.iter_content(chunk_size=4096), status=resp.status_code, headers=_out_headers(resp), content_type=resp.headers.get("Content-Type", ""), ) def _proxy_buffered(path): url = f"{LS_URL}{path}" if request.query_string: url += "?" + request.query_string.decode() return requests.request( method=request.method, url=url, headers=_fwd_headers(request), data=request.get_data(), cookies=_ls_cookies(request), allow_redirects=False, timeout=30, ) def _respond_buffered(resp, extra_exclude=()): skip = {"content-encoding", "content-length"} | {s.lower() for s in extra_exclude} return Response( resp.content, status=resp.status_code, headers=_out_headers(resp, extra_exclude=skip), content_type=resp.headers.get("Content-Type", ""), ) # --------------------------------------------------------------------------- # Custom /projects landing page # --------------------------------------------------------------------------- def get_projects_from_db(session_cookies=None): """Return active projects via LS API using session cookies (in-memory state). Falls back to DB token, then SQLite direct read. """ # Method 1: Use the current session cookies (most reliable โ€” uses LS in-memory state) if session_cookies: try: r = requests.get( f"{LS_URL}/api/projects/?page_size=100", cookies={k: v for k, v in session_cookies.items()}, timeout=10, ) if r.status_code == 200: data = r.json() results = [(p["id"], p["title"]) for p in data.get("results", [])] if results: return results except Exception as e: print(f"[projects] Session-cookie call failed: {e}", flush=True) # Method 2: SQLite direct read (may include stale rows โ€” filtered by _MANIFEST_IDS) try: conn = sqlite3.connect(DB_PATH) rows = conn.execute("SELECT id, title FROM project ORDER BY id").fetchall() conn.close() return rows except Exception: return [] def build_projects_page(user_email=""): projects = sorted(get_projects_from_db(session_cookies=request.cookies), key=lambda x: x[0]) cards = "" for pid, title in projects: if _MANIFEST_IDS and str(pid) not in _MANIFEST_IDS: continue code, lang = _flag_for_title(title) clean_title = _re.sub(r'^[\U0001F1E0-\U0001F1FF]{2}\s*', '', title) if code: flag_html = ( f'' ) else: flag_html = "" cards += f"""
{flag_html}
{clean_title}
""" user_bar = "" if user_email: user_bar = ( f'
' f'✓ Logged in as {user_email}' f'  Log out' f'
' ) return f""" SAGA โ€” Annotation Projects {FAVICON_TAG} {user_bar}

SAGA Annotation Projects

Click a project below to start annotating.

{cards}

Full Label Studio projects page

{PRIVACY_SCRIPT} """ # --------------------------------------------------------------------------- # Routes # --------------------------------------------------------------------------- @app.route("/proxy/headers") def proxy_headers(): """Dump all incoming headers and cookies โ€” for browser debugging.""" data = { "headers": dict(request.headers), "cookies": list(request.cookies.keys()), "method": request.method, "url": request.url, } return Response(json.dumps(data, indent=2), content_type="application/json") @app.route("/proxy/me") def proxy_me(): uid, admin = get_current_user(request) return Response(json.dumps({"user_id": uid, "is_admin": admin}), content_type="application/json; charset=utf-8") def _count_user_annotations(uid, proj): """Count distinct tasks annotated by uid in proj โ€” DB-first, LS API fallback.""" # Primary: query task_completion table directly (no network call needed) try: conn = sqlite3.connect(DB_PATH) done = conn.execute( "SELECT COUNT(DISTINCT tc.task_id) FROM task_completion tc " "JOIN task t ON tc.task_id=t.id " "WHERE tc.completed_by_id=? AND t.project_id=? AND tc.was_cancelled=0", (uid, int(proj)) ).fetchone()[0] conn.close() if done > 0: return done except Exception: pass # Fallback: use superuser token from DB to call LS API try: conn = sqlite3.connect(DB_PATH) # Get token for any superuser row = conn.execute( "SELECT at.key FROM authtoken_token at " "JOIN htx_user u ON at.user_id = u.id " "WHERE u.is_superuser=1 ORDER BY u.id LIMIT 1" ).fetchone() conn.close() admin_token = row[0] if row else None except Exception: admin_token = None if admin_token: try: r = requests.get( f"{LS_URL}/api/tasks/?project={proj}&page_size=500", headers={"Authorization": f"Token {admin_token}", "Host": "localhost:8080"}, timeout=10, ) if r.status_code == 200: data = r.json() tasks = data if isinstance(data, list) else data.get("tasks", data.get("results", [])) done = 0 for t in tasks: for ann in t.get("annotations", []): if ann.get("was_cancelled"): continue cb = ann.get("completed_by") cb_id = cb if isinstance(cb, int) else (cb.get("id") if isinstance(cb, dict) else None) if cb_id == uid: done += 1 break return done except Exception: pass return 0 @app.route("/proxy/init") def proxy_init(): """Single endpoint: returns {is_admin, done, total} in one round-trip.""" uid, admin = get_current_user(request) proj = request.args.get("project") done, total = 0, 0 if uid and proj: total = _MANIFEST_TASKS.get(str(proj), 0) done = _count_user_annotations(uid, proj) return Response( json.dumps({"is_admin": admin, "done": done, "total": total}), content_type="application/json; charset=utf-8" ) @app.route("/proxy/count-debug") def proxy_count_debug(): """Debug: test _count_user_annotations for a given uid+project (key-gated).""" if request.args.get("key") != "saga2026": return Response("forbidden", status=403) uid = int(request.args.get("uid", 0)) proj = request.args.get("project", "4") # Also test admin token retrieval try: conn = sqlite3.connect(DB_PATH) tok_row = conn.execute("SELECT user_id, key FROM authtoken_token WHERE user_id=1").fetchone() conn.close() admin_token = tok_row[1] if tok_row else None except Exception as e: admin_token = None tok_err = str(e) # Test LS API call ls_status, ls_tasks = None, None if admin_token: try: r = requests.get( f"{LS_URL}/api/tasks/?project={proj}&page_size=500", headers={"Authorization": f"Token {admin_token}", "Host": "localhost:8080"}, timeout=10, ) ls_status = r.status_code if r.status_code == 200: data = r.json() tasks = data if isinstance(data, list) else data.get("tasks", data.get("results", [])) ls_tasks = len(tasks) done = sum(1 for t in tasks for a in t.get("annotations", []) if not a.get("was_cancelled") and (a.get("completed_by") == uid or (isinstance(a.get("completed_by"), dict) and a["completed_by"].get("id") == uid))) else: done = -1 except Exception as e: done = -2 ls_status = str(e) else: done = -3 # Also check task_completion directly try: conn2 = sqlite3.connect(DB_PATH) db_done = conn2.execute( "SELECT COUNT(DISTINCT tc.task_id) FROM task_completion tc " "JOIN task t ON tc.task_id=t.id " "WHERE tc.completed_by_id=? AND t.project_id=? AND tc.was_cancelled=0", (uid, int(proj)) ).fetchone()[0] conn2.close() except Exception as e: db_done = str(e) # Find actual LS DB path import glob as _glob db_files = sorted(_glob.glob("/data/**/*.sqlite3", recursive=True) + _glob.glob("/label-studio/**/*.sqlite3", recursive=True)) # Count users in each candidate DB db_user_counts = {} for f in db_files[:10]: try: c = sqlite3.connect(f); n = c.execute("SELECT COUNT(*) FROM htx_user").fetchone()[0]; c.close() db_user_counts[f] = n except Exception: db_user_counts[f] = "err" result = {"uid": uid, "proj": proj, "admin_token_found": admin_token is not None, "ls_status": ls_status, "ls_tasks": ls_tasks, "done": done, "db_done": db_done, "ls_url": LS_URL, "DB_PATH": DB_PATH, "db_files": db_user_counts, "_count_result": _count_user_annotations(uid, proj)} return Response(json.dumps(result, indent=2), content_type="application/json; charset=utf-8") @app.route("/proxy/db-schema") def proxy_db_schema(): """Return schema of key tables for debugging (admin only via ?key=).""" if request.args.get("key") != "saga2026": return Response("forbidden", status=403) try: conn = sqlite3.connect(DB_PATH) tables = [r[0] for r in conn.execute( "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name" ).fetchall()] result = {"tables": tables} for t in tables: if "org" in t.lower() or "token" in t.lower(): cols = [r[1] for r in conn.execute(f"PRAGMA table_info({t})").fetchall()] result[t] = cols conn.close() return Response(json.dumps(result, indent=2), content_type="application/json; charset=utf-8") except Exception as e: return Response(json.dumps({"error": str(e)}), status=500, content_type="application/json; charset=utf-8") @app.route("/proxy/ls-token") def proxy_ls_token(): """Return the LS API token for the authenticated user (admins only).""" uid, admin = get_current_user(request) if not uid or not admin: return Response(json.dumps({"error": "unauthorized"}), status=401, content_type="application/json; charset=utf-8") try: conn = sqlite3.connect(DB_PATH) row = conn.execute( "SELECT key FROM authtoken_token WHERE user_id=?", (uid,) ).fetchone() conn.close() if row: return Response(json.dumps({"token": row[0]}), content_type="application/json; charset=utf-8") return Response(json.dumps({"error": "no token found"}), status=404, content_type="application/json; charset=utf-8") except Exception as e: return Response(json.dumps({"error": str(e)}), status=500, content_type="application/json; charset=utf-8") @app.route("/proxy/diagnostic") def proxy_diagnostic(): """Returns admin-user status and Space health info.""" lines = [] # Check LS health try: r = requests.get(f"{LS_URL}/health", timeout=5) lines.append(f"LS health: {r.status_code}") except Exception as e: lines.append(f"LS health: ERROR {e}") # Check admin user in DB try: conn = sqlite3.connect(DB_PATH, timeout=5) rows = conn.execute( "SELECT id, email, is_active, is_superuser, is_staff, " "substr(password,1,30) FROM htx_user" ).fetchall() conn.close() lines.append(f"htx_user rows: {len(rows)}") for row in rows: lines.append(f" id={row[0]} email={row[1]} active={row[2]} " f"super={row[3]} staff={row[4]} pw_prefix={row[5]}") except Exception as e: lines.append(f"DB check: ERROR {e}") # Verify changeme against stored hash (pure Python โ€” no Django setup needed) try: import hashlib, base64 conn = sqlite3.connect(DB_PATH, timeout=5) rows2 = conn.execute("SELECT email, password FROM htx_user").fetchall() conn.close() for email, pw_hash in rows2: # Format: pbkdf2_sha256$$$ parts = pw_hash.split("$", 3) if len(parts) == 4 and parts[0] == "pbkdf2_sha256": _, iters, salt, stored_b64 = parts dk = hashlib.pbkdf2_hmac("sha256", "changeme".encode(), salt.encode(), int(iters)) computed = base64.b64encode(dk).decode() ok = (computed == stored_b64) lines.append(f" verify('changeme', {email}): {ok} [iters={iters}]") else: lines.append(f" {email}: unrecognised hash format: {pw_hash[:20]}") except Exception as e: lines.append(f"verify error: {e}") # Show ensure_admin log log_path = "/data/ls/ensure_admin.log" try: if os.path.exists(log_path): with open(log_path) as f: lines.append("\n--- ensure_admin.log ---") lines.extend(f.read().splitlines()[-20:]) else: lines.append("\nensure_admin.log: not found") except Exception as e: lines.append(f"ensure_admin.log error: {e}") return Response("\n".join(lines) + "\n", status=200, content_type="text/plain") @app.route("/proxy/token-debug") def proxy_token_debug(): """Dump authtoken_token and test all tokens against LS API (admin only).""" uid, admin = get_current_user(request) if not uid or not admin: return Response(json.dumps({"error": "unauthorized"}), status=401, content_type="application/json; charset=utf-8") try: conn = sqlite3.connect(DB_PATH) rows = conn.execute( "SELECT at.key, at.user_id, u.email, u.is_superuser " "FROM authtoken_token at JOIN htx_user u ON at.user_id = u.id" ).fetchall() conn.close() results = [] for (key, uid2, email, su) in rows: # Test this token against internal LS API try: r = requests.get(f"{LS_URL}/api/projects/?page_size=1", headers={"Authorization": f"Token {key}"}, timeout=5) status = r.status_code n_projects = len(r.json().get("results", [])) if status == 200 else 0 except Exception as e: status = f"error: {e}" n_projects = 0 results.append({ "key_prefix": key[:8], "user_id": uid2, "email": email, "is_superuser": su, "ls_api_status": status, "ls_projects": n_projects, }) return Response(json.dumps(results, indent=2), content_type="application/json; charset=utf-8") except Exception as e: return Response(json.dumps({"error": str(e)}), status=500, content_type="application/json; charset=utf-8") @app.route("/proxy/manifest-check") def proxy_manifest_check(): """Return current _MANIFEST_IDS for debugging.""" uid, admin = get_current_user(request) if not uid or not admin: return Response(json.dumps({"error": "unauthorized"}), status=401, content_type="application/json; charset=utf-8") import os p1 = os.path.join(os.path.dirname(os.path.abspath(__file__)), "projects", "manifest.json") p2 = "/app/projects/manifest.json" return Response(json.dumps({ "manifest_ids": sorted(_MANIFEST_IDS) if _MANIFEST_IDS else None, "manifest_type": str(type(_MANIFEST_IDS)), "p1_exists": os.path.exists(p1), "p2_exists": os.path.exists(p2), "file": __file__, "proxy_version": "v6-manifest-check", }, ensure_ascii=False, indent=2), content_type="application/json; charset=utf-8") @app.route("/proxy/db-projects") def proxy_db_projects(): """Dump all rows from the project table plus debug info (admin only).""" uid, admin = get_current_user(request) if not uid or not admin: return Response(json.dumps({"error": "unauthorized"}), status=401, content_type="application/json; charset=utf-8") debug = {} # List DB-related files data_dir = "/data/ls" try: files = sorted(os.listdir(data_dir)) db_files = [f for f in files if f.endswith(".sqlite3") or "sqlite" in f.lower()] debug["db_files"] = db_files debug["db_path"] = DB_PATH debug["db_path_exists"] = os.path.exists(DB_PATH) # Find ALL sqlite files in /data/ import subprocess result = subprocess.run(["find", "/data", "-name", "*.sqlite3", "-type", "f"], capture_output=True, text=True, timeout=5) debug["all_sqlite_files"] = result.stdout.strip().split("\n") if result.stdout.strip() else [] # Also check /label-studio/data symlink target import os.path ls_data = "/label-studio/data" if os.path.islink(ls_data): debug["ls_data_link"] = os.readlink(ls_data) elif os.path.isdir(ls_data): debug["ls_data_dir"] = "is a directory (not symlink)" except Exception as e: debug["list_error"] = str(e) # Get journal mode try: conn = sqlite3.connect(DB_PATH) jm = conn.execute("PRAGMA journal_mode").fetchone() debug["journal_mode"] = jm[0] if jm else "unknown" rows = conn.execute( "SELECT id, title, is_draft, is_published FROM project ORDER BY id" ).fetchall() conn.close() result = { "debug": debug, "projects": [{"id": r[0], "title": r[1], "is_draft": r[2], "is_published": r[3]} for r in rows] } return Response(json.dumps(result, ensure_ascii=False, indent=2), content_type="application/json; charset=utf-8") except Exception as e: return Response(json.dumps({"error": str(e), "debug": debug}), status=500, content_type="application/json; charset=utf-8") @app.route("/proxy/reset-admin") def proxy_reset_admin(): """Force-reset admin password. Requires ?key=saga_reset secret.""" if request.args.get("key") != "saga_reset": return Response("Forbidden", status=403, content_type="text/plain") email = os.environ.get("LABEL_STUDIO_USERNAME", "admin@saga.is") password = "12345" # matches ensure_admin.py hardcoded value lines = [] # Show what password we're resetting to lines.append(f"LABEL_STUDIO_USERNAME={email}") lines.append(f"LABEL_STUDIO_PASSWORD={password} (len={len(password)})") try: import hashlib, base64 salt = base64.b64encode(os.urandom(12)).decode() iters = 870000 dk = hashlib.pbkdf2_hmac("sha256", password.encode(), salt.encode(), iters) pw_hash = f"pbkdf2_sha256${iters}${salt}${base64.b64encode(dk).decode()}" lines.append(f"Hash generated (pure-Python): {pw_hash[:20]}โ€ฆ") except Exception as e: lines.append(f"Hash generation failed: {e}") return Response("\n".join(lines) + "\n", status=500, content_type="text/plain") try: conn = sqlite3.connect(DB_PATH, timeout=10) row = conn.execute("SELECT id FROM htx_user WHERE email=?", (email,)).fetchone() if row: conn.execute( "UPDATE htx_user SET password=?, is_active=1, is_superuser=1, is_staff=1 " "WHERE email=?", (pw_hash, email) ) conn.commit() lines.append(f"Updated {email} in htx_user") else: lines.append(f"User {email} not found โ€” running Django ORM creation") os.environ.setdefault("DJANGO_SETTINGS_MODULE", "label_studio.core.settings.label_studio") import django; django.setup() from django.contrib.auth import get_user_model User = get_user_model() user, created = User.objects.get_or_create( email=email, defaults={"username": email, "is_superuser": True, "is_staff": True, "is_active": True}, ) user.username = email; user.is_superuser = True user.is_staff = True; user.is_active = True user.set_password(password); user.save() lines.append(f"ORM: {'created' if created else 'updated'} {email}") conn.close() except Exception as e: lines.append(f"ERROR: {e}") return Response("\n".join(lines) + "\n", status=200, content_type="text/plain") @app.route("/proxy/trigger-backup") def proxy_trigger_backup(): """Manually trigger a DB backup to HF. Requires admin session or ?key=saga2026.""" uid, admin = get_current_user(request) if not admin and request.args.get("key") != "saga2026": return Response("forbidden", status=403, content_type="text/plain") try: import subprocess result = subprocess.run( ["python3", "/app/db_sync.py", "backup"], timeout=120, capture_output=True, text=True, ) out = result.stdout + result.stderr return Response(f"exit={result.returncode}\n{out}", status=200, content_type="text/plain") except Exception as e: return Response(f"ERROR: {e}", status=500, content_type="text/plain") @app.route("/proxy/db-download") def proxy_db_download(): """Serve the live SQLite DB for download. Requires ?key=saga_dl_2026.""" if request.args.get("key") != "saga_dl_2026": return Response("forbidden", status=403, content_type="text/plain") if not os.path.exists(DB_PATH): return Response("DB not found", status=404, content_type="text/plain") return Response( open(DB_PATH, "rb").read(), status=200, content_type="application/octet-stream", headers={"Content-Disposition": "attachment; filename=label_studio.sqlite3"}, ) _LOGIN_BANNER = """
SAGA Annotation Platform
Sign in with your email and password  ยท  Create account
Coming from Prolific? Do not log in here โ€” use the Start button in your Prolific study link instead. Your login happens automatically when you click that link. If you lost the link, return to app.prolific.com and open the study from there.
""" @app.route("/user/login/", methods=["GET"]) @app.route("/user/login", methods=["GET"]) def login_get(): """If already logged in, skip the login page; otherwise inject a clear banner.""" uid, _ = get_current_user(request) if uid: next_url = request.args.get("next", "/projects/") if not next_url.startswith("/") or next_url in ("/", ""): next_url = "/projects/" return Response("", status=302, headers={"Location": next_url}) # Fetch LS login page and inject the SAGA banner at the top try: r = requests.get(f"{LS_URL}/user/login/", cookies=_ls_cookies(request), headers=_fwd_headers(request), timeout=10) except requests.exceptions.ConnectionError: return _ls_unavailable() if r.status_code != 200: return _ls_unavailable() html = _rewrite_hf_urls(r.text) body_match = _re.search(r'(]*>)', html, _re.IGNORECASE) if body_match: insert_pos = body_match.end() html = html[:insert_pos] + _LOGIN_BANNER + html[insert_pos:] else: html = _LOGIN_BANNER + html fwd = _out_headers(r, extra_exclude={"content-encoding", "content-length"}) return Response(html, status=200, headers=fwd, content_type="text/html; charset=utf-8") @app.route("/user/login/", methods=["POST"]) @app.route("/user/login", methods=["POST"]) def login_post(): """Open a fresh server-side LS session to handle CSRF, then forward cookies to browser. We never rely on the browser's csrftoken cookie because the Space runs inside an iframe on huggingface.co โ€” cross-site context blocks SameSite=Lax cookies. The proxy fetches a fresh CSRF token from LS directly and posts credentials server-to-server. Cookies are returned as SameSite=None;Secure so they work both in direct-URL access and in the HF iframe. """ try: sess = requests.Session() r0 = sess.get(f"{LS_URL}/user/login/", timeout=10) if r0.status_code != 200: return _ls_unavailable() csrf_cookie = sess.cookies.get("csrftoken", "") # Support any attribute order in the hidden input m = (_re.search(r'name="csrfmiddlewaretoken"[^>]*value="([^"]+)"', r0.text) or _re.search(r'value="([^"]+)"[^>]*name="csrfmiddlewaretoken"', r0.text)) csrf_token = m.group(1) if m else csrf_cookie form = request.form.to_dict(flat=True) form["csrfmiddlewaretoken"] = csrf_token # allow_redirects=True so we get the final sessionid after Django session rotation r1 = sess.post( f"{LS_URL}/user/login/", data=form, headers={"Referer": f"{LS_URL}/user/login/", "X-CSRFToken": csrf_token}, allow_redirects=True, timeout=15, ) if "/user/login" not in r1.url: next_url = request.args.get("next") or request.form.get("next") or "/projects/" if not next_url.startswith("/") or next_url in ("/", ""): next_url = "/projects/" out = Response("", status=302, headers={"Location": next_url}) # SameSite=None;Secure โ€” required for cross-site iframe on huggingface.co for name, value in sess.cookies.items(): out.set_cookie( name, value, path="/", max_age=1209600, samesite="None", httponly=(name == "sessionid"), secure=True, ) return out # Failure: wrong credentials โ€” re-render login page with banner err_html = r1.text if r1.text else "" body_match = _re.search(r'(]*>)', err_html, _re.IGNORECASE) if body_match: insert_pos = body_match.end() err_html = err_html[:insert_pos] + _LOGIN_BANNER + err_html[insert_pos:] else: err_html = _LOGIN_BANNER + err_html return Response(err_html, status=200, content_type="text/html; charset=utf-8") except requests.exceptions.ConnectionError: return _ls_unavailable() except Exception as e: return Response(f"Login error: {e}", status=502, content_type="text/plain") @app.route("/user/signup/", methods=["GET", "POST"]) @app.route("/user/signup", methods=["GET", "POST"]) def signup(): """Proxy sign-up and re-set cookies with SameSite=None;Secure so the proxy can read them.""" if request.method == "GET": return _proxy_stream("/user/signup/") try: sess = requests.Session() r0 = sess.get(f"{LS_URL}/user/signup/", timeout=10) if r0.status_code != 200: return _ls_unavailable() csrf_cookie = sess.cookies.get("csrftoken", "") m = (_re.search(r'name="csrfmiddlewaretoken"[^>]*value="([^"]+)"', r0.text) or _re.search(r'value="([^"]+)"[^>]*name="csrfmiddlewaretoken"', r0.text)) csrf_token = m.group(1) if m else csrf_cookie form = request.form.to_dict(flat=True) form["csrfmiddlewaretoken"] = csrf_token r1 = sess.post( f"{LS_URL}/user/signup/", data=form, headers={"Referer": f"{LS_URL}/user/signup/", "X-CSRFToken": csrf_token}, allow_redirects=True, timeout=15, ) if "/user/signup" not in r1.url: out = Response("", status=302, headers={"Location": "/projects/"}) for name, value in sess.cookies.items(): out.set_cookie(name, value, path="/", max_age=1209600, httponly=(name == "sessionid"), samesite="None", secure=True) return out # Sign-up failed (validation errors) โ€” return LS response as-is return Response(r1.text, status=r1.status_code, content_type=r1.headers.get("Content-Type", "text/html")) except requests.exceptions.ConnectionError: return _ls_unavailable() except Exception as e: return Response(f"Signup error: {e}", status=502, content_type="text/plain") @app.route("/user/logout/", methods=["GET", "POST"]) @app.route("/user/logout", methods=["GET", "POST"]) def logout(): """Clear the server-side LS session and wipe browser cookies.""" try: requests.get(f"{LS_URL}/user/logout/", cookies=_ls_cookies(request), headers=_fwd_headers(request), allow_redirects=True, timeout=5) except Exception: pass html = f""" SAGA โ€” Logged Out {FAVICON_TAG}

You have been logged out

Thank you for using SAGA Annotation.

Log in again
""" out = Response(html, status=200, content_type="text/html; charset=utf-8") for name in ("sessionid", "csrftoken", "prolific_pid", "prolific_cc"): for ss in ("None", "Lax"): out.set_cookie(name, "", path="/", max_age=0, expires=0, samesite=ss, secure=(ss == "None")) return out @app.route("/autologin", methods=["GET"]) def autologin(): """Log in a Prolific participant โ€” two modes: Credential mode (Prolific distributes pre-created accounts): /autologin?email=CRED_EMAIL&password=CRED_PW&pid=PID&next=/projects/5/ PID mode (auto-create throwaway account from Prolific PID): /autologin?pid=PROLIFIC_PID&next=/projects/4/ """ pid = request.args.get("pid") or request.cookies.get("prolific_pid", "") next_url = request.args.get("next", "/projects/") if not next_url.startswith("/"): next_url = "/projects/" # Credential mode: explicit email+password provided (pre-created accounts) cred_email = request.args.get("email", "").strip() cred_password = request.args.get("password", "").strip() if not pid and not cred_email: return Response("", status=302, headers={"Location": "/user/login/"}) # Already logged in โ†’ go straight to dest uid, _ = get_current_user(request) if uid: return Response("", status=302, headers={"Location": next_url}) if cred_email and cred_password: email = cred_email password = cred_password else: email = f"{pid}@prolific.saga" password = f"saga-{pid[:16]}" def _do_login(email, password): """Attempt LS login, return requests.Session on success or None on failure.""" sess = requests.Session() r0 = sess.get(f"{LS_URL}/user/login/", timeout=10) csrf = sess.cookies.get("csrftoken", "") m = (_re.search(r'name="csrfmiddlewaretoken"[^>]*value="([^"]+)"', r0.text) or _re.search(r'value="([^"]+)"[^>]*name="csrfmiddlewaretoken"', r0.text)) csrf_token = m.group(1) if m else csrf r1 = sess.post( f"{LS_URL}/user/login/", data={"email": email, "password": password, "csrfmiddlewaretoken": csrf_token}, headers={"Referer": f"{LS_URL}/user/login/", "X-CSRFToken": csrf_token}, allow_redirects=True, timeout=15, ) if sess.cookies.get("sessionid", "") and "/user/login" not in r1.url: return sess return None def _make_response_from_session(sess, next_url, pid): out = Response("", status=302, headers={"Location": next_url}) for name, value in sess.cookies.items(): out.set_cookie(name, value, path="/", max_age=1209600, httponly=(name == "sessionid"), samesite="None", secure=True) if pid: out.set_cookie("prolific_pid", pid, max_age=86400, samesite="Lax") return out # Try login first (account may already exist from a previous visit) try: sess = _do_login(email, password) if sess: return _make_response_from_session(sess, next_url, pid) except Exception: pass # Credential mode: login failed โ€” accounts created via admin API may have # plaintext/unhashed passwords. Self-heal: rewrite the DB hash, then retry. if cred_email and cred_password: try: import hashlib, base64 as _b64 _salt = _b64.b64encode(os.urandom(12)).decode() _iters = 870000 _dk = hashlib.pbkdf2_hmac("sha256", password.encode(), _salt.encode(), _iters) _pw = f"pbkdf2_sha256${_iters}${_salt}${_b64.b64encode(_dk).decode()}" _conn = sqlite3.connect(DB_PATH, timeout=10) _rows = _conn.execute( "UPDATE htx_user SET password=?, is_active=1 WHERE email=?", (_pw, email) ).rowcount _conn.commit(); _conn.close() print(f"[autologin] self-healed hash for {email} ({_rows} row)", flush=True) if _rows > 0: sess2 = _do_login(email, password) if sess2: return _make_response_from_session(sess2, next_url, pid) except Exception as _e: print(f"[autologin] self-heal failed: {_e}", flush=True) # Login failed โ†’ account doesn't exist yet, create it via signup try: sess2 = requests.Session() r2 = sess2.get(f"{LS_URL}/user/signup/", timeout=10) csrf_cookie2 = sess2.cookies.get("csrftoken", "") m2 = (_re.search(r'name="csrfmiddlewaretoken"[^>]*value="([^"]+)"', r2.text) or _re.search(r'value="([^"]+)"[^>]*name="csrfmiddlewaretoken"', r2.text)) csrf_token2 = m2.group(1) if m2 else csrf_cookie2 r3 = sess2.post( f"{LS_URL}/user/signup/", data={"email": email, "password": password, "password_confirm": password, "csrfmiddlewaretoken": csrf_token2}, headers={"Referer": f"{LS_URL}/user/signup/", "X-CSRFToken": csrf_token2}, allow_redirects=True, timeout=15, ) session_cookie2 = sess2.cookies.get("sessionid", "") if session_cookie2: out = Response("", status=302, headers={"Location": next_url}) for name, value in sess2.cookies.items(): out.set_cookie(name, value, path="/", max_age=1209600, httponly=(name == "sessionid"), samesite="None", secure=True) if pid: out.set_cookie("prolific_pid", pid, max_age=86400, samesite="Lax") return out except Exception: pass # Fallback โ€” send to manual login return Response("", status=302, headers={"Location": f"/user/login/?next={next_url}"}) @app.route("/projects", methods=["GET"]) @app.route("/projects/", methods=["GET"]) def projects_landing(): uid, _ = get_current_user(request) if not uid: return Response("", status=302, headers={"Location": "/user/login/?next=/projects/"}) # Check if user has a prolific_cc cookie โ€” if so, show completion page when done prolific_cc = request.cookies.get("prolific_cc", "") if prolific_cc: # Find which project this CC belongs to proj_id = None proj_meta = None for pid, meta in PROLIFIC_PROJECT_META.items(): if meta.get("cc") == prolific_cc: proj_id = pid proj_meta = meta break if proj_id and proj_meta: done = _count_user_annotations(uid, int(proj_id)) total = _MANIFEST_TASKS.get(str(proj_id), 0) if total > 0 and done >= total: flag = proj_meta.get("flag", "") lang = proj_meta.get("lang", "") task = proj_meta.get("task", "") html = f""" SAGA โ€” Thank You! {FAVICON_TAG}

✓ All Done!

{flag} {lang} โ€” {task}
Thank you for completing all {total} tasks.

Your Prolific completion code:
{prolific_cc}
Submit on Prolific →

Copy the code above if the button doesn't work.

""" return Response(html, status=200, content_type="text/html; charset=utf-8") # Look up user email for the status bar try: conn = sqlite3.connect(DB_PATH) row = conn.execute("SELECT email FROM htx_user WHERE id=?", (uid,)).fetchone() conn.close() user_email = row[0] if row else "" except Exception: user_email = "" return Response(build_projects_page(user_email), status=200, content_type="text/html; charset=utf-8") @app.route("/projects-ls/", methods=["GET"]) @app.route("/projects-ls", methods=["GET"]) def projects_ls(): return _proxy_stream("/projects/") @app.route("/api/tasks//", methods=["GET"]) @app.route("/api/tasks/", methods=["GET"]) def filter_task_data(task_id): """Strip other users' embedded annotations so LS shows a blank form to each annotator.""" uid, admin = get_current_user(request) resp = _proxy_buffered(f"/api/tasks/{task_id}/") if admin: return _respond_buffered(resp) try: data = resp.json() if "annotations" in data: def _cb_id(ann): cb = ann.get("completed_by") return cb.get("id") if isinstance(cb, dict) else cb if uid is not None: data["annotations"] = [a for a in data["annotations"] if _cb_id(a) == uid] else: data["annotations"] = [] return Response(json.dumps(data), status=resp.status_code, content_type="application/json; charset=utf-8") except Exception: return _respond_buffered(resp) @app.route("/api/tasks//annotations/", methods=["GET", "POST"]) @app.route("/api/tasks//annotations", methods=["GET", "POST"]) def filter_task_annotations(task_id): uid, admin = get_current_user(request) if request.method == "POST": resp = _proxy_buffered(f"/api/tasks/{task_id}/annotations/") # Send email notification on successful annotation submission if resp.status_code in (200, 201): try: ann = resp.json() project_id = ann.get("project") annotator_email = "unknown" project_title = f"project {project_id}" try: conn = sqlite3.connect(DB_PATH, timeout=3) row = conn.execute("SELECT email FROM htx_user WHERE id=?", (uid,)).fetchone() if row: annotator_email = row[0] prow = conn.execute("SELECT title FROM projects_project WHERE id=?", (project_id,)).fetchone() if prow: project_title = prow[0] conn.close() except Exception: pass # Email only when annotator finishes ALL tasks in this project try: conn2 = sqlite3.connect(DB_PATH, timeout=3) done_count = conn2.execute( "SELECT COUNT(*) FROM task_completion tc JOIN task t ON tc.task_id=t.id " "WHERE tc.completed_by_id=? AND t.project_id=? AND tc.was_cancelled=0", (uid, project_id) ).fetchone()[0] total_count = conn2.execute( "SELECT COUNT(*) FROM task t WHERE t.project_id=? " "AND NOT EXISTS (SELECT 1 FROM core_deletedrow d " "WHERE CAST(d.object_id AS INTEGER)=t.id)", (project_id,) ).fetchone()[0] conn2.close() if total_count > 0 and done_count >= total_count: threading.Thread( target=_send_completion_email, args=(project_title, annotator_email, done_count, total_count), daemon=True, ).start() except Exception as exc: print(f"[notify] completion check failed: {exc}", flush=True) except Exception as exc: print(f"[notify] parse error after POST annotation: {exc}", flush=True) # Backup DB immediately after each annotation threading.Thread( target=_backup_db, daemon=True ).start() return _respond_buffered(resp) # GET โ€” filter annotations by user for non-admins resp = _proxy_buffered(f"/api/tasks/{task_id}/annotations/") if admin or uid is None: return _respond_buffered(resp) try: data = resp.json() def _cb_id(ann): cb = ann.get("completed_by") return cb.get("id") if isinstance(cb, dict) else cb own = [ann for ann in data if _cb_id(ann) == uid] return Response(json.dumps(own), status=resp.status_code, content_type="application/json; charset=utf-8") except Exception: return _respond_buffered(resp) @app.route("/api/annotations//", methods=["DELETE"]) @app.route("/api/annotations/", methods=["DELETE"]) def guard_annotation_delete(ann_id): uid, admin = get_current_user(request) if admin: return _proxy_stream(f"/api/annotations/{ann_id}/") if uid is None: return Response('{"detail":"Forbidden"}', status=403, content_type="application/json") conn = sqlite3.connect(DB_PATH) row = conn.execute("SELECT completed_by_id FROM task_completion WHERE id=?", (ann_id,)).fetchone() conn.close() if row and row[0] == uid: return _proxy_stream(f"/api/annotations/{ann_id}/") return Response('{"detail":"Forbidden"}', status=403, content_type="application/json") @app.route("/proxy/progress") def proxy_progress(): uid, _ = get_current_user(request) proj = request.args.get("project") if not uid or not proj: return Response(json.dumps({"done": 0, "total": 0}), content_type="application/json") try: conn = sqlite3.connect(DB_PATH) done = conn.execute( "SELECT COUNT(*) FROM task_completion tc JOIN task t ON tc.task_id=t.id " "WHERE tc.completed_by_id=? AND t.project_id=? AND tc.was_cancelled=0", (uid, proj) ).fetchone()[0] total = _MANIFEST_TASKS.get(str(proj), 0) conn.close() except Exception: done, total = 0, 0 return Response(json.dumps({"done": done, "total": total}), content_type="application/json") @app.route("/proxy/session-check") def proxy_session_check(): """Debug: show session status and which cookies the proxy received.""" uid, admin = get_current_user(request) cookie_names = sorted(request.cookies.keys()) ls_cookie_names = sorted(_ls_cookies(request).keys()) has_sid = "sessionid" in request.cookies try: r = requests.get(f"{LS_URL}/api/current-user/whoami", cookies=_ls_cookies(request), timeout=5, allow_redirects=False) whoami = r.status_code whoami_body = r.text[:200] if r.status_code != 200 else r.json().get("email","?") except Exception as e: whoami, whoami_body = "ERR", str(e) result = { "uid": uid, "is_admin": admin, "cookies_received": cookie_names, "ls_cookies_forwarded": ls_cookie_names, "has_sessionid": has_sid, "whoami_status": whoami, "whoami_email": whoami_body, } return Response(json.dumps(result, indent=2), content_type="application/json") @app.route("/go//", methods=["GET"]) @app.route("/go/", methods=["GET"]) def go_to_project(project_id): """After autologin, redirect participant straight to their first available task.""" uid, _ = get_current_user(request) if not uid: return Response("", status=302, headers={"Location": f"/user/login/?next=/go/{project_id}/"}) try: conn = sqlite3.connect(DB_PATH) row = conn.execute( "SELECT at.key FROM authtoken_token at " "JOIN htx_user u ON at.user_id = u.id " "WHERE u.is_superuser=1 ORDER BY u.id LIMIT 1" ).fetchone() conn.close() admin_token = row[0] if row else None except Exception: admin_token = None if admin_token: try: r = requests.get( f"{LS_URL}/api/projects/{project_id}/next/", headers={"Authorization": f"Token {admin_token}", "Host": "localhost:8080"}, cookies=_ls_cookies(request), timeout=10, ) d = r.json() if r.status_code == 200 else {} if d.get("id"): return Response("", status=302, headers={"Location": f"/projects/{project_id}/data?task={d['id']}"}) except Exception: pass return Response("", status=302, headers={"Location": f"/projects/{project_id}/data"}) @app.route("/start", methods=["GET"]) def prolific_start(): """Prolific entry point. Single-project mode (one Prolific study per LS project): /start?project=4&pid=PROLIFIC_PID Legacy per-language mode (both tasks for a language): /start?lang=da&pid=PROLIFIC_PID """ pid = request.args.get("pid", "") project = request.args.get("project", "").strip() lang = request.args.get("lang", "").lower() cc_override = request.args.get("cc", "") cred_email = request.args.get("email", "").strip() cred_pw = request.args.get("password", "").strip() # โ”€โ”€ Single-project mode โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ if project: meta = PROLIFIC_PROJECT_META.get(project) if not meta: return Response("

Unknown project. Please contact the researcher.

", status=400, content_type="text/html") cc = cc_override or meta["cc"] flag = meta["flag"] lang_name = meta["lang"] task = meta["task"] est_min = meta.get("est_min", 30) short_lang = lang_name.split("/")[0].strip() n_tasks = _MANIFEST_TASKS.get(str(project), 0) html = f""" SAGA โ€” {lang_name} {task} {FAVICON_TAG}
{flag} {lang_name}

{task}

Tasks: {n_tasks} short text evaluations
Estimated time: approximately {est_min} minutes
When done: you will be sent back to Prolific automatically
{flag} Start → {task}

Your Prolific ID is recorded automatically. Do not share this link.

{PRIVACY_SCRIPT} """ resp = make_response(html) if pid: resp.set_cookie("prolific_pid", pid, max_age=86400, samesite="Lax") if cc: resp.set_cookie("prolific_cc", cc, max_age=86400, samesite="Lax") return resp # โ”€โ”€ Legacy per-language mode โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ cc = cc_override or PROLIFIC_CC.get(lang, "") if lang not in PROLIFIC_LANG_PROJECTS: return Response("

Unknown language. Please contact the researcher.

", status=400, content_type="text/html") projects = PROLIFIC_LANG_PROJECTS[lang] flag = {"is": "๐Ÿ‡ฎ๐Ÿ‡ธ", "da": "๐Ÿ‡ฉ๐Ÿ‡ฐ", "nb": "๐Ÿ‡ณ๐Ÿ‡ด", "sv": "๐Ÿ‡ธ๐Ÿ‡ช"}.get(lang, "") lang_name = {"is": "Icelandic / รslenska", "da": "Danish / Dansk", "nb": "Norwegian / Norsk", "sv": "Swedish / Svenska"}.get(lang, lang.upper()) task_links = "" for proj_id, label in projects: task_links += f""" {flag} {label} """ html = f""" SAGA โ€” {lang_name} Evaluation {FAVICON_TAG}

{flag} {lang_name} โ€” Parse Quality Study

Thank you for participating! You will evaluate a few sentence-level text completions.

Step 1: Complete each annotation task below (~5 min each).
Step 2: You will be redirected back to Prolific automatically when done.
{task_links}

Your Prolific ID is recorded automatically. Do not share this link.

""" resp = make_response(html) if pid: resp.set_cookie("prolific_pid", pid, max_age=86400, samesite="Lax") if cc: resp.set_cookie("prolific_cc", cc, max_age=86400, samesite="Lax") return resp @app.route("/", defaults={"path": ""}, methods=["GET","POST","PUT","PATCH","DELETE","OPTIONS"]) @app.route("/", methods=["GET","POST","PUT","PATCH","DELETE","OPTIONS"]) def catch_all(path): full_path = "/" + path if any(s in full_path for s in ["googletagmanager", "google-analytics", "hotjar"]): return Response("", status=204) if request.method != "GET": return _proxy_stream(full_path) url = f"{LS_URL}{full_path}" if request.query_string: url += "?" + request.query_string.decode() try: resp = requests.get(url, headers=_fwd_headers(request), cookies=_ls_cookies(request), allow_redirects=False, stream=True, timeout=30) except requests.exceptions.ConnectionError: return _ls_unavailable() except requests.exceptions.Timeout: return Response("Request timed out.", status=504, content_type="text/plain") # Intercept LS's "not authenticated" redirect โ€” forward it to our login page. # Without this the browser would follow the redirect silently and the user # would see the LS login form (which bypasses our proxy cookie logic). if resp.status_code in (301, 302, 303, 307, 308): loc = resp.headers.get("Location", "") # Normalise โ€” LS may return absolute or relative if loc.startswith(LS_URL): loc = loc[len(LS_URL):] if loc.startswith("/user/login"): # Expired / missing session โ€” send back to our custom login page return Response("", status=302, headers={"Location": f"/user/login/?next={full_path}"}) # All other LS redirects pass through unchanged return Response("", status=resp.status_code, headers=_out_headers(resp)) ct = resp.headers.get("Content-Type", "") if "text/html" in ct and resp.status_code == 200: html = resp.content.decode("utf-8", errors="replace") # Rewrite absolute HF Space URLs โ†’ relative so scripts/API calls stay on current domain html = _rewrite_hf_urls(html) html = _re.sub(r']*(?:googletagmanager|google-analytics)[^>]*>.*?', '', html, flags=_re.DOTALL | _re.IGNORECASE) tag = PRIVACY_SCRIPT if "" in html: html = html.replace("", FAVICON_TAG + tag + "", 1) elif "" in html: html = html.replace("", FAVICON_TAG + tag + "", 1) else: html += tag fwd = _out_headers(resp, extra_exclude={"content-encoding", "content-length"}) return Response(html, status=200, headers=fwd, content_type="text/html; charset=utf-8") return Response(resp.iter_content(chunk_size=4096), status=resp.status_code, headers=_out_headers(resp, extra_exclude={"content-encoding"}), content_type=ct) if __name__ == "__main__": print("SAGA proxy on :7860 โ†’ Label Studio on :8080", flush=True) app.run(host="0.0.0.0", port=7860, threaded=True)