62 lines
2.8 KiB
Python
62 lines
2.8 KiB
Python
"""One-time extraction: python scripts/extract_k1_journals.py <categories.pdf>.
|
||
|
||
Requires pdfplumber for this development operation only; the service reads JSON.
|
||
"""
|
||
import argparse
|
||
from collections import Counter
|
||
from datetime import datetime
|
||
import hashlib
|
||
import json
|
||
from pathlib import Path
|
||
import re
|
||
|
||
import pdfplumber
|
||
|
||
|
||
def extract(source: Path, category: str = "К1") -> dict:
|
||
if category not in {"К1", "К2"}:
|
||
raise ValueError("Unsupported category")
|
||
rows = []
|
||
with pdfplumber.open(source) as document:
|
||
for page in document.pages:
|
||
for table in page.extract_tables():
|
||
for row in table:
|
||
if row[0] and row[0].isdigit():
|
||
if len(row) != 4 or not all(row) or not set(row[2].split()) <= {"К1", "К2", "К3"}:
|
||
raise ValueError(f"Invalid category row: {row}")
|
||
# Some entries include previous categories; use the final category.
|
||
row[2] = row[2].split()[-1]
|
||
row[3] = row[3].splitlines()[-1]
|
||
datetime.strptime(row[3], "%d.%m.%Y")
|
||
rows.append(row)
|
||
if not rows or [int(row[0]) for row in rows] != list(range(1, len(rows) + 1)):
|
||
raise ValueError("Journal numbering is incomplete or duplicated")
|
||
journals = []
|
||
for row in rows:
|
||
if row[2] != category:
|
||
continue
|
||
name = " ".join(row[1].split())
|
||
journal = {"name": name, "categorized_on": row[3], "source_row": int(row[0])}
|
||
# Only an explicit administrative rename note supplies an alias.
|
||
# Parentheses in journal names (e.g. bilingual titles) stay unchanged.
|
||
rename = re.fullmatch(r"(.+?) \((до \d{2}\.\d{2}\.\d{4} г\. наименование в Перечне «(.+)»)\)", name)
|
||
if rename:
|
||
journal.update(name=rename[1], note=rename[2], aliases=[rename[3]])
|
||
journals.append(journal)
|
||
if not journals:
|
||
raise ValueError(f"No {category} journals found")
|
||
return {"source": source.name, "source_sha256": hashlib.sha256(source.read_bytes()).hexdigest(),
|
||
"source_rows": len(rows), "category_counts": dict(Counter(row[2] for row in rows)),
|
||
"journals": journals}
|
||
|
||
|
||
if __name__ == "__main__":
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument("source", type=Path)
|
||
parser.add_argument("--output", type=Path, default=Path("app/data/k1_journals.json"))
|
||
parser.add_argument("--category", choices=["К1", "К2"], default="К1")
|
||
args = parser.parse_args()
|
||
result = extract(args.source, args.category)
|
||
args.output.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", newline="\n")
|
||
print(f"Saved {len(result['journals'])} {args.category} journals; verified {result['source_rows']} source rows")
|