Blame for app/main.py

"""Reproducibility 独立双标系统 — 后端(0925: 云部署 + GitHub 存储版)

本地跑:  cd 03_annotation/app && pip install -r requirements.txt && python main.py
         → http://localhost:8420 (默认 local 后端 + 开放模式, 行为同 0817 本地版)
云部署:  见 ../README.md「云部署」一节(Dockerfile 在 03_annotation/)

设计要点:
- schema 驱动: 表单结构全部来自 schema 文件(SCHEMA_PATH;0817 = ../schema/schema_0817.json,1103 起 v7 = ../schema/schema_v7.json), 改 schema 不改代码
- 盲标: 不加载任何 LLM 预填; 标注人只能读写自己的仓库, 看不到别人的; 每条标注盖 schema_version + annotator + 时间戳章
- grounding: evidence quote 保存时按 ground_check 同款算法(NFKC + 空白折叠)回填 char_span(折叠坐标)
- 存储: 见 store.py。一篇 × 一人一份 JSON, 内存即时生效, 后台定时 commit 到该标注人的私有仓库;
  events 留全程操作痕迹, 每份标注记 papers_ref(论文快照 commit), 坐标系可追溯
"""
import json
import os
import re
import secrets
import shutil
import subprocess
import sys
import tempfile
import threading
import time
import unicodedata
import uuid
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Optional

from fastapi import Depends, FastAPI, File, Form, HTTPException, Request, Response, UploadFile
from fastapi.responses import FileResponse, HTMLResponse, PlainTextResponse
from pydantic import BaseModel

import httpx

import auth
from store import (DocCache, GitHubAnnotationStore, GitHubPapers, LocalAnnotationStore,
                   LocalPapers, env_flag)

HERE = Path(__file__).resolve().parent
BASE = HERE.parent                      # 03_annotation/
PROJECT = BASE.parent                   # 1.Reproducibility/
SCHEMA_PATH = Path(os.environ.get("SCHEMA_PATH", BASE / "schema" / "schema_0817.json"))
# 1003: 公开部署(HF Space)时 schema 不进镜像, 启动时从私有数据仓库读: SCHEMA_REPO_PATH=config/schema_0817.json
SCHEMA_REPO_PATH = os.environ.get("SCHEMA_REPO_PATH", "")

# ---------- 配置 ----------

BACKEND = os.environ.get("STORAGE_BACKEND", "local")          # local | github
ANNOTATORS = json.loads(os.environ.get("ANNOTATORS_JSON", "") or "{}")
# 0925: 密码默认关(只填 ID 即可进), REQUIRE_PASSWORD=1 才校验 ANNOTATORS_JSON 里的 pw; Docker 镜像默认开。
# 配了 ANNOTATORS_JSON 时只允许名单里的 ID(防手误多出一个标注人); 没配(仅本地)则任填 ID。
REQUIRE_PASSWORD = env_flag("REQUIRE_PASSWORD", False)
AUTH = REQUIRE_PASSWORD and bool(ANNOTATORS)
ID_RE = re.compile(r"^[A-Za-z0-9_-]{1,32}$")                # ID 进存储路径, 只许安全字符
SESSION_SECRET = os.environ.get("SESSION_SECRET") or (secrets.token_hex(32) if not AUTH else "")
COOKIE_SECURE = env_flag("COOKIE_SECURE", BACKEND == "github")
FLUSH_SECONDS = float(os.environ.get("FLUSH_SECONDS", "15"))

if AUTH and not SESSION_SECRET:
    raise SystemExit("SESSION_SECRET is required when REQUIRE_PASSWORD=1")

if BACKEND == "github":
    # 默认一个仓库管所有: DATA_REPO 里 papers/ + annotations/<annotator>/;
    # 某人配了 "repo" 则写进那个仓库的 annotations/(每人一仓库的旧方案, 仍支持)
    TOKEN = os.environ["GITHUB_TOKEN"]
    # 1103: v7 标注写进 annotations_v7/(ANNOT_DIR),0817 的 annotations/ 原样保留;论文同理用 PAPERS_DIR=papers_v7
    ANNOT_DIR = os.environ.get("ANNOT_DIR", "annotations")
    DATA_REPO = os.environ.get("DATA_REPO", "")
    if not ANNOTATORS:
        raise SystemExit("github backend needs ANNOTATORS_JSON")
    locations = {}
    for a, c in ANNOTATORS.items():
        if c.get("repo"):
            locations[a] = (c["repo"], "annotations")
        elif DATA_REPO:
            locations[a] = (DATA_REPO, f"{ANNOT_DIR}/{a}")
        else:
            raise SystemExit(f"annotator {a}: set DATA_REPO or give them a repo")
    STORE = GitHubAnnotationStore(TOKEN, locations, branch=os.environ.get("DATA_BRANCH") or None)
    PAPERS = GitHubPapers(TOKEN, os.environ.get("PAPERS_REPO") or DATA_REPO,
                          os.environ.get("DATA_BRANCH") or "main",
                          cache_dir=Path(os.environ.get("CACHE_DIR", "/tmp/repl_cache")),
                          subdir=os.environ.get("PAPERS_DIR", "papers"))
else:
    STORE = LocalAnnotationStore(Path(os.environ.get("LOCAL_STORE_DIR", BASE / "data" / "store")))
    PAPERS = LocalPapers(Path(os.environ.get("PAPERS_ROOT", PROJECT)),
                         Path(os.environ.get("PAPERS_MANIFEST", BASE / "papers" / "manifest.json")))

CACHE = DocCache(STORE, flush_seconds=FLUSH_SECONDS)

if SCHEMA_REPO_PATH and BACKEND == "github":
    _r = httpx.get(f"https://api.github.com/repos/{os.environ.get('PAPERS_REPO') or DATA_REPO}/contents/{SCHEMA_REPO_PATH}",
                   params={"ref": os.environ.get("DATA_BRANCH") or "main"}, timeout=60,
                   headers={"Authorization": f"Bearer {TOKEN}", "Accept": "application/vnd.github.raw+json"})
    _r.raise_for_status()
    SCHEMA = _r.json()
else:
    SCHEMA = json.loads(SCHEMA_PATH.read_text(encoding="utf-8"))
SCHEMA_VERSION = SCHEMA["schema_version"]


@asynccontextmanager
async def lifespan(_app):
    CACHE.start()
    yield
    CACHE.stop()          # 关机前把没同步的改动写回


app = FastAPI(title="Reproducibility Annotator", lifespan=lifespan)


# ---------- 会话 ----------

def is_admin(annotator: str) -> bool:
    return ANNOTATORS.get(annotator, {}).get("role") == "admin"


def is_imported(annotator: str) -> bool:
    """role = imported: 导入的标注(不在本系统里标), 不能登录、不出现在登录页, 只供 admin 只读查看。
    是什么来源不经任何接口暴露(盲审要求), 来源记录另存, 不在数据仓库。"""
    return ANNOTATORS.get(annotator, {}).get("role") == "imported"


LOGIN_IDS = [a for a in ANNOTATORS if not is_imported(a)]


def current_annotator(request: Request) -> str:
    a = auth.read_session(SESSION_SECRET, request.cookies.get(auth.COOKIE))
    if not a or (ANNOTATORS and a not in ANNOTATORS) or is_imported(a):
        raise HTTPException(401, "not logged in")
    return a


def allowed_papers(annotator: str) -> dict:
    """admin 看全部; 配了 papers 的人只看那些; 其余人看仓库里的论文池 papers/pool.json(1003, 大家同一个池子)。"""
    papers = PAPERS.manifest()["papers"]
    if is_admin(annotator):
        return papers
    only = ANNOTATORS.get(annotator, {}).get("papers")
    if only is None:
        only = PAPERS.pool()
    if only is not None:
        papers = {k: v for k, v in papers.items() if k in set(only)}
    return papers


def paper_entry(annotator: str, pid: str) -> dict:
    papers = allowed_papers(annotator)
    if pid not in papers:
        raise HTTPException(404, f"unknown paper {pid}")
    return papers[pid]


# ---------- 标注文档 ----------

def load_doc(annotator: str, paper: str) -> dict:
    doc = CACHE.get(annotator, paper, PAPERS.ref)
    with CACHE.lock:
        seen = doc.setdefault("papers_refs_seen", [doc.get("papers_ref")])
        if PAPERS.ref not in seen:
            # 论文快照换了: char_span 坐标系可能变, 记一笔, 仲裁 / 导出时可查
            seen.append(PAPERS.ref)
            doc["events"].append({"ts": time.time(), "kind": "papers_ref_changed", "detail": PAPERS.ref})
            CACHE.mark_dirty(annotator, paper, f"papers_ref_changed {PAPERS.ref[:12]}")
    return doc


def log_event(doc: dict, annotator: str, paper: str, kind: str, detail: str = ""):
    doc["events"].append({"ts": time.time(), "kind": kind, "detail": detail[:500]})
    CACHE.mark_dirty(annotator, paper, f"{kind} {detail[:80]}")


# ---------- grounding (ground_check 同款) ----------

def fold(t: str) -> str:
    t = unicodedata.normalize("NFKC", t)
    return re.sub(r"\s+", " ", t).strip()


_FOLDED = {}


def folded_fulltext(annotator: str, pid: str) -> str:
    rel = paper_entry(annotator, pid)["fulltext"]
    if rel not in _FOLDED:
        _FOLDED[rel] = fold(PAPERS.text(rel))
    return _FOLDED[rel]


def relaxed_find(text: str, q: str):
    """markdown 容忍匹配: 渲染视图选出的 quote 不含标记符(** ` | #), 源文件里有。
    在 quote 每个字符间允许穿插标记符, 空格处允许穿插空白+标记符, 返回源文件折叠坐标。"""
    parts = []
    for ch in q:
        if ch == " ":
            parts.append(r"[\s|*_`#]+")
        else:
            parts.append(re.escape(ch) + r"[|*_`]*")
    m = re.compile("".join(parts)).search(text)
    return (m.start(), m.end()) if m else None


_LAYOUT = {}


def fold_with_map(raw: str):
    """逐字符版 fold, 同时记每个折叠字符对应的原文下标。只用来把 quote 落到 layout 块上;
    char_span 仍按 fold() 整串口径算, 两者在组合字符处可能差一两位, 不影响定位到哪一块。"""
    out, idx = [], []
    for i, ch in enumerate(raw):
        n = unicodedata.normalize("NFKC", ch)
        if not n.strip():
            if out and out[-1] != " ":
                out.append(" ")
                idx.append(i)
            continue
        for c in n:
            out.append(c)
            idx.append(i)
    if out and out[-1] == " ":
        out.pop()
        idx.pop()
    return "".join(out), idx


def paper_layout(annotator: str, pid: str):
    """layout.json(1003 起): 人读 md 的每块 → PDF (page, bbox)。没有 layout 的论文(旧 txt / 0817 md)返回 None。"""
    e = paper_entry(annotator, pid)
    rel = e.get("layout")
    if not rel:
        return None
    if rel not in _LAYOUT:
        lay = json.loads(PAPERS.text(rel))
        folded, idx = fold_with_map(PAPERS.text(e["fulltext"]))
        _LAYOUT[rel] = {"raw": lay, "folded": folded, "idx": idx}
    return _LAYOUT[rel]


def pdf_anchor_for(lay: dict, quote: str):
    """quote → 覆盖它的 layout 块 → 这些块在 PDF 上的框。找不到返回 None。"""
    text, q = lay["folded"], fold(quote)
    i = text.find(q)
    sp = (i, i + len(q)) if i >= 0 else relaxed_find(text, q)
    if not sp or sp[1] <= sp[0]:
        return None
    a, b = lay["idx"][sp[0]], lay["idx"][sp[1] - 1] + 1
    blocks = [blk for blk in lay["raw"]["blocks"] if blk["start"] < b and blk["end"] > a]
    boxes = [bx for blk in blocks for bx in blk["boxes"]]
    return {"blocks": [blk["id"] for blk in blocks], "boxes": boxes} if blocks else None


def ground_evidence(annotator: str, pid: str, payload: dict) -> dict:
    """对 payload.evidence 里每条 quote 回填折叠坐标 char_span; 找不到标 grounded=false 并降级提示。
    两层: ① NFKC+折叠后精确子串(与 ground_check 同口径) ② markdown 容忍匹配(md 语料, 0817 加)。
    有 layout 的论文另回填 pdf_anchor(块 id + PDF 框); 在 PDF 上直接划选的证据(origin=pdf)
    保留前端记下的精确框, 即使文本里对不上(转换损坏)也不丢位置。"""
    text = folded_fulltext(annotator, pid)
    lay = paper_layout(annotator, pid)
    for ev in payload.get("evidence", []):
        q = fold(ev.get("quote", ""))
        if lay and q and ev.get("origin") != "pdf":
            ev["pdf_anchor"] = pdf_anchor_for(lay, q)
        if not q:
            ev["grounded"] = False
            ev["char_span"] = None
            continue
        i = text.find(q)
        if i >= 0:
            ev["char_span"] = [i, i + len(q)]
            ev["grounded"] = True
            ev.pop("match", None)
        else:
            sp = relaxed_find(text, q)
            if sp:
                ev["char_span"] = [sp[0], sp[1]]
                ev["grounded"] = True
                ev["match"] = "relaxed"   # 命中含 markdown 标记的源文本
            else:
                ev["char_span"] = None
                ev["grounded"] = False   # 失配降级, 前端提示核对
    return payload


# ---------- API: 登录 ----------

@app.get("/", response_class=HTMLResponse)
def index():
    return (HERE / "static" / "index.html").read_text(encoding="utf-8")


@app.get("/api/config")
def get_config():
    return {"auth": AUTH, "backend": BACKEND, "schema_version": SCHEMA_VERSION,
            "annotators": [] if AUTH else LOGIN_IDS}


class Login(BaseModel):
    annotator: str
    password: str = ""


@app.post("/api/login")
def login(body: Login, response: Response):
    a = body.annotator.strip()
    if not ID_RE.match(a):
        raise HTTPException(400, "annotator ID: letters, digits, _ or - only")
    if ANNOTATORS and a not in ANNOTATORS:
        raise HTTPException(401, "unknown annotator ID")
    if is_imported(a):
        raise HTTPException(403, "imported annotator — cannot sign in")
    if AUTH and not auth.verify_password(body.password, ANNOTATORS[a].get("pw", "")):
        time.sleep(1.0)       # 放慢暴力尝试
        raise HTTPException(401, "wrong annotator ID or password")
    response.set_cookie(auth.COOKIE, auth.sign_session(SESSION_SECRET, a), max_age=auth.SESSION_TTL,
                        httponly=True, samesite="lax", secure=COOKIE_SECURE)
    return {"annotator": a, "admin": is_admin(a)}


@app.post("/api/logout")
def logout(response: Response):
    response.delete_cookie(auth.COOKIE)
    return {"ok": True}


@app.get("/api/me")
def me(annotator: str = Depends(current_annotator)):
    out = {"annotator": annotator, "admin": is_admin(annotator)}
    if out["admin"]:
        out["annotators"] = list(ANNOTATORS)
    return out


# ---------- API: schema / 论文 ----------

@app.get("/api/schema")
def get_schema(_a: str = Depends(current_annotator)):
    return SCHEMA


def clean_title(pid: str, entry: dict) -> str:
    """manifest 里旧的 title 是全文首行, MICCAI 是 "<!-- image -->"、openreview 有页码; publish_papers 新版已写干净的。"""
    t = (entry.get("title") or "").strip()
    t = re.sub(r"^#+\s*", "", t)
    if not t or t.startswith("<!--") or t.isdigit():
        return pid
    return t


@app.get("/api/papers")
def get_papers(annotator: str = Depends(current_annotator)):
    papers = {pid: {**e, "title": clean_title(pid, e)} for pid, e in allowed_papers(annotator).items()}
    out = {"papers": papers, "papers_ref": PAPERS.ref}
    if is_admin(annotator):
        out["pool"] = PAPERS.pool()
    return out


def filled(payload: dict) -> bool:
    # 外键 / 轴 / no_anchor 也算填了(F2、M7、D24、R3、D27 常常只有这些)
    return bool(payload.get("status") or payload.get("value") or payload.get("evidence")
                or payload.get("fk") or payload.get("axes") or payload.get("no_anchor"))


@app.get("/api/overview")
def overview(annotator: Optional[str] = None, me_: str = Depends(current_annotator)):
    """首页卡片用: 每篇论文 × 当前标注人(admin 可指定别人)按记录类型的记录数 / 已填字段数。
    只加载存储里已有的文档(一次 list), 没标过的论文不打 GitHub。"""
    who = annotator if (annotator and is_admin(me_)) else me_
    existing = {Path(n).stem for n in STORE.list(who)}
    with CACHE.lock:
        existing |= {p for (a, p) in CACHE.docs if a == who}
    out = {}
    for pid in allowed_papers(me_):
        blocks, total, last = {}, 0, None
        if pid in existing:
            doc = load_doc(who, pid)
            with CACHE.lock:
                for uid, r in doc["records"].items():
                    if r.get("deleted"):
                        continue
                    n = sum(1 for f in doc["fields"].get(uid, {}).values() if filled(f["payload"]))
                    b = blocks.setdefault(r["rtype"], {"records": 0, "filled": 0})
                    b["records"] += 1
                    b["filled"] += n
                    total += n
                last = max((e["ts"] for e in doc["events"]), default=None)
        out[pid] = {"blocks": blocks, "filled": total, "last_ts": last}
    return {"annotator": who, "papers": out}


@app.get("/api/paper/{pid}/text", response_class=PlainTextResponse)
def get_text(pid: str, annotator: str = Depends(current_annotator)):
    return PAPERS.text(paper_entry(annotator, pid)["fulltext"])


@app.get("/api/paper/{pid}/layout")
def get_layout(pid: str, annotator: str = Depends(current_annotator)):
    lay = paper_layout(annotator, pid)
    if not lay:
        raise HTTPException(404, "no layout for this paper")
    return lay["raw"]


@app.get("/api/paper/{pid}/pdf")
def get_pdf(pid: str, annotator: str = Depends(current_annotator)):
    entry = paper_entry(annotator, pid)
    if not entry.get("pdf"):
        raise HTTPException(404, "no pdf for this paper")
    return FileResponse(PAPERS.pdf_path(entry["pdf"]), media_type="application/pdf")


# ---------- API: 标注 ----------

@app.get("/api/state")
def get_state(paper: str, annotator: Optional[str] = None, me_: str = Depends(current_annotator)):
    # 标注人只能读自己的; admin 可以只读查看别人的(仲裁用)
    who = annotator if (annotator and is_admin(me_)) else me_
    paper_entry(me_, paper)
    doc = load_doc(who, paper)
    with CACHE.lock:
        recs = sorted(({"uid": uid, "rtype": r["rtype"], "extra": r.get("extra", {}),
                        "created_ts": r["created_ts"]}
                       for uid, r in doc["records"].items() if not r.get("deleted")),
                      key=lambda r: r["created_ts"])
        flds = {uid: {fid: f["payload"] for fid, f in m.items()} for uid, m in doc["fields"].items()}
    return {"records": recs, "fields": flds, "schema_version": SCHEMA_VERSION,
            "annotator": who, "read_only": who != me_}


class RecordOp(BaseModel):
    paper: str
    action: str            # create / delete / extra
    annotator: Optional[str] = None     # 旧前端兼容, 忽略; 身份以会话为准
    rtype: Optional[str] = None
    uid: Optional[str] = None
    extra: Optional[dict] = None


@app.post("/api/record")
def record_op(op: RecordOp, annotator: str = Depends(current_annotator)):
    paper_entry(annotator, op.paper)
    doc = load_doc(annotator, op.paper)
    now = time.time()
    with CACHE.lock:
        if op.action == "create":
            uid = str(uuid.uuid4())[:8]
            doc["records"][uid] = {"rtype": op.rtype, "extra": op.extra or {}, "deleted": 0,
                                   "created_ts": now, "updated_ts": now}
            log_event(doc, annotator, op.paper, "record_create", f"{op.rtype}:{uid}")
            return {"uid": uid}
        rec = doc["records"].get(op.uid or "")
        if rec is None:
            raise HTTPException(404, "unknown record")
        if op.action == "delete":
            rec["deleted"] = 1
            rec["updated_ts"] = now
            log_event(doc, annotator, op.paper, "record_delete", op.uid)
            return {"ok": True}
        if op.action == "extra":
            rec["extra"] = op.extra or {}
            rec["updated_ts"] = now
            log_event(doc, annotator, op.paper, "record_extra", op.uid)
            return {"ok": True}
    raise HTTPException(400, "bad action")


class FieldSave(BaseModel):
    paper: str
    record_uid: str
    field_id: str
    payload: dict
    annotator: Optional[str] = None     # 同上, 忽略


@app.post("/api/field")
def field_save(s: FieldSave, annotator: str = Depends(current_annotator)):
    payload = ground_evidence(annotator, s.paper, s.payload)
    doc = load_doc(annotator, s.paper)
    with CACHE.lock:
        if s.record_uid not in doc["records"]:
            raise HTTPException(404, "unknown record")
        doc["fields"].setdefault(s.record_uid, {})[s.field_id] = {
            "payload": payload, "schema_version": SCHEMA_VERSION, "updated_ts": time.time()}
        log_event(doc, annotator, s.paper, "field_save", f"{s.record_uid}.{s.field_id}")
    return {"payload": payload}


@app.get("/api/progress")
def progress(paper: str, annotator: str = Depends(current_annotator)):
    doc = load_doc(annotator, paper)
    with CACHE.lock:
        n = sum(len(m) for m in doc["fields"].values())
    return {"fields_touched": n}


@app.get("/api/sync")
def sync_status(annotator: str = Depends(current_annotator)):
    return {**CACHE.status(annotator), "backend": BACKEND}


@app.post("/api/sync")
def sync_now(annotator: str = Depends(current_annotator)):
    CACHE.flush()
    return {**CACHE.status(annotator), "backend": BACKEND}


# ---------- API: admin ----------

@app.get("/api/admin/export")
def admin_export(annotator: str = Depends(current_annotator)):
    """全部标注人 × 全部论文的原始文档(先把内存里的改动写回, 再从存储读)。"""
    if not is_admin(annotator):
        raise HTTPException(403, "admin only")
    CACHE.flush()
    people = list(ANNOTATORS) if ANNOTATORS else sorted(
        p.name for p in getattr(STORE, "root", Path("/nonexistent")).glob("*") if p.is_dir())
    out = {}
    for a in people:
        for name in STORE.list(a):
            data, _ = STORE.read(a, name)
            if data:
                out.setdefault(a, {})[Path(name).stem] = json.loads(data)
    return {"exported": time.time(), "papers_ref": PAPERS.ref, "schema_version": SCHEMA_VERSION,
            "annotations": out}


# ---------- API: admin 上传论文(1003) ----------
# 上传 PDF 和 / 或填 PMCID / DOI / PMID → 子进程跑 textconv.py(PMC 开放获取有 XML 就用 XML, 否则 Docling)
# → 一个 commit 写进 papers/<pid>/(只增不改) → 可选加入论文池。一次只转一篇(Docling 吃内存)。

CONVERT_PYTHON = os.environ.get("CONVERT_PYTHON", sys.executable)
MAX_UPLOAD_MB = float(os.environ.get("MAX_UPLOAD_MB", "60"))
JOBS: dict = {}
_CONVERT_LOCK = threading.Lock()


def _abstract(md: str) -> str:
    paras = [p.strip() for p in md.split("\n\n") if p.strip()]
    for i, p in enumerate(paras):
        if re.match(r"^#+\s*abstract\b", p, re.I) and i + 1 < len(paras):
            a = paras[i + 1]
            break
        m = re.match(r"^abstract[.:\s]+(.{80,})", p, re.I | re.S)
        if m:
            a = m.group(1)
            break
    else:
        a = next((p for p in paras if not p.startswith(("#", "|", "[")) and len(p) > 200), "")
    a = re.sub(r"\s+", " ", a)
    return a[:400].rsplit(" ", 1)[0] + " …" if len(a) > 400 else a


def _new_pid(meta: dict, filename: str) -> str:
    if meta.get("pmcid"):
        return f"PMC_{meta['pmcid']}"
    stem = re.sub(r"[^A-Za-z0-9]+", "-", Path(filename or "paper").stem).strip("-")[:40] or "paper"
    pid, n, have = f"UP_{stem}", 2, PAPERS.manifest()["papers"]
    while pid in have:
        pid, n = f"UP_{stem}-{n}", n + 1
    return pid


def _run_upload(jid: str, work: Path, pdf: Optional[Path], ident: str, title: str, venue: str,
                to_pool: bool, who: str, filename: str):
    job = JOBS[jid]
    try:
        with _CONVERT_LOCK:
            job["status"] = "running"
            out = work / "out"
            cmd = [CONVERT_PYTHON, str(HERE / "textconv.py"), "job", "--out", str(out)]
            cmd += ["--pdf", str(pdf)] if pdf else []
            cmd += ["--ident", ident] if ident else []
            proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
            for line in proc.stdout:
                line = line.rstrip()
                if line and not re.search(r"warn|it/s|^\s|^\[|INFO|DEBUG|torch|conv2d", line, re.I):
                    job["log"].append(line[:300])
            if proc.wait() != 0 or not (out / "meta.json").exists():
                err = next((l for l in reversed(job["log"]) if l.startswith("ERROR")), "conversion failed")
                raise RuntimeError(err.replace("ERROR: ", ""))
        meta = json.loads((out / "meta.json").read_text(encoding="utf-8"))
        pid = _new_pid(meta, filename)
        if pid in PAPERS.manifest()["papers"]:
            raise RuntimeError(f"{pid} is already in the system")
        md = (out / "fulltext.md").read_text(encoding="utf-8")
        files = {"fulltext.md": md.encode("utf-8"), "layout.json": (out / "layout.json").read_bytes(),
                 "paper.pdf": (out / "paper.pdf").read_bytes()}
        entry = {"fulltext": "fulltext.md", "pdf": "paper.pdf", "layout": "layout.json",
                 "venue": venue or ("PMC" if meta.get("pmcid") else "Upload"),
                 "title": title or meta.get("title") or Path(filename or pid).stem,
                 "abstract": _abstract(md), "text_source": meta["source"],
                 "pmcid": meta.get("pmcid"), "doi": meta.get("doi"), "license": meta.get("license"),
                 "uploaded_by": who, "uploaded_ts": time.time(), "upload_name": filename or None}
        job["log"].append("saving to the data repository…")
        PAPERS.add_paper(pid, files, entry, message=f"papers: add {pid} ({meta['source']}, uploaded by {who})")
        if to_pool:
            pool = PAPERS.pool() or []
            if pid not in pool:
                PAPERS.set_pool(pool + [pid], message=f"papers: add {pid} to the annotator pool")
        job.update(status="done", pid=pid, title=entry["title"], source=meta["source"])
        job["log"].append(f"done → {pid}" + (" (added to the pool)" if to_pool else ""))
    except Exception as e:  # noqa: BLE001
        job.update(status="error", error=str(e))
    finally:
        shutil.rmtree(work, ignore_errors=True)


def require_admin(annotator: str = Depends(current_annotator)) -> str:
    if not is_admin(annotator):
        raise HTTPException(403, "admin only")
    return annotator


@app.post("/api/admin/upload")
async def admin_upload(pdf: Optional[UploadFile] = File(None), ident: str = Form(""), title: str = Form(""),
                       venue: str = Form(""), add_to_pool: bool = Form(True), who: str = Depends(require_admin)):
    ident = ident.strip()
    if not pdf and not ident:
        raise HTTPException(400, "upload a PDF or give a PMCID / DOI / PMID")
    work = Path(tempfile.mkdtemp(prefix="upload_"))
    path, filename = None, ""
    if pdf:
        data = await pdf.read()
        if len(data) > MAX_UPLOAD_MB * 1e6:
            raise HTTPException(413, f"PDF larger than {MAX_UPLOAD_MB:.0f} MB")
        if not data.startswith(b"%PDF"):
            raise HTTPException(400, "that file is not a PDF")
        filename = Path(pdf.filename or "paper.pdf").name
        path = work / "upload.pdf"
        path.write_bytes(data)
    jid = uuid.uuid4().hex[:10]
    JOBS[jid] = {"id": jid, "status": "queued", "log": [], "file": filename, "ident": ident,
                 "by": who, "ts": time.time()}
    threading.Thread(target=_run_upload, daemon=True,
                     args=(jid, work, path, ident, title.strip(), venue.strip(), add_to_pool, who, filename)).start()
    return JOBS[jid]


@app.get("/api/admin/jobs")
def admin_jobs(_a: str = Depends(require_admin)):
    return {"jobs": sorted(JOBS.values(), key=lambda j: -j["ts"])[:30]}


class PoolOp(BaseModel):
    paper: str
    include: bool


@app.post("/api/admin/pool")
def admin_pool(op: PoolOp, who: str = Depends(require_admin)):
    if op.paper not in PAPERS.manifest()["papers"]:
        raise HTTPException(404, "unknown paper")
    pool = list(PAPERS.pool() or [])
    if op.include and op.paper not in pool:
        pool.append(op.paper)
    elif not op.include and op.paper in pool:
        pool.remove(op.paper)
    else:
        return {"pool": pool}
    PAPERS.set_pool(pool, message=f"papers: {'add' if op.include else 'remove'} {op.paper} "
                                  f"{'to' if op.include else 'from'} the annotator pool ({who})")
    return {"pool": pool}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get("PORT", "8420")))