#!/usr/bin/env python3 from __future__ import annotations from io import BytesIO import csv, json, re, sys, zipfile import xml.etree.ElementTree as ET from collections import defaultdict LABELS = ["Species","Strain","Induction","Intervention","Comparator","Outcome"] COLS = { "Species":"species_spans", "Strain":"strain_spans", "Induction":"induction_spans", "Intervention":"intervention_spans", "Comparator":"comparator_spans", "Outcome":"outcome_spans", } CONF = {"HIGH","MEDIUM","LOW"} NS = "http://schemas.openxmlformats.org/spreadsheetml/2006/main" def norm(v): return "" if v is None else str(v).strip() def col_index(ref): letters = "".join(ch for ch in ref if ch.isalpha()).upper() n = 0 for ch in letters: n = n * 26 + ord(ch) - 64 return n - 1 def rows_from_xlsx(book_bytes): with zipfile.ZipFile(BytesIO(book_bytes)) as zf: shared = [] if "xl/sharedStrings.xml" in zf.namelist(): root = ET.fromstring(zf.read("xl/sharedStrings.xml")) for si in root.findall(f"{{{NS}}}si"): shared.append("".join((t.text or "") for t in si.iter(f"{{{NS}}}t"))) sheets = sorted(n for n in zf.namelist() if re.fullmatch(r"xl/worksheets/sheet\d+\.xml", n)) if not sheets: return [] root = ET.fromstring(zf.read(sheets[0])) rows = [] for row in root.iter(f"{{{NS}}}row"): vals, max_i = {}, -1 for c in row.findall(f"{{{NS}}}c"): idx = col_index(c.attrib.get("r", "A1")) max_i = max(max_i, idx) typ = c.attrib.get("t") if typ == "inlineStr": value = "".join((t.text or "") for t in c.iter(f"{{{NS}}}t")) else: v = c.find(f"{{{NS}}}v") raw = "" if v is None or v.text is None else v.text value = shared[int(raw)] if typ == "s" and raw else raw vals[idx] = value if max_i >= 0: rows.append([vals.get(i, "") for i in range(max_i + 1)]) return rows def dict_rows(rows): if not rows: return {}, [] headers = [norm(x).lower() for x in rows[0]] idx = {h:i for i,h in enumerate(headers)} out, dup = {}, [] for row in rows[1:]: if "case_id" not in idx or idx["case_id"] >= len(row): continue cid = norm(row[idx["case_id"]]).upper() if not cid: continue record = {h:(row[i] if i < len(row) else "") for h,i in idx.items()} if cid in out: dup.append(cid) else: out[cid] = record return out, dup def load_cases(path): out = {} with open(path, encoding="utf-8") as f: for line in f: if line.strip(): o = json.loads(line) out[str(o["case_id"]).upper()] = str(o["abstract_text"]) return out def load_gold(path): gold = defaultdict(lambda: defaultdict(set)) with open(path, encoding="utf-8", newline="") as f: for r in csv.DictReader(f): gold[r["case_id"].upper()][r["label"]].add((int(r["start_char"]), int(r["end_char"]))) return gold def parse_span_cell(raw, source_text): s = norm(raw) if not s: return set(), False, 0, 0 try: arr = json.loads(s) except Exception: return set(), False, 0, 0 if not isinstance(arr, list): return set(), False, 0, 0 spans, n_items, n_text_ok = set(), 0, 0 all_valid = True for item in arr: if not isinstance(item, dict): all_valid = False continue start, end = item.get("start"), item.get("end") if (isinstance(start,bool) or isinstance(end,bool) or not isinstance(start,int) or not isinstance(end,int) or not (0 <= start < end <= len(source_text))): all_valid = False continue n_items += 1 supplied = item.get("text") if isinstance(supplied, str) and source_text[start:end] == supplied: n_text_ok += 1 else: all_valid = False spans.add((start,end)) return spans, all_valid, n_items, n_text_ok def prf(tp, fp, fn): p = tp/(tp+fp) if tp+fp else (1.0 if fn == 0 else 0.0) r = tp/(tp+fn) if tp+fn else (1.0 if fp == 0 else 0.0) f = 2*p*r/(p+r) if p+r else 0.0 return p,r,f def score(candidate_bytes, gold_csv, case_jsonl): out, dup = dict_rows(rows_from_xlsx(candidate_bytes)) cases, gold = load_cases(case_jsonl), load_gold(gold_csv) global_tp = global_fp = global_fn = 0 label_counts = {lab:{"tp":0,"fp":0,"fn":0} for lab in LABELS} exact_docs = parseable = text_ok = conf_ok = note_ok = 0 span_items = span_text_ok = 0 details = [] for cid in sorted(cases): src = cases[cid] row = out.get(cid, {}) same_text = str(row.get("abstract_text","")) == src text_ok += int(same_text) case_tp = case_fp = case_fn = 0 parse_all, exact = True, same_text for lab in LABELS: pred, valid, n, n_ok = parse_span_cell(row.get(COLS[lab],""), src) parse_all = parse_all and valid span_items += n; span_text_ok += n_ok g = gold[cid][lab] tp, fp, fn = len(pred & g), len(pred - g), len(g - pred) global_tp += tp; global_fp += fp; global_fn += fn label_counts[lab]["tp"] += tp; label_counts[lab]["fp"] += fp; label_counts[lab]["fn"] += fn case_tp += tp; case_fp += fp; case_fn += fn if pred != g: exact = False parseable += int(parse_all) exact_docs += int(exact and parse_all) c_ok = norm(row.get("confidence")).upper() in CONF n_ok = len(norm(row.get("extraction_note")).split()) >= 5 conf_ok += int(c_ok); note_ok += int(n_ok) p,r,f1 = prf(case_tp,case_fp,case_fn) details.append({ "case_id":cid, "gold_entities":case_tp+case_fn, "predicted_entities":case_tp+case_fp, "tp":case_tp, "fp":case_fp, "fn":case_fn, "precision":round(p,6), "recall":round(r,6), "f1":round(f1,6), "exact_document_match":int(exact and parse_all), "all_span_cells_valid":int(parse_all), "input_text_integrity":int(same_text), "confidence_complete":int(c_ok), "note_complete":int(n_ok), }) n = len(cases) micro_p,micro_r,micro_f1 = prf(global_tp,global_fp,global_fn) per_label, label_f1s = {}, [] for lab in LABELS: c = label_counts[lab] p,r,f = prf(c["tp"],c["fp"],c["fn"]) per_label[lab] = {"precision":p,"recall":r,"f1":f,"tp":c["tp"],"fp":c["fp"],"fn":c["fn"]} label_f1s.append(f) macro = sum(label_f1s)/len(label_f1s) structural = min( parseable/n, text_ok/n, conf_ok/n, note_ok/n, 1.0 if span_items == 0 else span_text_ok/span_items, ) final_score = (0.60*macro + 0.40*micro_f1) * structural perfect_pass = exact_docs == n and structural == 1.0 and not dup return { "score":final_score, "perfect_pass":perfect_pass, "documents":n, "exact_documents":exact_docs, "exact_document_match_rate":exact_docs/n, "micro_span_precision":micro_p, "micro_span_recall":micro_r, "micro_span_f1":micro_f1, "macro_label_f1":macro, "structural_completeness":structural, "per_label":per_label, "duplicate_case_ids_in_output":dup, "non_exact_case_ids":[d["case_id"] for d in details if not d["exact_document_match"]], }, details def main(): if len(sys.argv) not in (4,6): raise SystemExit("Usage: python score_pico_submission.py OUTPUT.xlsx hidden_entities.csv case_texts.jsonl [--details results.csv]") with open(sys.argv[1],"rb") as f: candidate = f.read() summary, details = score(candidate, sys.argv[2], sys.argv[3]) print(json.dumps(summary, indent=2, sort_keys=True)) if len(sys.argv) == 6: if sys.argv[4] != "--details": raise SystemExit("Expected --details") with open(sys.argv[5],"w",encoding="utf-8",newline="") as f: w = csv.DictWriter(f, fieldnames=list(details[0].keys())) w.writeheader(); w.writerows(details) if __name__ == "__main__": main()