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>
304 lines
9.8 KiB
Python
304 lines
9.8 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Lädt Bundesgesetzblatt-Ausgaben ab 2023 vom amtlichen Portal recht.bund.de.
|
|
|
|
Lebende Quelle: ELI-Permalinks (PDF, optional XML). Schon vorhandene Ausgaben
|
|
werden übersprungen (idempotent, Resume).
|
|
|
|
python3 download_rechtbund.py --live-test
|
|
python3 download_rechtbund.py --year 2024 --year 2025
|
|
python3 download_rechtbund.py --from-year 2023
|
|
python3 download_rechtbund.py --issue 2026:1 --issue 2025:1
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import re
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from bgbl_util import (
|
|
META_DIR,
|
|
ROOT,
|
|
SESSION,
|
|
download_file,
|
|
fundstelle,
|
|
issue_id,
|
|
kind_to_teil,
|
|
load_json,
|
|
pdf_path,
|
|
polite_sleep,
|
|
save_json,
|
|
sha256_file,
|
|
)
|
|
|
|
ELI_PAGE = "https://www.recht.bund.de/eli/bund/{teil}/{year}/{number}/"
|
|
ELI_PDF = (
|
|
"https://www.recht.bund.de/eli/bund/{teil}/{year}/{number}/"
|
|
"regelungstext.pdf?__blob=publicationFile"
|
|
)
|
|
ELI_XML = "https://www.recht.bund.de/eli/bund/{teil}/{year}/{number}/regelungstext.xml"
|
|
|
|
LABEL_RE = re.compile(
|
|
r"<strong>\s*([^:<]+):\s*</strong>\s*<span>\s*([^<]+?)\s*</span>",
|
|
re.IGNORECASE | re.DOTALL,
|
|
)
|
|
TITLE_RE = re.compile(r'<meta\s+name="title"\s+content="([^"]+)"', re.IGNORECASE)
|
|
FUNDSTELLE_RE = re.compile(
|
|
r"BGBl\.\s+(\d{4})\s+([IV]+)\s+Nr\.\s+(\d+[a-z]?)\s+vom\s+(\d{2}\.\d{2}\.\d{4})",
|
|
re.IGNORECASE,
|
|
)
|
|
HASH_RE = re.compile(
|
|
r'c-tooltip__content--hash">\s*([0-9a-fA-F]{32})\s*<',
|
|
re.IGNORECASE,
|
|
)
|
|
H1_CONTENT_RE = re.compile(
|
|
r'<h1[^>]*class="[^"]*documentTitle[^"]*"[^>]*>(.*?)</h1>',
|
|
re.IGNORECASE | re.DOTALL,
|
|
)
|
|
|
|
MANIFEST_NAME = "rechtbund-manifest.json"
|
|
DEFAULT_STOP_AFTER_MISS = 3
|
|
|
|
|
|
def _strip_tags(html: str) -> str:
|
|
return re.sub(r"<[^>]+>", "", html).strip()
|
|
|
|
|
|
def _de_to_iso(value: str | None) -> str | None:
|
|
if not value:
|
|
return None
|
|
value = value.strip()
|
|
for fmt in ("%d.%m.%Y", "%Y-%m-%d"):
|
|
try:
|
|
return datetime.strptime(value, fmt).strftime("%Y-%m-%d")
|
|
except ValueError:
|
|
continue
|
|
return None
|
|
|
|
|
|
def parse_eli_html(html: str) -> dict:
|
|
"""Titel, Datum, Nummer, amtliche MD5 aus der ELI-Seite."""
|
|
labels = {}
|
|
for key, val in LABEL_RE.findall(html):
|
|
labels[re.sub(r"\s+", " ", key).strip()] = re.sub(r"\s+", " ", val).strip()
|
|
|
|
title = None
|
|
m = TITLE_RE.search(html)
|
|
if m:
|
|
title = m.group(1)
|
|
title = re.sub(r"^Bundesgesetzblatt Teil [IV]+\s*-\s*", "", title).strip()
|
|
title = title.rstrip(" -").strip()
|
|
h1 = H1_CONTENT_RE.search(html)
|
|
if h1:
|
|
title = _strip_tags(h1.group(1)) or title
|
|
|
|
fund = FUNDSTELLE_RE.search(html)
|
|
date = _de_to_iso(labels.get("Veröffentlichungsdatum"))
|
|
if not date and fund:
|
|
date = _de_to_iso(fund.group(4))
|
|
|
|
md5 = None
|
|
hm = HASH_RE.search(html)
|
|
if hm:
|
|
md5 = hm.group(1).lower()
|
|
|
|
return {
|
|
"titel": title,
|
|
"typ_amtlich": labels.get("Typ"),
|
|
"verkundung": date,
|
|
"ausfertigung": _de_to_iso(labels.get("Ausfertigungsdatum")),
|
|
"federfuehrung": labels.get("Federführung"),
|
|
"fna": labels.get("FNA"),
|
|
"sachgebiet": labels.get("Sachgebiet"),
|
|
"md5_amtlich": md5,
|
|
"fundstelle_seite": fund.group(0) if fund else None,
|
|
}
|
|
|
|
|
|
def fetch_issue(kind: str, year: int, number: int | str) -> dict | None:
|
|
"""Eine Ausgabe: HTML-Metadaten + PDF. None wenn die Nummer nicht existiert."""
|
|
teil = kind_to_teil(kind)
|
|
page_url = ELI_PAGE.format(teil=teil, year=year, number=number)
|
|
pdf_url = ELI_PDF.format(teil=teil, year=year, number=number)
|
|
xml_url = ELI_XML.format(teil=teil, year=year, number=number)
|
|
|
|
r = SESSION.get(page_url, timeout=45, allow_redirects=True)
|
|
if r.status_code == 404:
|
|
return None
|
|
if r.status_code != 200:
|
|
print(f" HTTP {r.status_code} {page_url}")
|
|
return None
|
|
if "text/html" not in (r.headers.get("content-type") or ""):
|
|
return None
|
|
|
|
meta = parse_eli_html(r.text)
|
|
dest = pdf_path(kind, year, number)
|
|
existed = dest.exists() and dest.stat().st_size > 0
|
|
if existed:
|
|
print(f" schon da: {dest.name}")
|
|
else:
|
|
print(f" lade PDF {kind} {year} Nr. {number}")
|
|
download_file(pdf_url, dest)
|
|
polite_sleep(0.3)
|
|
|
|
checksum = sha256_file(dest) if dest.exists() else None
|
|
xml_dest = dest.with_suffix(".xml")
|
|
xml_ok = False
|
|
if not xml_dest.exists():
|
|
xr = SESSION.get(xml_url, timeout=30)
|
|
if xr.status_code == 200 and (xr.headers.get("content-type") or "").startswith(
|
|
("application/xml", "text/xml", "application/xhtml")
|
|
):
|
|
xml_dest.write_bytes(xr.content)
|
|
xml_ok = True
|
|
polite_sleep(0.15)
|
|
else:
|
|
xml_ok = True
|
|
|
|
item = {
|
|
"id": issue_id(kind, year, number),
|
|
"kind": kind,
|
|
"year": year,
|
|
"number": int(str(number).rstrip("abcdefghijklmnopqrstuvwxyz"))
|
|
if str(number).rstrip("abcdefghijklmnopqrstuvwxyz").isdigit()
|
|
else number,
|
|
"number_raw": str(number),
|
|
"page": None,
|
|
"date": meta.get("verkundung"),
|
|
"title": meta.get("titel"),
|
|
"titel": meta.get("titel"),
|
|
"typ_amtlich": meta.get("typ_amtlich"),
|
|
"ausfertigung": meta.get("ausfertigung"),
|
|
"federfuehrung": meta.get("federfuehrung"),
|
|
"fna": meta.get("fna"),
|
|
"sachgebiet": meta.get("sachgebiet"),
|
|
"document_url": pdf_url,
|
|
"url": page_url,
|
|
"xml_url": xml_url if xml_ok else None,
|
|
"quelle": "recht.bund.de",
|
|
"fundstelle": fundstelle(kind, year, number, None),
|
|
"sha256": checksum,
|
|
"md5_amtlich": meta.get("md5_amtlich"),
|
|
"pdf": str(dest.relative_to(ROOT)) if dest.exists() else None,
|
|
"bytes": dest.stat().st_size if dest.exists() else 0,
|
|
"skipped": existed,
|
|
}
|
|
return item
|
|
|
|
|
|
def upsert_manifest(items: list[dict]) -> Path:
|
|
path = META_DIR / MANIFEST_NAME
|
|
existing = load_json(path, default=[]) or []
|
|
by_id = {e.get("id"): e for e in existing if e.get("id")}
|
|
for item in items:
|
|
by_id[item["id"]] = item
|
|
merged = sorted(
|
|
by_id.values(),
|
|
key=lambda e: (e.get("year") or 0, e.get("kind") or "", e.get("number") or 0, e.get("id") or ""),
|
|
)
|
|
save_json(path, merged)
|
|
year_path = None
|
|
years = {i["year"] for i in items}
|
|
kinds = {i["kind"] for i in items}
|
|
for kind in kinds:
|
|
for year in years:
|
|
subset = [i for i in merged if i.get("kind") == kind and i.get("year") == year]
|
|
year_path = META_DIR / f"{kind}-{year}-rechtbund.json"
|
|
save_json(year_path, subset)
|
|
return path
|
|
|
|
|
|
def scan_year(kind: str, year: int, *, limit: int | None, stop_after: int) -> list[dict]:
|
|
items = []
|
|
misses = 0
|
|
n = 1
|
|
while misses < stop_after:
|
|
if limit and len(items) >= limit:
|
|
break
|
|
print(f" prüfe {kind} {year} Nr. {n}")
|
|
item = fetch_issue(kind, year, n)
|
|
if item is None:
|
|
misses += 1
|
|
else:
|
|
misses = 0
|
|
items.append(item)
|
|
n += 1
|
|
polite_sleep(0.2)
|
|
return items
|
|
|
|
|
|
def parse_issue_arg(raw: str) -> tuple[int, str]:
|
|
if ":" not in raw:
|
|
raise argparse.ArgumentTypeError("Format Jahr:Nummer, z. B. 2025:1")
|
|
year_s, number = raw.split(":", 1)
|
|
return int(year_s), number
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="BGBl ab 2023 von recht.bund.de laden")
|
|
parser.add_argument("--year", type=int, action="append", help="Jahrgang scannen (ab 2023)")
|
|
parser.add_argument("--from-year", type=int, default=None, help="Scan ab diesem Jahr")
|
|
parser.add_argument("--to-year", type=int, default=None, help="Scan bis zu diesem Jahr")
|
|
parser.add_argument("--kind", choices=["bgbl1", "bgbl2"], default="bgbl1")
|
|
parser.add_argument("--limit", type=int, default=0, help="Max. Nummern je Jahr")
|
|
parser.add_argument("--issue", action="append", type=parse_issue_arg, help="Einzelausgabe Jahr:Nummer")
|
|
parser.add_argument(
|
|
"--live-test",
|
|
action="store_true",
|
|
help="Drei Ausgaben wirklich laden: 2024/1, 2025/1, 2026/1",
|
|
)
|
|
parser.add_argument("--stop-after-miss", type=int, default=DEFAULT_STOP_AFTER_MISS)
|
|
args = parser.parse_args()
|
|
|
|
META_DIR.mkdir(parents=True, exist_ok=True)
|
|
collected: list[dict] = []
|
|
|
|
if args.live_test:
|
|
args.issue = (args.issue or []) + [(2024, "1"), (2025, "1"), (2026, "1")]
|
|
|
|
if args.issue:
|
|
for year, number in args.issue:
|
|
if year < 2023:
|
|
print(f"Übersprungen {year}:{number} — vor 2023 gehört zu OffeneGesetze")
|
|
continue
|
|
print(f"=== {args.kind} {year} Nr. {number} ===")
|
|
item = fetch_issue(args.kind, year, number)
|
|
if item:
|
|
collected.append(item)
|
|
else:
|
|
print(f" nicht gefunden: {year} Nr. {number}")
|
|
polite_sleep(0.2)
|
|
|
|
years: list[int] = list(args.year or [])
|
|
if args.from_year:
|
|
ende = args.to_year or datetime.now().year
|
|
years.extend(range(max(args.from_year, 2023), ende + 1))
|
|
years = sorted(set(y for y in years if y >= 2023))
|
|
|
|
for year in years:
|
|
print(f"=== Scan {args.kind} {year} ===")
|
|
collected.extend(
|
|
scan_year(
|
|
args.kind,
|
|
year,
|
|
limit=args.limit or None,
|
|
stop_after=args.stop_after_miss,
|
|
)
|
|
)
|
|
|
|
if not collected and not args.issue and not years:
|
|
parser.error("Bitte --live-test, --issue, --year oder --from-year angeben")
|
|
|
|
if collected:
|
|
path = upsert_manifest(collected)
|
|
neu = sum(1 for i in collected if not i.get("skipped"))
|
|
print(f"\n{len(collected)} Ausgaben, {neu} neu geladen → {path}")
|
|
else:
|
|
print("Keine Ausgaben geladen.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|