Blame for server.py

#!/usr/bin/env python3
"""User-study server: static pages + a small JSON API. Python standard library only
(huggingface_hub is optional, for backing up data/ to a private HF dataset).

    python3 server.py [--port 8000] [--host 0.0.0.0]

Env:
  PART1_N (12), PART2_N (10)  questions per participant
  ADMIN_KEY                   enables GET /api/admin/summary?key=... and /api/admin/export?key=...
  HF_TOKEN + BACKUP_REPO      e.g. BACKUP_REPO=user/us_responses (private dataset): restore data/ on start,
                              upload it every BACKUP_EVERY seconds (120) when it changed

Data (data/):
  participants.json   pid -> {created, questions: [...with the hidden method order...], done: bool}
  responses.jsonl     one line per answer (raw labels + resolved methods)
"""
import argparse
import json
import os
import random
import re
import threading
import time
from http import HTTPStatus
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlparse

HERE = os.path.dirname(os.path.abspath(__file__))
DATA = os.environ.get("US_DATA", os.path.join(HERE, "data"))
STATIC = os.path.join(HERE, "static")
PART1_N = int(os.environ.get("PART1_N", 12))
PART2_N = int(os.environ.get("PART2_N", 10))
ADMIN_KEY = os.environ.get("ADMIN_KEY", "")
PID_RE = re.compile(r"^[A-Za-z0-9]{1,5}$")
LABELS = "ABCDE"
# a finished ID that has not been opened again for this long is taken: entering it again is refused
TAKEN_AFTER = int(os.environ.get("TAKEN_AFTER", 30 * 60))

def load_manifest():
    """private/manifest.json if present (local runs); otherwise manifest.json from the private BACKUP_REPO dataset,
    so the method <-> video mapping is never part of a public Space repo."""
    p = os.path.join(HERE, "private", "manifest.json")
    if os.path.exists(p):
        return json.load(open(p))
    from huggingface_hub import hf_hub_download
    return json.load(open(hf_hub_download(os.environ["BACKUP_REPO"], "manifest.json", repo_type="dataset",
                                          token=os.environ["HF_TOKEN"])))


MAN = load_manifest()
LOCK = threading.Lock()
os.makedirs(DATA, exist_ok=True)
P_FILE = os.path.join(DATA, "participants.json")
R_FILE = os.path.join(DATA, "responses.jsonl")
DIRTY = threading.Event()


# ---------------------------------------------------------------- optional HF backup
def hf_restore():
    tok, repo = os.environ.get("HF_TOKEN"), os.environ.get("BACKUP_REPO")
    if not (tok and repo):
        return
    try:
        from huggingface_hub import hf_hub_download
        for name in ("participants.json", "responses.jsonl"):
            if not os.path.exists(os.path.join(DATA, name)):
                try:
                    p = hf_hub_download(repo, name, repo_type="dataset", token=tok, local_dir=DATA)
                    print("[backup] restored", p, flush=True)
                except Exception as e:  # first run: nothing there yet
                    print("[backup] no %s in %s (%s)" % (name, repo, type(e).__name__), flush=True)
    except ImportError:
        print("[backup] huggingface_hub not installed; backup disabled", flush=True)


def hf_backup_loop():
    tok, repo = os.environ.get("HF_TOKEN"), os.environ.get("BACKUP_REPO")
    if not (tok and repo):
        return
    from huggingface_hub import HfApi
    api, every = HfApi(token=tok), int(os.environ.get("BACKUP_EVERY", 120))
    while True:
        time.sleep(every)
        if not DIRTY.is_set():
            continue
        DIRTY.clear()
        try:
            with LOCK:
                snap = {n: open(os.path.join(DATA, n), "rb").read() for n in ("participants.json", "responses.jsonl")
                        if os.path.exists(os.path.join(DATA, n))}
            from huggingface_hub import CommitOperationAdd
            api.create_commit(repo, [CommitOperationAdd(n, b) for n, b in snap.items()], repo_type="dataset",
                              commit_message="backup %s" % time.strftime("%Y-%m-%d %H:%M:%S"))
            print("[backup] uploaded", list(snap), flush=True)
        except Exception as e:
            DIRTY.set()
            print("[backup] failed:", e, flush=True)


# ---------------------------------------------------------------- state
def load_participants():
    if os.path.exists(P_FILE):
        return json.load(open(P_FILE))
    return {}


def save_participants(ps):
    tmp = P_FILE + ".tmp"
    with open(tmp, "w") as f:
        json.dump(ps, f)
    os.replace(tmp, P_FILE)
    DIRTY.set()


def exposure(ps, part, assignable=False):
    excl = set(MAN.get("exclude", {}).get(part, [])) if assignable else set()
    c = {s: 0 for s in MAN[part] if s not in excl}
    for p in ps.values():
        for q in p["questions"]:
            if q["part"] == part and q["scene"] in c:
                c[q["scene"]] += 1
    return c


def pick(counts, n, exclude, rng):
    pool = [s for s in counts if s not in exclude]
    rng.shuffle(pool)                               # random tie-break
    pool.sort(key=lambda s: counts[s])              # least-shown scenes first (stable sort keeps the shuffle)
    return pool[:n]


def new_assignment(ps, rng):
    """Part 2 first (smaller pool, 27 scenes), then part 1 from the scenes not used in part 2."""
    s2 = pick(exposure(ps, "part2", assignable=True), PART2_N, set(), rng)
    s1 = pick(exposure(ps, "part1", assignable=True), PART1_N, set(s2), rng)
    rng.shuffle(s1)
    rng.shuffle(s2)
    qs = []
    for part, scenes in (("part1", s1), ("part2", s2)):
        for s in scenes:
            methods = list(MAN[part][s])
            rng.shuffle(methods)                    # display order, per question
            qs.append({"part": part, "scene": s, "order": methods})
    return qs


def last_seen(p):
    return p.get("last_seen") or max([q.get("t", 0) for q in p["questions"]] + [p["created"]])


def is_taken(p):
    return bool(p.get("done")) and time.time() - last_seen(p) > TAKEN_AFTER


def public_questions(p):
    out = []
    for i, q in enumerate(p["questions"]):
        vids = [{"label": LABELS[k], "url": "/static/v/%s.mp4" % MAN[q["part"]][q["scene"]][m]}
                for k, m in enumerate(q["order"])]
        out.append({"qi": i, "part": q["part"], "prompt": MAN["prompts"][q["scene"]], "videos": vids,
                    "answered": "answer" in q, "answer": q.get("answer")})
    return out


# ---------------------------------------------------------------- summary
def summary():
    ps = load_participants()
    m1 = {}
    m2 = {}
    n_done = sum(1 for p in ps.values() if p.get("done"))
    for p in ps.values():
        for q in p["questions"]:
            a = q.get("answer")
            if a is None:
                continue
            if q["part"] == "part1":
                for lab, r in a.items():
                    meth = q["order"][LABELS.index(lab)]
                    d = m1.setdefault(meth, {"n": 0, "rank_sum": 0, "top1": 0})
                    d["n"] += 1
                    d["rank_sum"] += r
                    d["top1"] += int(r == 1)
            else:
                d = m2.setdefault(" vs ".join(sorted(q["order"])), {})
                key = "tie" if a == "tie" else q["order"][LABELS.index(a)]
                d[key] = d.get(key, 0) + 1
    part1 = {m: {"mean_rank": round(d["rank_sum"] / d["n"], 3), "top1_rate": round(d["top1"] / d["n"], 3),
                 "n": d["n"]} for m, d in sorted(m1.items())}
    per_scene1 = {s: c for s, c in exposure(ps, "part1").items()}
    per_scene2 = {s: c for s, c in exposure(ps, "part2").items()}
    return {"participants": len(ps), "completed": n_done, "part1": part1, "part2": m2,
            "assigned_per_scene": {"part1": per_scene1, "part2": per_scene2}}


# ---------------------------------------------------------------- HTTP
class Handler(SimpleHTTPRequestHandler):
    def translate_path(self, path):
        p = urlparse(path).path
        p = p[len("/static"):] if p.startswith("/static/") else "/__none__"
        return super().translate_path(p)        # drops '..' segments, stays inside static/

    def __init__(self, *a, **kw):
        super().__init__(*a, directory=STATIC, **kw)   # only static/ is ever served

    def log_message(self, fmt, *args):
        if "/static/v/" not in str(args[0] if args else ""):
            super().log_message(fmt, *args)

    def end_headers(self):
        if self.path.startswith("/static/v/"):
            self.send_header("Cache-Control", "public, max-age=86400")
            if not self.headers.get("Range"):
                self.send_header("Accept-Ranges", "bytes")
        else:
            self.send_header("Cache-Control", "no-store")
        super().end_headers()

    def send_json(self, obj, code=200):
        b = json.dumps(obj).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(b)))
        self.end_headers()
        self.wfile.write(b)

    def do_GET(self):
        u = urlparse(self.path)
        if u.path in ("/", "/index.html"):
            self.path = "/static/index.html"
        elif u.path in ("/survey", "/survey.html"):
            self.path = "/static/survey.html"
        elif u.path.startswith("/api/admin/"):
            key = parse_qs(u.query).get("key", [""])[0]
            if not ADMIN_KEY or key != ADMIN_KEY:
                return self.send_json({"error": "forbidden"}, 403)
            if u.path == "/api/admin/summary":
                return self.send_json(summary())
            if u.path == "/api/admin/export":
                with LOCK:
                    return self.send_json({"participants": load_participants(),
                                           "responses": [json.loads(l) for l in open(R_FILE)] if os.path.exists(R_FILE) else []})
            return self.send_json({"error": "not found"}, 404)
        elif not u.path.startswith("/static/"):         # never serve data/ or private/
            return self.send_json({"error": "not found"}, 404)
        if u.path.startswith("/static/v/") and self.headers.get("Range"):
            return self.send_range(u.path)
        return super().do_GET()

    def send_range(self, path):
        """HTTP Range for the videos (Safari / iOS will not play an mp4 without 206 responses)."""
        fp = self.translate_path(path)
        if not os.path.isfile(fp):
            return self.send_json({"error": "not found"}, 404)
        size = os.path.getsize(fp)
        m = re.match(r"bytes=(\d*)-(\d*)$", self.headers["Range"].strip())
        if not m or (m.group(1) == "" and m.group(2) == ""):
            self.send_response(HTTPStatus.REQUESTED_RANGE_NOT_SATISFIABLE)
            self.send_header("Content-Range", "bytes */%d" % size)
            self.end_headers()
            return
        if m.group(1) == "":                         # suffix range: last N bytes
            start, end = max(0, size - int(m.group(2))), size - 1
        else:
            start = int(m.group(1))
            end = min(int(m.group(2)), size - 1) if m.group(2) else size - 1
        if start >= size or start > end:
            self.send_response(HTTPStatus.REQUESTED_RANGE_NOT_SATISFIABLE)
            self.send_header("Content-Range", "bytes */%d" % size)
            self.end_headers()
            return
        self.send_response(HTTPStatus.PARTIAL_CONTENT)
        self.send_header("Content-Type", "video/mp4")
        self.send_header("Accept-Ranges", "bytes")
        self.send_header("Content-Range", "bytes %d-%d/%d" % (start, end, size))
        self.send_header("Content-Length", str(end - start + 1))
        self.end_headers()
        with open(fp, "rb") as f:
            f.seek(start)
            self.wfile.write(f.read(end - start + 1))

    def do_POST(self):
        u = urlparse(self.path)
        try:
            n = int(self.headers.get("Content-Length", 0))
            body = json.loads(self.rfile.read(min(n, 65536)) or b"{}")
        except Exception:
            return self.send_json({"error": "bad request"}, 400)
        pid = str(body.get("pid", "")).strip()
        if not PID_RE.match(pid):
            return self.send_json({"error": "ID must be 1-5 letters or digits"}, 400)
        key = pid.upper()
        if u.path == "/api/check":                # landing page: is this ID free / resumable / taken?
            with LOCK:
                p = load_participants().get(key)
            st = "new" if p is None else "taken" if is_taken(p) else "done" if p.get("done") else "resume"
            return self.send_json({"status": st})
        if u.path == "/api/start":
            with LOCK:
                ps = load_participants()
                if key in ps and is_taken(ps[key]):
                    return self.send_json({"error": "taken"}, 409)
                if key not in ps:
                    ps[key] = {"pid": pid, "created": time.time(), "ua": self.headers.get("User-Agent", "")[:200],
                               "questions": new_assignment(ps, random.Random()), "done": False}
                ps[key]["last_seen"] = time.time()
                save_participants(ps)
                p = ps[key]
            return self.send_json({"pid": p["pid"], "done": p["done"], "questions": public_questions(p)})
        if u.path == "/api/answer":
            try:
                qi = int(body["qi"])
                ans = body["answer"]
                ms = int(body.get("ms", 0))
            except Exception:
                return self.send_json({"error": "bad request"}, 400)
            with LOCK:
                ps = load_participants()
                if key not in ps or not 0 <= qi < len(ps[key]["questions"]):
                    return self.send_json({"error": "unknown participant or question"}, 400)
                q = ps[key]["questions"][qi]
                k = len(q["order"])
                if q["part"] == "part1":
                    ok = isinstance(ans, dict) and sorted(ans) == list(LABELS[:k]) and \
                        sorted(ans.values()) == list(range(1, k + 1))
                else:
                    ok = ans in ("A", "B", "tie")
                if not ok:
                    return self.send_json({"error": "invalid answer"}, 400)
                if is_taken(ps[key]):
                    return self.send_json({"error": "taken"}, 409)
                q["answer"], q["ms"], q["t"] = ans, ms, time.time()
                ps[key]["last_seen"] = q["t"]
                if q["part"] == "part1":
                    resolved = {q["order"][LABELS.index(l)]: r for l, r in ans.items()}
                else:
                    resolved = "tie" if ans == "tie" else q["order"][LABELS.index(ans)]
                ps[key]["done"] = all("answer" in x for x in ps[key]["questions"])
                save_participants(ps)
                with open(R_FILE, "a") as f:
                    f.write(json.dumps({"pid": key, "qi": qi, "part": q["part"], "scene": q["scene"],
                                        "order": q["order"], "answer": ans, "resolved": resolved, "ms": ms,
                                        "t": q["t"]}) + "\n")
                done = ps[key]["done"]
            return self.send_json({"ok": True, "done": done})
        return self.send_json({"error": "not found"}, 404)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--host", default="0.0.0.0")
    ap.add_argument("--port", type=int, default=int(os.environ.get("PORT", 8000)))
    a = ap.parse_args()
    hf_restore()
    threading.Thread(target=hf_backup_loop, daemon=True).start()
    print("part1 %d scenes, part2 %d scenes; %d + %d questions per participant; http://%s:%d" % (
        len(MAN["part1"]), len(MAN["part2"]), PART1_N, PART2_N, a.host, a.port), flush=True)
    ThreadingHTTPServer((a.host, a.port), Handler).serve_forever()


if __name__ == "__main__":
    main()