Michael Hermann
2bb70ae0ed
Historie 1949–2022 bleibt in lawgit, bis der GPU-Lauf endet; Schnitt ohne Lücke. Co-authored-by: Cursor <cursoragent@cursor.com>
115 lines
3.6 KiB
Python
115 lines
3.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Extrahiert Text aus BGBl-PDFs (Textschicht). Idempotent: vorhandene JSON überspringen.
|
|
|
|
python3 extract_bgbl.py --year 2025
|
|
python3 extract_bgbl.py --pdf bgbl/bgbl1/2025/bgbl1_2025_1.pdf
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
from concurrent.futures import ProcessPoolExecutor, as_completed
|
|
from pathlib import Path
|
|
|
|
from pypdf import PdfReader
|
|
|
|
from bgbl_util import BGBl_DIR, TEXT_DIR
|
|
|
|
MIN_CHARS_PER_PAGE = 80
|
|
|
|
|
|
def extract_pdf(pdf_path: Path) -> dict:
|
|
reader = PdfReader(str(pdf_path))
|
|
pages = []
|
|
for i, page in enumerate(reader.pages, start=1):
|
|
text = page.extract_text() or ""
|
|
pages.append({"page": i, "text": text, "chars": len(text.strip())})
|
|
total_chars = sum(p["chars"] for p in pages)
|
|
n = max(len(pages), 1)
|
|
quality = "ok" if total_chars / n >= MIN_CHARS_PER_PAGE else "needs_vision"
|
|
return {
|
|
"pdf": str(pdf_path),
|
|
"pages": len(pages),
|
|
"chars": total_chars,
|
|
"quality": quality,
|
|
"page_texts": [{"page": p["page"], "text": p["text"]} for p in pages],
|
|
"page_chars": [p["chars"] for p in pages],
|
|
}
|
|
|
|
|
|
def out_path_for(pdf_path: Path) -> Path:
|
|
try:
|
|
rel = pdf_path.resolve().relative_to(BGBl_DIR.resolve())
|
|
except ValueError:
|
|
rel = Path(pdf_path.name)
|
|
return TEXT_DIR / rel.with_suffix(".json")
|
|
|
|
|
|
def process(pdf_path: Path) -> tuple[str, str]:
|
|
dest = out_path_for(pdf_path)
|
|
try:
|
|
data = extract_pdf(pdf_path)
|
|
except Exception as exc:
|
|
return str(pdf_path), f"fehler: {type(exc).__name__}"
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
dest.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
|
|
return str(pdf_path), data["quality"]
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Text aus BGBl-PDFs extrahieren")
|
|
parser.add_argument("--year", type=int, action="append")
|
|
parser.add_argument("--kind", default="bgbl1")
|
|
parser.add_argument("--pdf", type=Path)
|
|
parser.add_argument("--workers", type=int, default=max(os.cpu_count() or 4, 2))
|
|
parser.add_argument("--force", action="store_true")
|
|
args = parser.parse_args()
|
|
|
|
pdfs: list[Path] = []
|
|
if args.pdf:
|
|
pdfs = [args.pdf]
|
|
elif args.year:
|
|
for year in args.year:
|
|
pdfs.extend(sorted((BGBl_DIR / args.kind / str(year)).glob("*.pdf")))
|
|
else:
|
|
pdfs = sorted(BGBl_DIR.rglob("*.pdf"))
|
|
|
|
if not pdfs:
|
|
print("Keine PDFs. Zuerst: python3 download_rechtbund.py --live-test")
|
|
return
|
|
|
|
if not args.force:
|
|
offen = [p for p in pdfs if not out_path_for(p).exists()]
|
|
if len(offen) < len(pdfs):
|
|
print(f"{len(pdfs) - len(offen)} schon extrahiert, übersprungen")
|
|
pdfs = offen
|
|
if not pdfs:
|
|
print("Nichts zu tun.")
|
|
return
|
|
|
|
print(f"{len(pdfs)} PDFs, {args.workers} Prozesse")
|
|
zaehler = {"ok": 0, "needs_vision": 0}
|
|
fehler = []
|
|
with ProcessPoolExecutor(max_workers=args.workers) as pool:
|
|
futures = [pool.submit(process, p) for p in pdfs]
|
|
for i, fut in enumerate(as_completed(futures), 1):
|
|
pfad, status = fut.result()
|
|
if status.startswith("fehler"):
|
|
fehler.append((pfad, status))
|
|
else:
|
|
zaehler[status] = zaehler.get(status, 0) + 1
|
|
if i % 50 == 0 or i == len(pdfs):
|
|
print(f" {i}/{len(pdfs)} — {zaehler}, {len(fehler)} Fehler", flush=True)
|
|
|
|
print(
|
|
f"Fertig: {zaehler['ok']} mit Textschicht, "
|
|
f"{zaehler['needs_vision']} brauchen OCR, {len(fehler)} Fehler"
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|