| """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 |
| PROJECT = BASE.parent |
| SCHEMA_PATH = Path(os.environ.get("SCHEMA_PATH", BASE / "schema" / "schema_0817.json")) |
| |
| SCHEMA_REPO_PATH = os.environ.get("SCHEMA_REPO_PATH", "") |
|
|
| |
|
|
| BACKEND = os.environ.get("STORAGE_BACKEND", "local") |
| ANNOTATORS = json.loads(os.environ.get("ANNOTATORS_JSON", "") or "{}") |
| |
| |
| REQUIRE_PASSWORD = env_flag("REQUIRE_PASSWORD", False) |
| AUTH = REQUIRE_PASSWORD and bool(ANNOTATORS) |
| ID_RE = re.compile(r"^[A-Za-z0-9_-]{1,32}$") |
| 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": |
| |
| |
| TOKEN = os.environ["GITHUB_TOKEN"] |
| |
| 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: |
| |
| 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]}") |
|
|
|
|
| |
|
|
| 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" |
| else: |
| ev["char_span"] = None |
| ev["grounded"] = False |
| return payload |
|
|
|
|
| |
|
|
| @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 |
|
|
|
|
| |
|
|
| @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: |
| |
| 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") |
|
|
|
|
| |
|
|
| @app.get("/api/state") |
| def get_state(paper: str, annotator: Optional[str] = None, me_: str = Depends(current_annotator)): |
| |
| 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 |
| 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} |
|
|
|
|
| |
|
|
| @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} |
|
|
|
|
| |
| |
| |
|
|
| 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: |
| 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"))) |
|
|
