Dateien nach "/" hochladen

2026-05-20 12:27:39 +02:00
parent e38b3f48f6
commit 38132e6511
1 changed files with 539 additions and 0 deletions
@@ -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())