193 lines
10 KiB
Python
193 lines
10 KiB
Python
"""Owner-run CSV enrichment; no app startup, .env loading, DDL or deletes."""
|
||
|
||
import argparse
|
||
import csv
|
||
import hashlib
|
||
import json
|
||
import re
|
||
import unicodedata
|
||
from collections import Counter, defaultdict
|
||
from pathlib import Path
|
||
|
||
|
||
def normalized(value):
|
||
return " ".join(unicodedata.normalize("NFKC", value or "").casefold().replace("ё", "е").split())
|
||
|
||
|
||
def doi(value):
|
||
return re.sub(r"^(?:https?://(?:dx\.)?doi\.org/|doi:\s*)", "", (value or "").strip(), flags=re.I).lower()
|
||
|
||
|
||
def read_rows(path):
|
||
with path.open(encoding="utf-8-sig", newline="") as source:
|
||
reader = csv.DictReader(source, delimiter=";")
|
||
required = {"Название публикации", "Год", "DOI", "Авторы НИУ ВШЭ"}
|
||
if not required <= set(reader.fieldnames or []):
|
||
raise ValueError("Missing required CSV headers")
|
||
rows = []
|
||
for number, row in enumerate(reader, 2):
|
||
if None in row or any(v is None for v in row.values()):
|
||
raise ValueError(f"Malformed CSV record {number}")
|
||
rows.append((number, {k.strip(): v.strip() for k, v in row.items()}))
|
||
return rows
|
||
|
||
|
||
def plan(rows, employees, publications):
|
||
names = defaultdict(list)
|
||
by_employee = defaultdict(list)
|
||
for employee in employees:
|
||
if normalized(employee["full_name"]):
|
||
names[normalized(employee["full_name"])].append(employee["id"])
|
||
for publication in publications:
|
||
by_employee[publication["employee_id"]].append(dict(publication))
|
||
actions, audit, seen = [], [], set()
|
||
for number, row in rows:
|
||
title, identifier = row["Название публикации"], doi(row["DOI"])
|
||
author_names = sorted(set(normalized(a) for a in row["Авторы НИУ ВШЭ"].split(";")) - {""})
|
||
year = int(row["Год"]) if row["Год"].isdigit() else None
|
||
if not title or not year or not 1000 <= year <= 9999 or not author_names or (identifier and not re.fullmatch(r"10\.\d{4,9}/\S+", identifier)):
|
||
audit.append({"row": number, "employee_id": None, "result": "skip", "reason": "invalid_fields"})
|
||
continue
|
||
for author in author_names:
|
||
ids = names.get(author, [])
|
||
entry = {"row": number, "employee_id": ids[0] if len(ids) == 1 else None}
|
||
def skip(reason):
|
||
audit.append({**entry, "result": "skip", "reason": reason})
|
||
if len(ids) != 1:
|
||
skip("employee_not_found" if not ids else "ambiguous_employee")
|
||
continue
|
||
employee_id = ids[0]
|
||
candidates = by_employee[employee_id]
|
||
exact = [p for p in candidates if identifier and doi(p.get("doi_url")) == identifier]
|
||
title_matches = [p for p in candidates if normalized(p["title"]) == normalized(title) and p.get("year") in (None, year)]
|
||
matches = {p["id"]: p for p in exact + title_matches}
|
||
if len(matches) > 1:
|
||
skip("ambiguous_publication")
|
||
continue
|
||
existing = next(iter(matches.values()), None)
|
||
if existing and identifier and doi(existing.get("doi_url")) and doi(existing["doi_url"]) != identifier:
|
||
skip("doi_conflict")
|
||
continue
|
||
if existing and existing.get("year") and existing["year"] != year:
|
||
skip("year_conflict")
|
||
continue
|
||
key = (employee_id, identifier or (normalized(title), year))
|
||
duplicate = key in seen
|
||
seen.add(key)
|
||
metadata = {k: v for k, v in row.items() if v}
|
||
raw = existing.get("raw_data") if existing else None
|
||
if raw is not None and not isinstance(raw, dict):
|
||
skip("unsupported_raw_data")
|
||
continue
|
||
raw = dict(raw or {})
|
||
old_metadata = raw.get("csv_enrichment", {})
|
||
if not isinstance(old_metadata, dict):
|
||
skip("unsupported_csv_metadata")
|
||
continue
|
||
conflicts = sorted(k for k, v in metadata.items() if old_metadata.get(k) and old_metadata[k] != v)
|
||
if conflicts:
|
||
entry["conflict_fields"] = conflicts
|
||
merged = dict(old_metadata)
|
||
for k, v in metadata.items():
|
||
if not merged.get(k):
|
||
merged[k] = v
|
||
raw["csv_enrichment"] = merged
|
||
values = {"raw_data": raw}
|
||
if identifier and (not existing or not existing.get("doi_url")):
|
||
values["doi_url"] = "https://doi.org/" + identifier
|
||
if not existing or not existing.get("year"):
|
||
values["year"] = year
|
||
if not existing or not existing.get("title"):
|
||
values["title"] = title
|
||
if existing:
|
||
values = {k: v for k, v in values.items() if existing.get(k) != v}
|
||
if not values:
|
||
skip("metadata_conflict_first_value_kept" if conflicts else "csv_duplicate" if duplicate else "already_present")
|
||
continue
|
||
if isinstance(existing["id"], str):
|
||
actions[int(existing["id"].split("-")[1])]["values"].update(values)
|
||
existing.update(values)
|
||
skip("metadata_conflict_first_value_kept" if conflicts else "merged_csv_duplicate")
|
||
continue
|
||
action = {"result": "update", "id": existing["id"], "values": values}
|
||
existing.update(values)
|
||
else:
|
||
source_hash = hashlib.sha256(json.dumps(key, ensure_ascii=False, sort_keys=True).encode()).hexdigest()
|
||
values.update(employee_id=employee_id, source_hash=source_hash)
|
||
action = {"result": "insert", "values": values}
|
||
by_employee[employee_id].append({**values, "id": f"planned-{len(actions)}"})
|
||
actions.append(action)
|
||
audit.append({**entry, "result": action["result"], "reason": "metadata_conflict_first_value_kept" if conflicts else ""})
|
||
return actions, audit
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument("file", type=Path)
|
||
parser.add_argument("--apply", action="store_true", help="Commit; default is read-only dry-run")
|
||
parser.add_argument("--report", type=Path, required=True)
|
||
args = parser.parse_args()
|
||
pending_report = args.report.with_suffix(args.report.suffix + ".pending")
|
||
if args.file.resolve() in (args.report.resolve(), pending_report.resolve()):
|
||
raise ValueError("Report must not overwrite input")
|
||
source_digest = hashlib.sha256(args.file.read_bytes()).hexdigest()
|
||
rows = read_rows(args.file)
|
||
# libpq environment/service only; never load application settings or .env.
|
||
import psycopg
|
||
from psycopg.rows import dict_row
|
||
from psycopg.types.json import Jsonb
|
||
from psycopg import sql
|
||
|
||
# Check report access before any possible commit; pending file is not a success report.
|
||
with pending_report.open("x", encoding="utf-8"):
|
||
pass
|
||
with psycopg.connect("", row_factory=dict_row) as connection:
|
||
connection.execute("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE" + ("" if args.apply else " READ ONLY"))
|
||
connection.execute("SET LOCAL search_path = pg_catalog, public")
|
||
connection.execute("SET LOCAL statement_timeout = '60s'")
|
||
connection.execute("SET LOCAL lock_timeout = '5s'")
|
||
if args.apply:
|
||
locked = connection.execute("SELECT pg_try_advisory_xact_lock(716203841)").fetchone()
|
||
if not next(iter(locked.values())):
|
||
raise RuntimeError("Another import is running")
|
||
employees = connection.execute("SELECT id, full_name FROM public.employees").fetchall()
|
||
publications = connection.execute("SELECT id, employee_id, title, year, doi_url, raw_data FROM public.employee_publications").fetchall()
|
||
actions, audit = plan(rows, employees, publications)
|
||
if args.apply:
|
||
for action in actions:
|
||
values = dict(action["values"])
|
||
if "raw_data" in values:
|
||
values["raw_data"] = Jsonb(values["raw_data"])
|
||
columns = list(values)
|
||
if action["result"] == "insert":
|
||
query = sql.SQL("INSERT INTO public.employee_publications ({}, created_at, updated_at) VALUES ({}, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)").format(
|
||
sql.SQL(", ").join(map(sql.Identifier, columns)), sql.SQL(", ").join(sql.Placeholder() for _ in columns))
|
||
connection.execute(query, list(values.values()))
|
||
else:
|
||
query = sql.SQL("UPDATE public.employee_publications SET {}, updated_at = CURRENT_TIMESTAMP WHERE id = %s").format(
|
||
sql.SQL(", ").join(sql.SQL("{} = %s").format(sql.Identifier(k)) for k in columns))
|
||
result = connection.execute(query, [*values.values(), action["id"]])
|
||
if result.rowcount != 1:
|
||
raise RuntimeError("Publication changed concurrently")
|
||
if hashlib.sha256(args.file.read_bytes()).hexdigest() != source_digest:
|
||
raise RuntimeError("Input changed during import")
|
||
summary = {"mode": "apply" if args.apply else "dry-run", "csv_rows": len(rows),
|
||
"sha256": source_digest,
|
||
"counts": {key: sum(item["result"] == key for item in audit) for key in ("insert", "update", "skip")},
|
||
"metadata_conflicts": sum(bool(item.get("conflict_fields")) for item in audit),
|
||
"skip_reasons": dict(Counter(item["reason"] for item in audit if item["result"] == "skip"))}
|
||
pending_report.write_text(json.dumps({"summary": summary, "records": audit}, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
if not args.apply:
|
||
connection.rollback()
|
||
pending_report.replace(args.report)
|
||
print(json.dumps(summary, ensure_ascii=False))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
main()
|
||
except Exception as error:
|
||
# Driver errors may contain connection credentials or personal data.
|
||
print(f"Import failed ({type(error).__name__}); transaction not partially applied. Check configuration and permissions.")
|
||
raise SystemExit(1)
|