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>
99 lines
2.7 KiB
Python
99 lines
2.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Gemeinsame Pfade, HTTP-Session und Fundstellen für das gesetzblatt-Repo."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
|
|
ROOT = Path(__file__).resolve().parent
|
|
BGBl_DIR = ROOT / "bgbl"
|
|
META_DIR = ROOT / "timeline" / "meta"
|
|
TEXT_DIR = ROOT / "timeline" / "text"
|
|
PARSED_DIR = ROOT / "timeline" / "parsed"
|
|
DATA_DIR = ROOT / "data"
|
|
|
|
LAWGIT_ROOT = ROOT.parent / "lawgit"
|
|
LAWGIT_BGBL = LAWGIT_ROOT / "bgbl"
|
|
|
|
USER_AGENT = (
|
|
"Gesetzblatt/0.1 (+https://git.coded.law/michaelkole/gesetzblatt; "
|
|
"Forschungsparser, höflich, kein Massen-Crawler)"
|
|
)
|
|
|
|
SESSION = requests.Session()
|
|
SESSION.headers.update({"User-Agent": USER_AGENT})
|
|
|
|
|
|
def kind_to_teil(kind: str) -> str:
|
|
return "BGBl-1" if kind == "bgbl1" else "BGBl-2"
|
|
|
|
|
|
def kind_to_rom(kind: str) -> str:
|
|
return "I" if kind == "bgbl1" else "II"
|
|
|
|
|
|
def fundstelle(kind: str, year: int, number: int | str, page: int | None = None) -> str:
|
|
teil = kind_to_rom(kind)
|
|
if year >= 2023:
|
|
return f"BGBl. {year} {teil} Nr. {number}"
|
|
if page:
|
|
return f"BGBl. {year} {teil} S. {page}"
|
|
return f"BGBl. {year} {teil} Nr. {number}"
|
|
|
|
|
|
def issue_id(kind: str, year: int, number: int | str) -> str:
|
|
return f"{kind}-{year}-{number}-1"
|
|
|
|
|
|
def pdf_path(kind: str, year: int, number: int | str) -> Path:
|
|
return BGBl_DIR / kind / str(year) / f"{kind}_{year}_{number}.pdf"
|
|
|
|
|
|
def save_json(path: Path, obj) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(obj, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
|
|
|
|
def load_json(path: Path, default=None):
|
|
if not path.exists():
|
|
return default
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def sha256_file(path: Path) -> str:
|
|
h = hashlib.sha256()
|
|
with path.open("rb") as f:
|
|
for chunk in iter(lambda: f.read(1 << 16), b""):
|
|
h.update(chunk)
|
|
return h.hexdigest()
|
|
|
|
|
|
def sha256_bytes(data: bytes) -> str:
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def download_file(url: str, dest: Path, *, timeout: int = 120) -> bool:
|
|
"""Datei holen. True = neu geschrieben, False = schon da oder Fehler."""
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
if dest.exists() and dest.stat().st_size > 0:
|
|
return False
|
|
r = SESSION.get(url, timeout=timeout, stream=True)
|
|
if r.status_code != 200:
|
|
print(f" Fehler HTTP {r.status_code}: {url}")
|
|
return False
|
|
tmp = dest.with_suffix(dest.suffix + ".part")
|
|
with open(tmp, "wb") as f:
|
|
for chunk in r.iter_content(chunk_size=1 << 16):
|
|
f.write(chunk)
|
|
tmp.rename(dest)
|
|
return True
|
|
|
|
|
|
def polite_sleep(seconds: float = 0.25) -> None:
|
|
time.sleep(seconds)
|