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.
49 lines
1.2 KiB
Python
49 lines
1.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from io import StringIO
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from artenschutz_router.ingest import cli
|
|
|
|
|
|
SAMPLE_BNATSCHG = """§ 44 Vorschriften
|
|
|
|
(1) Es ist verboten …
|
|
|
|
§ 45b Schutz wild lebender Vögel
|
|
|
|
(1) Beim Betrieb von WEA …
|
|
"""
|
|
|
|
|
|
def test_cli_dry_run_emits_json(tmp_path: Path, capsys: pytest.CaptureFixture):
|
|
sources_root = tmp_path / "sources"
|
|
bn_dir = sources_root / "bnatschg"
|
|
bn_dir.mkdir(parents=True)
|
|
(bn_dir / "bnatschg.txt").write_text(SAMPLE_BNATSCHG, encoding="utf-8")
|
|
|
|
rc = cli.main(
|
|
[
|
|
"--source",
|
|
"bnatschg",
|
|
"--sources-root",
|
|
str(sources_root),
|
|
"--dry-run",
|
|
]
|
|
)
|
|
assert rc == 0
|
|
out = capsys.readouterr().out.strip().splitlines()
|
|
assert len(out) == 2
|
|
for line in out:
|
|
record = json.loads(line)
|
|
assert record["payload"]["kind"] == "fact"
|
|
assert record["payload"]["fact_type"] == "legal"
|
|
assert record["content"]
|
|
|
|
|
|
def test_cli_rejects_unknown_source(tmp_path: Path):
|
|
with pytest.raises(SystemExit):
|
|
cli.main(["--source", "nonsense", "--sources-root", str(tmp_path), "--dry-run"])
|