""" OpenAlex to KAR (EPrints) export script Version: 1.15 (weekly KAR CSV lookup, RIS export preserved) 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 - 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 # 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": "", # Obtain a free OpenAlex account and enter email address used here (recommended) "openalex_api_key": "", # 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": 30, # 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", # 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']}", ] # ============================================================================= # LOGGING # ============================================================================= logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)-8s %(message)s", datefmt="%Y-%m-%d %H:%M:%S", handlers=[ logging.FileHandler(CONFIG["log_file"], encoding="utf-8"), logging.StreamHandler(sys.stdout), ], ) 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) # ============================================================================= # 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.15 (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", "open_access", "type", "biblio", ] ), "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]) -> str: work_type = (work.get("type") or "").strip().lower() title = (work.get("title") or "").strip().lower() abstract = (record.get("abstract") or "").strip().lower() if work_type == "dataset" or title.startswith("dataset for:") or "dataset" in title or "raw dataset" in abstract: return "datasets" if work_type == "book-chapter": return "book_chapters" if work_type in {"book", "monograph"}: 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("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, "errors": 0, "ref_action_needed": 0, "matched_by_doi": 0, "matched_by_title": 0, } 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 == "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) 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"], ) except Exception as e: log.warning("Could not write summary CSV report(s): %s", e) save_run_date() print_summary(counts, from_date) # ============================================================================= # ENTRY POINT # ============================================================================= if __name__ == "__main__": main()