fegger
e990488273
Move the licensed PDF exports and the Layer-1 full texts from the project root into the dot directory .lexis360/, keeping the root listing to versioned content plus the tool checkouts. The location also encodes provenance, distinct from .firecrawl/: .lexis360/ holds user-supplied licensed exports that cannot be re-fetched by an agent - if missing, stop and ask the user (documented in MEMORY), whereas .firecrawl/ remains the home of reconstructible web fetches. - plain move of the unversioned directory (batch2/ subfolder spotted and left untouched for its own intake run) - .gitignore, tool constants/messages, source.pdf/source.text frontmatter paths of all 55 curated entries, README/RUNBOOK/MEMORY mentions; kb.json/INDEX.md regenerated - batch 2 (53 PDFs) recorded in MEMORY as staged in .lexis360/batch2/ Validated end to end: --extract 55/55 with stable ids, --registry 55 entries/6 clusters, --check 0 problems (every frontmatter path resolves at the new location), no stale path references remain.
574 lines
21 KiB
Python
574 lines
21 KiB
Python
#!/usr/bin/env python3
|
|
"""Build the Lexis360 knowledge base (Layer-1 extraction + kb.json registry).
|
|
|
|
The .lexis360/ folder holds PDF exports of "Lexis Briefings Personalrecht"
|
|
chapters (LexisNexis AT), exported from a licensed Lexis 360 subscription.
|
|
They are the raw source layer for the curated knowledge base under
|
|
personalverrechnung/wissensbasis/ (development reference and future
|
|
copilot corpus).
|
|
|
|
Licensing (decision D1, 2026-09-10): the PDF exports and the extracted
|
|
full texts stay local and unversioned (.lexis360/ is gitignored), like
|
|
.firecrawl/. Only the curated entries under wissensbasis/ (own words,
|
|
short quotes, source citations) are versioned.
|
|
|
|
Run from the repository root:
|
|
|
|
python3 personalverrechnung/tools/build_lexis_kb.py --extract # Layer 1
|
|
python3 personalverrechnung/tools/build_lexis_kb.py --registry # kb.json
|
|
python3 personalverrechnung/tools/build_lexis_kb.py --check # QA pass
|
|
|
|
Outputs:
|
|
Layer 1 (unversioned, licensed raw texts):
|
|
.lexis360/md/<slug>.md full text + extraction frontmatter
|
|
.lexis360/md/_catalog.json parsed metadata of all sources
|
|
Layer 2 (versioned, curated by hand):
|
|
personalverrechnung/wissensbasis/dokumente/<slug>.md
|
|
personalverrechnung/wissensbasis/kb.json (generated)
|
|
|
|
Frontmatter schema of curated entries (subset, no PyYAML needed):
|
|
id, batch, title, work, chapter, topic (cluster slug), author,
|
|
stand (YYYY-MM), source{pdf,text}, legal_bases[], tags[], cross_refs[]
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import datetime
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import unicodedata
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
PDF_DIR = ROOT / ".lexis360"
|
|
MD_DIR = PDF_DIR / "md"
|
|
CURATED_DIR = ROOT / "personalverrechnung" / "wissensbasis" / "dokumente"
|
|
KB_JSON = CURATED_DIR.parent / "kb.json"
|
|
|
|
WORK = "Lexis Briefings Personalrecht"
|
|
BATCH = 1
|
|
|
|
MONTHS = {
|
|
"jänner": 1, "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,
|
|
}
|
|
|
|
# Explicit breadcrumb-topic -> cluster mapping (pinned from a first
|
|
# --extract pass; keep deterministic and auditable). Keys are the
|
|
# ascii_slug() form of the parsed breadcrumb topic: diacritics STRIPPED
|
|
# (ü->u, ä->a, ß->ss), non-alnum -> hyphen — matching the umlaut-dropped
|
|
# style of the Lexis360 export filenames ("beschaftigung", not
|
|
# "beschaeftigung").
|
|
TOPIC_MAP: dict[str, str] = {
|
|
"altersteilzeit": "atz",
|
|
"lehrlinge": "leh",
|
|
"schnupperlehre": "leh",
|
|
"jugendliche": "jug",
|
|
"kinder": "jug",
|
|
"ferialpraktikanten-volontare": "pra",
|
|
"geringfugige-beschaftigung": "bes",
|
|
"fallweise-beschaftigung": "bes",
|
|
"freier-dienstvertrag": "bes",
|
|
"werkvertrag": "bes",
|
|
"echtes-dienstverhaltnis": "bes",
|
|
"abgrenzung-arbeiter-angestellte": "bes",
|
|
"teilzeitbeschaftigung": "tzb",
|
|
"bildungsteilzeit": "tzb",
|
|
}
|
|
|
|
# Fallback keyword scan over the ASCII topic + slug, checked after the
|
|
# explicit map. Order matters (first match wins).
|
|
KEYWORDS: list[tuple[str, str]] = [
|
|
("altersteilzeit", "atz"),
|
|
("lehrling", "leh"),
|
|
("schnupperlehre", "leh"),
|
|
("lehrverhaltnis", "leh"),
|
|
("jugend", "jug"),
|
|
("kinder", "jug"),
|
|
("ferialpraktikant", "pra"),
|
|
("ferialpraktikum", "pra"),
|
|
("volontar", "pra"),
|
|
("au-pair", "pra"),
|
|
("bildungsteilzeit", "tzb"),
|
|
("teilzeit", "tzb"),
|
|
("geringfugig", "bes"),
|
|
("fallweise", "bes"),
|
|
("dienstvertrag", "bes"),
|
|
("werkvertrag", "bes"),
|
|
("arbeitsverhaltnis", "bes"),
|
|
]
|
|
|
|
# Cluster registry: ID prefix -> display name (kb.json, INDEX).
|
|
CLUSTERS: dict[str, str] = {
|
|
"atz": "Altersteilzeit",
|
|
"leh": "Lehrverhältnis / Lehrlinge",
|
|
"jug": "Jugendarbeit / Jugendschutz",
|
|
"pra": "Ferialpraktikanten, Volontäre, Au-pair",
|
|
"bes": "Beschäftigungsformen & Abgrenzung",
|
|
"tzb": "Teilzeit & Bildungsteilzeit",
|
|
}
|
|
|
|
# Frontmatter `topic` values (descriptive ASCII slugs, per approved D2
|
|
# schema example) -> cluster ID prefix. The 3-letter prefix lives in the
|
|
# id; `topic` carries the human-filterable slug.
|
|
TOPIC_TO_PREFIX: dict[str, str] = {
|
|
"altersteilzeit": "atz",
|
|
"lehrlinge": "leh",
|
|
"jugendliche": "jug",
|
|
"ferialpraktikanten": "pra",
|
|
"beschaftigungsformen": "bes",
|
|
"teilzeit": "tzb",
|
|
}
|
|
|
|
HEADER_RE = re.compile(r"^\s*Lexis 360®\s*$")
|
|
MARKER = "Lexis Briefings Personalrecht"
|
|
# Breadcrumb: "Kapitel > Thema · Autor(en) · <Monat> <Jahr>". Dossier-style
|
|
# exports merge the breadcrumb and the body into ONE line, so trailing
|
|
# text is allowed and ignored by the (?P<rest>...) tail.
|
|
BC_LINE_RE = re.compile(
|
|
r"^\s*(?P<chapter>[^>·]+?)\s*>\s*(?P<topic>.+?)\s*·\s*(?P<author>.+?)"
|
|
r"\s*·\s*(?P<month>[A-ZÄÖÜ][a-zäöüß]+)\s+(?P<year>\d{4})(?P<rest>.*)$"
|
|
)
|
|
STAND_RE = re.compile(r"^\d{4}-\d{2}$")
|
|
ID_RE = re.compile(r"^lb-[a-z]{3}-\d{2}$")
|
|
|
|
|
|
def ascii_slug(text: str) -> str:
|
|
"""ASCII form used for topic matching: diacritics stripped (NFKD,
|
|
ü->u, ä->a), ß->ss, lowercase, non-alphanumeric -> hyphen. This
|
|
matches the umlaut-dropped style of the Lexis360 export filenames."""
|
|
text = unicodedata.normalize("NFKD", text)
|
|
text = "".join(c for c in text if not unicodedata.combining(c))
|
|
text = text.lower().replace("ß", "ss")
|
|
text = re.sub(r"[^a-z0-9]+", "-", text).strip("-")
|
|
return text
|
|
|
|
|
|
def parse_pdf(pdf: Path) -> tuple[str, int, str]:
|
|
"""Return (text, pages, pdfinfo title)."""
|
|
proc = subprocess.run(
|
|
["pdftotext", "-layout", str(pdf), "-"],
|
|
capture_output=True, text=True, check=True,
|
|
)
|
|
pages = 0
|
|
title = ""
|
|
info = subprocess.run(
|
|
["pdfinfo", str(pdf)], capture_output=True, text=True, check=True,
|
|
)
|
|
for line in info.stdout.splitlines():
|
|
if line.startswith("Pages:"):
|
|
pages = int(line.split(":", 1)[1].strip())
|
|
if line.startswith("Title:"):
|
|
title = line.split(":", 1)[1].strip()
|
|
return proc.stdout, pages, title
|
|
|
|
|
|
def clean_text(raw: str) -> str:
|
|
lines = [ln for ln in raw.splitlines() if not HEADER_RE.match(ln)]
|
|
text = "\n".join(lines).replace("\x0c", "")
|
|
text = re.sub(r"\n{3,}", "\n\n", text)
|
|
return text.strip() + "\n"
|
|
|
|
|
|
def parse_meta(text: str, pdf_title: str = "") -> dict:
|
|
"""Parse title + breadcrumb from the first page.
|
|
|
|
Layout: a line with the work marker, the chapter title, then the
|
|
breadcrumb "Kapitel > Thema · Autor · <Monat> <Jahr>". Dossier-style
|
|
exports merge breadcrumb and body text into one line; the regex tail
|
|
(?P<rest>) tolerates that. If no separate title line exists, fall
|
|
back to the pdfinfo Title.
|
|
"""
|
|
lines = text.splitlines()
|
|
marker_idx = None
|
|
for i, ln in enumerate(lines[:20]):
|
|
if MARKER in ln:
|
|
marker_idx = i
|
|
break
|
|
if marker_idx is None:
|
|
raise ValueError("marker 'Lexis Briefings Personalrecht' not found")
|
|
title_lines: list[str] = []
|
|
m = None
|
|
scan = lines[marker_idx + 1:]
|
|
for pos, ln in enumerate(scan):
|
|
cand = BC_LINE_RE.match(ln)
|
|
if cand is None and " · " in ln:
|
|
# dossier-style exports wrap the breadcrumb ("· Juli\n2026"):
|
|
# retry with up to two following lines joined in
|
|
joined = ln.rstrip()
|
|
for nxt in scan[pos + 1:pos + 3]:
|
|
joined = f"{joined} {nxt.strip()}".strip()
|
|
cand = BC_LINE_RE.match(joined)
|
|
if cand:
|
|
break
|
|
if cand:
|
|
m = cand
|
|
break
|
|
if not ln.strip():
|
|
continue
|
|
title_lines.append(ln.strip())
|
|
if len(title_lines) > 4:
|
|
raise ValueError("breadcrumb not found near page header")
|
|
if m is None:
|
|
raise ValueError("breadcrumb not found")
|
|
title = " ".join(t for t in title_lines if t) or pdf_title
|
|
if not title:
|
|
raise ValueError("no title (neither in text nor in pdfinfo)")
|
|
month_no = MONTHS[m.group("month").lower()]
|
|
return {
|
|
"title": title,
|
|
"chapter": m.group("chapter").strip(),
|
|
"topic_group": m.group("topic").strip(),
|
|
"author": m.group("author").strip(),
|
|
"stand": f"{m.group('year')}-{month_no:02d}",
|
|
"stand_human": f"{m.group('month')} {m.group('year')}",
|
|
}
|
|
|
|
|
|
def resolve_cluster(topic_ascii: str, slug: str) -> str | None:
|
|
if topic_ascii in TOPIC_MAP:
|
|
return TOPIC_MAP[topic_ascii]
|
|
for keyword, cluster in KEYWORDS:
|
|
if keyword in topic_ascii or keyword in slug:
|
|
return cluster
|
|
return None
|
|
|
|
|
|
def load_previous_ids() -> dict[str, tuple[str, int]]:
|
|
"""slug -> (id, batch) from the previous _catalog.json.
|
|
|
|
IDs are frozen once assigned: a later batch must never renumber
|
|
existing entries, because cross_refs and kb.json consumers rely on
|
|
stable IDs. New documents continue after the highest existing
|
|
number in their cluster; numbers of removed documents are retired,
|
|
never reused.
|
|
"""
|
|
path = MD_DIR / "_catalog.json"
|
|
if not path.is_file():
|
|
return {}
|
|
try:
|
|
prev = json.loads(path.read_text(encoding="utf-8"))
|
|
except json.JSONDecodeError:
|
|
return {}
|
|
out = {}
|
|
for e in prev.get("entries", []):
|
|
if "slug" in e and "id" in e:
|
|
out[e["slug"]] = (e["id"], int(e.get("batch", 1)))
|
|
return out
|
|
|
|
|
|
def assign_ids(catalog: list[dict], prev: dict[str, tuple[str, int]]) -> None:
|
|
counters: dict[str, int] = {}
|
|
for id_, _batch in prev.values():
|
|
m = ID_RE.match(id_)
|
|
if m:
|
|
prefix, num = id_[3:6], int(id_[-2:])
|
|
counters[prefix] = max(counters.get(prefix, 0), num)
|
|
for entry in catalog:
|
|
if slug_prev := prev.get(entry["slug"]):
|
|
entry["id"], entry["batch"] = slug_prev
|
|
for entry in sorted(catalog, key=lambda e: (e["cluster"] or "zzz", e["slug"])):
|
|
if entry.get("id") or entry["cluster"] is None:
|
|
continue
|
|
cluster = entry["cluster"]
|
|
counters[cluster] = counters.get(cluster, 0) + 1
|
|
entry["id"] = f"lb-{cluster}-{counters[cluster]:02d}"
|
|
|
|
|
|
def write_layer1(entry: dict, text: str) -> None:
|
|
md = MD_DIR / f"{entry['slug']}.md"
|
|
fm = [
|
|
"---",
|
|
f"id: {entry['id']}",
|
|
f"batch: {entry['batch']}",
|
|
f'title: "{entry["title"]}"',
|
|
f'work: "{WORK}"',
|
|
f'chapter: "{entry["chapter"]}"',
|
|
f'topic_group: "{entry["topic_group"]}"',
|
|
f"cluster: {entry['cluster']}",
|
|
f'author: "{entry["author"]}"',
|
|
f"stand: {entry['stand']}",
|
|
f'pdf: "{entry["pdf_rel"]}"',
|
|
f"pages: {entry['pages']}",
|
|
"---",
|
|
"",
|
|
]
|
|
md.write_text("\n".join(fm) + text, encoding="utf-8")
|
|
|
|
|
|
def extract() -> int:
|
|
if not PDF_DIR.is_dir():
|
|
sys.exit(f"missing {PDF_DIR}")
|
|
MD_DIR.mkdir(exist_ok=True)
|
|
pdfs = sorted(PDF_DIR.glob("Lexis360_*.pdf"))
|
|
catalog: list[dict] = []
|
|
texts: dict[str, str] = {}
|
|
warnings: list[str] = []
|
|
for pdf in pdfs:
|
|
slug = pdf.stem[len("Lexis360_"):]
|
|
raw, pages, pdf_title = parse_pdf(pdf)
|
|
text = clean_text(raw)
|
|
texts[slug] = text
|
|
try:
|
|
meta = parse_meta(text, pdf_title)
|
|
except ValueError as exc:
|
|
warnings.append(f"{pdf.name}: {exc}")
|
|
continue
|
|
entry = {
|
|
"slug": slug,
|
|
"pdf_rel": f".lexis360/{pdf.name}",
|
|
"md_rel": f".lexis360/md/{slug}.md",
|
|
"pages": pages,
|
|
"chars": len(text),
|
|
"batch": BATCH,
|
|
**meta,
|
|
}
|
|
topic_ascii = ascii_slug(entry["topic_group"])
|
|
entry["topic_group_ascii"] = topic_ascii
|
|
entry["cluster"] = resolve_cluster(topic_ascii, slug)
|
|
if entry["cluster"] is None:
|
|
warnings.append(f"{slug}: no cluster for topic {topic_ascii!r}")
|
|
catalog.append(entry)
|
|
assign_ids(catalog, load_previous_ids())
|
|
for entry in catalog:
|
|
if entry["cluster"] is not None:
|
|
write_layer1(entry, texts[entry["slug"]])
|
|
report = {
|
|
"generated_at": datetime.datetime.now(datetime.UTC).isoformat(timespec="seconds"),
|
|
"n_pdfs": len(pdfs),
|
|
"n_entries": len(catalog),
|
|
"n_unclustered": sum(1 for e in catalog if e["cluster"] is None),
|
|
"entries": catalog,
|
|
"warnings": warnings,
|
|
}
|
|
(MD_DIR / "_catalog.json").write_text(
|
|
json.dumps(report, ensure_ascii=False, indent=1), encoding="utf-8")
|
|
print(f"extracted {len(catalog)}/{len(pdfs)} "
|
|
f"({report['n_unclustered']} unclustered)")
|
|
topics = sorted({e["topic_group"] for e in catalog})
|
|
print(f"distinct breadcrumb topics ({len(topics)}):")
|
|
for t in topics:
|
|
print(f" - {t}")
|
|
for w in warnings:
|
|
print(f"WARNING: {w}")
|
|
return 0 if not warnings else 1
|
|
|
|
|
|
# --- frontmatter (controlled YAML subset) ----------------------------------
|
|
|
|
def fm_value(raw: str):
|
|
raw = raw.strip()
|
|
if raw.startswith('"') and raw.endswith('"'):
|
|
return raw[1:-1]
|
|
if raw.startswith("[") and raw.endswith("]"):
|
|
inner = raw[1:-1].strip()
|
|
if not inner:
|
|
return []
|
|
return [v.strip().strip('"') for v in inner.split(",")]
|
|
if re.fullmatch(r"-?\d+", raw):
|
|
return int(raw)
|
|
return raw
|
|
|
|
|
|
def parse_frontmatter(text: str) -> dict:
|
|
if not text.startswith("---\n"):
|
|
raise ValueError("no frontmatter")
|
|
end = text.index("\n---\n", 4)
|
|
lines = text[4:end].splitlines()
|
|
data: dict = {}
|
|
current = None
|
|
for ln in lines:
|
|
if not ln.strip():
|
|
continue
|
|
if ln.startswith(" ") and current:
|
|
key, _, raw = ln.strip().partition(":")
|
|
data[current][key] = fm_value(raw)
|
|
elif ln.startswith("\t"):
|
|
raise ValueError("tabs not allowed in frontmatter")
|
|
else:
|
|
key, _, raw = ln.partition(":")
|
|
key = key.strip()
|
|
if raw.strip() == "":
|
|
data[key] = {}
|
|
current = key
|
|
else:
|
|
data[key] = fm_value(raw)
|
|
current = None
|
|
return data
|
|
|
|
|
|
REQUIRED = [
|
|
"id", "batch", "title", "work", "chapter", "topic", "author", "stand",
|
|
"legal_bases", "tags",
|
|
]
|
|
|
|
|
|
def validate(entry: dict, all_ids: set[str], source: str) -> list[str]:
|
|
errors = []
|
|
for key in REQUIRED:
|
|
if key not in entry:
|
|
errors.append(f"{source}: missing key {key!r}")
|
|
src = entry.get("source", {})
|
|
for key in ("pdf", "text"):
|
|
if key not in src:
|
|
errors.append(f"{source}: missing source.{key}")
|
|
if not ID_RE.match(str(entry.get("id", ""))):
|
|
errors.append(f"{source}: bad id {entry.get('id')!r}")
|
|
if str(entry.get("stand", "")) and not STAND_RE.match(str(entry["stand"])):
|
|
errors.append(f"{source}: stand must be YYYY-MM, got {entry['stand']!r}")
|
|
for ref in entry.get("cross_refs", []):
|
|
if ref not in all_ids:
|
|
errors.append(f"{source}: dangling cross_ref {ref!r}")
|
|
if entry.get("batch") != BATCH:
|
|
errors.append(f"{source}: batch must be {BATCH}")
|
|
if entry.get("topic") not in TOPIC_TO_PREFIX:
|
|
errors.append(
|
|
f"{source}: topic must be one of {sorted(TOPIC_TO_PREFIX)}, "
|
|
f"got {entry.get('topic')!r}")
|
|
if str(entry.get("id", ""))[3:6] != TOPIC_TO_PREFIX.get(str(entry.get("topic")), ""):
|
|
errors.append(
|
|
f"{source}: id prefix does not match topic {entry.get('topic')!r}")
|
|
return errors
|
|
|
|
|
|
def write_index(entries: list[dict], clusters: dict) -> None:
|
|
"""Generate wissensbasis/INDEX.md from curated frontmatter (derived,
|
|
never hand-edited — regenerate with --registry)."""
|
|
today = datetime.date.today().isoformat()
|
|
header = (
|
|
f"# Index — Wissensbasis „Lexis Briefings Personalrecht“\n\n"
|
|
f"Generiert am {today} · Batch 1 · {len(entries)} Einträge · "
|
|
f"{len(clusters)} Cluster. Quelle: LexisNexis Lexis 360 "
|
|
"(lizenzierter Export; Layer 1 lokal/unversioniert). "
|
|
"Werte gelten je zum Quell-Stand (`stand` im Frontmatter); "
|
|
"Widersprüche und offene Punkte sind mit ⚠/❓ in den Einträgen "
|
|
"selbst dokumentiert. Schema & Konventionen: `README.md`.\n"
|
|
)
|
|
lines = [header]
|
|
by_topic: dict[str, list[dict]] = {}
|
|
for e in entries:
|
|
by_topic.setdefault(str(e["topic"]), []).append(e)
|
|
for topic, cluster_entries in sorted(by_topic.items()):
|
|
stands = sorted(str(e["stand"]) for e in cluster_entries)
|
|
prefix = TOPIC_TO_PREFIX.get(topic)
|
|
name = CLUSTERS.get(prefix, topic)
|
|
lines.append(
|
|
f"\n## {name} (`topic: {topic}`) — {len(cluster_entries)} "
|
|
f"Einträge · Stand {stands[0]} bis {stands[-1]}\n")
|
|
lines.append("| ID | Titel | Autor | Stand | Datei |")
|
|
lines.append("|---|---|---|---|---|")
|
|
for e in sorted(cluster_entries, key=lambda x: str(x["id"])):
|
|
fname = str(e["source"]["text"]).rsplit("/", 1)[-1]
|
|
lines.append(
|
|
f"| {e['id']} | {e['title']} | {e['author']} | {e['stand']} "
|
|
f"| {fname} |")
|
|
path = CURATED_DIR.parent / "INDEX.md"
|
|
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
print(f"INDEX.md written: {len(entries)} entries")
|
|
|
|
|
|
def registry() -> int:
|
|
CURATED_DIR.mkdir(parents=True, exist_ok=True)
|
|
docs = sorted(CURATED_DIR.glob("*.md"))
|
|
if not docs:
|
|
sys.exit(f"no curated docs in {CURATED_DIR}")
|
|
entries = []
|
|
errors: list[str] = []
|
|
for doc in docs:
|
|
try:
|
|
fm = parse_frontmatter(doc.read_text(encoding="utf-8"))
|
|
except ValueError as exc:
|
|
errors.append(f"{doc.name}: {exc}")
|
|
continue
|
|
entries.append(fm)
|
|
all_ids = {str(e.get("id")) for e in entries if e.get("id")}
|
|
for fm, doc in zip(entries, docs):
|
|
errors.extend(validate(fm, all_ids, doc.name))
|
|
if errors:
|
|
for e in errors:
|
|
print(f"ERROR: {e}")
|
|
sys.exit(1)
|
|
entries.sort(key=lambda e: str(e.get("id")))
|
|
clusters = {}
|
|
for e in entries:
|
|
clusters[e["topic"]] = clusters.get(e["topic"], 0) + 1
|
|
out = {
|
|
"generated_at": datetime.datetime.now(datetime.UTC).isoformat(timespec="seconds"),
|
|
"n_entries": len(entries),
|
|
"layers": {
|
|
"curated": "personalverrechnung/wissensbasis/dokumente",
|
|
"fulltext": ".lexis360/md (local, unversioned - licensed source texts)",
|
|
"pdf": ".lexis360 (local, unversioned - licensed PDF exports)",
|
|
},
|
|
"schema": {
|
|
"id": "lb-<cluster-prefix>-<nn>",
|
|
"stand": "YYYY-MM (Stand der Lexis-Briefing-Ausgabe)",
|
|
"legal_bases": "Rechtsgrundlagen as cited in the source briefing",
|
|
"cross_refs": "ids of related kb entries",
|
|
},
|
|
"clusters": [
|
|
{"slug": slug,
|
|
"name": CLUSTERS.get(TOPIC_TO_PREFIX.get(slug, ""), slug),
|
|
"n": n}
|
|
for slug, n in sorted(clusters.items())
|
|
],
|
|
"entries": entries,
|
|
}
|
|
KB_JSON.parent.mkdir(parents=True, exist_ok=True)
|
|
KB_JSON.write_text(
|
|
json.dumps(out, ensure_ascii=False, indent=1) + "\n", encoding="utf-8")
|
|
write_index(entries, clusters)
|
|
print(f"kb.json written: {len(entries)} entries, {len(clusters)} clusters")
|
|
return 0
|
|
|
|
|
|
def check() -> int:
|
|
"""QA: curated docs 1:1 with extraction catalog, sources exist."""
|
|
problems: list[str] = []
|
|
catalog_path = MD_DIR / "_catalog.json"
|
|
if not catalog_path.is_file():
|
|
sys.exit("run --extract first (missing .lexis360/md/_catalog.json)")
|
|
catalog = json.loads(catalog_path.read_text(encoding="utf-8"))
|
|
cat_slugs = {e["slug"]: e for e in catalog["entries"]}
|
|
cur_slugs = {p.stem for p in CURATED_DIR.glob("*.md")} if CURATED_DIR.is_dir() else set()
|
|
for slug in sorted(set(cat_slugs) - cur_slugs):
|
|
problems.append(f"not curated: {slug}")
|
|
for slug in sorted(cur_slugs - set(cat_slugs)):
|
|
problems.append(f"curated but no source: {slug}")
|
|
for slug in sorted(set(cat_slugs) & cur_slugs):
|
|
text_path = ROOT / cat_slugs[slug]["md_rel"]
|
|
if not text_path.is_file():
|
|
problems.append(f"layer-1 text missing: {text_path}")
|
|
pdf_path = ROOT / cat_slugs[slug]["pdf_rel"]
|
|
if not pdf_path.is_file():
|
|
problems.append(f"source pdf missing: {pdf_path}")
|
|
for p in problems:
|
|
print(f"ERROR: {p}")
|
|
print(f"check: {len(cat_slugs)} sources, {len(cur_slugs)} curated, "
|
|
f"{len(problems)} problems")
|
|
return 1 if problems else 0
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
|
group = parser.add_mutually_exclusive_group(required=True)
|
|
group.add_argument("--extract", action="store_true",
|
|
help="extract Lexis360 PDFs into .lexis360/md/")
|
|
group.add_argument("--registry", action="store_true",
|
|
help="generate kb.json from curated frontmatter")
|
|
group.add_argument("--check", action="store_true",
|
|
help="QA: curated docs vs extraction catalog")
|
|
args = parser.parse_args()
|
|
if args.extract:
|
|
return extract()
|
|
if args.registry:
|
|
return registry()
|
|
return check()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main()) |