181 lines
9.9 KiB
Python
181 lines
9.9 KiB
Python
"""Extract K1 journal specialties from the supplied VAK list (development only).
|
||
|
||
python scripts/extract_k1_specialties.py <list.pdf> [--rows-cache <json>]
|
||
Requires pdfplumber only for regeneration; requests read the generated JSON.
|
||
"""
|
||
import argparse
|
||
from collections import defaultdict
|
||
from datetime import date, datetime
|
||
import hashlib
|
||
import json
|
||
from pathlib import Path
|
||
import re
|
||
import unicodedata
|
||
|
||
SPECIALTIES = [
|
||
("1.2.2", "Математическое моделирование, численные методы и комплексы программ"),
|
||
("2.2.2", "Электронная компонентная база микро- и наноэлектроники, квантовых устройств"),
|
||
("2.3.6", "Методы и системы защиты информации, информационная безопасность"),
|
||
("2.3.1", "Системный анализ, управление и обработка информации, статистика"),
|
||
("2.3.7", "Компьютерное моделирование и автоматизация проектирования"),
|
||
("1.3.8", "Физика конденсированного состояния"),
|
||
("2.2.15", "Системы, сети и устройства телекоммуникаций"),
|
||
("1.1.2", "Дифференциальные уравнения и математическая физика"),
|
||
("2.2.13", "Радиотехника, в том числе системы и устройства телевидения"),
|
||
("2.3.2", "Вычислительные системы и их элементы"),
|
||
("2.2.14", "Антенны, СВЧ-устройства и их технологии"),
|
||
("2.3.5", "Математическое и программное обеспечение вычислительных систем, комплексов и компьютерных сетей"),
|
||
]
|
||
CODE = re.compile(r"(?<![\d.])([1-6]\.\d{1,2}\.\d{1,2})\.")
|
||
|
||
|
||
def journal_key(value):
|
||
# Administrative notes are separate from the title; do not remove ordinary parentheses.
|
||
value = re.split(r"\s*\(\s*(?:до\s+\d|перевод)", value, flags=re.I)[0]
|
||
return " ".join(re.findall(r"\w+", unicodedata.normalize("NFKC", value).casefold().replace("ё", "е")))
|
||
|
||
|
||
def extract_rows(source):
|
||
import pdfplumber
|
||
rows = []
|
||
with pdfplumber.open(source) as document:
|
||
for number, page in enumerate(document.pages, 1):
|
||
for table in page.find_tables():
|
||
# Ignore tiny tables detected inside the journal-name cell.
|
||
if table.bbox[2] - table.bbox[0] < 500:
|
||
continue
|
||
for row, values in zip(table.rows, table.extract()):
|
||
columns = [""] * 5
|
||
for cell, value in zip(row.cells, values):
|
||
if cell and value:
|
||
# This source uses five fixed columns; some pages contain
|
||
# additional cell borders within those columns.
|
||
column = sum(cell[0] >= edge - 1 for edge in (71.424, 234.387, 295.39, 498.48))
|
||
columns[column] += ("\n" if columns[column] else "") + value
|
||
rows.append({"page": number, "cells": columns})
|
||
page.close()
|
||
return rows
|
||
|
||
|
||
def parse_rows(rows):
|
||
journals = []
|
||
for row in rows:
|
||
cells = row["cells"]
|
||
if len(cells) != 5:
|
||
raise ValueError("Expected five VAK columns")
|
||
if re.fullmatch(r"\d+\.", cells[0].strip()):
|
||
journals.append({"row": int(cells[0].strip(". ")), "name": cells[1], "issn": cells[2],
|
||
"pages": [], "assignments": []})
|
||
elif not journals:
|
||
continue # The first-page header.
|
||
elif cells[0].strip():
|
||
raise ValueError(f"Unexpected journal number on page {row['page']}")
|
||
elif cells[1]:
|
||
journals[-1]["name"] += " " + cells[1]
|
||
journal = journals[-1]
|
||
if row["page"] not in journal["pages"]:
|
||
journal["pages"].append(row["page"])
|
||
if cells[2] and not re.fullmatch(r"\d+\.", cells[0].strip()):
|
||
journal["issn"] += " " + cells[2]
|
||
if cells[3]:
|
||
if cells[4]:
|
||
dates = re.sub(r"\s+", "", cells[4]).lower().replace("c", "с")
|
||
dates = dates.removeprefix("с")
|
||
bounds = dates.split("по")
|
||
if len(bounds) > 2:
|
||
raise ValueError(f"Unrecognized VAK dates: {cells[4]}")
|
||
def parse_date(value):
|
||
try:
|
||
return datetime.strptime(value, "%d.%m.%Y").date().isoformat()
|
||
except ValueError:
|
||
return None
|
||
start = parse_date(bounds[0])
|
||
end = parse_date(bounds[1]) if len(bounds) == 2 else None
|
||
if len(bounds) == 2 and end is None:
|
||
raise ValueError(f"Unrecognized end date: {cells[4]}")
|
||
assignment = {"codes": [], "valid_from": start, "valid_until": end,
|
||
"source_dates": cells[4], "pages": []}
|
||
if start is None:
|
||
# Preserve source typos, without inventing a historical start date.
|
||
assignment["date_note"] = "Дата начала неполная в источнике; текущее назначение определяется по отсутствию истёкшего срока окончания в Перечне."
|
||
journal["assignments"].append(assignment)
|
||
elif not journal["assignments"]:
|
||
raise ValueError(f"Missing initial dates in VAK row {journal['row']}")
|
||
assignment = journal["assignments"][-1]
|
||
assignment["codes"] = sorted(set(assignment["codes"]) | set(CODE.findall(cells[3])))
|
||
if row["page"] not in assignment["pages"]:
|
||
assignment["pages"].append(row["page"])
|
||
elif cells[4]:
|
||
raise ValueError("Dates without specialty text")
|
||
if not journals or [j["row"] for j in journals] != list(range(1, len(journals) + 1)):
|
||
raise ValueError("VAK numbering is incomplete or duplicated")
|
||
for journal in journals:
|
||
journal["name"] = " ".join(journal["name"].split())
|
||
journal["issn"] = " ".join(journal["issn"].split())
|
||
return journals
|
||
|
||
|
||
def active_codes(journal, as_of):
|
||
return sorted({code for a in journal["assignments"]
|
||
if (a["valid_from"] is None or a["valid_from"] <= as_of)
|
||
and (a["valid_until"] is None or as_of <= a["valid_until"])
|
||
for code in a["codes"]})
|
||
|
||
|
||
def build_reference(journals, k1, overrides, as_of):
|
||
by_name = defaultdict(list)
|
||
by_row = {j["row"]: j for j in journals}
|
||
for journal in journals:
|
||
by_name[journal_key(journal["name"])].append(journal)
|
||
reviewed = {o["k1_name"]: o for o in overrides}
|
||
if len(reviewed) != len(overrides) or not set(reviewed) <= {j["name"] for j in k1["journals"]}:
|
||
raise ValueError("Invalid or stale reviewed K1 matches")
|
||
result = []
|
||
for journal in k1["journals"]:
|
||
entry = {"name": journal["name"], "specialty_codes": [], "status": "not_found", "vak_entries": []}
|
||
matches = by_name[journal_key(journal["name"])]
|
||
if journal["name"] in reviewed:
|
||
override = reviewed[journal["name"]]
|
||
candidate = by_row[override["vak_row"]]
|
||
if candidate["name"] != override["vak_name"]:
|
||
raise ValueError("Reviewed source journal title changed")
|
||
matches = [candidate]
|
||
entry["match_note"] = override["reason"]
|
||
active = [j for j in matches if active_codes(j, as_of)]
|
||
if len(active) == 1:
|
||
entry.update(specialty_codes=active_codes(active[0], as_of), status="matched")
|
||
elif len(active) > 1:
|
||
entry["status"] = "ambiguous" # Do not union different journals with identical titles.
|
||
elif matches:
|
||
entry["status"] = "no_current_specialties"
|
||
entry["vak_entries"] = matches
|
||
result.append(entry)
|
||
return result
|
||
|
||
|
||
if __name__ == "__main__":
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument("source", type=Path)
|
||
parser.add_argument("--k1", type=Path, default=Path("app/data/k1_journals.json"))
|
||
parser.add_argument("--matches", type=Path, default=Path("scripts/k1_specialty_matches.json"))
|
||
parser.add_argument("--output", type=Path, default=Path("app/data/k1_specialties.json"))
|
||
parser.add_argument("--rows-cache", type=Path)
|
||
parser.add_argument("--as-of", default="2026-09-15")
|
||
args = parser.parse_args()
|
||
date.fromisoformat(args.as_of)
|
||
rows = json.loads(args.rows_cache.read_text(encoding="utf-8")) if args.rows_cache else extract_rows(args.source)
|
||
journals = parse_rows(rows)
|
||
k1 = json.loads(args.k1.read_text(encoding="utf-8"))
|
||
entries = build_reference(journals, k1, json.loads(args.matches.read_text(encoding="utf-8")), args.as_of)
|
||
result = {"schema_version": 1, "source": args.source.name,
|
||
"source_sha256": hashlib.sha256(args.source.read_bytes()).hexdigest(), "as_of": args.as_of,
|
||
"k1_source_sha256": hashlib.sha256(args.k1.read_bytes().replace(b"\r\n", b"\n")).hexdigest(), "k1_source_hash_eol": "LF", "source_rows": len(journals),
|
||
"specialties": [{"code": code, "name": name,
|
||
"branches": ["физико-математические"] if code == "1.1.2" else
|
||
["технические"] if code == "2.3.2" else
|
||
["технические", "физико-математические"]}
|
||
for code, name in SPECIALTIES], "journals": entries}
|
||
args.output.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", newline="\n")
|
||
from collections import Counter
|
||
print(len(entries), dict(Counter(j['status'] for j in entries)))
|