341 lines
20 KiB
Python
341 lines
20 KiB
Python
"""Backfill HSE quartiles and resolve remaining DOI journals. No .env or DDL."""
|
|
|
|
import argparse
|
|
from collections import Counter, defaultdict
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from datetime import date, datetime, timezone
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import sys
|
|
import time
|
|
from urllib.parse import quote
|
|
|
|
import requests
|
|
from sqlalchemy import MetaData, Table, create_engine, select, update
|
|
from openpyxl import load_workbook
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
from app.services.council_source import _doi, _name, _journal_names, _description_journals
|
|
from app.services.council_catalog import issn
|
|
from app.services.publication_quartiles import catalog_entries, entries, hse_entries, merge, sjr_entries, index_sjr, with_title_history, identity_issues
|
|
from app.services.publication_vak import assess_vak
|
|
|
|
SOURCE_TITLES_URL = "https://downloads.ctfassets.net/o78em1y1w4i4/7xtaTxNiNcWRTeZkV86eNy/69cf2d506c905dc299531fdc93049dbb/ext_list_Aug_2026.xlsx"
|
|
SOURCE_TITLES_SHA256 = "11e81f686401c89fbef28de31d1880388bb7122f35c89e09db6e1848237e9afb"
|
|
|
|
|
|
def load_title_history(path):
|
|
if hashlib.sha256(path.read_bytes()).hexdigest() != SOURCE_TITLES_SHA256:
|
|
raise ValueError("Scopus title-list snapshot checksum mismatch")
|
|
by_issn, by_title = defaultdict(list), defaultdict(list)
|
|
workbook = load_workbook(path, read_only=True, data_only=True)
|
|
try:
|
|
rows = workbook.worksheets[0].iter_rows(values_only=True)
|
|
columns = next(rows)
|
|
required = {"Source Title", "ISSN", "EISSN", "Source Type", "Title History Indication", "Related Title 1"}
|
|
if not required <= set(columns):
|
|
raise ValueError("Invalid Scopus title list")
|
|
for row in rows:
|
|
data = dict(zip(columns, row))
|
|
if data["Source Type"] != "Journal":
|
|
continue
|
|
identifiers = {issn(data.get(key)) for key in ("ISSN", "EISSN")} - {""}
|
|
identifiers = {identifier[:4] + "-" + identifier[4:] for identifier in identifiers if len(identifier) == 8}
|
|
record = {"title": str(data["Source Title"] or ""), "issns": identifiers,
|
|
"history": data["Title History Indication"],
|
|
"related": [str(value) for key, value in data.items() if key and "Related Title" in key and value]}
|
|
by_title[_name(record["title"])].append(record)
|
|
for identifier in identifiers:
|
|
by_issn[issn(identifier)].append(record)
|
|
finally:
|
|
workbook.close()
|
|
return dict(by_issn), dict(by_title)
|
|
|
|
|
|
def load_sjr(identifiers, years, cache):
|
|
"""Read the university's licensed SCImago dataset for the requested ISSNs."""
|
|
rankings = {}
|
|
failures = []
|
|
for year in sorted(set(years) & set(range(2009, 2026))):
|
|
records = {}
|
|
for offset in range(0, len(identifiers), 200):
|
|
batch = identifiers[offset:offset + 200]
|
|
digest = hashlib.sha256(json.dumps([year, batch]).encode()).hexdigest()
|
|
path = cache / f"sjr-{digest}.json"
|
|
if path.exists():
|
|
bindings = json.loads(path.read_text(encoding="utf8"))
|
|
else:
|
|
values = " ".join(json.dumps(value) for value in batch)
|
|
query = f'''PREFIX bibo: <http://purl.org/ontology/bibo/>
|
|
PREFIX dct: <http://purl.org/dc/terms/>
|
|
PREFIX ou: <https://opendata.unex.es/def/ontouniversidad#>
|
|
SELECT DISTINCT ?uri ?title ?issn ?eissn ?quartile ?categories WHERE {{
|
|
VALUES ?wanted {{ {values} }}
|
|
{{ ?uri bibo:issn ?wanted }} UNION {{ ?uri bibo:eissn ?wanted }}
|
|
?uri a bibo:Journal; dct:title ?title; dct:identifier ?id; ou:sjrMejorCuartil ?quartile .
|
|
FILTER(STRSTARTS(STR(?id), "{year}-"))
|
|
OPTIONAL {{ ?uri bibo:issn ?issn }} OPTIONAL {{ ?uri bibo:eissn ?eissn }}
|
|
OPTIONAL {{ ?uri ou:categoriaSJR ?categories }}
|
|
}} LIMIT 10001'''
|
|
try:
|
|
response = requests.get("https://opendata.unex.es/sparql/", timeout=45,
|
|
params={"query": query, "format": "application/sparql-results+json"},
|
|
headers={"Accept": "application/sparql-results+json"})
|
|
response.raise_for_status()
|
|
bindings = response.json()["results"]["bindings"]
|
|
if len(bindings) >= 10000:
|
|
raise ValueError("Truncated ranking query")
|
|
path.write_text(json.dumps(bindings, ensure_ascii=False), encoding="utf8")
|
|
except (requests.RequestException, ValueError, KeyError, TypeError):
|
|
failures.append({"year": year, "batch": offset, "status": "ranking_source_unavailable"})
|
|
continue
|
|
for item in bindings:
|
|
value = lambda field: item.get(field, {}).get("value", "")
|
|
url = value("uri")
|
|
if not url.startswith(f"https://opendata.unex.es/recurso/ciencia-tecnologia/investigacion/revistas/Revista/{year}-"):
|
|
raise ValueError("Ranking year/identity mismatch")
|
|
record = {"journal": value("title"), "issns": sorted({value("issn"), value("eissn")} - {""}),
|
|
"quartile": value("quartile"), "categories": value("categories"), "source_url": url}
|
|
if record["quartile"] not in {"Q1", "Q2", "Q3", "Q4"}:
|
|
continue
|
|
previous = records.get(url)
|
|
if previous:
|
|
record["journal"] = min(previous["journal"], record["journal"])
|
|
record["issns"] = sorted(set(previous["issns"]) | set(record["issns"]))
|
|
record["categories"] = "; ".join(sorted({category.strip() for value in
|
|
(previous["categories"], record["categories"]) for category in value.split(";") if category.strip()}))
|
|
record["conflict"] = previous.get("conflict", False) or previous["quartile"] != record["quartile"]
|
|
records[url] = record
|
|
rankings[year] = list(records.values())
|
|
print(f"SJR {year}: {len(records)} matching journals", flush=True)
|
|
return index_sjr(rankings), failures
|
|
|
|
|
|
def resolve_datacite(doi, timeout):
|
|
url = "https://api.datacite.org/dois/" + quote(doi, safe="")
|
|
result = {"doi": doi, "status": "doi_lookup_failed", "source_url": url, "datacite_checked": True}
|
|
for attempt in range(3):
|
|
try:
|
|
response = requests.get(url, timeout=timeout, allow_redirects=False,
|
|
headers={"User-Agent": "MIEM-workers/0.18 publication-quartiles", "Accept": "application/json"})
|
|
if response.status_code == 404:
|
|
return {**result, "status": "doi_not_found"}
|
|
if response.status_code == 429 or response.status_code >= 500:
|
|
time.sleep(min(10, 2 ** attempt))
|
|
continue
|
|
response.raise_for_status()
|
|
metadata = response.json()["data"]["attributes"]
|
|
if _doi(metadata.get("doi", "")) != doi:
|
|
return {**result, "status": "doi_identity_conflict"}
|
|
types = metadata.get("types", {})
|
|
general = types.get("resourceTypeGeneral")
|
|
journal_article = general == "JournalArticle" or (general in {None, "Text"} and types.get("citeproc") == "article-journal")
|
|
container = metadata.get("container") or {}
|
|
identifiers = []
|
|
if container.get("identifierType") == "ISSN":
|
|
identifiers.append(container.get("identifier", ""))
|
|
for related in metadata.get("relatedIdentifiers", []):
|
|
if related.get("relatedIdentifierType") == "ISSN" and related.get("relationType") in {"IsPartOf", "IsPublishedIn"}:
|
|
identifiers.append(related.get("relatedIdentifier", ""))
|
|
identifiers = sorted({value[:4] + "-" + value[-4:] for value in identifiers if isinstance(value, str)
|
|
and re.fullmatch(r"\d{4}-?\d{3}[\dX]", value, re.I)})
|
|
year = metadata.get("publicationYear")
|
|
years = [int(year)] if str(year).isdigit() else []
|
|
return {**result, "status": "resolved", "type": "journal-article" if journal_article else "non-journal-material",
|
|
"issns": identifiers, "journals": [container["title"]] if isinstance(container.get("title"), str) else [], "years": years}
|
|
except (requests.RequestException, ValueError, KeyError, TypeError, AttributeError):
|
|
if attempt < 2:
|
|
time.sleep(2 ** attempt)
|
|
return result
|
|
|
|
|
|
def resolve_doi(doi, cache, timeout=20):
|
|
path = cache / (hashlib.sha256(doi.encode()).hexdigest() + ".json")
|
|
if path.exists():
|
|
result = json.loads(path.read_text(encoding="utf8"))
|
|
if result.get("doi") == doi and (result.get("status") == "resolved"
|
|
or (result.get("status") == "doi_not_found" and result.get("datacite_checked"))):
|
|
return result
|
|
url = "https://api.crossref.org/works/" + quote(doi, safe="")
|
|
result = {"doi": doi, "status": "doi_lookup_failed", "source_url": url}
|
|
for attempt in range(3):
|
|
try:
|
|
response = requests.get(url, timeout=timeout, allow_redirects=False,
|
|
headers={"User-Agent": "MIEM-workers/0.18 publication-quartiles",
|
|
"Accept": "application/json"})
|
|
if response.status_code == 404:
|
|
result = resolve_datacite(doi, timeout)
|
|
break
|
|
if response.status_code == 429 or response.status_code >= 500:
|
|
time.sleep(min(10, 2 ** attempt))
|
|
continue
|
|
response.raise_for_status()
|
|
message = response.json()["message"]
|
|
if _doi(message.get("DOI", "")) != doi:
|
|
result["status"] = "doi_identity_conflict"
|
|
break
|
|
years = set()
|
|
for field in ("published", "published-print", "published-online", "issued"):
|
|
for parts in message.get(field, {}).get("date-parts", []):
|
|
if parts and type(parts[0]) is int:
|
|
years.add(parts[0])
|
|
result.update(status="resolved", type=message.get("type"),
|
|
journals=[v for v in message.get("container-title", []) if isinstance(v, str)],
|
|
issns=[v for v in message.get("ISSN", []) if isinstance(v, str)
|
|
and re.fullmatch(r"\d{4}-\d{3}[\dX]", v, re.I)], years=sorted(years))
|
|
break
|
|
except (requests.RequestException, ValueError, KeyError, TypeError, AttributeError):
|
|
if attempt < 2:
|
|
time.sleep(2 ** attempt)
|
|
if result["status"] in {"resolved", "doi_not_found"}:
|
|
temporary = path.with_suffix(".tmp")
|
|
temporary.write_text(json.dumps(result, ensure_ascii=False), encoding="utf8")
|
|
temporary.replace(path)
|
|
return result
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--apply", action="store_true", help="Commit changes; default is read-only")
|
|
parser.add_argument("--lookup-doi", action="store_true")
|
|
parser.add_argument("--all-systems", action="store_true", help="Collect additional systems/categories even for publications with known quartiles")
|
|
parser.add_argument("--check-vak", action="store_true", help="Check the supplied K1/K2 snapshots")
|
|
parser.add_argument("--allow-latest", action="store_true", help="Use latest available metric if the publication year is unavailable")
|
|
parser.add_argument("--sjr-history", action="store_true", help="Check UNEx/SCImago historical quartiles by DOI ISSNs")
|
|
parser.add_argument("--scopus-title-history", type=Path, help="Official Scopus August 2026 source-title XLSX for explicit ISSN succession")
|
|
parser.add_argument("--cache-dir", type=Path, required=True)
|
|
parser.add_argument("--report", type=Path, required=True)
|
|
parser.add_argument("--workers", type=int, choices=range(1, 9), default=4)
|
|
args = parser.parse_args()
|
|
pending = args.report.with_suffix(args.report.suffix + ".pending")
|
|
if args.scopus_title_history and args.scopus_title_history.resolve() in {args.report.resolve(), pending.resolve()}:
|
|
raise ValueError("Report must not overwrite the source title list")
|
|
args.cache_dir.mkdir(parents=True, exist_ok=True)
|
|
args.report.parent.mkdir(parents=True, exist_ok=True)
|
|
# Verify report destination before changing any data.
|
|
pending.write_text("", encoding="utf8")
|
|
engine = create_engine(os.environ["DATABASE_URL"])
|
|
table = Table("employee_publications", MetaData(), autoload_with=engine)
|
|
if "quartiles" not in table.c:
|
|
raise ValueError("Apply migration 010 before enrichment")
|
|
fields = ["id", "year", "doi_url", "raw_data", "quartiles", "description"]
|
|
if args.check_vak:
|
|
if "vak_categories" not in table.c:
|
|
raise ValueError("Apply migration 011 before VAK enrichment")
|
|
fields.extend(["publication_type", "vak_categories"])
|
|
columns = [table.c[field] for field in fields]
|
|
with engine.connect() as connection:
|
|
rows = [dict(row) for row in connection.execute(select(*columns)).mappings()]
|
|
def needs_lookup(row):
|
|
raw = row["raw_data"] if isinstance(row["raw_data"], dict) else {}
|
|
names = _journal_names(raw)
|
|
description = row["description"] if isinstance(row["description"], dict) else raw.get("description", {})
|
|
if isinstance(description, dict):
|
|
names.extend(_description_journals(description))
|
|
ids = [raw[key] for key in ("issn", "eissn", "ISSN", "EISSN") if isinstance(raw.get(key), str)]
|
|
payload = merge(row["quartiles"], hse_entries(row["raw_data"], row["year"]))
|
|
return (args.all_systems or not entries(payload) or (payload.get("lookup") or {}).get("status") == "found_latest"
|
|
or bool(identity_issues(payload, row["doi_url"], names, ids)))
|
|
identifiers = sorted({_doi(row["doi_url"] or "") for row in rows
|
|
if needs_lookup(row)} - {""})
|
|
resolved = {}
|
|
if args.lookup_doi:
|
|
with ThreadPoolExecutor(max_workers=args.workers) as pool:
|
|
tasks = {pool.submit(resolve_doi, doi, args.cache_dir): doi for doi in identifiers}
|
|
for task in as_completed(tasks):
|
|
resolved[tasks[task]] = task.result()
|
|
if len(resolved) % 100 == 0 or len(resolved) == len(identifiers):
|
|
print(f"DOI checked: {len(resolved)}/{len(identifiers)}", flush=True)
|
|
rankings, ranking_failures = {}, []
|
|
if args.scopus_title_history:
|
|
by_issn, by_title = load_title_history(args.scopus_title_history)
|
|
resolved = {doi: with_title_history(metadata, by_issn, by_title, SOURCE_TITLES_URL)
|
|
for doi, metadata in resolved.items()}
|
|
if args.lookup_doi and args.sjr_history:
|
|
issns = sorted({value for metadata in resolved.values() if metadata.get("type") == "journal-article"
|
|
for value in metadata.get("ranking_issns", metadata.get("issns", []))})
|
|
years = {row["year"] for row in rows if needs_lookup(row) and row["year"] is not None}
|
|
if args.allow_latest:
|
|
years.add(2025)
|
|
rankings, ranking_failures = load_sjr(issns, years, args.cache_dir)
|
|
counts = Counter()
|
|
report = []
|
|
for row in rows:
|
|
doi = _doi(row["doi_url"] or "")
|
|
previous_lookup = (row["quartiles"].get("lookup") or {}) if row["quartiles"] else {}
|
|
metadata = resolved.get(doi, previous_lookup if doi and _doi(previous_lookup.get("doi") or "") == doi else {})
|
|
additions = hse_entries(row["raw_data"], row["year"])
|
|
payload = merge(row["quartiles"], additions)
|
|
if additions:
|
|
counts["hse_quartiles"] += 1
|
|
if args.lookup_doi and needs_lookup(row):
|
|
doi = _doi(row["doi_url"] or "")
|
|
metadata = resolved.get(doi, {"status": "invalid_doi" if row["doi_url"] else "missing_doi"})
|
|
status = metadata["status"]
|
|
if status == "resolved":
|
|
found, status = catalog_entries(metadata, row["year"], args.allow_latest)
|
|
if rankings:
|
|
sjr, sjr_status = sjr_entries(metadata, row["year"], rankings, args.allow_latest)
|
|
found.extend(sjr)
|
|
if not found and status == "historical_quartile_unavailable":
|
|
status = sjr_status
|
|
if found:
|
|
status = "found_latest" if any(item["metric_year"] != row["year"] for item in found) else "found"
|
|
payload = merge(payload, found)
|
|
payload["lookup"] = {**metadata, "status": status, "requested_year": row["year"],
|
|
"checked_on": date.today().isoformat()}
|
|
counts[status] += 1
|
|
counts["with_quartile" if entries(payload) else "without_quartile"] += 1
|
|
vak_payload = row.get("vak_categories")
|
|
if args.check_vak:
|
|
raw = row["raw_data"] if isinstance(row["raw_data"], dict) else {}
|
|
imported = raw.get("csv_enrichment", {})
|
|
names = [*_journal_names(raw), *(_journal_names(imported) if isinstance(imported, dict) else []),
|
|
*(metadata.get("journals", []) if metadata.get("type") in {"journal-article", "journal"} else [])]
|
|
if row["publication_type"] == "ARTICLE" and metadata.get("type") in {None, "journal-article", "journal"}:
|
|
description = row["description"] if isinstance(row["description"], dict) else raw.get("description", {})
|
|
if isinstance(description, dict):
|
|
names.extend(_description_journals(description))
|
|
identifiers = list(metadata.get("issns", []))
|
|
for source in (raw, imported):
|
|
if isinstance(source, dict):
|
|
identifiers.extend(source[key] for key in ("issn", "eissn", "ISSN", "EISSN") if isinstance(source.get(key), str))
|
|
vak_payload = assess_vak(list(dict.fromkeys(names)), list(dict.fromkeys(identifiers)))
|
|
if row.get("vak_categories") is not None and row["vak_categories"] != vak_payload:
|
|
counts["vak_existing_value_kept"] += 1
|
|
vak_payload = row["vak_categories"]
|
|
counts["vak_" + vak_payload["status"]] += 1
|
|
if payload == row["quartiles"] and vak_payload == row.get("vak_categories"):
|
|
continue
|
|
if args.apply:
|
|
# Only this field is changed; parser-owned data and existing evidence survive.
|
|
with engine.begin() as connection:
|
|
current = dict(connection.execute(select(*columns).where(table.c.id == row["id"])
|
|
.with_for_update()).mappings().one())
|
|
if any(current[field] != row[field] for field in fields if field != "id"):
|
|
counts["concurrent_change_skipped"] += 1
|
|
continue
|
|
values = {"quartiles": payload, "updated_at": datetime.now(timezone.utc)}
|
|
if args.check_vak:
|
|
values["vak_categories"] = vak_payload
|
|
connection.execute(update(table).where(table.c.id == row["id"]).values(**values))
|
|
counts["changed"] += 1
|
|
report.append({"id": row["id"], "quartiles": payload, **({"vak_categories": vak_payload} if args.check_vak else {})})
|
|
summary = {"mode": "apply" if args.apply else "dry-run", "publications": len(rows),
|
|
"unique_doi_checked": len(resolved), "counts": dict(counts), "ranking_failures": ranking_failures}
|
|
pending.write_text(json.dumps({"summary": summary, "records": report}, ensure_ascii=False, indent=2), encoding="utf8")
|
|
pending.replace(args.report)
|
|
print(json.dumps(summary, ensure_ascii=False), flush=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except Exception as error:
|
|
# Connection errors can contain credentials; do not print their text.
|
|
print(f"Enrichment failed ({type(error).__name__}); completed records are retained; retry is safe.", flush=True)
|
|
raise SystemExit(1)
|