app.py

#!/usr/bin/env python3
"""Static server + ruling sink for human_labeling.

Serves index.html / queue.json and appends POSTed rulings to
annotations/rulings.jsonl. Stdlib only. The frontend feature-detects
POST /api/rulings: static hosts (github.io) 404 it and fall back to
file export; Spaces (Dockerfile) runs this file so writes land on disk.

    uv run python human_labeling/app.py [--port 7860]
"""

import argparse
import datetime
import http.server
import json
import socketserver
from pathlib import Path
HERE = Path(__file__).resolve().parent
ANNOT = HERE / "annotations"
RULINGS = ANNOT / "rulings.jsonl"
ANN_REPO = "rafmacalaba/data-use-annotations"
_hub_lines = None

class Handler(http.server.SimpleHTTPRequestHandler):
    def __init__(self, *a, **kw):
        super().__init__(*a, directory=str(HERE), **kw)

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

    def do_GET(self):
        if self.path == "/api/queue":
            items = []
            for name in ("queue_gliner.json",):
                p = HERE / name
                if p.exists():
                    items.extend(json.loads(r) for r in p.read_text().splitlines() if r.strip())
            return self._json({"items": items})
        if self.path == "/api/health":
            return self._json({"ok": True, "rulings": _count()})
        return super().do_GET()

    def do_POST(self):
        if self.path != "/api/rulings":
            return self._json({"error": "not found"}, 404)
        try:
            n = int(self.headers.get("Content-Length", 0))
        except ValueError:
            return self._json({"error": "bad length"}, 400)
        try:
            ruling = json.loads(self.rfile.read(n) or b"{}")
        except json.JSONDecodeError:
            return self._json({"error": "bad json"}, 400)
        if not ruling.get("key") or ruling.get("ruling") not in ("DATA_MENTION", "NON_MENTION"):
            return self._json({"error": "need key + ruling"}, 400)
        if not (ruling.get("annotator") or "").strip():
            return self._json({"error": "need annotator name"}, 400)
        ruling["annotator"] = ruling["annotator"].strip()
        ruling.setdefault("ts", datetime.datetime.now(datetime.timezone.utc).isoformat())
        ANNOT.mkdir(exist_ok=True)
        with open(RULINGS, "a") as f:
            f.write(json.dumps(ruling) + "\n")
        pushed = _push_hub(ruling)
        return self._json({"ok": True, "n": _count(), "pushed": pushed})


def _count() -> int:
    if not RULINGS.exists():
        return 0
    with open(RULINGS) as f:
        return sum(1 for line in f if line.strip())


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--port", type=int, default=7860)
    a = ap.parse_args()
    with socketserver.TCPServer(("", a.port), Handler) as httpd:
        print(f"human_labeling on :{a.port} (rulings -> {RULINGS})")
        httpd.serve_forever()
def _push_hub(ruling) -> bool:
    """Merge one ruling into the annotator's public Hub file
    (rulings/<annotator>.jsonl). Last write per (queue, key) wins.
    Local file stays source of truth; Hub failures never fail."""
    global _hub_lines
    try:
        from huggingface_hub import HfApi, hf_hub_download
        api = HfApi()
        who = "".join(c if c.isalnum() or c in "-_" else "_" for c in ruling["annotator"])
        path = f"rulings/{who}.jsonl"
        if _hub_lines is None:
            _hub_lines = {}
        if who not in _hub_lines:
            try:
                p = hf_hub_download(ANN_REPO, path, repo_type="dataset")
                with open(p) as f:
                    _hub_lines[who] = [json.loads(l) for l in f if l.strip()]
            except Exception:
                _hub_lines[who] = []
        rows = [r for r in _hub_lines[who]
                if not (r.get("queue") == ruling.get("queue")
                        and r.get("key") == ruling.get("key"))]
        rows.append(ruling)
        _hub_lines[who] = rows
        api.upload_file(
            path_or_fileobj="\n".join(json.dumps(r) for r in rows).encode(),
            path_in_repo=path, repo_id=ANN_REPO, repo_type="dataset",
            commit_message=f"ruling {who}: {ruling.get('key')}={ruling.get('ruling')}",
        )
        return True
    except Exception as exc:  # noqa: BLE001
        print(f"hub push failed: {exc}")
        return False


if __name__ == "__main__":
    main()