feat(kb): hierarchical heading_path for statutes (חלק>פרק>סימן>סעיף)
Before: chunk_statute produced heading_path = section_ref = "ס' 195".
After: heading_path walks the four Israeli-statute levels (part, chapter,
sub-chapter, section) and renders them as "חלק ט׳ — ביטוח נכות > פרק ב׳
— ועדות רפואיות > סימן א׳ > ס' 195". section_ref stays the leaf so
citations stay precise.
- New _RE_PART / _RE_CHAPTER / _RE_SUBCHAPTER regexes with Hebrew ordinal
support including multi-letter gershayim ("י״א", "ט״ז").
- _collect_markers emits all markers ordered by position; chunk_statute
walks them tracking active_part/chapter/subchapter and attaches the
full path to each section chunk.
Refs Task Master #2
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
+99
-18
@@ -19,11 +19,31 @@ _TOKENS_PER_CHAR = 0.55
|
|||||||
# and a batch is capped at 120K tokens. Keep chunks small so batches fit.
|
# and a batch is capped at 120K tokens. Keep chunks small so batches fit.
|
||||||
_MAX_TOKENS_PER_CHUNK = 1500
|
_MAX_TOKENS_PER_CHUNK = 1500
|
||||||
|
|
||||||
# A "section" in Israeli statutes: "סעיף 12", "תקנה 5", "5א.", "12(ב)"
|
# Hierarchy markers in Israeli statutes. Each regex captures the full heading line
|
||||||
_SECTION_RE = re.compile(
|
# and is ordered from largest container to smallest. The section-level match
|
||||||
r"(?m)^\s*(?:סעיף\s+|תקנה\s+|§\s*)?(\d+[א-ת]?\.?)\s*(?:\(([א-ת\d]+)\))?\s*[\.\:]",
|
# (_RE_SECTION) also captures the section number and optional sub-letter.
|
||||||
|
#
|
||||||
|
# Hebrew ordinal numbers: "א", "ב", … "ט׳", "י״א", "י״ב", … "ט״ז", etc.
|
||||||
|
# Pattern: one or two Hebrew letters with optional gershayim/geresh between them,
|
||||||
|
# and optionally followed by a trailing geresh.
|
||||||
|
_HEB_ORD = r"[א-ת](?:[\"״][א-ת])?[׳'\"]?"
|
||||||
|
|
||||||
|
_RE_PART = re.compile(rf"(?m)^\s*חלק\s+({_HEB_ORD})\s*[:\-–—]?\s*(.*?)\s*$")
|
||||||
|
_RE_CHAPTER = re.compile(rf"(?m)^\s*פרק\s+({_HEB_ORD})\s*[:\-–—]?\s*(.*?)\s*$")
|
||||||
|
_RE_SUBCHAPTER = re.compile(rf"(?m)^\s*סימן\s+({_HEB_ORD})\s*[:\-–—]?\s*(.*?)\s*$")
|
||||||
|
# Section header: "סעיף 12.", "12א.", "תקנה 5.", at start of a line, followed by
|
||||||
|
# optional sub-letter "(ב)" and then a dot / colon / dash or newline.
|
||||||
|
_RE_SECTION = re.compile(
|
||||||
|
r"(?m)^\s*(?:סעיף\s+|תקנה\s+|§\s*)?(\d+[א-ת]?)\s*(?:\(([א-ת\d]+)\))?\s*[\.:\-–—]"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
_STATUTE_MARKERS = [
|
||||||
|
("part", _RE_PART),
|
||||||
|
("chapter", _RE_CHAPTER),
|
||||||
|
("subchapter", _RE_SUBCHAPTER),
|
||||||
|
("section", _RE_SECTION),
|
||||||
|
]
|
||||||
|
|
||||||
# Header in circulars — markdown-ish headings, or numbered "נושא: ..." / "מס' N. ..."
|
# Header in circulars — markdown-ish headings, or numbered "נושא: ..." / "מס' N. ..."
|
||||||
_HEADER_RE = re.compile(
|
_HEADER_RE = re.compile(
|
||||||
r"(?m)^\s*(?:#{1,4}\s+|מס['׳]\s*\d+\s*[\.\:]\s*|[A-Zא-ת][^\n]{0,80}\:\s*$)",
|
r"(?m)^\s*(?:#{1,4}\s+|מס['׳]\s*\d+\s*[\.\:]\s*|[A-Zא-ת][^\n]{0,80}\:\s*$)",
|
||||||
@@ -83,34 +103,95 @@ def _split_oversized(chunks: list[Chunk]) -> list[Chunk]:
|
|||||||
return result
|
return result
|
||||||
|
|
||||||
|
|
||||||
|
def _collect_markers(text: str) -> list[dict]:
|
||||||
|
"""Collect all hierarchy markers from the text, sorted by position.
|
||||||
|
|
||||||
|
Each marker carries: {level, start, end, number, title, heading}.
|
||||||
|
level ∈ {'chapter', 'subchapter', 'section'}.
|
||||||
|
"""
|
||||||
|
markers: list[dict] = []
|
||||||
|
for level, rx in _STATUTE_MARKERS:
|
||||||
|
for m in rx.finditer(text):
|
||||||
|
g1 = m.group(1) if m.groups() else ""
|
||||||
|
if level == "section":
|
||||||
|
number = g1
|
||||||
|
sub = m.group(2) if m.lastindex and m.lastindex >= 2 else None
|
||||||
|
ref = f"ס' {number}" + (f"({sub})" if sub else "")
|
||||||
|
heading = ref
|
||||||
|
title = ""
|
||||||
|
else:
|
||||||
|
prefix = {"part": "חלק", "chapter": "פרק", "subchapter": "סימן"}[level]
|
||||||
|
title = m.group(2).strip() if m.lastindex and m.lastindex >= 2 else ""
|
||||||
|
number = g1
|
||||||
|
ref = f"{prefix} {number}"
|
||||||
|
heading = f"{ref}{(' — ' + title) if title else ''}"
|
||||||
|
markers.append({
|
||||||
|
"level": level,
|
||||||
|
"start": m.start(),
|
||||||
|
"end": m.end(),
|
||||||
|
"number": number,
|
||||||
|
"ref": ref,
|
||||||
|
"heading": heading,
|
||||||
|
"title": title,
|
||||||
|
})
|
||||||
|
markers.sort(key=lambda x: x["start"])
|
||||||
|
return markers
|
||||||
|
|
||||||
|
|
||||||
def chunk_statute(text: str) -> list[dict]:
|
def chunk_statute(text: str) -> list[dict]:
|
||||||
"""Law / regulation: one chunk per section. Preserves section numbering."""
|
"""Law / regulation: one chunk per section, with hierarchical heading_path.
|
||||||
|
|
||||||
|
heading_path format: "חלק ט' — ביטוח נכות > פרק ב' > סימן א' > ס' 195(א)".
|
||||||
|
section_ref stays the leaf ("ס' 195(א)") for precise citation.
|
||||||
|
"""
|
||||||
text = text.strip()
|
text = text.strip()
|
||||||
if not text:
|
if not text:
|
||||||
return []
|
return []
|
||||||
|
|
||||||
matches = list(_SECTION_RE.finditer(text))
|
markers = _collect_markers(text)
|
||||||
if not matches:
|
section_markers = [m for m in markers if m["level"] == "section"]
|
||||||
# Fallback: treat the whole thing as one chunk.
|
if not section_markers:
|
||||||
return _as_dicts(
|
return _as_dicts([Chunk(0, "", "", text, _count_tokens(text))])
|
||||||
[Chunk(0, "", "", text, _count_tokens(text))]
|
|
||||||
)
|
# Walk markers in order, tracking the active part/chapter/subchapter.
|
||||||
|
active_part: dict | None = None
|
||||||
|
active_chapter: dict | None = None
|
||||||
|
active_subchapter: dict | None = None
|
||||||
|
|
||||||
chunks: list[Chunk] = []
|
chunks: list[Chunk] = []
|
||||||
for i, m in enumerate(matches):
|
for i, m in enumerate(markers):
|
||||||
start = m.start()
|
if m["level"] == "part":
|
||||||
end = matches[i + 1].start() if i + 1 < len(matches) else len(text)
|
active_part = m
|
||||||
|
active_chapter = None
|
||||||
|
active_subchapter = None
|
||||||
|
continue
|
||||||
|
if m["level"] == "chapter":
|
||||||
|
active_chapter = m
|
||||||
|
active_subchapter = None
|
||||||
|
continue
|
||||||
|
if m["level"] == "subchapter":
|
||||||
|
active_subchapter = m
|
||||||
|
continue
|
||||||
|
# section marker
|
||||||
|
start = m["start"]
|
||||||
|
end = markers[i + 1]["start"] if i + 1 < len(markers) else len(text)
|
||||||
content = text[start:end].strip()
|
content = text[start:end].strip()
|
||||||
if not content:
|
if not content:
|
||||||
continue
|
continue
|
||||||
number = m.group(1).rstrip(".")
|
parts: list[str] = []
|
||||||
sub = m.group(2)
|
if active_part:
|
||||||
section_ref = f"ס' {number}" + (f"({sub})" if sub else "")
|
parts.append(active_part["heading"])
|
||||||
|
if active_chapter:
|
||||||
|
parts.append(active_chapter["heading"])
|
||||||
|
if active_subchapter:
|
||||||
|
parts.append(active_subchapter["heading"])
|
||||||
|
parts.append(m["ref"])
|
||||||
|
heading_path = " > ".join(parts)
|
||||||
chunks.append(
|
chunks.append(
|
||||||
Chunk(
|
Chunk(
|
||||||
chunk_index=len(chunks),
|
chunk_index=len(chunks),
|
||||||
heading_path=section_ref,
|
heading_path=heading_path,
|
||||||
section_ref=section_ref,
|
section_ref=m["ref"],
|
||||||
content=content,
|
content=content,
|
||||||
token_count=_count_tokens(content),
|
token_count=_count_tokens(content),
|
||||||
)
|
)
|
||||||
|
|||||||
Reference in New Issue
Block a user