#!/usr/bin/env python3 """ Extract civil society organisations (CSOs) from PDFs flagged YES/MAYBE in master_bibliography_disciplines_enriched.csv, using Gemini. Output is LONG format: one row per (paper, CSO). Columns: Title | Author | Discipline | Citations | Year | Where | When | Query | File Path | Name in text | Canonical name | Type | Mechanism | Norms/values/interests | Verbatim quote | Page `Where` and `When` are paper-level fields generated by Gemini from the PDF contents (not taken from the bibliography). They repeat across every row for a given paper. Resume-safe: PDFs already in the output CSV are skipped. Usage: python extract_csos.py python extract_csos.py --limit 20 python extract_csos.py --workers 5 python extract_csos.py --retry-errors python extract_csos.py --watch --poll 120 """ import argparse import csv import json import logging import os import re import signal import sys import threading import time from concurrent.futures import ThreadPoolExecutor, as_completed # ═══════════════════════════════════════════════════════════════════════════════ # CONFIGURATION # ═══════════════════════════════════════════════════════════════════════════════ GEMINI_API_KEY = "" # set via environment or paste your key GEMINI_MODEL = "gemini-2.5-flash" # Bibliography lives in the PhD project (shared source of truth) MASTER_CSV = ( '/Users/edekeulenaar/Projects/PhDs/PhD 2020-2025/Publications 📇/' 'Censorship and moderation/Analyses/' 'master_bibliography_disciplines_enriched.csv' ) # Output lives in the PoP project (this script's home) BASE_DIR = ('/Users/edekeulenaar/Projects/Postdocs/PoP/' 'PoP - Historical literature review') EXTRACT_CSV = os.path.join(BASE_DIR, 'cso_extraction_results.csv') # Additional PDF sources beyond master_bibliography_disciplines_enriched.csv. # Every PDF in these folders is treated as a YES paper (Discipline = "AI / model card"). EXTRA_PDF_DIRS = [ ('/Users/edekeulenaar/Projects/Postdocs/PoP/' 'PoP - Alignment actors/Model cards', 'AI / model card'), ] POLL_SECS = 60 # ═══════════════════════════════════════════════════════════════════════════════ # EXTRACTION PROMPT # ═══════════════════════════════════════════════════════════════════════════════ EXTRACTION_PROMPT = """\ You are a research assistant identifying CIVIL SOCIETY ORGANISATIONS (CSOs) mentioned in academic literature on censorship, moderation and AI alignment. DEFINITIONS - CSO: a non-state, non-corporate-commercial collective actor that participates in public life. The category is HISTORICALLY VARIABLE — many readings cover earlier periods (e.g. sixteenth-century England, early-modern France, nineteenth-century Germany, Cold-War USSR), and what counts as a "CSO" today may correspond there to guilds, confraternities, printers' companies, learned societies, parish associations, salons, reading societies, lodges, samizdat circles, etc. Include all such historically-specific civil or interest groups. Modern examples include: trade associations, professional bodies, special- interest / advocacy groups, religious associations, youth or student associations, NGOs, civic foundations, watchdog/transparency groups, community-run platforms or collectives, social movements with formal organisational form, trusted-flagger networks, fact-checker coalitions. Historical examples include: guilds and trade companies (e.g. Stationers' Company), religious orders and confraternities, learned/scientific societies, reading clubs, salons, masonic lodges, abolitionist or temperance societies, samizdat circles, dissident networks, parish councils. EXCLUDES: state agencies, regulators, courts, monarchs, individual censors acting in state capacity, for-profit platforms (Meta, Google, X, etc.), individual scholars, political parties when in government. If borderline (party-affiliated foundation, state-chartered guild, public broadcaster, established church acting as state organ), include it and explain in `notes`. - Moderation / censorship: interpreted broadly — content moderation on platforms, moderation of public debates, political moderation, self- regulation, gatekeeping, licensing, imprimatur, pre-publication review, index-making, takedowns, labelling, amplification/demotion, social shaming, norm enforcement. ALSO INCLUDES AI ALIGNMENT and model behaviour control: RLHF / RLAIF, constitutional AI, red-teaming, refusal training, safety fine-tuning, output filtering / classifier-based moderation of model responses, jailbreak mitigation, evals, model-card disclosure of restrictions, usage policies, deprecation of unsafe behaviours, watermarking of generated content. Treat these as forms of moderation/censorship of machine speech. DOCUMENT TYPES The PDF may be an academic article, book chapter, report, OR a MODEL CARD / system card / responsible-use document for an AI system (e.g. GPT-4 system card, Claude model card, Gemini technical report). For model cards, mine the document for any organisations cited as participating in the system's alignment, safety, evaluation, or deployment governance: external red-teaming firms, safety evaluators, civil-society reviewers, standards bodies, industry consortia (e.g. Frontier Model Forum, MLCommons, Partnership on AI), academic safety labs treated as collective actors (e.g. METR, Apollo Research, ARC Evals), and so on. Treat the publishing AI lab itself as a CSO ONLY if it is a non-profit (e.g. EleutherAI, the original OpenAI nonprofit); otherwise exclude commercial labs but include the third-party CSOs they engage. TASK First, summarise the PAPER as a whole: - paper_where: the SETTING the paper focuses on. This may be a country or polity ("England", "Holy Roman Empire", "USSR"), a place / region ("Bay Area", "colonial Bengal"), a PLATFORM or set of platforms ("Facebook", "Twitter and YouTube", "early bulletin board systems"), a MEDIA TYPE ("printed pamphlets", "broadcast television", "samizdat"), or an AI SYSTEM / MODEL FAMILY ("GPT-4", "Claude 3 family", "Llama 2"). Combine if relevant ("Reddit, USA"). Return "n/a" if there is no clear setting. - paper_when: the temporal scope the PAPER focuses on. Use a century ("16th century"), a period label ("early modern", "Weimar", "Cold War", "post-2010 platform era"), or a year range ("1789–1815"). If the paper is purely theoretical or has no clear temporal focus, return "n/a". Then, for each CSO substantively discussed in the PDF, extract: - name_in_text: verbatim, exactly as first written in the document - canonical_name: standard / full name if known (else same as name_in_text) - type: one of [trade association / guild, professional body, advocacy/special-interest group, religious association, youth/student association, NGO, foundation, watchdog/transparency, community/collective, social movement, learned/scientific society, salon/reading society, fraternal order/lodge, dissident/samizdat network, AI safety / evals organisation, AI industry consortium / standards body, external red-team / auditor, other] - mechanism: 1–3 sentence paraphrase of HOW this CSO participates in censorship, moderation, or AI alignment (lobbying, drafting codes of conduct, licensing presses, running flaggers / trusted-flagger programmes, operating own moderation, litigation, norm-setting, public pressure, capacity- building, maintaining indexes of forbidden books, social shaming, underground circulation; for AI: red-teaming a model, running pre-deployment evaluations, defining safety benchmarks, certifying compliance, drafting voluntary commitments, advising on refusal policies, etc.) - norms_values: if the text describes them, what NORMS, VALUES or INTERESTS the CSO defends (e.g. orthodoxy, professional honour, free speech, public morals, trade monopoly, national identity, workers' rights). Empty string if not described. - evidence_quote: short verbatim quote (≤40 words) supporting the mechanism - page: page number of the evidence quote, as a string RULES - Only include CSOs actually named in the text. Do not infer or add orgs the paper does not mention. - Prefer CSOs whose role is substantively discussed; skip orgs only listed in passing as examples. - One entry per CSO. If the same CSO appears multiple times, pick the most substantive passage for evidence_quote/page. - paper_where and paper_when refer to the WORLD DESCRIBED by the text, not the world the author is writing in. A 2024 article on sixteenth-century England → where "England", when "16th century". - If no CSOs are discussed, still return paper_where and paper_when, with organisations: []. Return a SINGLE JSON object wrapped in ```json fences: ```json { "paper_where": "...", "paper_when": "...", "organisations": [ { "name_in_text": "...", "canonical_name": "...", "type": "...", "mechanism": "...", "norms_values": "...", "country_guess": "...", "epoch_guess": "...", "evidence_quote": "...", "page": "N" } ] } ``` """ EXTRACT_FIELDS = [ 'Title', 'Author', 'Discipline', 'Citations', 'Year', 'Where', 'When', 'Query', 'File Path', 'Name in text', 'Canonical name', 'Type', 'Mechanism', 'Norms/values/interests', 'Verbatim quote', 'Page', ] # ═══════════════════════════════════════════════════════════════════════════════ # LOGGING # ═══════════════════════════════════════════════════════════════════════════════ logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s', force=True, handlers=[logging.StreamHandler(sys.stdout)], ) log = logging.getLogger(__name__) # ═══════════════════════════════════════════════════════════════════════════════ # BIBLIOGRAPHY HELPERS # ═══════════════════════════════════════════════════════════════════════════════ def load_master_rows() -> list[dict]: rows: list[dict] = [] if not os.path.exists(MASTER_CSV): log.warning(f'master CSV not found at {MASTER_CSV}') return rows with open(MASTER_CSV, newline='', encoding='utf-8') as f: for row in csv.DictReader(f): verdict = (row.get('Relevant') or '').strip().upper() if verdict in ('YES', 'MAYBE'): rows.append(row) return rows def load_extra_pdf_rows() -> list[dict]: """Build synthetic bibliography rows for PDFs in EXTRA_PDF_DIRS. Each PDF gets a row with Relevant=YES, Title=filename (stem), and the Discipline label from the (dir, label) pair. Local PDF Path points at the file so the rest of the pipeline treats it like any other entry. """ rows: list[dict] = [] for folder, label in EXTRA_PDF_DIRS: if not os.path.isdir(folder): log.warning(f'extra PDF dir not found: {folder}') continue for fname in sorted(os.listdir(folder)): if not fname.lower().endswith('.pdf'): continue path = os.path.join(folder, fname) rows.append({ 'Title': os.path.splitext(fname)[0], 'Author': '', 'Publication Year': '', 'Discipline': label, 'Citations': '', 'Query': '', 'Local PDF Path': path, 'Relevant': 'YES', }) return rows def _get_pdf_path(row: dict) -> str | None: p = (row.get('Local PDF Path') or '').strip() if p and os.path.isfile(p): return os.path.normpath(p) attachments = (row.get('File Attachments') or '').strip() for att in re.split(r'[;,]', attachments): att = att.strip() if att.lower().endswith('.pdf') and os.path.isfile(att): return os.path.normpath(att) return None def _row_to_paper_meta(row: dict) -> dict: year = (row.get('Publication Year') or row.get('Scholar Year') or '').strip() try: year = str(int(float(year))) except (ValueError, TypeError): pass return { 'Title': (row.get('Title') or '').strip(), 'Author': (row.get('Author') or '').strip(), 'Discipline': (row.get('Discipline') or '').strip(), 'Citations': (row.get('Citations') or '').strip(), 'Year': year, 'Query': (row.get('Query') or '').strip(), } # ═══════════════════════════════════════════════════════════════════════════════ # JSON UTILITIES # ═══════════════════════════════════════════════════════════════════════════════ def parse_json_response(text: str) -> dict | None: m = re.search(r'```(?:json)?\s*([\s\S]*?)\s*```', text) if m: try: return json.loads(m.group(1)) except json.JSONDecodeError: pass try: return json.loads(text) except json.JSONDecodeError: return None # ═══════════════════════════════════════════════════════════════════════════════ # GEMINI EXTRACTOR # ═══════════════════════════════════════════════════════════════════════════════ def _ascii_display_name(fname: str) -> str: import unicodedata nfkd = unicodedata.normalize('NFKD', fname) ascii_name = nfkd.encode('ascii', errors='ignore').decode('ascii') ascii_name = re.sub(r'\s+', ' ', ascii_name).strip() return ascii_name or 'document.pdf' def extract_with_gemini(pdf_path: str, model_name: str, max_retries: int = 4) -> str: from google import genai from google.genai import types as genai_types client = genai.Client(api_key=GEMINI_API_KEY) fname = os.path.basename(pdf_path) safe_name = _ascii_display_name(fname) log.info(f' [Gemini] Uploading {fname!r}…') for attempt in range(1, max_retries + 1): uploaded = None try: with open(pdf_path, 'rb') as _fh: uploaded = client.files.upload( file=_fh, config=genai_types.UploadFileConfig( mime_type='application/pdf', display_name=safe_name, ), ) while uploaded.state == genai_types.FileState.PROCESSING: time.sleep(4) uploaded = client.files.get(name=uploaded.name) if uploaded.state == genai_types.FileState.FAILED: raise RuntimeError('Gemini file processing failed') response = client.models.generate_content( model=model_name, contents=[uploaded, EXTRACTION_PROMPT], ) return response.text except Exception as e: err = str(e) if uploaded is not None: try: client.files.delete(name=uploaded.name) except Exception: pass transient = ( 'server disconnected' in err.lower() or 'connection' in err.lower() or '429' in err or '500' in err or '503' in err or 'quota' in err.lower() or 'rate' in err.lower() or 'timeout' in err.lower() ) if transient and attempt < max_retries: wait = 30 * attempt log.warning(f' [Gemini] transient error ' f'(attempt {attempt}/{max_retries}) — ' f'{err[:80]} — retrying in {wait}s…') time.sleep(wait) else: raise raise RuntimeError('Max retries exceeded') # ═══════════════════════════════════════════════════════════════════════════════ # PROCESSED-SET HELPERS # ═══════════════════════════════════════════════════════════════════════════════ def load_processed(retry_errors: bool = False) -> set[str]: """Return set of normalised PDF paths already extracted. Because output is long-format (multiple rows per PDF), we identify "already processed" by File Path + at least one row whose Raw JSON is not an error. """ seen: set[str] = set() if not os.path.exists(EXTRACT_CSV): return seen # utf-8-sig: strip BOM if Numbers/Excel re-saved the file. with open(EXTRACT_CSV, newline='', encoding='utf-8-sig') as f: sample = f.read(8192); f.seek(0) try: dialect = csv.Sniffer().sniff(sample, delimiters=',;\t') except csv.Error: dialect = csv.excel all_rows = list(csv.DictReader(f, dialect=dialect)) # Group rows by File Path by_path: dict[str, list[dict]] = {} for r in all_rows: p = os.path.normpath((r.get('File Path') or '').strip()) if r.get('File Path', '').strip() else '' if p: by_path.setdefault(p, []).append(r) def _is_error_row(r: dict) -> bool: # A failed extraction produces a single placeholder row with no # CSO name and no paper-level Where/When (we only emit empty # Where/When on error; "no CSOs found" still has them filled). return ( not (r.get('Name in text') or '').strip() and not (r.get('Where') or '').strip() and not (r.get('When') or '').strip() ) error_paths: set[str] = set() for p, rows in by_path.items(): all_errors = all(_is_error_row(r) for r in rows) if retry_errors and all_errors: error_paths.add(p) else: seen.add(p) if retry_errors and error_paths: good_rows = [ r for r in all_rows if os.path.normpath((r.get('File Path') or '').strip()) not in error_paths ] with open(EXTRACT_CSV, 'w', newline='', encoding='utf-8') as f: w = csv.DictWriter(f, fieldnames=EXTRACT_FIELDS, extrasaction='ignore') w.writeheader() w.writerows(good_rows) log.info(f'--retry-errors: removed {len(error_paths)} failed PDF(s); ' f'they will be re-processed') return seen # ═══════════════════════════════════════════════════════════════════════════════ # MAIN # ═══════════════════════════════════════════════════════════════════════════════ def main(): parser = argparse.ArgumentParser( description='Extract CSOs from YES/MAYBE PDFs using Gemini.') parser.add_argument('--limit', type=int, default=None, metavar='N') parser.add_argument('--start-from', type=int, default=0, metavar='N') parser.add_argument('--model', default=GEMINI_MODEL) parser.add_argument('--watch', action='store_true') parser.add_argument('--poll', type=int, default=POLL_SECS, metavar='SECS') parser.add_argument('--retry-errors', action='store_true') parser.add_argument('--workers', type=int, default=3, metavar='N') args = parser.parse_args() os.makedirs(BASE_DIR, exist_ok=True) processed: set[str] = load_processed(retry_errors=args.retry_errors) processed_lock = threading.Lock() log.info(f'Already extracted: {len(processed):,} PDFs') csv_lock = threading.Lock() if not os.path.exists(EXTRACT_CSV) or os.path.getsize(EXTRACT_CSV) == 0: with open(EXTRACT_CSV, 'w', newline='', encoding='utf-8') as f: csv.DictWriter(f, fieldnames=EXTRACT_FIELDS, extrasaction='ignore').writeheader() def write_rows(rows: list[dict]) -> None: if not rows: return with csv_lock: with open(EXTRACT_CSV, 'a', newline='', encoding='utf-8') as f: w = csv.DictWriter(f, fieldnames=EXTRACT_FIELDS, extrasaction='ignore') for r in rows: w.writerow(r) def _process_one(task, task_n: int, total: int) -> list[dict]: pdf_path, paper_meta = task fname = os.path.basename(pdf_path) title_display = (paper_meta.get('Title') or fname)[:70] log.info(f'[{task_n}/{total}] {title_display}') raw = '' data: dict = {} try: raw = extract_with_gemini(pdf_path, args.model) data = parse_json_response(raw) or {} if not data: log.warning(f' [{fname[:40]}] Could not parse JSON') except Exception as e: log.error(f' [{fname[:40]}] Error: {e}') raw = f'Error: {e}' data = {} orgs = data.get('organisations') or [] base = { **paper_meta, 'Where': (data.get('paper_where') or '').strip() if isinstance(data, dict) else '', 'When': (data.get('paper_when') or '').strip() if isinstance(data, dict) else '', 'File Path': pdf_path, } if not orgs: # Emit a single placeholder row so we know the PDF was processed. # On error: Where/When stay empty so --retry-errors can find it. # On success-with-no-CSOs: Where/When are populated. return [{ **base, 'Name in text': '', 'Canonical name': '', 'Type': '', 'Mechanism': '', 'Norms/values/interests': '', 'Verbatim quote': '', 'Page': '', }] out = [] for org in orgs: if not isinstance(org, dict): continue out.append({ **base, 'Name in text': org.get('name_in_text', ''), 'Canonical name': org.get('canonical_name', ''), 'Type': org.get('type', ''), 'Mechanism': org.get('mechanism', ''), 'Norms/values/interests': org.get('norms_values', ''), 'Verbatim quote': org.get('evidence_quote', ''), 'Page': str(org.get('page', '')), }) return out _shutdown = threading.Event() def _handle_interrupt(sig, frame): log.info('\nInterrupted — waiting for in-flight workers to finish…') _shutdown.set() signal.signal(signal.SIGINT, _handle_interrupt) signal.signal(signal.SIGTERM, _handle_interrupt) iteration = 0 completed = 0 while not _shutdown.is_set(): iteration += 1 master_rows = load_master_rows() + load_extra_pdf_rows() todo: list[tuple[str, dict]] = [] no_pdf_count = 0 with processed_lock: for row in master_rows: pdf_path = _get_pdf_path(row) if not pdf_path: no_pdf_count += 1 continue if pdf_path in processed: continue todo.append((pdf_path, _row_to_paper_meta(row))) if iteration == 1 or todo: log.info(f'YES/MAYBE rows: {len(master_rows):,}') log.info(f' with local PDF: {len(master_rows) - no_pdf_count:,}') log.info(f' without local PDF: {no_pdf_count:,}') log.info(f' already extracted: {len(processed):,}') log.info(f' to process now: {len(todo):,}') if iteration == 1: if args.start_from: todo = todo[args.start_from:] if args.limit is not None: todo = todo[:args.limit] if not todo: if not args.watch: log.info('Nothing to do.') break log.info(f'No new PDFs — sleeping {args.poll}s…') time.sleep(args.poll) continue total = len(todo) log.info(f'Starting {total} extractions with {args.workers} workers…') with ThreadPoolExecutor(max_workers=args.workers) as executor: futures = { executor.submit(_process_one, task, i, total): task for i, task in enumerate(todo, 1) } for future in as_completed(futures): if _shutdown.is_set(): for f in futures: f.cancel() break try: rows = future.result() except Exception as e: pdf_path, paper_meta = futures[future] log.error(f' Unhandled error for {os.path.basename(pdf_path)}: {e}') rows = [{ **paper_meta, 'Where': '', 'When': '', 'File Path': pdf_path, **{k: '' for k in ( 'Name in text','Canonical name','Type','Mechanism', 'Norms/values/interests','Verbatim quote','Page')}, }] write_rows(rows) with processed_lock: if rows: processed.add(rows[0]['File Path']) completed += 1 ok = rows and bool( (rows[0].get('Where') or '').strip() or (rows[0].get('When') or '').strip() or (rows[0].get('Name in text') or '').strip() ) status = '✓' if ok else '✗' title = rows[0].get('Title', '')[:55] if rows else '' n_orgs = sum(1 for r in rows if r.get('Name in text')) log.info(f' {status} done ({completed} total, {n_orgs} CSOs) — {title}') if _shutdown.is_set() or not args.watch: break log.info(f'Batch done — sleeping {args.poll}s…') time.sleep(args.poll) log.info(f'\nDone — {completed} PDFs processed → {EXTRACT_CSV}') if __name__ == '__main__': main()