#!/usr/bin/env python3 """ doi2ris.py — Konvertiert eine Liste von Literaturquellen in eine RIS-Datei für Zotero. Funktionsweise: 1. Liest eine Eingabedatei (eine Quelle pro Zeile; freier Text, BibTeX-Schnipsel, URLs oder reine DOIs sind erlaubt). 2. Extrahiert DOIs per Regex. Zeilen ohne DOI werden über die Crossref-Titelsuche aufgelöst (erster Treffer mit hohem Score). 3. Lädt bibliographische Metadaten von Crossref (Fallback: DataCite). 4. Schreibt eine Zotero-kompatible .ris-Datei. Abhängigkeiten: nur `requests` (sonst Standardbibliothek). Beispiel: python doi2ris.py quellen.txt -o bibliothek.ris --email you@example.com """ from __future__ import annotations import argparse import json import logging import re import sys import time from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass, field from pathlib import Path from typing import Any, Iterable, Optional import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry # --------------------------------------------------------------------------- # Konstanten # --------------------------------------------------------------------------- DOI_REGEX = re.compile(r"\b10\.\d{4,9}/[^\s,;\]<>\"']+", re.IGNORECASE) # Häufige Trailing-Zeichen, die Regex sonst mit einsaugt (z. B. ".", ")", "}") TRAILING_PUNCT = ".,);:]}>\"'" CROSSREF_WORKS = "https://api.crossref.org/works/{doi}" CROSSREF_QUERY = "https://api.crossref.org/works" DATACITE_DOI = "https://api.datacite.org/dois/{doi}" # Crossref-Typ → RIS-Typ TYPE_MAP = { "journal-article": "JOUR", "book": "BOOK", "book-chapter": "CHAP", "book-section": "CHAP", "book-part": "CHAP", "monograph": "BOOK", "edited-book": "BOOK", "reference-book": "BOOK", "proceedings-article": "CPAPER", "proceedings": "CONF", "conference-paper": "CPAPER", "report": "RPRT", "report-component": "RPRT", "dissertation": "THES", "thesis": "THES", "dataset": "DATA", "posted-content": "GEN", # u. a. Preprints "preprint": "GEN", "standard": "STAND", "reference-entry": "DICT", "component": "GEN", "other": "GEN", } # --------------------------------------------------------------------------- # Datenklasse # --------------------------------------------------------------------------- @dataclass class Record: """Ein bibliographischer Eintrag im neutralen Zwischenformat.""" raw_input: str doi: Optional[str] = None ris_type: str = "GEN" title: str = "" authors: list[str] = field(default_factory=list) editors: list[str] = field(default_factory=list) journal: str = "" book_title: str = "" publisher: str = "" publisher_place: str = "" volume: str = "" issue: str = "" page_start: str = "" page_end: str = "" year: str = "" date_full: str = "" # YYYY/MM/DD abstract: str = "" keywords: list[str] = field(default_factory=list) url: str = "" issn: list[str] = field(default_factory=list) isbn: list[str] = field(default_factory=list) language: str = "" source_api: str = "" error: str = "" # --------------------------------------------------------------------------- # HTTP-Setup mit Retries # --------------------------------------------------------------------------- def build_session(email: str) -> requests.Session: s = requests.Session() retry = Retry( total=5, backoff_factor=1.5, status_forcelist=(429, 500, 502, 503, 504), allowed_methods=("GET",), respect_retry_after_header=True, ) s.mount("https://", HTTPAdapter(max_retries=retry, pool_connections=20, pool_maxsize=20)) ua = f"doi2ris/1.0 (mailto:{email})" if email else "doi2ris/1.0" s.headers.update({"User-Agent": ua, "Accept": "application/json"}) return s # --------------------------------------------------------------------------- # DOI-Extraktion und Eingabeparsing # --------------------------------------------------------------------------- def extract_doi(line: str) -> Optional[str]: """Findet die erste DOI in einer Zeile und säubert Trailing-Zeichen.""" m = DOI_REGEX.search(line) if not m: return None doi = m.group(0) # Trailing-Punktuation wegschneiden, aber Klammerpaare ausbalanciert lassen while doi and doi[-1] in TRAILING_PUNCT: # Schließende Klammer nur entfernen, wenn keine öffnende vorhanden ist if doi[-1] == ")" and doi.count("(") >= doi.count(")"): break doi = doi[:-1] return doi.lower() def read_input(path: Path) -> list[str]: """Liest die Eingabedatei und liefert nicht-leere, getrimmte Zeilen.""" raw = path.read_text(encoding="utf-8", errors="replace") lines = [ln.strip() for ln in raw.splitlines() if ln.strip() and not ln.strip().startswith("#")] return lines # --------------------------------------------------------------------------- # Cache # --------------------------------------------------------------------------- class Cache: def __init__(self, path: Optional[Path]): self.path = path self.data: dict[str, dict] = {} if path and path.exists(): try: self.data = json.loads(path.read_text(encoding="utf-8")) except json.JSONDecodeError: logging.warning("Cache-Datei beschädigt, starte mit leerem Cache.") self.data = {} def get(self, key: str) -> Optional[dict]: return self.data.get(key) def set(self, key: str, value: dict) -> None: self.data[key] = value def save(self) -> None: if self.path: self.path.write_text(json.dumps(self.data, ensure_ascii=False, indent=2), encoding="utf-8") # --------------------------------------------------------------------------- # Metadaten-Fetcher # --------------------------------------------------------------------------- def fetch_crossref(doi: str, session: requests.Session) -> Optional[dict]: r = session.get(CROSSREF_WORKS.format(doi=requests.utils.quote(doi, safe="/")), timeout=30) if r.status_code == 404: return None r.raise_for_status() return r.json().get("message") def fetch_datacite(doi: str, session: requests.Session) -> Optional[dict]: r = session.get(DATACITE_DOI.format(doi=requests.utils.quote(doi, safe="/")), timeout=30) if r.status_code == 404: return None r.raise_for_status() return r.json().get("data", {}).get("attributes") def crossref_search_by_title(query: str, session: requests.Session) -> Optional[str]: """Letzter Rettungsanker: Crossref-Titelsuche.""" params = {"query.bibliographic": query, "rows": 3} r = session.get(CROSSREF_QUERY, params=params, timeout=30) if not r.ok: return None items = r.json().get("message", {}).get("items", []) if not items: return None # Crossref liefert Treffer nach Relevanz sortiert. top = items[0] score = top.get("score", 0) if score < 40: # erfahrungsgemäß sind Treffer < 40 unzuverlässig return None return top.get("DOI", "").lower() or None # --------------------------------------------------------------------------- # Parser: Crossref / DataCite → Record # --------------------------------------------------------------------------- def _clean_text(s: Optional[str]) -> str: if not s: return "" # JATS-Tags und überflüssige Whitespaces entfernen s = re.sub(r"<[^>]+>", "", s) s = re.sub(r"\s+", " ", s).strip() return s def _name(item: dict) -> str: fam = item.get("family", "").strip() giv = item.get("given", "").strip() if fam and giv: return f"{fam}, {giv}" return fam or giv or item.get("name", "").strip() def _date_parts(d: Optional[dict]) -> tuple[str, str]: """(jahr, vollständiges Datum YYYY/MM/DD//) aus Crossref date-parts.""" if not d: return "", "" parts = (d.get("date-parts") or [[]])[0] if not parts: return "", "" year = str(parts[0]) month = f"{parts[1]:02d}" if len(parts) > 1 else "" day = f"{parts[2]:02d}" if len(parts) > 2 else "" ris_date = f"{year}/{month}/{day}/" # RIS-Format YYYY/MM/DD/other return year, ris_date def parse_crossref(rec: Record, msg: dict) -> Record: rec.source_api = "crossref" cr_type = msg.get("type", "other") rec.ris_type = TYPE_MAP.get(cr_type, "GEN") titles = msg.get("title") or [] rec.title = _clean_text(titles[0]) if titles else "" rec.authors = [_name(a) for a in msg.get("author", []) if _name(a)] rec.editors = [_name(a) for a in msg.get("editor", []) if _name(a)] container = msg.get("container-title") or [] container_main = _clean_text(container[0]) if container else "" if rec.ris_type in ("CHAP", "CPAPER", "CONF"): rec.book_title = container_main else: rec.journal = container_main rec.publisher = _clean_text(msg.get("publisher", "")) rec.publisher_place = _clean_text(msg.get("publisher-location", "")) rec.volume = str(msg.get("volume", "") or "") rec.issue = str(msg.get("issue", "") or "") page = msg.get("page", "") or "" if page: parts = re.split(r"[-–]", page, maxsplit=1) rec.page_start = parts[0].strip() rec.page_end = parts[1].strip() if len(parts) > 1 else "" # Datum: published > published-print > published-online > issued > created for key in ("published", "published-print", "published-online", "issued", "created"): if key in msg: y, d = _date_parts(msg[key]) if y: rec.year, rec.date_full = y, d break rec.abstract = _clean_text(msg.get("abstract", "")) subjects = msg.get("subject") or [] rec.keywords = [s for s in (s.strip() for s in subjects) if s] rec.url = msg.get("URL", "") or (f"https://doi.org/{rec.doi}" if rec.doi else "") rec.issn = list(msg.get("ISSN", []) or []) rec.isbn = list(msg.get("ISBN", []) or []) rec.language = msg.get("language", "") or "" return rec def parse_datacite(rec: Record, attrs: dict) -> Record: rec.source_api = "datacite" types = attrs.get("types", {}) or {} resource_type = (types.get("resourceTypeGeneral") or "").lower() mapping = { "dataset": "DATA", "text": "GEN", "journalarticle": "JOUR", "book": "BOOK", "bookchapter": "CHAP", "conferencepaper": "CPAPER", "report": "RPRT", "dissertation": "THES", "software": "COMP", "preprint": "GEN", } rec.ris_type = mapping.get(resource_type.replace(" ", "").lower(), "GEN") titles = attrs.get("titles") or [] rec.title = _clean_text(titles[0].get("title")) if titles else "" creators = attrs.get("creators") or [] rec.authors = [] for c in creators: if c.get("familyName") and c.get("givenName"): rec.authors.append(f"{c['familyName']}, {c['givenName']}") elif c.get("name"): rec.authors.append(c["name"]) rec.publisher = _clean_text(attrs.get("publisher", "")) year = attrs.get("publicationYear") if year: rec.year = str(year) rec.date_full = f"{year}///" descs = attrs.get("descriptions") or [] if descs: rec.abstract = _clean_text(descs[0].get("description", "")) subjects = attrs.get("subjects") or [] rec.keywords = [s["subject"] for s in subjects if s.get("subject")] rec.url = attrs.get("url") or (f"https://doi.org/{rec.doi}" if rec.doi else "") rec.language = attrs.get("language", "") or "" return rec # --------------------------------------------------------------------------- # Worker # --------------------------------------------------------------------------- def resolve_record(raw_line: str, session: requests.Session, cache: Cache) -> Record: rec = Record(raw_input=raw_line) doi = extract_doi(raw_line) if not doi: # Versuch: Titelsuche candidate = crossref_search_by_title(raw_line, session) if candidate: doi = candidate else: rec.error = "Keine DOI gefunden und Titelsuche ohne Treffer" return rec rec.doi = doi cached = cache.get(doi) if cached: try: if cached.get("_api") == "crossref": return parse_crossref(rec, cached["data"]) if cached.get("_api") == "datacite": return parse_datacite(rec, cached["data"]) except Exception as e: # pragma: no cover logging.debug("Cache-Eintrag für %s unbrauchbar: %s", doi, e) # Crossref zuerst try: msg = fetch_crossref(doi, session) if msg: cache.set(doi, {"_api": "crossref", "data": msg}) return parse_crossref(rec, msg) except requests.RequestException as e: logging.warning("Crossref-Fehler für %s: %s", doi, e) # DataCite-Fallback try: attrs = fetch_datacite(doi, session) if attrs: cache.set(doi, {"_api": "datacite", "data": attrs}) return parse_datacite(rec, attrs) except requests.RequestException as e: logging.warning("DataCite-Fehler für %s: %s", doi, e) rec.error = "Keine Metadaten bei Crossref oder DataCite gefunden" return rec # --------------------------------------------------------------------------- # RIS-Writer # --------------------------------------------------------------------------- def ris_lines(rec: Record) -> list[str]: """Erzeugt die RIS-Zeilen für einen Datensatz (ohne Zeilentrenner).""" lines: list[str] = [] add = lines.append add(f"TY - {rec.ris_type}") for a in rec.authors: add(f"AU - {a}") for e in rec.editors: add(f"ED - {e}") if rec.title: add(f"TI - {rec.title}") if rec.journal: add(f"T2 - {rec.journal}") add(f"JO - {rec.journal}") if rec.book_title: add(f"T2 - {rec.book_title}") if rec.year: add(f"PY - {rec.year}") if rec.date_full: add(f"DA - {rec.date_full}") if rec.volume: add(f"VL - {rec.volume}") if rec.issue: add(f"IS - {rec.issue}") if rec.page_start: add(f"SP - {rec.page_start}") if rec.page_end: add(f"EP - {rec.page_end}") if rec.publisher: add(f"PB - {rec.publisher}") if rec.publisher_place: add(f"CY - {rec.publisher_place}") for issn in rec.issn: add(f"SN - {issn}") for isbn in rec.isbn: add(f"SN - {isbn}") if rec.doi: add(f"DO - {rec.doi}") if rec.url: add(f"UR - {rec.url}") if rec.abstract: add(f"AB - {rec.abstract}") for kw in rec.keywords: add(f"KW - {kw}") if rec.language: add(f"LA - {rec.language}") add("ER - ") return lines def write_ris(records: Iterable[Record], out_path: Path) -> int: n = 0 # RIS-Konvention: CRLF-Zeilenenden, Leerzeile zwischen Einträgen with out_path.open("w", encoding="utf-8", newline="") as f: for rec in records: if rec.error or not rec.title: continue f.write("\r\n".join(ris_lines(rec)) + "\r\n\r\n") n += 1 return n # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main(argv: Optional[list[str]] = None) -> int: p = argparse.ArgumentParser( description="Konvertiert eine Liste von Literaturquellen in eine RIS-Datei.") p.add_argument("input", type=Path, help="Eingabedatei mit einer Quelle pro Zeile") p.add_argument("-o", "--output", type=Path, default=Path("library.ris"), help="Pfad zur RIS-Ausgabedatei (Default: library.ris)") p.add_argument("--email", default="", help="E-Mail-Adresse für den Crossref Polite-Pool (empfohlen)") p.add_argument("--workers", type=int, default=8, help="Anzahl paralleler API-Requests (Default: 8)") p.add_argument("--cache", type=Path, default=Path(".doi_cache.json"), help="JSON-Cache-Datei (Default: .doi_cache.json; leer = aus)") p.add_argument("--unresolved", type=Path, default=Path("unresolved.txt"), help="Datei für nicht auflösbare Zeilen (Default: unresolved.txt)") p.add_argument("-v", "--verbose", action="store_true") args = p.parse_args(argv) logging.basicConfig( format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%H:%M:%S", level=logging.DEBUG if args.verbose else logging.INFO, ) if not args.input.exists(): logging.error("Eingabedatei nicht gefunden: %s", args.input) return 2 lines = read_input(args.input) logging.info("%d Eingabezeilen gelesen.", len(lines)) cache_path = args.cache if str(args.cache) else None cache = Cache(cache_path) session = build_session(args.email) results: list[Record] = [] with ThreadPoolExecutor(max_workers=args.workers) as pool: futures = {pool.submit(resolve_record, ln, session, cache): ln for ln in lines} for i, fut in enumerate(as_completed(futures), 1): try: rec = fut.result() except Exception as e: # pragma: no cover rec = Record(raw_input=futures[fut], error=f"Worker-Exception: {e}") results.append(rec) status = "OK" if not rec.error and rec.title else "FAIL" logging.info("[%3d/%d] %s %s %s", i, len(lines), status, (rec.doi or "—")[:50].ljust(50), (rec.title or rec.error)[:80]) cache.save() # Stabile Reihenfolge wie in der Eingabe order = {ln: i for i, ln in enumerate(lines)} results.sort(key=lambda r: order.get(r.raw_input, 0)) n_written = write_ris(results, args.output) failed = [r for r in results if r.error or not r.title] if failed: with args.unresolved.open("w", encoding="utf-8") as f: for r in failed: f.write(f"{r.raw_input}\t{r.doi or '—'}\t{r.error or 'kein Titel'}\n") logging.info("Fertig: %d Einträge nach %s geschrieben, %d nicht auflösbar.", n_written, args.output, len(failed)) if failed: logging.info("Nicht aufgelöste Zeilen siehe %s", args.unresolved) return 0 if n_written else 1 if __name__ == "__main__": sys.exit(main())