443a56a04a
One-shot script that fetches the consolidated National Insurance Law from
Hebrew Wikisource and emits plain text with heading lines the chunker can
parse (פרק / סימן / סעיף).
Key quirks of the source markup:
- {{ח:קטע2}} carries the level name inside the title parameter ("פרק א׳:
…"), not in the template name. Emit the title verbatim and let the
chunker's existing prefix regex handle it.
- Named args (e.g. 'אחר=[פרק א׳]') must be filtered — the original
positional parser was treating them as positional titles and producing
gibberish "חלק אחר=[פרק א׳]" headings.
- No חלק level exists in this law — only פרק > סימן > סעיף.
Run as: python3 scripts/fetch_national_insurance_law.py <out_path>
Then upload the output .txt to s3://insurance-kb/inbox/law/ and invoke
POST /admin/kb/scan-inbox.
Refs Task Master #2
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
236 lines
7.9 KiB
Python
Executable File
236 lines
7.9 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Fetch חוק הביטוח הלאומי from Hebrew Wikisource and convert to plain text
|
|
with the headers the shira-hermes chunker expects (חלק / פרק / סימן / סעיף)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import sys
|
|
import urllib.request
|
|
import urllib.parse
|
|
|
|
WIKI_PAGE = "חוק_הביטוח_הלאומי"
|
|
API = "https://he.wikisource.org/w/api.php"
|
|
|
|
|
|
def fetch_wikitext(page: str) -> str:
|
|
url = f"{API}?action=parse&page={urllib.parse.quote(page)}&prop=wikitext&format=json"
|
|
req = urllib.request.Request(url, headers={"User-Agent": "shira-kb/1.0"})
|
|
with urllib.request.urlopen(req) as r:
|
|
data = json.load(r)
|
|
return data["parse"]["wikitext"]["*"]
|
|
|
|
|
|
# ---- template handlers ----------------------------------------------------
|
|
|
|
# {{ח:קטע2|סימולא|שם החלק}} → "חלק <number> — <title>"
|
|
# Wikisource numbers parts/chapters in their own way; sometimes the number is
|
|
# embedded in the title. We emit the raw text of the second param and let the
|
|
# chunker pick up the prefix from there.
|
|
def _section_header(prefix: str, args: list[str]) -> str:
|
|
args = _positional_args(args)
|
|
if not args:
|
|
return ""
|
|
# The title is usually the last positional arg; for קטע templates the
|
|
# title already contains its own prefix ("פרק א׳: ...") so we don't
|
|
# add our own. Otherwise prepend the given prefix.
|
|
title = (args[-1] if args[-1].strip() else (args[0] if args else "")).strip()
|
|
if not title:
|
|
return ""
|
|
if prefix:
|
|
return f"\n\n{prefix} {title}\n\n"
|
|
return f"\n\n{title}\n\n"
|
|
|
|
|
|
# Templates that produce a statute heading line.
|
|
# Note: {{ח:קטע2}} / {{ח:קטע3}} don't directly encode the level name —
|
|
# the actual level is embedded in the *title* argument (e.g. "פרק א׳: ...").
|
|
# So we emit the title verbatim and let the chunker's existing regexes
|
|
# pick it up (which they do, for פרק / סימן prefixes).
|
|
_HEADER_TEMPLATES = {
|
|
"ח:קטע2": "",
|
|
"ח:קטע3": "",
|
|
"ח:קטע4": "",
|
|
}
|
|
|
|
|
|
def _positional_args(args: list[str]) -> list[str]:
|
|
"""Filter out named arguments (e.g. 'אחר=[פרק א׳]') from a template arg list."""
|
|
return [a for a in args if "=" not in a.split("\n", 1)[0][:40]]
|
|
|
|
|
|
def _render_section(args: list[str]) -> str:
|
|
"""{{ח:סעיף|number|title|anchor?}} → "סעיף <number>. <title>" """
|
|
args = _positional_args(args)
|
|
if not args:
|
|
return ""
|
|
number = args[0].strip()
|
|
title = (args[1] if len(args) > 1 else "").strip()
|
|
if title:
|
|
return f"\n\nסעיף {number}. {title}\n\n"
|
|
return f"\n\nסעיף {number}.\n\n"
|
|
|
|
|
|
def _render_subsection(args: list[str]) -> str:
|
|
"""{{ח:ת|א|text}} → "(א) text" on its own line"""
|
|
args = _positional_args(args)
|
|
if not args:
|
|
return ""
|
|
letter = args[0].strip()
|
|
body = (args[1] if len(args) > 1 else "").strip()
|
|
return f"\n({letter}) {body}"
|
|
|
|
|
|
def _render_linkish(args: list[str]) -> str:
|
|
"""{{ח:פנימי|anchor|label}} / {{ח:חיצוני|url|label}} / {{ח:תיבה|...}} →
|
|
just the label text, dropping the link itself."""
|
|
if not args:
|
|
return ""
|
|
# label is typically the second arg; if only one, return that.
|
|
return args[1].strip() if len(args) > 1 else args[0].strip()
|
|
|
|
|
|
def _drop(args: list[str]) -> str:
|
|
return ""
|
|
|
|
|
|
# Map template name → handler. Unknown templates are silently stripped.
|
|
_HANDLERS = {
|
|
"ח:סעיף": _render_section,
|
|
"ח:סעיף*": _render_section,
|
|
"ח:ת": _render_subsection,
|
|
"ח:תת": _render_subsection,
|
|
"ח:תתת": _render_subsection,
|
|
"ח:תתתת": _render_subsection,
|
|
"ח:תתתתת": _render_subsection,
|
|
"ח:תתתתתת": _render_subsection,
|
|
"ח:פנימי": _render_linkish,
|
|
"ח:חיצוני": _render_linkish,
|
|
"ח:תיבה": _render_linkish,
|
|
# No body — drop silently
|
|
"ח:התחלה": _drop,
|
|
"ח:סוף": _drop,
|
|
"ח:כותרת": _drop,
|
|
"ח:מאגר": _drop,
|
|
"ח:פתיח-התחלה": _drop,
|
|
"ח:מפריד": _drop,
|
|
"ח:סוגר": _drop,
|
|
"ח:מבוא": _drop,
|
|
"ח:חתימות": _drop,
|
|
"ח:הערה": _drop, # footnotes — not needed for RAG
|
|
}
|
|
|
|
|
|
def _parse_template_args(raw: str) -> list[str]:
|
|
"""Split a template body by '|' at depth 0 (respecting nested templates)."""
|
|
args: list[str] = []
|
|
buf: list[str] = []
|
|
depth = 0
|
|
i = 0
|
|
while i < len(raw):
|
|
c = raw[i]
|
|
if c == "{" and i + 1 < len(raw) and raw[i + 1] == "{":
|
|
depth += 1
|
|
buf.append("{{")
|
|
i += 2
|
|
continue
|
|
if c == "}" and i + 1 < len(raw) and raw[i + 1] == "}":
|
|
depth -= 1
|
|
buf.append("}}")
|
|
i += 2
|
|
continue
|
|
if c == "[" and i + 1 < len(raw) and raw[i + 1] == "[":
|
|
depth += 1
|
|
buf.append("[[")
|
|
i += 2
|
|
continue
|
|
if c == "]" and i + 1 < len(raw) and raw[i + 1] == "]":
|
|
depth -= 1
|
|
buf.append("]]")
|
|
i += 2
|
|
continue
|
|
if c == "|" and depth == 0:
|
|
args.append("".join(buf))
|
|
buf = []
|
|
i += 1
|
|
continue
|
|
buf.append(c)
|
|
i += 1
|
|
args.append("".join(buf))
|
|
return args
|
|
|
|
|
|
def _expand_templates(text: str) -> str:
|
|
"""Recursively resolve {{ח:...}} templates, depth-first."""
|
|
out: list[str] = []
|
|
i = 0
|
|
while i < len(text):
|
|
if text[i : i + 2] == "{{":
|
|
# find matching }}
|
|
depth = 1
|
|
j = i + 2
|
|
while j < len(text) and depth > 0:
|
|
if text[j : j + 2] == "{{":
|
|
depth += 1
|
|
j += 2
|
|
elif text[j : j + 2] == "}}":
|
|
depth -= 1
|
|
j += 2
|
|
else:
|
|
j += 1
|
|
body = text[i + 2 : j - 2] # inside {{ ... }}
|
|
# recurse into body first
|
|
expanded_body = _expand_templates(body)
|
|
args = _parse_template_args(expanded_body)
|
|
name = args[0].strip() if args else ""
|
|
if name in _HEADER_TEMPLATES:
|
|
out.append(_section_header(_HEADER_TEMPLATES[name], args[1:]))
|
|
elif name in _HANDLERS:
|
|
out.append(_HANDLERS[name](args[1:]))
|
|
else:
|
|
# Unknown template — keep the last arg if it looks like text.
|
|
if len(args) > 1:
|
|
candidate = args[-1].strip()
|
|
if candidate and len(candidate) < 200:
|
|
out.append(candidate)
|
|
i = j
|
|
else:
|
|
out.append(text[i])
|
|
i += 1
|
|
return "".join(out)
|
|
|
|
|
|
def _cleanup(text: str) -> str:
|
|
# Drop wiki tables/images we don't care about.
|
|
text = re.sub(r"\[\[קובץ:[^\]]*\]\]", "", text)
|
|
text = re.sub(r"\[\[(?:[^\]|]*\|)?([^\]]*)\]\]", r"\1", text) # [[link|label]] → label
|
|
# Drop HTML-ish refs (footnotes, comments).
|
|
text = re.sub(r"<ref[^>]*>.*?</ref>", "", text, flags=re.DOTALL)
|
|
text = re.sub(r"<!--.*?-->", "", text, flags=re.DOTALL)
|
|
text = re.sub(r"</?[a-zA-Z][^>]*>", "", text)
|
|
# Collapse 3+ newlines to 2; strip trailing spaces.
|
|
text = re.sub(r"[ \t]+\n", "\n", text)
|
|
text = re.sub(r"\n{3,}", "\n\n", text)
|
|
# Drop bold/italic wiki markup.
|
|
text = text.replace("'''", "").replace("''", "")
|
|
return text.strip()
|
|
|
|
|
|
def main() -> int:
|
|
out_path = sys.argv[1] if len(sys.argv) > 1 else "/tmp/national_insurance_law.txt"
|
|
wt = fetch_wikitext(WIKI_PAGE)
|
|
print(f"[fetch] raw wikitext: {len(wt):,} chars", file=sys.stderr)
|
|
expanded = _expand_templates(wt)
|
|
print(f"[expand] after templates: {len(expanded):,} chars", file=sys.stderr)
|
|
clean = _cleanup(expanded)
|
|
print(f"[cleanup] final: {len(clean):,} chars", file=sys.stderr)
|
|
with open(out_path, "w", encoding="utf-8") as f:
|
|
f.write(clean)
|
|
print(f"[write] {out_path}", file=sys.stderr)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|