From 38132e6511d59624ee853501882863e89f0e2394 Mon Sep 17 00:00:00 2001 From: riegera Date: Wed, 20 May 2026 12:27:39 +0200 Subject: [PATCH] Dateien nach "/" hochladen --- doi2ris.py | 539 +++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 539 insertions(+) create mode 100644 doi2ris.py diff --git a/doi2ris.py b/doi2ris.py new file mode 100644 index 0000000..af67411 --- /dev/null +++ b/doi2ris.py @@ -0,0 +1,539 @@ +#!/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())