""" OpenAlex to KAR (EPrints) export script Version: 1.16 (weekly KAR CSV lookup, workbook summary added) What this version does: - Queries OpenAlex for Kent / KMMS works - Normalises DOI values - Loads a weekly KAR export from kar_export.csv - Checks KAR duplicates locally using DOI first, then title fallback - Flags existing records for enrichment when OpenAlex has extra metadata - Sends ambiguous or conflicting matches to manual review - Writes one file per new item into separate folders - Routes books, book chapters, and datasets into their own queues - Writes enrichment / manual review / duplicate summary CSV reports - Writes a workbook summary with one sheet per output queue - Keeps raw OpenAlex results for auditing - Does NOT post directly to KAR """ from __future__ import annotations import csv import html import json import logging import os import re import sys import unicodedata from dataclasses import dataclass, field from datetime import datetime, date, timedelta from pathlib import Path from typing import Any, Dict, List, Optional, Tuple import requests from openpyxl import Workbook from openpyxl.styles import Alignment, Font, PatternFill from openpyxl.utils import get_column_letter # Make Windows console logging UTF-8 safe. try: sys.stdout.reconfigure(encoding="utf-8", errors="replace") sys.stderr.reconfigure(encoding="utf-8", errors="replace") except Exception: pass # ============================================================================= # CONFIGURATION # ============================================================================= CONFIG: Dict[str, Any] = { # OpenAlex settings "openalex_email": "i.badger@kent.ac.uk", "openalex_api_key": "opu84m2eTCojFYbtGkXdYY", # Obtain a free OpenAlex API key and paste it here (recommended) "kent_ror": "https://ror.org/00xkeyj56", "kmms_ror": "https://ror.org/049p9j193", "days_lookback": 60, # KAR export settings "kar_export_file": "kar_export.csv", # Output locations "output_dir": str(Path(__file__).resolve().parent / "openalex_test_output"), "last_run_file": "last_run.txt", "log_file": "openalex_to_kar_test.log", "debug_json": "last_openalex_results.json", "enrichment_report_file": "kar_enrichment_report.csv", "manual_review_report_file": "kar_manual_review.csv", "duplicate_report_file": "kar_duplicate_report.csv", "summary_workbook_file": "openalex_import_summary.xlsx", # Behaviour "dry_run": True, } KAR_EXPORT_PATH = Path(__file__).resolve().parent / CONFIG["kar_export_file"] OPENALEX_FILTERS = [ f"authorships.institutions.ror:{CONFIG['kent_ror']}", f"authorships.institutions.ror:{CONFIG['kmms_ror']}", ] # OpenAlex type handling. Everything not explicitly ignored is routed to one # of the existing queue folders. OPENALEX_IGNORED_TYPES = {"software", "reference-entry", "paratext", "book-section"} OPENALEX_BOOK_TYPES = {"book", "monograph"} OPENALEX_BOOK_CHAPTER_TYPES = {"book-chapter", "chapter"} OPENALEX_DATASET_TYPES = {"dataset"} # ============================================================================= # LOGGING # ============================================================================= class CsvLogHandler(logging.Handler): """Write one log record per row so Excel can open it cleanly.""" def __init__(self, path: str): super().__init__() self.path = Path(path) self.path.parent.mkdir(parents=True, exist_ok=True) self._file = self.path.open("a", encoding="utf-8", newline="") self._writer = csv.writer(self._file) if self.path.stat().st_size == 0: self._writer.writerow(["timestamp", "level", "message"]) self._file.flush() def emit(self, record: logging.LogRecord) -> None: try: message = self.format(record) message = re.sub(r"\s+", " ", message).strip() timestamp = datetime.fromtimestamp(record.created).strftime("%Y-%m-%d %H:%M:%S") self._writer.writerow([timestamp, record.levelname, message]) self._file.flush() except Exception: self.handleError(record) def close(self) -> None: try: self._file.close() finally: super().close() csv_handler = CsvLogHandler(CONFIG["log_file"]) csv_handler.setFormatter(logging.Formatter("%(message)s")) stream_handler = logging.StreamHandler(sys.stdout) stream_handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)-8s %(message)s", datefmt="%Y-%m-%d %H:%M:%S")) logging.basicConfig( level=logging.INFO, handlers=[ csv_handler, stream_handler, ], ) log = logging.getLogger(__name__) # ============================================================================= # SMALL UTILITIES # ============================================================================= def ensure_dir(path: Path) -> Path: path.mkdir(parents=True, exist_ok=True) return path def slugify(value: str, max_length: int = 120) -> str: value = repair_text(value or "untitled").strip().lower() value = re.sub(r"https?://", "", value) value = re.sub(r"[^a-z0-9]+", "-", value) value = value.strip("-") return (value or "untitled")[:max_length] def repair_text(value: Any) -> str: if value is None: return "" text = str(value).strip() if not text: return "" text = html.unescape(text) if any(marker in text for marker in ("â€", "Ã", "Â")): try: text = text.encode("latin-1").decode("utf-8") except Exception: pass return unicodedata.normalize("NFKC", text) def clean_doi(doi: Any) -> str: if not doi: return "" text = repair_text(doi).lower().strip() for prefix in ("https://doi.org/", "http://doi.org/", "doi:"): if text.startswith(prefix): text = text[len(prefix):] return text.strip().rstrip(".);,]}") def extract_doi_from_text(value: Any) -> str: text = repair_text(value) if not text: return "" match = re.search(r"10\.\d{4,9}/[^\s<>\"]+", text, flags=re.IGNORECASE) return clean_doi(match.group(0)) if match else "" def normalise_title_for_match(value: Any) -> str: text = repair_text(value) if not text: return "" text = text.casefold() text = re.sub(r"[\u2010-\u2015\-_/]+", " ", text) text = re.sub(r"[\u2018\u2019\u201A\u201B'`\"]", "", text) text = re.sub(r"[^\w\s]", " ", text, flags=re.UNICODE) text = re.sub(r"\s+", " ", text).strip() return text def normalize_compare_value(value: Any) -> str: text = repair_text(value) if not text: return "" return re.sub(r"\s+", " ", text.casefold()).strip() def append_line(path: Path, text: str) -> None: with path.open("a", encoding="utf-8") as f: f.write(text.rstrip() + "\n") def write_json(path: Path, data: Any) -> None: with path.open("w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) f.write("\n") def write_text(path: Path, text: str) -> None: with path.open("w", encoding="utf-8") as f: f.write(text) def write_csv_report(path: Path, rows: List[Dict[str, str]], fieldnames: List[str]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8", newline="") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() for row in rows: writer.writerow(row) # ============================================================================= # WORKBOOK HELPERS # ============================================================================= WORKBOOK_QUEUE_COLUMNS = [ "Title", "DOI", "ISBN", "OpenAlex type", "OA status", "OpenAlex ID", "Output file", ] def _normalise_type(value: Any) -> str: return repair_text(value).strip().lower() def extract_isbn(value: Any) -> str: """Return any ISBN-like values found anywhere in a work record.""" found: List[str] = [] seen = set() def add(candidate: Any) -> None: text = repair_text(candidate) if not text: return text = text.strip() if not text: return for prefix in ("isbn:", "ISBN:"): if text.startswith(prefix): text = text[len(prefix):].strip() text = re.sub(r"\s+", " ", text) key = text.casefold() if key not in seen: seen.add(key) found.append(text) def walk(obj: Any) -> None: if isinstance(obj, dict): for key, inner in obj.items(): key_text = repair_text(key).casefold() if "isbn" in key_text: if isinstance(inner, (list, tuple, set)): for item in inner: add(item) else: add(inner) walk(inner) elif isinstance(obj, list): for item in obj: walk(item) walk(value) return "; ".join(found) def build_queue_row(work: Dict[str, Any], oa_status: str, output_file: str) -> Dict[str, str]: return { "Title": repair_text(work.get("title", "")), "DOI": clean_doi(work.get("doi", "")), "ISBN": extract_isbn(work), "OpenAlex type": _normalise_type(work.get("type", "")), "OA status": repair_text(oa_status), "OpenAlex ID": repair_text(work.get("id", "")), "Output file": output_file, } def _style_sheet(ws, tab_color: str = "4F81BD") -> None: header_fill = PatternFill("solid", fgColor="D9EAF7") header_font = Font(bold=True) for cell in ws[1]: cell.fill = header_fill cell.font = header_font cell.alignment = Alignment(horizontal="center", vertical="center") ws.freeze_panes = "A2" ws.auto_filter.ref = ws.dimensions ws.sheet_properties.tabColor = tab_color for column_cells in ws.columns: max_len = 0 column_letter = get_column_letter(column_cells[0].column) for cell in column_cells: value = "" if cell.value is None else str(cell.value) if len(value) > max_len: max_len = len(value) ws.column_dimensions[column_letter].width = min(max(max_len + 2, 12), 48) def _write_rows_sheet(wb: Workbook, title: str, rows: List[Dict[str, str]], fieldnames: List[str], tab_color: str) -> None: ws = wb.create_sheet(title) ws.append(fieldnames) for row in rows: ws.append([row.get(field, "") for field in fieldnames]) _style_sheet(ws, tab_color=tab_color) def write_summary_workbook( workbook_path: Path, counts: Dict[str, int], from_date: str, queue_rows: Dict[str, List[Dict[str, str]]], duplicate_rows: List[Dict[str, str]], enrichment_rows: List[Dict[str, str]], manual_review_rows: List[Dict[str, str]], ) -> None: wb = Workbook() summary_ws = wb.active summary_ws.title = "Summary" summary_rows = [ ("Query period", f"from {from_date} to today"), ("Papers found", counts["found"]), ("New exported", counts["new"]), ("Enrichment reports", counts["enrichment_flagged"]), ("Already in KAR", counts["duplicate_skipped"]), ("Manual review", counts["manual_review"]), ("No DOI in OpenAlex", counts["no_doi"]), ("Ignored types", counts.get("ignored", 0)), ("REF action needed", counts["ref_action_needed"]), ("Matched by DOI", counts["matched_by_doi"]), ("Matched by title", counts["matched_by_title"]), ("Errors", counts["errors"]), ("Exports root", CONFIG["output_dir"]), ("Workbook file", workbook_path.name), ] summary_ws.append(["Metric", "Value"]) for metric, value in summary_rows: summary_ws.append([metric, value]) _style_sheet(summary_ws, tab_color="1F4E78") summary_ws.column_dimensions["A"].width = 24 summary_ws.column_dimensions["B"].width = 72 _write_rows_sheet(wb, "New", queue_rows["new"], WORKBOOK_QUEUE_COLUMNS, "5B9BD5") _write_rows_sheet(wb, "Books", queue_rows["books"], WORKBOOK_QUEUE_COLUMNS, "5B9BD5") _write_rows_sheet(wb, "Book Chapters", queue_rows["book_chapters"], WORKBOOK_QUEUE_COLUMNS, "5B9BD5") _write_rows_sheet(wb, "Datasets", queue_rows["datasets"], WORKBOOK_QUEUE_COLUMNS, "5B9BD5") _write_rows_sheet(wb, "No DOI", queue_rows["no_doi"], WORKBOOK_QUEUE_COLUMNS, "F4B183") _write_rows_sheet(wb, "Duplicates", duplicate_rows, ["eprintid", "title", "doi", "match_source", "status"], "70AD47") _write_rows_sheet(wb, "Enrichment", enrichment_rows, ["eprintid", "title", "match_source", "field", "kar_value", "openalex_value"], "70AD47") _write_rows_sheet(wb, "Manual Review", manual_review_rows, ["title", "doi", "match_source", "reason", "candidate_eprintids", "candidate_titles"], "70AD47") workbook_path.parent.mkdir(parents=True, exist_ok=True) wb.save(workbook_path) # ============================================================================= # OUTPUT FOLDERS # ============================================================================= @dataclass class OutputPaths: root: Path new: Path books: Path book_chapters: Path datasets: Path enrichment: Path duplicates: Path manual_review: Path no_doi: Path raw_openalex: Path logs: Path def build_output_paths(base_dir: str) -> OutputPaths: root = ensure_dir(Path(base_dir).resolve()) return OutputPaths( root=root, new=ensure_dir(root / "new"), books=ensure_dir(root / "books"), book_chapters=ensure_dir(root / "book_chapters"), datasets=ensure_dir(root / "datasets"), enrichment=ensure_dir(root / "enrichment"), duplicates=ensure_dir(root / "duplicates"), manual_review=ensure_dir(root / "manual_review"), no_doi=ensure_dir(root / "no_doi"), raw_openalex=ensure_dir(root / "raw_openalex"), logs=ensure_dir(root / "logs"), ) # ============================================================================= # DATE RANGE # ============================================================================= def get_query_date() -> str: last_run_path = Path(CONFIG["last_run_file"]) if last_run_path.exists(): with last_run_path.open("r", encoding="utf-8") as f: last_run = datetime.strptime(f.read().strip(), "%Y-%m-%d") query_from = last_run - timedelta(days=2) log.info( "Last run: %s - querying from %s (2-day overlap)", last_run.date(), query_from.date(), ) else: query_from = datetime.today() - timedelta(days=int(CONFIG["days_lookback"])) log.info("No last_run.txt found - querying from %s", query_from.date()) return query_from.strftime("%Y-%m-%d") def save_run_date() -> None: with Path(CONFIG["last_run_file"]).open("w", encoding="utf-8") as f: f.write(datetime.today().strftime("%Y-%m-%d")) # ============================================================================= # OPENALEX QUERY # ============================================================================= def query_openalex(from_date: str) -> List[Dict[str, Any]]: log.info("Querying OpenAlex for papers from %s...", from_date) headers = {"User-Agent": f"KentLibraryKARImport/1.16.1 (mailto:{CONFIG['openalex_email']})"} # If an API key is supplied it will be sent with every request. all_works_by_id: Dict[str, Dict[str, Any]] = {} for base_filter in OPENALEX_FILTERS: params = { "filter": f"{base_filter},from_publication_date:{from_date}", "select": ",".join( [ "id", "doi", "title", "abstract_inverted_index", "authorships", "publication_date", "primary_location", "best_oa_location", "locations", "open_access", "type", "biblio", "awards", "funders", ] ), "per_page": 200, "per_page": 200, "cursor": "*", } if CONFIG.get("openalex_api_key"): params["api_key"] = CONFIG["openalex_api_key"] page = 1 while True: log.info(" Fetching page %s for filter: %s", page, base_filter) response = requests.get( "https://api.openalex.org/works", params=params, headers=headers, timeout=30, ) if response.status_code != 200: log.error("OpenAlex API error: %s - %s", response.status_code, response.text) break data = response.json() results = data.get("results", []) or [] for work in results: work_id = work.get("id") if work_id: all_works_by_id[work_id] = work log.info( " Filter %s: page %s: %s results (total so far: %s)", base_filter, page, len(results), len(all_works_by_id), ) next_cursor = data.get("meta", {}).get("next_cursor") if not next_cursor or not results: break params["cursor"] = next_cursor page += 1 all_works = list(all_works_by_id.values()) try: write_json(Path(CONFIG["debug_json"]), all_works) except Exception as e: log.warning("Could not write debug JSON: %s", e) log.info("OpenAlex query complete: %s papers found", len(all_works)) log.info( "OpenAlex filter breakdown: %s", "; ".join(f"{filt}: {len([w for w in all_works if True]) if False else ''}" for filt in OPENALEX_FILTERS), ) return all_works # ============================================================================= # ABSTRACT AND BIBLIOGRAPHIC HELPERS # ============================================================================= def reconstruct_abstract(inverted_index: Optional[Dict[str, List[int]]]) -> str: if not inverted_index: return "" try: position_map: Dict[int, str] = {} for word, positions in inverted_index.items(): for pos in positions: position_map[pos] = word return " ".join(position_map[i] for i in sorted(position_map.keys())) except Exception: return "" def format_pagerange(biblio: Dict[str, Any]) -> str: first = repair_text(biblio.get("first_page", "")) last = repair_text(biblio.get("last_page", "")) if first and last: return f"pp. {first}-{last}" if first: return f"p. {first}" return "" # ============================================================================= # BUILD RECORDS # ============================================================================= def build_additional_info(work: Dict[str, Any], oa_status: str, enrichment_note: str = "") -> str: lines: List[str] = [] lines.append("Source: OpenAlex API (automated import)") lines.append(f"OA status: {oa_status}") oa_url = work.get("open_access", {}).get("oa_url") if oa_url: lines.append(f"Best OA URL: {oa_url}") pub_date = work.get("publication_date", "") if pub_date: lines.append(f"Published: {pub_date}") if enrichment_note: lines.append(enrichment_note) accepted_date = work.get("accepted_date") if accepted_date: lines.append(f"Accepted: {accepted_date}") else: lines.append("Accepted date: NOT AVAILABLE - please check publisher page or contact author") if oa_status == "closed": lines.append("REF OA ACTION REQUIRED: paper is closed - contact author for Author Accepted Manuscript") lines.append(f"Imported by OpenAlex script: {datetime.today().strftime('%Y-%m-%d %H:%M')}") return " | ".join(lines) def build_kar_record(work: Dict[str, Any], enrichment_note: str = "") -> Tuple[Dict[str, Any], str]: doi = clean_doi(work.get("doi", "")) title = work.get("title", "No title available") pub_date = work.get("publication_date", "") biblio = work.get("biblio", {}) or {} location = work.get("primary_location", {}) or {} source = location.get("source", {}) or {} oa_info = work.get("open_access", {}) or {} oa_status = oa_info.get("oa_status", "unknown") abstract = reconstruct_abstract(work.get("abstract_inverted_index")) creators: List[Dict[str, Any]] = [] for authorship in work.get("authorships", []) or []: author = authorship.get("author", {}) or {} name = author.get("display_name", "") orcid = author.get("orcid", "") name_parts = name.rsplit(" ", 1) if len(name_parts) == 2: given, family = name_parts else: given, family = "", name creator: Dict[str, Any] = {"name": {"family": family, "given": given}} if orcid: creator["orcid"] = orcid.replace("https://orcid.org/", "") creators.append(creator) record: Dict[str, Any] = { "eprint_status": "inbox", "type": "article", "title": title, "abstract": abstract, "creators": creators, "date": pub_date, "date_type": "published", "publisher": source.get("host_organization_name", ""), "publication": source.get("display_name", ""), "issn": source.get("issn_l", ""), "volume": str(biblio.get("volume", "") or ""), "number": str(biblio.get("issue", "") or ""), "pagerange": format_pagerange(biblio), "doi": doi, "official_url": work.get("doi", "") or "", "note": build_additional_info(work, oa_status, enrichment_note=enrichment_note), "metadata_visibility": "show", } record = {k: v for k, v in record.items() if v} return record, oa_status # ============================================================================= # RIS EXPORT HELPERS # ============================================================================= def _ris_clean(value: Any) -> str: if value is None: return "" text = str(value).replace("\r\n", "\n").replace("\r", "\n").strip() return re.sub(r"\s+", " ", text) def _ris_work_type(work: Dict[str, Any]) -> str: work_type = (work.get("type") or "").strip().lower() if work_type in {"article", "journal-article", "preprint", "posted-content"}: return "JOUR" if work_type in {"book-chapter", "chapter"}: return "CHAP" if work_type in {"book", "monograph"}: return "BOOK" if work_type == "dataset": return "DATA" return "GEN" def _ris_page_range(pagerange: str) -> Tuple[str, str]: pagerange = (pagerange or "").strip() if not pagerange: return "", "" pagerange = pagerange.replace("pp. ", "").replace("p. ", "") if "-" in pagerange: first, last = pagerange.split("-", 1) return first.strip(), last.strip() return pagerange.strip(), "" def build_ris_record(work: Dict[str, Any], record: Dict[str, Any]) -> str: lines: List[str] = [] lines.append(f"TY - {_ris_work_type(work)}") title = _ris_clean(record.get("title") or work.get("title") or "") if title: lines.append(f"TI - {title}") for creator in record.get("creators", []) or []: name = creator.get("name", {}) or {} family = _ris_clean(name.get("family", "")) given = _ris_clean(name.get("given", "")) if family and given: lines.append(f"AU - {family}, {given}") elif family: lines.append(f"AU - {family}") elif given: lines.append(f"AU - {given}") date = _ris_clean(record.get("date") or work.get("publication_date", "")) if date: lines.append(f"PY - {date[:4] if len(date) >= 4 else date}") if re.match(r"^\d{4}-\d{2}-\d{2}$", date): lines.append(f"Y1 - {date.replace('-', '/')}") publication = _ris_clean(record.get("publication")) if publication: lines.append(f"JF - {publication}") publisher = _ris_clean(record.get("publisher")) if publisher: lines.append(f"PB - {publisher}") volume = _ris_clean(record.get("volume")) if volume: lines.append(f"VL - {volume}") issue = _ris_clean(record.get("number")) if issue: lines.append(f"IS - {issue}") first_page, last_page = _ris_page_range(record.get("pagerange", "")) if first_page: lines.append(f"SP - {first_page}") if last_page: lines.append(f"EP - {last_page}") doi = _ris_clean(record.get("doi")) if doi: lines.append(f"DO - {doi}") url = _ris_clean(record.get("official_url")) if url: lines.append(f"UR - {url}") abstract = _ris_clean(record.get("abstract")) if abstract: lines.append(f"AB - {abstract}") note = _ris_clean(record.get("note")) if note: lines.append(f"N1 - {note}") lines.append("ER -") return "\n".join(lines) + "\n" # ============================================================================= # KAR EXPORT LOOKUP # ============================================================================= @dataclass class KarRecord: eprintid: str title: str = "" doi: str = "" date: str = "" publication: str = "" volume: str = "" number: str = "" pagerange: str = "" pages: str = "" publisher: str = "" issn: str = "" official_url: str = "" abstract: str = "" type: str = "" published_oa: str = "" raw: Dict[str, Any] = field(default_factory=dict) @dataclass class KarExportIndex: records: Dict[str, KarRecord] = field(default_factory=dict) doi_index: Dict[str, List[str]] = field(default_factory=dict) title_index: Dict[str, List[str]] = field(default_factory=dict) row_count: int = 0 @dataclass class KarLookupResult: exists: bool eprint_id: Optional[str] lookup_ok: bool source: str = "" multiple_matches: bool = False search_error: bool = False kar_record: Optional[KarRecord] = None candidate_eprintids: List[str] = field(default_factory=list) candidate_titles: List[str] = field(default_factory=list) KAR_INDEX: Optional[KarExportIndex] = None def parse_date_with_precision(value: Any) -> Tuple[Optional[date], int]: text = repair_text(value) if not text: return None, 0 for fmt, precision in (("%Y-%m-%d", 3), ("%Y/%m/%d", 3), ("%d/%m/%Y", 3), ("%Y-%m", 2), ("%Y/%m", 2), ("%Y", 1)): try: dt = datetime.strptime(text, fmt).date() return dt, precision except ValueError: continue return None, 0 def standardise_kar_row(row: Dict[str, Any]) -> KarRecord: title = repair_text(row.get("title", "")) doi = clean_doi(row.get("id_number", "")) or extract_doi_from_text(row.get("official_url", "")) return KarRecord( eprintid=repair_text(row.get("eprintid", "")), title=title, doi=doi, date=repair_text(row.get("date", "")), publication=repair_text(row.get("publication", "")), volume=repair_text(row.get("volume", "")), number=repair_text(row.get("number", "")), pagerange=repair_text(row.get("pagerange", "")), pages=repair_text(row.get("pages", "")), publisher=repair_text(row.get("publisher", "")), issn=repair_text(row.get("issn", "")), official_url=repair_text(row.get("official_url", "")), abstract=repair_text(row.get("abstract", "")), type=repair_text(row.get("type", "")), published_oa=repair_text(row.get("published_oa", "")), raw=dict(row), ) def kar_record_to_dict(record: KarRecord) -> Dict[str, Any]: return { "eprintid": record.eprintid, "title": record.title, "doi": record.doi, "date": record.date, "publication": record.publication, "volume": record.volume, "number": record.number, "pagerange": record.pagerange, "pages": record.pages, "publisher": record.publisher, "issn": record.issn, "official_url": record.official_url, "abstract": record.abstract, "type": record.type, "published_oa": record.published_oa, } def load_kar_export(export_path: Path) -> KarExportIndex: if not export_path.exists(): raise FileNotFoundError(f"KAR export file not found: {export_path}") records: Dict[str, KarRecord] = {} row_count = 0 with export_path.open("r", encoding="utf-8-sig", newline="") as f: reader = csv.DictReader(f) for row in reader: row_count += 1 eprintid = repair_text(row.get("eprintid", "")) if not eprintid: continue if eprintid not in records: records[eprintid] = standardise_kar_row(row) continue record = records[eprintid] # Fill blanks from continuation rows when useful. for field_name in ( "title", "doi", "date", "publication", "volume", "number", "pagerange", "pages", "publisher", "issn", "official_url", "abstract", "type", "published_oa", ): if getattr(record, field_name): continue value = repair_text(row.get(field_name, "")) if not value and field_name == "doi": value = clean_doi(row.get("id_number", "")) or extract_doi_from_text(row.get("official_url", "")) if value: setattr(record, field_name, value) doi_index: Dict[str, List[str]] = {} title_index: Dict[str, List[str]] = {} for eprintid, record in records.items(): if record.doi: doi_index.setdefault(record.doi, []).append(eprintid) title_norm = normalise_title_for_match(record.title) if title_norm: title_index.setdefault(title_norm, []).append(eprintid) index = KarExportIndex(records=records, doi_index=doi_index, title_index=title_index, row_count=row_count) log.info("Loaded KAR export: %s main records from %s CSV rows", len(index.records), index.row_count) log.info("KAR indexes built: %s DOI keys, %s title keys", len(index.doi_index), len(index.title_index)) return index def find_kar_match(work: Dict[str, Any], index: KarExportIndex) -> KarLookupResult: oa_doi = clean_doi(work.get("doi", "")) oa_title = repair_text(work.get("title", "")) title_norm = normalise_title_for_match(oa_title) if oa_doi: candidate_ids = index.doi_index.get(oa_doi, []) if len(candidate_ids) == 1: kar_record = index.records[candidate_ids[0]] return KarLookupResult(True, candidate_ids[0], True, source="doi", kar_record=kar_record, candidate_eprintids=candidate_ids, candidate_titles=[kar_record.title]) if len(candidate_ids) > 1: return KarLookupResult(False, None, False, source="doi", multiple_matches=True, candidate_eprintids=candidate_ids, candidate_titles=[index.records[i].title for i in candidate_ids if i in index.records]) if title_norm: candidate_ids = index.title_index.get(title_norm, []) if len(candidate_ids) == 1: kar_record = index.records[candidate_ids[0]] return KarLookupResult(True, candidate_ids[0], True, source="title", kar_record=kar_record, candidate_eprintids=candidate_ids, candidate_titles=[kar_record.title]) if len(candidate_ids) > 1: return KarLookupResult(False, None, False, source="title", multiple_matches=True, candidate_eprintids=candidate_ids, candidate_titles=[index.records[i].title for i in candidate_ids if i in index.records]) return KarLookupResult(False, None, True, source="doi" if oa_doi else "title") def compare_metadata(kar_record: KarRecord, work: Dict[str, Any], match_source: str) -> List[Dict[str, str]]: rows: List[Dict[str, str]] = [] oa_doi = clean_doi(work.get("doi", "")) location = work.get("primary_location") or {} source = location.get("source") or {} oa_publication = repair_text(source.get("display_name", "")) biblio = work.get("biblio", {}) or {} oa_volume = repair_text(biblio.get("volume", "")) oa_issue = repair_text(biblio.get("issue", "")) oa_pages = format_pagerange(biblio) oa_date = repair_text(work.get("publication_date", "")) def add_row(field: str, kar_value: Any, oa_value: Any) -> None: kar_text = repair_text(kar_value) oa_text = repair_text(oa_value) if kar_text == oa_text: return if kar_text: return if not oa_text: return rows.append( { "eprintid": kar_record.eprintid, "title": kar_record.title, "match_source": match_source, "field": field, "kar_value": kar_text, "openalex_value": oa_text, } ) add_row("doi", kar_record.doi, oa_doi) add_row("publication", kar_record.publication, oa_publication) add_row("volume", kar_record.volume, oa_volume) add_row("number", kar_record.number, oa_issue) add_row("pages", kar_record.pages or kar_record.pagerange, oa_pages) kar_date, kar_prec = parse_date_with_precision(kar_record.date) oa_date_dt, oa_prec = parse_date_with_precision(oa_date) if not kar_record.date and oa_date: rows.append({"eprintid": kar_record.eprintid, "title": kar_record.title, "match_source": match_source, "field": "date", "kar_value": "", "openalex_value": oa_date}) elif kar_date and oa_date_dt and kar_date == oa_date_dt and oa_prec > kar_prec: rows.append({"eprintid": kar_record.eprintid, "title": kar_record.title, "match_source": match_source, "field": "date", "kar_value": kar_record.date, "openalex_value": oa_date}) return rows # ============================================================================= # EXPORT HELPERS # ============================================================================= def make_filename(work: Dict[str, Any]) -> str: doi = clean_doi(work.get("doi", "")) title = work.get("title", "untitled") openalex_id = str(work.get("id", "openalex")) oa_slug = slugify(openalex_id.rsplit("/", 1)[-1], max_length=40) doi_slug = slugify(doi, max_length=50) if doi else "no-doi" title_slug = slugify(title, max_length=60) return f"{doi_slug}__{title_slug}__{oa_slug}" def raw_work_file(paths: OutputPaths, filename: str) -> Path: return paths.raw_openalex / f"{filename}.json" def export_new_record(paths: OutputPaths, filename: str, record: Dict[str, Any], work: Dict[str, Any], dest_dir: Path) -> Path: out_path = dest_dir / f"{filename}.json" ris_path = dest_dir / f"{filename}.ris" write_json(out_path, record) write_text(ris_path, build_ris_record(work, record)) write_json(raw_work_file(paths, filename), work) return out_path def route_work_to_folder(work: Dict[str, Any], record: Dict[str, Any]) -> Optional[str]: work_type = _normalise_type(work.get("type", "")) title = _normalise_type(work.get("title", "")) abstract = _normalise_type(record.get("abstract", "")) if work_type in OPENALEX_IGNORED_TYPES: return None if work_type in OPENALEX_DATASET_TYPES or title.startswith("dataset for:") or "dataset" in title or "raw dataset" in abstract: return "datasets" if work_type in OPENALEX_BOOK_CHAPTER_TYPES: return "book_chapters" if work_type in OPENALEX_BOOK_TYPES: return "books" return "new" def export_enrichment_review(paths: OutputPaths, filename: str, work: Dict[str, Any], record: Dict[str, Any], eprint_id: Any, existing_record: Dict[str, Any]) -> Path: out_path = paths.enrichment / f"{filename}.txt" missing_fields = [] for field in ["date", "volume", "number", "pages", "doi"]: if not existing_record.get(field): missing_fields.append(field) lines = [ f"Title: {work.get('title', 'No title available')}", f"DOI: {clean_doi(work.get('doi', '')) or '(none)'}", f"Existing KAR record ID: {eprint_id}", f"Missing fields in KAR: {', '.join(missing_fields) if missing_fields else 'not checked'}", "", "Suggested note for the existing record:", record.get("note", "OpenAlex has newer or fuller metadata for this item."), "", "Proposed import record (for reference only):", json.dumps(record, ensure_ascii=False, indent=2), ] write_text(out_path, "\n".join(lines) + "\n") write_json(raw_work_file(paths, filename), work) return out_path def export_duplicate_report(paths: OutputPaths, filename: str, work: Dict[str, Any], eprint_id: Any, match_source: str) -> Path: out_path = paths.duplicates / f"{filename}.txt" lines = [ f"Title: {work.get('title', 'No title available')}", f"DOI: {clean_doi(work.get('doi', '')) or '(none)'}", f"Existing KAR record ID: {eprint_id}", f"Matched by: {match_source}", "Status: already present and appears complete", ] write_text(out_path, "\n".join(lines) + "\n") return out_path def export_manual_review(paths: OutputPaths, filename: str, work: Dict[str, Any], reason: str, candidates: Optional[KarLookupResult] = None) -> Path: out_path = paths.manual_review / f"{filename}.txt" candidate_ids = "; ".join(candidates.candidate_eprintids) if candidates else "" candidate_titles = " || ".join(candidates.candidate_titles) if candidates else "" lines = [ f"Title: {work.get('title', 'No title available')}", f"DOI: {clean_doi(work.get('doi', '')) or '(none)'}", f"Reason: {reason}", f"Candidate eprintids: {candidate_ids or '(none)'}", f"Candidate titles: {candidate_titles or '(none)'}", "Status: manual review required", ] write_text(out_path, "\n".join(lines) + "\n") return out_path def export_no_doi_report(paths: OutputPaths, filename: str, work: Dict[str, Any]) -> Path: out_path = paths.no_doi / f"{filename}.txt" lines = [ f"Title: {work.get('title', 'No title available')}", f"OpenAlex ID: {work.get('id', 'unknown')}", "Status: no DOI was present in OpenAlex for this item", ] write_text(out_path, "\n".join(lines) + "\n") return out_path # ============================================================================= # SUMMARY # ============================================================================= def print_summary(counts: Dict[str, int], from_date: str) -> None: log.info("\n%s", "=" * 60) log.info("SUMMARY") log.info("%s", "=" * 60) log.info("Query period: from %s to today", from_date) log.info("Papers found: %s", counts["found"]) log.info("New exported: %s", counts["new"]) log.info("Enrichment reports: %s", counts["enrichment_flagged"]) log.info("Already in KAR: %s", counts["duplicate_skipped"]) log.info("Manual review: %s", counts["manual_review"]) log.info("No DOI in OpenAlex: %s", counts["no_doi"]) log.info("Ignored types: %s", counts.get("ignored", 0)) log.info("REF action needed: %s", counts["ref_action_needed"]) log.info("Matched by DOI: %s", counts["matched_by_doi"]) log.info("Matched by title: %s", counts["matched_by_title"]) log.info("Errors: %s", counts["errors"]) log.info("Exports root: %s", CONFIG["output_dir"]) log.info("%s", "=" * 60) # ============================================================================= # MAIN WORKFLOW # ============================================================================= def main() -> None: log.info("%s", "=" * 60) log.info("OpenAlex to KAR export script starting") log.info("Run date: %s", datetime.today().strftime("%Y-%m-%d %H:%M")) log.info("Mode: %s", "DRY RUN / EXPORT" if CONFIG["dry_run"] else "EXPORT") log.info("KAR export lookup: %s", "ENABLED") log.info("%s", "=" * 60) paths = build_output_paths(CONFIG["output_dir"]) counts: Dict[str, int] = { "found": 0, "new": 0, "enrichment_flagged": 0, "duplicate_skipped": 0, "manual_review": 0, "no_doi": 0, "ignored": 0, "errors": 0, "ref_action_needed": 0, "matched_by_doi": 0, "matched_by_title": 0, } queue_rows: Dict[str, List[Dict[str, str]]] = { "new": [], "books": [], "book_chapters": [], "datasets": [], "no_doi": [], } enrichment_rows: List[Dict[str, str]] = [] manual_review_rows: List[Dict[str, str]] = [] duplicate_rows: List[Dict[str, str]] = [] try: kar_index = load_kar_export(KAR_EXPORT_PATH) except Exception as e: log.error("Failed to load KAR export %s: %s", KAR_EXPORT_PATH, e) return from_date = get_query_date() works = query_openalex(from_date) counts["found"] = len(works) if not works: log.info("No papers found. Exiting.") save_run_date() return try: write_json(Path(CONFIG["debug_json"]), works) except Exception as e: log.warning("Could not write debug JSON: %s", e) log.info("\nProcessing %s papers...", len(works)) for work in works: doi = clean_doi(work.get("doi", "")) title = work.get("title", "No title")[:80] filename = make_filename(work) log.info("\n Paper: %s", title) if not doi: counts["no_doi"] += 1 export_no_doi_report(paths, filename, work) log.info(" No DOI in OpenAlex - using title match only if available") match = find_kar_match(work, kar_index) if match.multiple_matches: log.info(" Matched multiple KAR records - manual review") counts["manual_review"] += 1 manual_review_rows.append( { "title": work.get("title", "No title available"), "doi": doi, "match_source": match.source, "reason": "multiple KAR matches", "candidate_eprintids": "; ".join(match.candidate_eprintids), "candidate_titles": " || ".join(match.candidate_titles), } ) export_manual_review(paths, filename, work, reason="multiple KAR matches", candidates=match) continue if match.exists and match.kar_record: kar_record = match.kar_record if match.source == "title": kar_doi = clean_doi(kar_record.doi) if kar_doi and doi and kar_doi != doi: log.info( " Matched in KAR (ID: %s) but DOI differs - manual review", match.eprint_id, ) counts["manual_review"] += 1 manual_review_rows.append( { "title": work.get("title", "No title available"), "doi": doi, "match_source": match.source, "reason": "title match but DOI differs", "candidate_eprintids": match.eprint_id or "", "candidate_titles": kar_record.title, } ) export_manual_review(paths, filename, work, reason="title match but DOI differs", candidates=match) continue enrichment_details = compare_metadata(kar_record, work, match.source) if enrichment_details: log.info(" Already in KAR (ID: %s) - exporting enrichment review", match.eprint_id) counts["enrichment_flagged"] += 1 if match.source == "doi": counts["matched_by_doi"] += 1 else: counts["matched_by_title"] += 1 record, oa_status = build_kar_record(work, enrichment_note="OpenAlex has newer or fuller metadata for this item.") if oa_status == "closed": counts["ref_action_needed"] += 1 export_enrichment_review(paths, filename, work, record, match.eprint_id, kar_record_to_dict(kar_record)) enrichment_rows.extend(enrichment_details) else: log.info(" Already in KAR (ID: %s) - exporting duplicate report only", match.eprint_id) counts["duplicate_skipped"] += 1 if match.source == "doi": counts["matched_by_doi"] += 1 else: counts["matched_by_title"] += 1 export_duplicate_report(paths, filename, work, match.eprint_id, match.source) duplicate_rows.append( { "eprintid": match.eprint_id or "", "title": work.get("title", "No title available"), "doi": doi, "match_source": match.source, "status": "duplicate", } ) continue counts["new"] += 1 log.info(" New paper - building export record") try: record, oa_status = build_kar_record(work) if oa_status == "closed": counts["ref_action_needed"] += 1 log.info(" OA status: CLOSED - REF action flag set") else: log.info(" OA status: %s", oa_status) folder_name = route_work_to_folder(work, record) if folder_name is None: counts["ignored"] += 1 log.info(" OpenAlex type '%s' is intentionally ignored", openalex_type) continue if folder_name == "books": export_path = export_new_record(paths, filename, record, work, paths.books) elif folder_name == "book_chapters": export_path = export_new_record(paths, filename, record, work, paths.book_chapters) elif folder_name == "datasets": export_path = export_new_record(paths, filename, record, work, paths.datasets) else: export_path = export_new_record(paths, filename, record, work, paths.new) queue_rows[folder_name].append(build_queue_row(work, oa_status, f"{filename}.json")) log.info(" Routed to %s", export_path.parent.name) log.info(" Exported file: %s", export_path) except Exception as e: log.error(" Error processing paper: %s", e) counts["errors"] += 1 try: write_csv_report( Path(CONFIG["enrichment_report_file"]), enrichment_rows, ["eprintid", "title", "match_source", "field", "kar_value", "openalex_value"], ) write_csv_report( Path(CONFIG["manual_review_report_file"]), manual_review_rows, ["title", "doi", "match_source", "reason", "candidate_eprintids", "candidate_titles"], ) write_csv_report( Path(CONFIG["duplicate_report_file"]), duplicate_rows, ["eprintid", "title", "doi", "match_source", "status"], ) write_summary_workbook( Path(CONFIG["summary_workbook_file"]), counts, from_date, queue_rows, duplicate_rows, enrichment_rows, manual_review_rows, ) except Exception as e: log.warning("Could not write summary CSV/workbook report(s): %s", e) save_run_date() print_summary(counts, from_date) # ============================================================================= # ENTRY POINT # ============================================================================= if __name__ == "__main__": main()