m2 (AI Agent)
6da874b2e6
Phase 1: router skeleton + Qdrant read path - FastAPI app with /health, /search/generic, /ingest endpoints - Qdrant reader (hybrid dense+sparse w/ RRF, plus payload scroll) - BGE-M3 TEI client for query-side embedding - memory-api client proxying writes to /memory/store (m2-memory untouched) - Pydantic Fact + Exemplar payloads, discriminated by `kind` - Dockerfile + docker-compose joining the coolify Docker network Phase 2: authoritative-source ingest pipeline - PDF text extraction + paragraph-aware (§ N) and size chunkers - Loaders for bnatschg, mhbasp, biotopwertliste, state-<slug> - CLI: artenschutz-ingest --source <name> [--states ...] [--dry-run] - Helper script to fetch BNatSchG from gesetze-im-internet.de - End-to-end smoke script (scripts/smoke.sh) for Coolify validation 22/22 unit tests pass. Real-world dry-runs verified against mhbasp Anhang 4, biotopwertlisteNEU and Berlin Kartierstandards PDFs from GST-DATA. See COOLIFY-DEPLOY.md for staging deploy + smoke procedure.
112 lines
4.1 KiB
Python
Executable file
112 lines
4.1 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Fetch the consolidated text of BNatSchG §44 and §45b from
|
|
gesetze-im-internet.de and emit a plain-text file suitable for the
|
|
bnatschg ingest loader.
|
|
|
|
Usage:
|
|
python scripts/fetch_bnatschg.py [--out sources/bnatschg/bnatschg.txt]
|
|
|
|
The output uses one heading line per section ("§ N ..." or "Anlage N ...")
|
|
exactly as the bnatschg loader's paragraph splitter expects.
|
|
|
|
Best-effort HTML scrape. If gesetze-im-internet.de changes layout, paste
|
|
text manually instead (see sources/README.md).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import re
|
|
import sys
|
|
from html.parser import HTMLParser
|
|
from pathlib import Path
|
|
from urllib.request import Request, urlopen
|
|
|
|
# §s most relevant for Artenschutzgutachten. Add more as needed.
|
|
SECTIONS = [
|
|
("§ 44 Vorschriften für besonders geschützte und bestimmte andere Tier- und Pflanzenarten",
|
|
"https://www.gesetze-im-internet.de/bnatschg_2009/__44.html"),
|
|
("§ 45 Ausnahmen; Ermächtigung zum Erlass von Rechtsverordnungen",
|
|
"https://www.gesetze-im-internet.de/bnatschg_2009/__45.html"),
|
|
("§ 45b Schutz wild lebender Vogelarten bei der Errichtung und dem Betrieb von Windenergieanlagen an Land",
|
|
"https://www.gesetze-im-internet.de/bnatschg_2009/__45b.html"),
|
|
("Anlage 1 (zu § 45b Absatz 1 bis 5) Liste der bei der Errichtung und dem Betrieb von Windenergieanlagen an Land kollisionsgefährdeten Brutvogelarten",
|
|
"https://www.gesetze-im-internet.de/bnatschg_2009/anlage_1.html"),
|
|
]
|
|
|
|
|
|
class TextExtractor(HTMLParser):
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self._buf: list[str] = []
|
|
self._skip = 0 # depth of <script>/<style>
|
|
|
|
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
|
|
if tag in {"script", "style"}:
|
|
self._skip += 1
|
|
elif tag in {"br", "p", "div", "li", "tr"}:
|
|
self._buf.append("\n")
|
|
|
|
def handle_endtag(self, tag: str) -> None:
|
|
if tag in {"script", "style"} and self._skip:
|
|
self._skip -= 1
|
|
elif tag in {"p", "div", "li", "tr"}:
|
|
self._buf.append("\n")
|
|
|
|
def handle_data(self, data: str) -> None:
|
|
if not self._skip:
|
|
self._buf.append(data)
|
|
|
|
def value(self) -> str:
|
|
raw = "".join(self._buf)
|
|
# Collapse excess blank lines while preserving paragraph breaks.
|
|
raw = re.sub(r"[ \t]+", " ", raw)
|
|
raw = re.sub(r"\n{3,}", "\n\n", raw)
|
|
return raw.strip()
|
|
|
|
|
|
def fetch(url: str, timeout: float = 30.0) -> str:
|
|
req = Request(url, headers={"User-Agent": "artenschutz-router/0.1 (+fetch_bnatschg)"})
|
|
with urlopen(req, timeout=timeout) as resp:
|
|
raw_bytes = resp.read()
|
|
# gesetze-im-internet.de serves ISO-8859-1; sniff from header if present.
|
|
charset = resp.headers.get_content_charset() or "iso-8859-1"
|
|
return raw_bytes.decode(charset, errors="replace")
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(prog="fetch_bnatschg")
|
|
parser.add_argument(
|
|
"--out",
|
|
type=Path,
|
|
default=Path("sources/bnatschg/bnatschg.txt"),
|
|
help="Output path (default: %(default)s)",
|
|
)
|
|
args = parser.parse_args(argv)
|
|
args.out.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
chunks: list[str] = []
|
|
for heading, url in SECTIONS:
|
|
print(f" fetching {heading.split('(')[0].strip()} ...", file=sys.stderr)
|
|
try:
|
|
html = fetch(url)
|
|
except Exception as exc: # noqa: BLE001
|
|
print(f" FAILED ({exc}) — skipping", file=sys.stderr)
|
|
continue
|
|
extractor = TextExtractor()
|
|
extractor.feed(html)
|
|
body = extractor.value()
|
|
# Anchor each section with the expected heading on its own line so the
|
|
# paragraph chunker recognises it.
|
|
chunks.append(f"{heading}\n\n{body}\n")
|
|
|
|
if not chunks:
|
|
print("No sections fetched — refusing to write empty output", file=sys.stderr)
|
|
return 2
|
|
|
|
args.out.write_text("\n\n".join(chunks), encoding="utf-8")
|
|
print(f"Wrote {args.out} ({sum(len(c) for c in chunks):,} chars)", file=sys.stderr)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|