michaelkole/gesetzblatt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initiales gesetzblatt-Repo: Parser-Zweig ab 2023 von recht.bund.de.

Browse Source 
Historie 1949–2022 bleibt in lawgit, bis der GPU-Lauf endet; Schnitt ohne Lücke.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Michael Hermann
2026-08-16 11:56:05 +02:00
commit 2bb70ae0ed
23 changed files with 1870 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
.env.example Normal file
View File

@@ -0,0 +1,7 @@
# Optional. Ohne diese Werte läuft der Parser nur heuristisch.
# LLM-Aufrufe gegen RunPod nicht starten, solange der Historien-Parse in lawgit läuft.
# LLM_BASE_URL=https://<pod-id>-8000.proxy.runpod.net/v1
# LLM_API_KEY=not-needed
# LLM_MODEL=Qwen/Qwen2.5-72B-Instruct-AWQ
# LLM_TIMEOUT=180

36
.gitignore vendored Normal file
View File

@@ -0,0 +1,36 @@
# Secrets
.env
.env.*
!.env.example
# Python
__pycache__/
*.py[cod]
*$py.class
.Python
.venv/
venv/
env/
*.egg-info/
# Binaries / Rohdaten (PDFs, Archive, XML-Dumps)
*.pdf
*.zip
*.tar.bz2
*.tar.gz
bgbl/**/*.xml
data/
timeline/text/
timeline/parsed/
# IDE / OS
.vscode/
.idea/
.DS_Store
*.swp
*~
# Logs
*.log
*.tmp
*.bak

53
ARCHIVE.md Normal file
View File

@@ -0,0 +1,53 @@
# Archivierung von lawgit (vorbereitet, noch nicht ausführen)
**lawgit bleibt lebendig**, solange der GPU-Parse 1949–2022 auf RunPod läuft.
Dieses Dokument beschreibt nur, was *danach* zu tun ist. Jetzt weder das
Gitea-Repo auf „archived“ stellen noch Pods/Orchestrator anfassen.
## Warum ein Nachfolger?
lawgit ist der alte Arbeitsstand: GII-Markdown, UI, RunPod-Orchestrator,
OffeneGesetze-Download und der laufende Historien-Parse. Ab 2023 ist die
einzige amtliche Quelle **recht.bund.de**. Das gehört in dieses Repo
(`gesetzblatt`), ohne den Großeinsatz zu stören.
| | lawgit (alt) | gesetzblatt (dieses Repo) |
|---|---|---|
| Rolle jetzt | Historien-Parse 1949–2022, GII-Spiegel, UI | Parser-Zweig 2023+, sauberes Download/Parse |
| Nach `--holen` | archivieren | lebende Quelle + übernommene Historie |
## Wann archivieren?
Erst wenn **alle** Punkte zutreffen:
1. `parse_bgbl.py` auf dem Pod ist fertig (kein laufender Job).
2. Ergebnisse sind mit `deploy_unabhaengig.py --holen` nach `lawgit/timeline/parsed/` geholt.
3. Niemand braucht das lawgit-Remote mehr schreibend.
## Schritte nach dem Holen (dann, nicht jetzt)
Lokal, im **alten** Repo:
```bash
# nur zur Erinnerung — nicht ausführen, solange der Parse läuft
cd /Users/michaelhermann/Source/law/coded/lawgit
# Ergebnisse sichern, dann Gitea-UI: Repository → Settings → Archive
```
In Gitea (`https://git.coded.law/michaelkole/lawgit`):
1. Beschreibung setzen: „Archiv. Nachfolger: michaelkole/gesetzblatt.“
2. Repository auf **Archived** stellen (verhindert Pushes, löscht nichts).
3. Nicht löschen: der 1949–2022-Bestand (`bgbl/`, `timeline/parsed/`) ist die Historie.
Historie hierher andocken (nach Archivierung):
```bash
# PDFs nicht noch einmal von OffeneGesetze laden
rsync -a --dry-run /Users/michaelhermann/Source/law/coded/lawgit/bgbl/bgbl1/ \
/Users/michaelhermann/Source/law/coded/gesetzblatt/bgbl/bgbl1/
# parsed JSON der Historie analog nach timeline/parsed/ (nur 1949–2022)
```
Pods, Orchestrator und `parse_bgbl.py` auf RunPod bleiben unangetastet,
bis der Lauf von selbst endet.

110
README.md Normal file
View File

@@ -0,0 +1,110 @@
# gesetzblatt
Parser und Spiegel des **Bundesgesetzblatts**.
- **1949–2022:** [OffeneGesetze.de](https://offenegesetze.de/daten) (Historie, Bestand liegt in lawgit)
- **ab 2023:** [recht.bund.de](https://www.recht.bund.de) (einzige lebende amtliche Quelle)
Kein Nachbau der lawgit-UI und kein RunPod-Orchestrator. Ziel ist ein
schmaler, idempotenter Zweig: Download → Text → Event-JSON, an den Git-Spiegel
und eine spätere Oberfläche andocken können.
## Schnitt, keine Lücke
2022 ist das letzte OffeneGesetze-Jahr. 2023 I Nr. 1 ist die erste amtliche
elektronische Nummer. Die Jahre werden nicht doppelt geladen. Details:
[docs/QUELLEN.md](docs/QUELLEN.md).
lawgit bleibt der alte Arbeitsstand plus laufender GPU-Parse 1949–2022.
Archivieren erst nach `--holen` — siehe [ARCHIVE.md](ARCHIVE.md).
## Schnellstart
```bash
cd /Users/michaelhermann/Source/law/coded/gesetzblatt
python3 -m pip install -r requirements.txt
# Live-Test: drei echte Ausgaben (2024/1, 2025/1, 2026/1)
python3 sync.py --live-test
# Heuristik-Test ohne Netz
python3 -m unittest tests.test_parse_heuristik
```
## Nachziehen ab 2023
```bash
python3 sync.py --from-year 2023
```
Das ist der Befehl für den laufenden Betrieb: fehlende Nummern von
recht.bund.de holen, Text extrahieren, Event-JSON schreiben. Schon vorhandene
Ausgaben werden übersprungen.
Einzelne Schritte:
```bash
python3 download_rechtbund.py --from-year 2023
python3 extract_bgbl.py --year 2023 --year 2024 --year 2025 --year 2026
python3 parse_bgbl.py --von 2023
```
`--llm` (RunPod/vLLM über `LLM_BASE_URL`) ist eingebaut, aber **nicht** der
Standard. Nicht setzen, solange lawgit auf der GPU parst. Ohne Flag bleibt
es bei der Heuristik.
## Historie OffeneGesetze
```bash
python3 download_offenegesetze.py --hinweis
```
Schreibt nur ein Quellen-Manifest. Die ~5500 PDFs in lawgit werden nicht
kopiert. Nach Archivierung von lawgit den Bestand hierher ziehen (siehe
ARCHIVE.md), nicht erneut von OffeneGesetze laden.
## Event-JSON
Gleicher Schnitt wie bisher in lawgit (`timeline/parsed/{id}.json`):
| Feld | Bedeutung |
|---|---|
| `id` | z. B. `bgbl1-2025-1-1` |
| `fundstelle` | `BGBl. 2025 I Nr. 1` |
| `verkundung` | ISO-Datum |
| `titel` | amtlicher Titel |
| `typ` | Erlass / Änderung / Verordnung / … |
| `inkrafttreten` | Liste `{datum, geltung, hinweis?}` |
| `betroffene_normen` | Stammgesetze mit `kurz`, `operation` |
| `quelle` | `recht.bund.de` oder `offenegesetze` |
| `parser` | `heuristik` oder `llm` |
| `sha256` | Prüfsumme der PDF (ab 2023) |
## Struktur
```
gesetzblatt/
├── download_rechtbund.py # ab 2023, recht.bund.de
├── download_offenegesetze.py # 1949–2022, Hinweis + optionaler Nachzug
├── extract_bgbl.py
├── parse_bgbl.py
├── llm_client.py
├── sync.py
├── bgbl/ # PDFs, gitignoriert
├── timeline/meta/ # Manifeste (klein, im Git)
├── timeline/text/ # Extrakte, gitignoriert
└── timeline/parsed/ # Event-JSON, gitignoriert
```
## Remote
```
https://git.coded.law/michaelkole/gesetzblatt.git
```
Falls das Remote noch fehlt:
```bash
git remote add origin https://git.coded.law/michaelkole/gesetzblatt.git
git push -u origin main
```

3
bgbl/README.md Normal file
View File

@@ -0,0 +1,3 @@
Roh-PDFs (gitignoriert). Ab 2023: `bgbl1/{Jahr}/bgbl1_{Jahr}_{Nummer}.pdf` von recht.bund.de.
Historie 1949–2022 nicht hier duplizieren — sie liegt in lawgit, siehe `docs/QUELLEN.md`.

98
bgbl_util.py Normal file
View File

@@ -0,0 +1,98 @@
#!/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)

43
docs/QUELLEN.md Normal file
View File

@@ -0,0 +1,43 @@
# Quellen und Schnitt 2022/2023
Zwei getrennte Lieferanten, **keine überlappenden Jahre**.
| Zeitraum | Quelle | Skript | Was liegt schon vor |
|---|---|---|---|
| 1949–2022 | [OffeneGesetze.de](https://offenegesetze.de/daten) (API, Jahresarchive, PDFs) | `download_offenegesetze.py` | ~5500 PDFs Teil I in **lawgit** unter `bgbl/bgbl1/1949`–`2022` |
| 2023–heute | [recht.bund.de](https://www.recht.bund.de) (amtliche elektronische Verkündung) | `download_rechtbund.py` | hier nachziehen |
Ab 1. Januar 2023 erscheint das Bundesgesetzblatt nur noch elektronisch. Die Fundstelle ist **Jahr + Teil + Nummer** (nicht mehr die Seitenzahl). OffeneGesetze deckt diesen Amtsteil nicht ab — deshalb der neue Zweig.
## recht.bund.de (lebende Quelle)
Amtliche Doku: [Datenabruf](https://www.recht.bund.de/de/service/webservice/webservice_node.html).
ELI-Schema:
```
https://www.recht.bund.de/eli/bund/BGBl-1/{Jahr}/{Nummer}/
https://www.recht.bund.de/eli/bund/BGBl-1/{Jahr}/{Nummer}/regelungstext.pdf?__blob=publicationFile
```
Existenzprüfung über den HTTP-Status der ELI-Seite (404 = Nummer gibt es nicht). Nummern laufen je Jahr bei 1 los; Buchstabenzusätze (`1a`) sind möglich.
XML (LegalDocML.de) ist angekündigt, aber noch nicht flächendeckend. Der Downloader versucht die XML-URL und ignoriert 404.
Metadaten von der ELI-HTML-Seite: Titel, Veröffentlichungsdatum, Ausfertigungsdatum, Typ, Federführung, amtliche MD5. Zusätzlich SHA-256 der gespeicherten PDF.
## OffeneGesetze (Historie)
API: `https://api.offenegesetze.de/v1/veroeffentlichung/`
Die PDFs **nicht** ein zweites Mal herunterladen, solange sie in lawgit liegen. Nach dem Ende des GPU-Laufs 1949–2022 und der Archivierung von lawgit: Bestand hierher übernehmen (kopieren/verschieben), dann `timeline/parsed/` aus lawgit als Historie andocken.
```bash
python3 download_offenegesetze.py --hinweis
```
## Lückenfreiheit
1. Historie endet mit **BGBl. 2022** (OffeneGesetze / lawgit-Parse).
2. Der neue Zweig beginnt mit **BGBl. 2023 I Nr. 1** (recht.bund.de).
3. `sync.py --from-year 2023` füllt 2023, 2024, 2025, … ohne die Historie anzufassen.

175
download_offenegesetze.py Normal file
View File

@@ -0,0 +1,175 @@
#!/usr/bin/env python3
"""
Historische BGBl-PDFs 1949–2022 von OffeneGesetze.de.
Der Bestand liegt bereits lokal in lawgit (`bgbl/bgbl1/1949`–`2022`, ~5500 PDFs).
Dieses Skript dupliziert ihn nicht. Es schreibt ein Quellen-Manifest und kann
einzelne Jahrgänge nachziehen, falls sie hier wirklich gebraucht werden.
python3 download_offenegesetze.py --hinweis
python3 download_offenegesetze.py --year 1949 # nur wenn nötig
python3 download_offenegesetze.py --from-year 1949 --to-year 1950
"""
from __future__ import annotations
import argparse
import time
from pathlib import Path
from urllib.parse import urlparse
from bgbl_util import (
BGBl_DIR,
LAWGIT_BGBL,
LAWGIT_ROOT,
META_DIR,
SESSION,
download_file,
fundstelle,
issue_id,
save_json,
)
API_URL = "https://api.offenegesetze.de/v1/veroeffentlichung/"
MEDIA_BASE = "https://media.offenegesetze.de"
HISTORIE_BIS = 2022
HISTORIE_VON = 1949
def fetch_api(params: dict) -> list[dict]:
results = []
url = API_URL
params = {**params, "limit": 100}
while url:
r = SESSION.get(url, params=params if url == API_URL else None, timeout=60)
r.raise_for_status()
data = r.json()
results.extend(data.get("results") or [])
url = data.get("next")
params = None
time.sleep(0.15)
return results
def schreibe_hinweis() -> Path:
"""Kein PDF-Duplikat: nur dokumentieren, wo die Historie schon liegt."""
pdfs = 0
jahre: list[str] = []
quelle = LAWGIT_BGBL / "bgbl1"
if quelle.is_dir():
jahre = sorted(p.name for p in quelle.iterdir() if p.is_dir() and p.name.isdigit())
pdfs = sum(1 for p in quelle.rglob("*.pdf"))
manifest = {
"quelle": "offenegesetze.de",
"api": API_URL,
"jahre": f"{HISTORIE_VON}–{HISTORIE_BIS}",
"schnitt": "2022 endet hier; ab 2023 nur noch recht.bund.de (download_rechtbund.py)",
"lawgit_bestand": {
"pfad": str(quelle) if quelle.exists() else None,
"jahre": f"{jahre[0]}–{jahre[-1]}" if jahre else None,
"pdfs": pdfs,
"hinweis": (
"Nach Archivierung von lawgit diese PDFs hierher übernehmen "
"(kopieren oder verschieben), nicht ein zweites Mal von OffeneGesetze laden."
),
},
"nicht_duplizieren": True,
}
dest = META_DIR / "offenegesetze-quelle.json"
save_json(dest, manifest)
print("Historie 1949–2022: OffeneGesetze.de")
print(f" Schnitt: 2022 letzte OffeneGesetze-Jahr, 2023+ nur recht.bund.de")
if pdfs:
print(f" Bestand in lawgit: {pdfs} PDFs ({jahre[0]}–{jahre[-1]})")
print(f" Pfad: {quelle}")
print(" Keine Kopie angelegt — nach Archivierung von lawgit hierher ziehen.")
else:
print(f" lawgit-Bestand nicht gefunden unter {LAWGIT_ROOT}")
print(" Einzelne Jahre: python3 download_offenegesetze.py --year 1949")
print(f" Manifest: {dest}")
return dest
def download_year(kind: str, year: int) -> list[dict]:
if year > HISTORIE_BIS:
print(f"{year} gehört zu recht.bund.de — download_rechtbund.py nutzen")
return []
if year < HISTORIE_VON:
print(f"{year}: OffeneGesetze beginnt {HISTORIE_VON}")
return []
print(f"Lade Metadaten {kind} {year} …")
items = fetch_api({"year": year, "kind": kind})
normiert = []
for raw in items:
number = raw.get("number")
page = raw.get("page")
normiert.append(
{
"id": raw.get("id") or issue_id(kind, year, number or 0),
"kind": kind,
"year": year,
"number": number,
"page": page,
"date": (raw.get("date") or "")[:10] or None,
"title": raw.get("title"),
"titel": raw.get("title"),
"document_url": raw.get("document_url"),
"url": raw.get("url"),
"quelle": "offenegesetze",
"fundstelle": fundstelle(kind, year, number or 0, page),
}
)
save_json(META_DIR / f"{kind}-{year}.json", normiert)
print(f" {len(normiert)} Veröffentlichungen")
archive_url = f"{MEDIA_BASE}/{kind}/{year}.tar.bz2"
archive_path = BGBl_DIR / "archives" / f"{kind}-{year}.tar.bz2"
print(f"Lade Archiv {archive_url} …")
download_file(archive_url, archive_path)
if archive_path.exists() and archive_path.stat().st_size > 0:
print(f" Archiv liegt unter {archive_path} (nicht automatisch entpackt)")
else:
print(" Kein Jahresarchiv, lade Einzel-PDFs …")
seen = set()
for item in items:
doc = (item.get("document_url") or "").split("#")[0]
if not doc or doc in seen:
continue
seen.add(doc)
name = Path(urlparse(doc).path).name
dest = BGBl_DIR / kind / str(year) / name
print(f" PDF {name}")
download_file(doc, dest)
time.sleep(0.2)
return normiert
def main() -> None:
parser = argparse.ArgumentParser(description="BGBl 1949–2022 von OffeneGesetze.de")
parser.add_argument("--hinweis", action="store_true", help="Nur Quellen-Manifest, keine PDFs")
parser.add_argument("--year", type=int, action="append")
parser.add_argument("--from-year", type=int)
parser.add_argument("--to-year", type=int)
parser.add_argument("--kind", choices=["bgbl1", "bgbl2"], default="bgbl1")
args = parser.parse_args()
META_DIR.mkdir(parents=True, exist_ok=True)
years = list(args.year or [])
if args.from_year:
years.extend(range(args.from_year, (args.to_year or HISTORIE_BIS) + 1))
years = sorted(set(y for y in years if HISTORIE_VON <= y <= HISTORIE_BIS))
if args.hinweis or not years:
schreibe_hinweis()
if not years:
return
for year in years:
download_year(args.kind, year)
if __name__ == "__main__":
main()

303
download_rechtbund.py Normal file
View File

@@ -0,0 +1,303 @@
#!/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()

114
extract_bgbl.py Normal file
View File

@@ -0,0 +1,114 @@
#!/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()

128
llm_client.py Normal file
View File

@@ -0,0 +1,128 @@
#!/usr/bin/env python3
"""
OpenAI-kompatibler Client für RunPod (vLLM) oder andere Endpunkte.
Nur aufrufen mit `parse_bgbl.py --llm` / `sync.py --llm`. Standardläufe
bleiben heuristisch und kosten nichts.
Umgebungsvariablen (siehe .env.example):
LLM_BASE_URL z.B. https://<pod-id>-8000.proxy.runpod.net/v1
LLM_API_KEY RunPod-API-Key oder "not-needed"
LLM_MODEL z.B. Qwen/Qwen2.5-72B-Instruct-AWQ
LLM_TIMEOUT Sekunden (Standard 180)
"""
from __future__ import annotations
import json
import os
import time
from pathlib import Path
from typing import Any
import requests
_ENV_LOADED = False
def load_env(path: str | Path = ".env") -> None:
global _ENV_LOADED
if _ENV_LOADED:
return
_ENV_LOADED = True
env_file = Path(path)
if not env_file.exists():
return
for line in env_file.read_text(encoding="utf-8").splitlines():
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, _, value = line.partition("=")
value = value.strip().strip('"').strip("'")
if value:
os.environ.setdefault(key.strip(), value)
def llm_configured() -> bool:
load_env()
return bool(os.environ.get("LLM_BASE_URL") or os.environ.get("RUNPOD_BASE_URL"))
def _endpoint() -> tuple[str, str, str]:
load_env()
base = os.environ.get("LLM_BASE_URL") or os.environ.get("RUNPOD_BASE_URL") or ""
base = base.rstrip("/")
if not base.endswith("/v1"):
base = base + "/v1"
key = os.environ.get("LLM_API_KEY") or os.environ.get("RUNPOD_API_KEY") or "not-needed"
model = os.environ.get("LLM_MODEL") or "Qwen/Qwen2.5-72B-Instruct-AWQ"
return base, key, model
def _lies_strom(response: requests.Response) -> str:
stuecke: list[str] = []
for zeile in response.iter_lines(decode_unicode=True):
if not zeile or not zeile.startswith("data:"):
continue
rest = zeile[5:].strip()
if rest == "[DONE]":
break
try:
happen = json.loads(rest)
except json.JSONDecodeError:
continue
for wahl in happen.get("choices") or []:
inhalt = (wahl.get("delta") or {}).get("content")
if inhalt:
stuecke.append(inhalt)
if not stuecke:
raise ValueError("leere Antwort im Datenstrom")
return "".join(stuecke)
def chat(messages: list[dict[str, Any]], *, json_mode: bool = True, max_tokens: int = 4096) -> str:
base, key, model = _endpoint()
if not base.startswith("http"):
raise RuntimeError("LLM_BASE_URL / RUNPOD_BASE_URL ist nicht gesetzt")
timeout = int(os.environ.get("LLM_TIMEOUT", "180"))
versuche = int(os.environ.get("LLM_RETRIES", "3"))
payload: dict[str, Any] = {
"model": model,
"messages": messages,
"temperature": 0.1,
"max_tokens": max_tokens,
"stream": True,
}
if json_mode:
payload["response_format"] = {"type": "json_object"}
letzter: Exception | None = None
for versuch in range(1, versuche + 1):
try:
with requests.post(
f"{base}/chat/completions",
headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"},
json=payload,
timeout=timeout,
stream=True,
) as r:
if r.status_code >= 500 or r.status_code == 429:
raise requests.HTTPError(f"HTTP {r.status_code}", response=r)
r.raise_for_status()
return _lies_strom(r)
except (requests.RequestException, KeyError, ValueError) as exc:
letzter = exc
if versuch < versuche:
time.sleep(min(2 ** versuch * 5, 40))
raise RuntimeError(f"LLM nach {versuche} Versuchen nicht erreichbar: {letzter}")
def chat_json(messages: list[dict[str, Any]], **kwargs) -> dict:
raw = chat(messages, json_mode=True, **kwargs)
raw = raw.strip()
if raw.startswith("```"):
raw = raw.split("\n", 1)[-1]
if raw.endswith("```"):
raw = raw[: raw.rfind("```")]
return json.loads(raw)

446
parse_bgbl.py Normal file
View File

@@ -0,0 +1,446 @@
#!/usr/bin/env python3
"""
Strukturiert BGBl-Texte zu Ereignissen der Normzeitachse.
Standard: Heuristik (kein GPU, keine Kosten).
Optional: --llm (LLM_BASE_URL in .env) — nicht starten, solange lawgit den
Historien-Parse auf RunPod fährt.
python3 parse_bgbl.py --year 2025
python3 parse_bgbl.py --year 2025 --llm # nur bewusst, kostet GPU-Zeit
"""
from __future__ import annotations
import argparse
import json
import re
from datetime import datetime, timedelta
from pathlib import Path
from bgbl_util import META_DIR, PARSED_DIR, TEXT_DIR, fundstelle
TYP_ERLASS = "Erlass"
TYP_AENDERUNG = "Änderung"
TYP_NEUBEKANNT = "Neubekanntmachung"
TYP_AUFHEBUNG = "Aufhebung"
TYP_VERORDNUNG = "Verordnung"
TYP_BEKANNT = "Bekanntmachung"
TYP_SONSTIGE = "Sonstige"
ALLE_TYPEN = {
TYP_ERLASS, TYP_AENDERUNG, TYP_NEUBEKANNT, TYP_AUFHEBUNG,
TYP_VERORDNUNG, TYP_BEKANNT, TYP_SONSTIGE,
}
STAMMGESETZE = [
(r"Grundgesetz", "GG", "gg"),
(r"Bürgerlichen Gesetzbuchs|Bürgerliche(?:s)? Gesetzbuch", "BGB", "bgb"),
(r"Strafgesetzbuch", "StGB", "stgb"),
(r"Zivilprozessordnung", "ZPO", "zpo"),
(r"Strafprozessordnung", "StPO", "stpo"),
(r"Handelsgesetzbuch", "HGB", "hgb"),
(r"Aktiengesetz", "AktG", "aktg"),
(r"GmbH-Gesetz|Gesetz betreffend die Gesellschaften mit beschränkter Haftung", "GmbHG", "gmbhg"),
(r"Verwaltungsverfahrensgesetz", "VwVfG", "vwvfg"),
(r"Verwaltungsgerichtsordnung", "VwGO", "vwgo"),
(r"Abgabenordnung", "AO", "ao"),
(r"Einkommensteuergesetz", "EStG", "estg"),
(r"Umsatzsteuergesetz", "UStG", "ustg"),
(r"Sozialgesetzbuch", "SGB", None),
(r"Aufenthaltsgesetz", "AufenthG", "aufenthg"),
(r"Asylgesetz", "AsylG", "asylvfg_1992"),
(r"Bundes-Immissionsschutzgesetz", "BImSchG", "bimschg"),
(r"Straßenverkehrsgesetz", "StVG", "stvg"),
(r"Straßenverkehrs-Ordnung", "StVO", "stvo_2013"),
]
INKRAFT_RE = re.compile(
r"(?:tritt|treten)\s+"
r"(?:am\s+)?"
r"("
r"mit\s+Ablauf\s+des\s+Tages\s+der\s+Verkündung"
r"|Tage?s?\s+der\s+Verkündung"
r"|Tage?s?\s+nach\s+(?:der\s+)?Verkündung"
r"|\d{1,2}\.\s*(?:Januar|Februar|März|April|Mai|Juni|Juli|August|September|Oktober|November|Dezember)\s+\d{4}"
r"|\d{1,2}\.\s*\d{1,2}\.\s*\d{4}"
r")"
r"[^.]*?in\s+Kraft",
re.IGNORECASE | re.DOTALL,
)
DATE_DE = re.compile(
r"(\d{1,2})\.\s*(Januar|Februar|März|April|Mai|Juni|Juli|August|September|Oktober|November|Dezember)\s+(\d{4})"
)
DATE_NUM = re.compile(r"(\d{1,2})\.\s*(\d{1,2})\.\s*(\d{4})")
MONTHS = {
"januar": 1, "februar": 2, "märz": 3, "april": 4, "mai": 5, "juni": 6,
"juli": 7, "august": 8, "september": 9, "oktober": 10, "november": 11, "dezember": 12,
}
SYSTEM_PROMPT = """Du bist ein Parser für das deutsche Bundesgesetzblatt.
Aus dem Text einer Verkündung extrahierst du strukturierte JSON-Daten für eine Normzeitachse.
Antworte NUR mit einem JSON-Objekt.
Schema:
{
"titel": "amtlicher Titel",
"typ": "Erlass|Änderung|Neubekanntmachung|Aufhebung|Verordnung|Bekanntmachung|Sonstige",
"ausfertigung": "YYYY-MM-DD oder null",
"inkrafttreten": [
{"datum": "YYYY-MM-DD oder 'Tag nach der Verkündung'", "geltung": "ganzes Gesetz oder z.B. Art. 2"}
],
"betroffene_normen": [
{
"name": "Bürgerliches Gesetzbuch",
"kurz": "BGB",
"operation": "erlass|aenderung|aufhebung|neubekanntmachung",
"aenderungen": [
{"stelle": "§ 433", "aktion": "neufassung|einfuegung|streichung|aenderung|wortlaut", "kurzbeschreibung": "ein Satz"}
]
}
],
"zusammenfassung": "1-2 Sätze, verständlich für Juristinnen ohne IT-Hintergrund"
}
Regeln:
- Änderungsgesetze ändern oft MEHRERE Stammgesetze (Artikelgesetz). Jedes Stammgesetz einzeln auflisten.
- Stelle immer mit § oder Art. angeben.
- Keine Git-Begriffe. Keine Spekulation: unsichere Felder null / leere Liste.
"""
def normalize_ocr(text: str) -> str:
t = text.replace("\u00ad", "")
t = re.sub(r"-\s*\n\s*", "", t)
t = re.sub(r"(?<=[A-Za-zÄÖÜäöüß])\n(?=[a-zäöüß])", "", t)
t = t.replace("~", "m")
t = re.sub(r"\s+", " ", t)
return t
def parse_date_de(text: str) -> str | None:
m = DATE_DE.search(text)
if m:
d, mon, y = int(m.group(1)), MONTHS[m.group(2).lower()], int(m.group(3))
return f"{y:04d}-{mon:02d}-{d:02d}"
m = DATE_NUM.search(text)
if m:
d, mo, y = int(m.group(1)), int(m.group(2)), int(m.group(3))
if 1 <= mo <= 12 and 1 <= d <= 31:
return f"{y:04d}-{mo:02d}-{d:02d}"
return None
def classify_typ(title: str) -> str:
t = title.lower()
if "bekanntmachung der neufassung" in t or "neufassung" in t:
return TYP_NEUBEKANNT
if "aufhebung" in t:
return TYP_AUFHEBUNG
if "änderung" in t or "aenderung" in t:
return TYP_AENDERUNG
if t.startswith("verordnung") or " verordnung " in f" {t} ":
return TYP_VERORDNUNG
if t.startswith("bekanntmachung") or "bekanntmachung" in t:
return TYP_BEKANNT
if t.startswith("gesetz") or "gesetz" in t:
return TYP_ERLASS
return TYP_SONSTIGE
def _norm_hit(kurz: str, gii: str | None, operation: str) -> dict:
return {"name": kurz, "kurz": kurz, "gii": gii, "operation": operation, "aenderungen": []}
def find_betroffene(title: str, text: str, typ: str) -> list[dict]:
found: list[dict] = []
seen: set[str] = set()
op_title = "aenderung" if typ == TYP_AENDERUNG else "erlass"
for pattern, kurz, gii in STAMMGESETZE:
if re.search(pattern, title, re.IGNORECASE) and kurz not in seen:
seen.add(kurz)
found.append(_norm_hit(kurz, gii, op_title))
head = text[:8000]
for pattern, kurz, gii in STAMMGESETZE:
if kurz in seen:
continue
aend = re.search(
rf"(?:Änderung|Anderung|Abänderung)\s+(?:des|der|von)\s+(?:[^\n.]{{0,40}})?{pattern}",
head,
re.IGNORECASE,
)
geaendert = re.search(
rf"{pattern}[^.{{]{{0,120}}wird wie folgt geändert",
head,
re.IGNORECASE,
)
if aend or geaendert:
seen.add(kurz)
found.append(_norm_hit(kurz, gii, "aenderung"))
return found
def find_inkrafttreten(text: str, verkundung: str | None) -> list[dict]:
hits = []
seen = set()
blob = normalize_ocr(text)
for m in INKRAFT_RE.finditer(blob):
raw = re.sub(r"\s+", " ", m.group(1)).strip()
low = raw.lower()
hinweis = None
datum: str | None = None
if "verkündung" in low:
if verkundung:
try:
d = datetime.fromisoformat(verkundung[:10])
if "ablauf" in low:
datum = (d + timedelta(days=1)).strftime("%Y-%m-%d")
hinweis = "mit Ablauf des Verkündungstages, gilt ab dem Folgetag"
elif "nach" in low:
datum = (d + timedelta(days=1)).strftime("%Y-%m-%d")
hinweis = "Tag nach der Verkündung"
else:
datum = d.strftime("%Y-%m-%d")
hinweis = "Tag der Verkündung"
except ValueError:
datum = raw
else:
datum = raw
else:
datum = parse_date_de(raw) or raw
key = (datum, hinweis)
if key in seen:
continue
seen.add(key)
rec = {"datum": datum, "geltung": "ganzes Gesetz"}
if hinweis:
rec["hinweis"] = hinweis
hits.append(rec)
return hits
def load_meta() -> list[dict]:
items = []
if not META_DIR.exists():
return items
seen: set[str] = set()
for path in sorted(META_DIR.glob("*.json")):
if path.name in {"offenegesetze-quelle.json"}:
continue
data = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(data, list):
continue
for item in data:
key = item.get("id")
if not key or key in seen:
continue
seen.add(key)
items.append(item)
return items
def _load_extracted(item: dict) -> dict | None:
kind = item.get("kind") or "bgbl1"
year = item.get("year")
number = item.get("number_raw") or item.get("number")
candidates = [
TEXT_DIR / kind / str(year) / f"{kind}_{year}_{number}.json",
]
doc = (item.get("document_url") or "").split("#")[0]
if doc:
name = Path(doc).name
candidates.append(TEXT_DIR / kind / str(year) / name.replace(".pdf", ".json"))
candidates.append(TEXT_DIR / kind / str(year) / Path(name).with_suffix(".json").name)
pdf = item.get("pdf")
if pdf:
candidates.append(TEXT_DIR / kind / str(year) / Path(pdf).with_suffix(".json").name)
for c in candidates:
if c.exists():
return json.loads(c.read_text(encoding="utf-8"))
return None
def text_for_item(item: dict) -> dict | None:
extracted = _load_extracted(item)
if not extracted:
return None
page_texts = extracted.get("page_texts") or []
if not page_texts:
return extracted if extracted.get("text") else None
start = item.get("pdf_page") or 1
n = item.get("num_pages") or 0
seiten = [p for p in page_texts if start <= p["page"] < start + n] if n else page_texts
return {
**extracted,
"text": "\n\n".join(f"--- Seite {p['page']} ---\n{p['text']}" for p in seiten),
"pages": len(seiten),
}
def heuristic_parse(item: dict, extracted: dict | None) -> dict:
title = item.get("title") or item.get("titel") or ""
text = (extracted or {}).get("text") or ""
date = (item.get("date") or item.get("verkundung") or "")[:10] or None
kind = item.get("kind") or "bgbl1"
year = item.get("year")
number = item.get("number_raw") or item.get("number")
page = item.get("page")
fs = item.get("fundstelle") or fundstelle(kind, year or 0, number or 0, page)
typ = classify_typ(title)
body = normalize_ocr(text)
if typ == TYP_ERLASS and re.search(r"wird wie folgt geändert", body, re.IGNORECASE):
typ = TYP_AENDERUNG
return {
"id": item.get("id"),
"fundstelle": fs,
"verkundung": date,
"titel": title,
"typ": typ,
"ausfertigung": item.get("ausfertigung") or parse_date_de(title) or date,
"inkrafttreten": find_inkrafttreten(text, date),
"betroffene_normen": find_betroffene(title, text, typ),
"zusammenfassung": title,
"pdf": (extracted or {}).get("pdf") or item.get("pdf"),
"text_quality": (extracted or {}).get("quality"),
"quelle": item.get("quelle") or ("offenegesetze" if year and year < 2023 else "recht.bund.de"),
"parser": "heuristik",
"kind": kind,
"year": year,
"number": number,
"page": page,
"url": item.get("url"),
"document_url": item.get("document_url"),
"sha256": item.get("sha256"),
}
def llm_parse(item: dict, extracted: dict, base: dict) -> dict:
import llm_client
meta = json.dumps(
{k: item.get(k) for k in ("id", "title", "titel", "date", "year", "number", "kind")},
ensure_ascii=False,
)
text = (extracted.get("text") or "")[:18000]
daten = llm_client.chat_json(
[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": f"Metadaten: {meta}\n\nText der Verkündung:\n{text}"},
],
max_tokens=4000,
)
event = dict(base)
for feld in ("titel", "typ", "ausfertigung", "inkrafttreten", "zusammenfassung", "betroffene_normen"):
if daten.get(feld):
if feld == "typ" and daten[feld] not in ALLE_TYPEN:
continue
event[feld] = daten[feld]
event["parser"] = "llm"
return event
def parse_item(item: dict, use_llm: bool) -> dict:
extracted = text_for_item(item)
event = heuristic_parse(item, extracted)
if use_llm and extracted and extracted.get("text"):
try:
return llm_parse(item, extracted, event)
except Exception as exc:
event["parser"] = "heuristik"
event["parser_fehler"] = [f"{type(exc).__name__}: {exc}"]
return event
def _schreibe(event: dict) -> Path:
dest = PARSED_DIR / f"{event['id']}.json"
dest.parent.mkdir(parents=True, exist_ok=True)
dest.write_text(json.dumps(event, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
return dest
def baue_index() -> Path:
alle = []
for f in sorted(PARSED_DIR.glob("bgbl*.json")):
try:
alle.append(json.loads(f.read_text(encoding="utf-8")))
except (json.JSONDecodeError, OSError) as exc:
print(f" übersprungen: {f.name} ({exc})")
alle.sort(key=lambda e: (e.get("verkundung") or "", e.get("id") or ""))
index = PARSED_DIR / "index.json"
index.write_text(json.dumps(alle, ensure_ascii=False), encoding="utf-8")
return index
def main() -> None:
parser = argparse.ArgumentParser(description="BGBl-Texte zu Zeitachsen-Ereignissen parsen")
parser.add_argument("--year", type=int, action="append")
parser.add_argument("--von", type=int)
parser.add_argument("--bis", type=int)
parser.add_argument(
"--llm",
action="store_true",
help="RunPod/vLLM verwenden (LLM_BASE_URL). Standard ist Heuristik — --llm nicht "
"starten, solange der lawgit-Historien-Parse auf der GPU läuft.",
)
parser.add_argument("--limit", type=int, default=0)
parser.add_argument("--force", action="store_true")
parser.add_argument("--kein-index", action="store_true")
args = parser.parse_args()
items = load_meta()
if args.year:
years = set(args.year)
items = [i for i in items if i.get("year") in years]
if args.von:
items = [i for i in items if (i.get("year") or 0) >= args.von]
if args.bis:
items = [i for i in items if (i.get("year") or 9999) <= args.bis]
items.sort(key=lambda i: (i.get("year") or 0, i.get("number") or 0, i.get("id") or ""))
if args.limit:
items = items[: args.limit]
if not items:
print("Keine Metadaten. Zuerst: python3 download_rechtbund.py --live-test")
return
use_llm = args.llm
if use_llm:
import llm_client
if not llm_client.llm_configured():
print("LLM_BASE_URL ist nicht gesetzt — falle auf Heuristik zurück.")
use_llm = False
PARSED_DIR.mkdir(parents=True, exist_ok=True)
if not args.force:
offen = []
for item in items:
dest = PARSED_DIR / f"{item.get('id')}.json"
if dest.exists():
continue
offen.append(item)
if len(offen) < len(items):
print(f"{len(items) - len(offen)} bereits geparst, übersprungen")
items = offen
if not items:
print("Nichts zu tun.")
return
for i, item in enumerate(items, 1):
print(f"[{i}/{len(items)}] {item.get('id')} {(item.get('title') or item.get('titel') or '')[:70]}")
_schreibe(parse_item(item, use_llm))
if args.kein_index:
print(f"{len(items)} Ereignisse geschrieben, Index nicht neu gebaut")
return
index = baue_index()
print(f"{len(items)} in diesem Lauf → {index}")
if __name__ == "__main__":
main()

2
requirements.txt Normal file
View File

@@ -0,0 +1,2 @@
requests>=2.31.0
pypdf>=4.0.0

102
sync.py Normal file
View File

@@ -0,0 +1,102 @@
#!/usr/bin/env python3
"""
Inkrementeller Abgleich ab 2023: laden → Text → Heuristik-Parse.
Idempotent und wiederaufsetzbar: vorhandene PDFs, Extrakte und Events
werden übersprungen. LLM bleibt aus, außer mit --llm (kostet, nicht
parallel zum lawgit-GPU-Lauf starten).
python3 sync.py --live-test
python3 sync.py --from-year 2023
python3 sync.py --year 2026
"""
from __future__ import annotations
import argparse
import subprocess
import sys
from datetime import datetime
from pathlib import Path
from bgbl_util import META_DIR, ROOT, save_json
PY = sys.executable
def run(args: list[str]) -> None:
print("+", " ".join(args), flush=True)
subprocess.run([PY, *args], cwd=ROOT, check=True)
def main() -> None:
parser = argparse.ArgumentParser(description="BGBl ab 2023 nachziehen (recht.bund.de)")
parser.add_argument("--live-test", action="store_true", help="2024/1, 2025/1, 2026/1")
parser.add_argument("--year", type=int, action="append")
parser.add_argument("--from-year", type=int)
parser.add_argument("--to-year", type=int)
parser.add_argument("--kind", choices=["bgbl1", "bgbl2"], default="bgbl1")
parser.add_argument("--limit", type=int, default=0)
parser.add_argument(
"--llm",
action="store_true",
help="Nach dem Download mit LLM parsen. Standard: Heuristik. "
"Nicht setzen, solange lawgit auf RunPod parst.",
)
parser.add_argument("--force", action="store_true")
args = parser.parse_args()
dl = ["download_rechtbund.py", "--kind", args.kind]
if args.live_test:
dl.append("--live-test")
for y in args.year or []:
dl.extend(["--year", str(y)])
if args.from_year:
dl.extend(["--from-year", str(args.from_year)])
if args.to_year:
dl.extend(["--to-year", str(args.to_year)])
if args.limit:
dl.extend(["--limit", str(args.limit)])
if not args.live_test and not args.year and not args.from_year:
parser.error("Bitte --live-test, --year oder --from-year angeben")
run(dl)
ex = ["extract_bgbl.py", "--kind", args.kind]
years = list(args.year or [])
if args.live_test:
years.extend([2024, 2025, 2026])
if args.from_year:
years.extend(range(args.from_year, (args.to_year or datetime.now().year) + 1))
years = sorted(set(years))
for y in years:
ex.extend(["--year", str(y)])
if args.force:
ex.append("--force")
run(ex)
pr = ["parse_bgbl.py"]
for y in years:
pr.extend(["--year", str(y)])
if args.llm:
pr.append("--llm")
if args.force:
pr.append("--force")
run(pr)
save_json(
META_DIR / "letzter-sync.json",
{
"zeit": datetime.now().isoformat(timespec="seconds"),
"jahre": years,
"kind": args.kind,
"llm": bool(args.llm),
"live_test": bool(args.live_test),
},
)
print("\nSync fertig. Nächster voller Nachzug ab 2023:")
print(" python3 sync.py --from-year 2023")
if __name__ == "__main__":
main()

19
tests/fixtures/aenderung_bgb.json vendored Normal file
View File

@@ -0,0 +1,19 @@
{
"item": {
"id": "bgbl1-2024-fixture-1",
"kind": "bgbl1",
"year": 2024,
"number": 99,
"date": "2024-03-15",
"title": "Gesetz zur Änderung des Bürgerlichen Gesetzbuchs",
"titel": "Gesetz zur Änderung des Bürgerlichen Gesetzbuchs",
"quelle": "recht.bund.de",
"fundstelle": "BGBl. 2024 I Nr. 99",
"document_url": "https://www.recht.bund.de/eli/bund/BGBl-1/2024/99/regelungstext.pdf"
},
"extracted": {
"pdf": "tests/fixtures/aenderung_bgb.pdf",
"quality": "ok",
"text": "Gesetz zur Änderung des Bürgerlichen Gesetzbuchs\nVom 14. März 2024\n\nDer Bundestag hat das folgende Gesetz beschlossen:\n\nArtikel 1\nÄnderung des Bürgerlichen Gesetzbuchs\nDas Bürgerliche Gesetzbuch in der Fassung der Bekanntmachung vom 2. Januar 2002 wird wie folgt geändert:\n§ 433 Absatz 1 wird wie folgt gefasst: Der Verkäufer einer Sache ist verpflichtet, dem Käufer die Sache zu übergeben.\n\nArtikel 2\nInkrafttreten\nDieses Gesetz tritt am Tage nach der Verkündung in Kraft.\n"
}
}

42
tests/test_parse_heuristik.py Normal file
View File

@@ -0,0 +1,42 @@
#!/usr/bin/env python3
"""Heuristik-Test ohne Netz und ohne LLM — Event-JSON-Schnitt wie in lawgit."""
from __future__ import annotations
import json
import sys
import unittest
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from parse_bgbl import heuristic_parse # noqa: E402
class HeuristikTest(unittest.TestCase):
def setUp(self) -> None:
raw = json.loads((ROOT / "tests/fixtures/aenderung_bgb.json").read_text(encoding="utf-8"))
self.event = heuristic_parse(raw["item"], raw["extracted"])
def test_id_und_fundstelle(self) -> None:
self.assertEqual(self.event["id"], "bgbl1-2024-fixture-1")
self.assertEqual(self.event["fundstelle"], "BGBl. 2024 I Nr. 99")
self.assertEqual(self.event["verkundung"], "2024-03-15")
self.assertEqual(self.event["quelle"], "recht.bund.de")
self.assertEqual(self.event["parser"], "heuristik")
def test_typ_aenderung(self) -> None:
self.assertEqual(self.event["typ"], "Änderung")
def test_betroffene_norm_bgb(self) -> None:
kurznamen = {n["kurz"] for n in self.event["betroffene_normen"]}
self.assertIn("BGB", kurznamen)
def test_inkrafttreten_folgetag(self) -> None:
daten = [e["datum"] for e in self.event["inkrafttreten"]]
self.assertIn("2024-03-16", daten)
if __name__ == "__main__":
unittest.main()

1
timeline/meta/README.md Normal file
View File

@@ -0,0 +1 @@
Metadaten der geladenen Ausgaben (JSON). PDFs liegen unter `bgbl/` und sind gitignoriert.

28
timeline/meta/bgbl1-2024-rechtbund.json Normal file
View File

@@ -0,0 +1,28 @@
[
{
"id": "bgbl1-2024-1-1",
"kind": "bgbl1",
"year": 2024,
"number": 1,
"number_raw": "1",
"page": null,
"date": "2024-01-04",
"title": "Besondere Gebührenverordnung des Bundesministeriums für Digitales und Verkehr für telekommunikationsrechtliche Tätigkeiten",
"titel": "Besondere Gebührenverordnung des Bundesministeriums für Digitales und Verkehr für telekommunikationsrechtliche Tätigkeiten",
"typ_amtlich": "Verordnung",
"ausfertigung": "2023-12-20",
"federfuehrung": "Bundesministerium für Digitales und Verkehr",
"fna": "neu: 202-5-25",
"sachgebiet": "Verwaltungsgebühren",
"document_url": "https://www.recht.bund.de/eli/bund/BGBl-1/2024/1/regelungstext.pdf?__blob=publicationFile",
"url": "https://www.recht.bund.de/eli/bund/BGBl-1/2024/1/",
"xml_url": null,
"quelle": "recht.bund.de",
"fundstelle": "BGBl. 2024 I Nr. 1",
"sha256": "48e987cf1ea55689503e72a9ec009f96599e1f4c2643064a660c76c63b697857",
"md5_amtlich": "19cb6316fb5e82e0b2234e50b0a27715",
"pdf": "bgbl/bgbl1/2024/bgbl1_2024_1.pdf",
"bytes": 410933,
"skipped": false
}
]

28
timeline/meta/bgbl1-2025-rechtbund.json Normal file
View File

@@ -0,0 +1,28 @@
[
{
"id": "bgbl1-2025-1-1",
"kind": "bgbl1",
"year": 2025,
"number": 1,
"number_raw": "1",
"page": null,
"date": "2025-01-08",
"title": "Dreißigste Verordnung zur Änderung der Hundertneunundfünfzigsten Durchführungsverordnung zur Luftverkehrs-Ordnung (Festlegung von Flugverfahren für An- und Abflüge nach Instrumentenflugregeln zum und vom Flughafen Saarbrücken)",
"titel": "Dreißigste Verordnung zur Änderung der Hundertneunundfünfzigsten Durchführungsverordnung zur Luftverkehrs-Ordnung (Festlegung von Flugverfahren für An- und Abflüge nach Instrumentenflugregeln zum und vom Flughafen Saarbrücken)",
"typ_amtlich": "Verordnung",
"ausfertigung": "2025-01-03",
"federfuehrung": "Bundesaufsichtsamt für Flugsicherung",
"fna": "96-1-2-159",
"sachgebiet": "Luftverkehr",
"document_url": "https://www.recht.bund.de/eli/bund/BGBl-1/2025/1/regelungstext.pdf?__blob=publicationFile",
"url": "https://www.recht.bund.de/eli/bund/BGBl-1/2025/1/",
"xml_url": null,
"quelle": "recht.bund.de",
"fundstelle": "BGBl. 2025 I Nr. 1",
"sha256": "0be0159df3ba7fce4b70d7dcf911c774a9cb3cd2d0f069a7d877d992e2d6e8c5",
"md5_amtlich": "010f95da4586a4dc5b4c0a7fdd32379a",
"pdf": "bgbl/bgbl1/2025/bgbl1_2025_1.pdf",
"bytes": 477937,
"skipped": false
}
]

28
timeline/meta/bgbl1-2026-rechtbund.json Normal file
View File

@@ -0,0 +1,28 @@
[
{
"id": "bgbl1-2026-1-1",
"kind": "bgbl1",
"year": 2026,
"number": 1,
"number_raw": "1",
"page": null,
"date": "2026-01-05",
"title": "Verordnung zur Bestimmung eines anderen Netzes des Bundes zum Datenaustausch über das nationale Once-only-technical-System im Anwendungsbereich des Onlinezugangsgesetzes",
"titel": "Verordnung zur Bestimmung eines anderen Netzes des Bundes zum Datenaustausch über das nationale Once-only-technical-System im Anwendungsbereich des Onlinezugangsgesetzes",
"typ_amtlich": "Verordnung",
"ausfertigung": "2025-12-29",
"federfuehrung": "Bundesministerium für Digitales und Staatsmodernisierung",
"fna": "neu: 206-1-1",
"sachgebiet": "Öffentliche Informationstechnik",
"document_url": "https://www.recht.bund.de/eli/bund/BGBl-1/2026/1/regelungstext.pdf?__blob=publicationFile",
"url": "https://www.recht.bund.de/eli/bund/BGBl-1/2026/1/",
"xml_url": null,
"quelle": "recht.bund.de",
"fundstelle": "BGBl. 2026 I Nr. 1",
"sha256": "4716d4620b1a5b1ef6a614292d20bf74ef1b1562afb8aa1143e4e1c6bfac2e2c",
"md5_amtlich": "bc5c30b528a5b5208f26ab36642e03f8",
"pdf": "bgbl/bgbl1/2026/bgbl1_2026_1.pdf",
"bytes": 259584,
"skipped": false
}
]

11
timeline/meta/letzter-sync.json Normal file
View File

@@ -0,0 +1,11 @@
{
"zeit": "2026-08-16T11:55:38",
"jahre": [
2024,
2025,
2026
],
"kind": "bgbl1",
"llm": false,
"live_test": true
}

13
timeline/meta/offenegesetze-quelle.json Normal file
View File

@@ -0,0 +1,13 @@
{
"quelle": "offenegesetze.de",
"api": "https://api.offenegesetze.de/v1/veroeffentlichung/",
"jahre": "1949–2022",
"schnitt": "2022 endet hier; ab 2023 nur noch recht.bund.de (download_rechtbund.py)",
"lawgit_bestand": {
"pfad": "/Users/michaelhermann/Source/law/coded/lawgit/bgbl/bgbl1",
"jahre": "1949–2022",
"pdfs": 5526,
"hinweis": "Nach Archivierung von lawgit diese PDFs hierher übernehmen (kopieren oder verschieben), nicht ein zweites Mal von OffeneGesetze laden."
},
"nicht_duplizieren": true
}

80
timeline/meta/rechtbund-manifest.json Normal file
View File

@@ -0,0 +1,80 @@
[
{
"id": "bgbl1-2024-1-1",
"kind": "bgbl1",
"year": 2024,
"number": 1,
"number_raw": "1",
"page": null,
"date": "2024-01-04",
"title": "Besondere Gebührenverordnung des Bundesministeriums für Digitales und Verkehr für telekommunikationsrechtliche Tätigkeiten",
"titel": "Besondere Gebührenverordnung des Bundesministeriums für Digitales und Verkehr für telekommunikationsrechtliche Tätigkeiten",
"typ_amtlich": "Verordnung",
"ausfertigung": "2023-12-20",
"federfuehrung": "Bundesministerium für Digitales und Verkehr",
"fna": "neu: 202-5-25",
"sachgebiet": "Verwaltungsgebühren",
"document_url": "https://www.recht.bund.de/eli/bund/BGBl-1/2024/1/regelungstext.pdf?__blob=publicationFile",
"url": "https://www.recht.bund.de/eli/bund/BGBl-1/2024/1/",
"xml_url": null,
"quelle": "recht.bund.de",
"fundstelle": "BGBl. 2024 I Nr. 1",
"sha256": "48e987cf1ea55689503e72a9ec009f96599e1f4c2643064a660c76c63b697857",
"md5_amtlich": "19cb6316fb5e82e0b2234e50b0a27715",
"pdf": "bgbl/bgbl1/2024/bgbl1_2024_1.pdf",
"bytes": 410933,
"skipped": false
},
{
"id": "bgbl1-2025-1-1",
"kind": "bgbl1",
"year": 2025,
"number": 1,
"number_raw": "1",
"page": null,
"date": "2025-01-08",
"title": "Dreißigste Verordnung zur Änderung der Hundertneunundfünfzigsten Durchführungsverordnung zur Luftverkehrs-Ordnung (Festlegung von Flugverfahren für An- und Abflüge nach Instrumentenflugregeln zum und vom Flughafen Saarbrücken)",
"titel": "Dreißigste Verordnung zur Änderung der Hundertneunundfünfzigsten Durchführungsverordnung zur Luftverkehrs-Ordnung (Festlegung von Flugverfahren für An- und Abflüge nach Instrumentenflugregeln zum und vom Flughafen Saarbrücken)",
"typ_amtlich": "Verordnung",
"ausfertigung": "2025-01-03",
"federfuehrung": "Bundesaufsichtsamt für Flugsicherung",
"fna": "96-1-2-159",
"sachgebiet": "Luftverkehr",
"document_url": "https://www.recht.bund.de/eli/bund/BGBl-1/2025/1/regelungstext.pdf?__blob=publicationFile",
"url": "https://www.recht.bund.de/eli/bund/BGBl-1/2025/1/",
"xml_url": null,
"quelle": "recht.bund.de",
"fundstelle": "BGBl. 2025 I Nr. 1",
"sha256": "0be0159df3ba7fce4b70d7dcf911c774a9cb3cd2d0f069a7d877d992e2d6e8c5",
"md5_amtlich": "010f95da4586a4dc5b4c0a7fdd32379a",
"pdf": "bgbl/bgbl1/2025/bgbl1_2025_1.pdf",
"bytes": 477937,
"skipped": false
},
{
"id": "bgbl1-2026-1-1",
"kind": "bgbl1",
"year": 2026,
"number": 1,
"number_raw": "1",
"page": null,
"date": "2026-01-05",
"title": "Verordnung zur Bestimmung eines anderen Netzes des Bundes zum Datenaustausch über das nationale Once-only-technical-System im Anwendungsbereich des Onlinezugangsgesetzes",
"titel": "Verordnung zur Bestimmung eines anderen Netzes des Bundes zum Datenaustausch über das nationale Once-only-technical-System im Anwendungsbereich des Onlinezugangsgesetzes",
"typ_amtlich": "Verordnung",
"ausfertigung": "2025-12-29",
"federfuehrung": "Bundesministerium für Digitales und Staatsmodernisierung",
"fna": "neu: 206-1-1",
"sachgebiet": "Öffentliche Informationstechnik",
"document_url": "https://www.recht.bund.de/eli/bund/BGBl-1/2026/1/regelungstext.pdf?__blob=publicationFile",
"url": "https://www.recht.bund.de/eli/bund/BGBl-1/2026/1/",
"xml_url": null,
"quelle": "recht.bund.de",
"fundstelle": "BGBl. 2026 I Nr. 1",
"sha256": "4716d4620b1a5b1ef6a614292d20bf74ef1b1562afb8aa1143e4e1c6bfac2e2c",
"md5_amtlich": "bc5c30b528a5b5208f26ab36642e03f8",
"pdf": "bgbl/bgbl1/2026/bgbl1_2026_1.pdf",
"bytes": 259584,
"skipped": false
}
]