540 lines
18 KiB
Python
540 lines
18 KiB
Python
#!/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())
|