chaim
1739fa1211
ספר הליקויים produced 698 chunks, most of them junk. A line like "3 . . . . . . . . . . . . . . . . . . . . . . . . . . הוראות החוק" from its table of contents was being treated as section 3. Phone numbers like "02-6709701" became section 02. Both of these polluted the index and broke the browse view's readability. Tighten _RE_SECTION so it requires either the explicit prefix (סעיף / תקנה / §) OR real text immediately after the terminator — a row of dots or more digits disqualifies. The scraped חוק הביטוח הלאומי still chunks correctly because the scraper emits "סעיף N. ..." with the prefix. Refs Task Master #2 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
304 lines
10 KiB
Python
304 lines
10 KiB
Python
"""Text chunking — strategies per source kind.
|
||
|
||
All strategies return list[dict] with keys:
|
||
chunk_index, heading_path, section_ref, content, token_count
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from dataclasses import dataclass
|
||
from typing import Literal
|
||
|
||
# Approx tokens per char for Hebrew/mixed text (voyage-multilingual-2 tokenizer).
|
||
# Voyage tends to split Hebrew into 1.5-2 tokens per word; 0.55 tokens/char
|
||
# is a conservative estimate that matches what the API actually reports.
|
||
_TOKENS_PER_CHAR = 0.55
|
||
|
||
# Hard safety ceiling per chunk. Voyage accepts up to 32K tokens per item,
|
||
# and a batch is capped at 120K tokens. Keep chunks small so batches fit.
|
||
_MAX_TOKENS_PER_CHUNK = 1500
|
||
|
||
# Hierarchy markers in Israeli statutes. Each regex captures the full heading line
|
||
# and is ordered from largest container to smallest. The section-level match
|
||
# (_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, then either EOL or
|
||
# real Hebrew/English text (NOT a row of dots / a phone number / more digits).
|
||
#
|
||
# The key change vs the original: the tail `[\.:\-–—]` must be followed by
|
||
# at least one letter within the same short lookahead. This filters out:
|
||
# - phone numbers like "02-6709701"
|
||
# - table-of-contents rows like "3 . . . . . . . . . . . . . . ."
|
||
# - page-number-only lines like "23."
|
||
# while still catching real section headers like "סעיף 12. הגדרות" or
|
||
# "195. חישוב קצבת נכות".
|
||
_RE_SECTION = re.compile(
|
||
r"(?m)^\s*(?:סעיף\s+|תקנה\s+|§\s*)" # prefix is now REQUIRED (one of סעיף / תקנה / §)
|
||
r"(\d+[א-ת]?)\s*(?:\(([א-ת\d]+)\))?"
|
||
r"\s*[\.:\-–—]\s*"
|
||
r"(?=[^\d\.\s])" # next non-space char must not be digit/dot — i.e. real text follows
|
||
)
|
||
|
||
_STATUTE_MARKERS = [
|
||
("part", _RE_PART),
|
||
("chapter", _RE_CHAPTER),
|
||
("subchapter", _RE_SUBCHAPTER),
|
||
("section", _RE_SECTION),
|
||
]
|
||
|
||
# Header in circulars — markdown-ish headings, or numbered "נושא: ..." / "מס' N. ..."
|
||
_HEADER_RE = re.compile(
|
||
r"(?m)^\s*(?:#{1,4}\s+|מס['׳]\s*\d+\s*[\.\:]\s*|[A-Zא-ת][^\n]{0,80}\:\s*$)",
|
||
)
|
||
|
||
|
||
@dataclass
|
||
class Chunk:
|
||
chunk_index: int
|
||
heading_path: str
|
||
section_ref: str
|
||
content: str
|
||
token_count: int
|
||
|
||
|
||
def _count_tokens(text: str) -> int:
|
||
return max(1, int(len(text) * _TOKENS_PER_CHAR))
|
||
|
||
|
||
def _as_dicts(chunks: list[Chunk]) -> list[dict]:
|
||
return [
|
||
{
|
||
"chunk_index": c.chunk_index,
|
||
"heading_path": c.heading_path or None,
|
||
"section_ref": c.section_ref or None,
|
||
"content": c.content,
|
||
"token_count": c.token_count,
|
||
}
|
||
for c in chunks
|
||
]
|
||
|
||
|
||
def _split_oversized(chunks: list[Chunk]) -> list[Chunk]:
|
||
"""Split any chunk whose token_count exceeds the ceiling into fixed-size pieces.
|
||
|
||
Preserves heading_path/section_ref on each piece and re-indexes.
|
||
"""
|
||
max_chars = int(_MAX_TOKENS_PER_CHUNK / _TOKENS_PER_CHAR)
|
||
result: list[Chunk] = []
|
||
for c in chunks:
|
||
if c.token_count <= _MAX_TOKENS_PER_CHUNK:
|
||
c.chunk_index = len(result)
|
||
result.append(c)
|
||
continue
|
||
text = c.content
|
||
for start in range(0, len(text), max_chars):
|
||
piece = text[start : start + max_chars]
|
||
result.append(
|
||
Chunk(
|
||
chunk_index=len(result),
|
||
heading_path=c.heading_path,
|
||
section_ref=c.section_ref,
|
||
content=piece,
|
||
token_count=_count_tokens(piece),
|
||
)
|
||
)
|
||
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]:
|
||
"""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()
|
||
if not text:
|
||
return []
|
||
|
||
markers = _collect_markers(text)
|
||
section_markers = [m for m in markers if m["level"] == "section"]
|
||
if not section_markers:
|
||
return _as_dicts([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] = []
|
||
for i, m in enumerate(markers):
|
||
if m["level"] == "part":
|
||
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()
|
||
if not content:
|
||
continue
|
||
parts: list[str] = []
|
||
if active_part:
|
||
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(
|
||
Chunk(
|
||
chunk_index=len(chunks),
|
||
heading_path=heading_path,
|
||
section_ref=m["ref"],
|
||
content=content,
|
||
token_count=_count_tokens(content),
|
||
)
|
||
)
|
||
return _as_dicts(chunks)
|
||
|
||
|
||
def chunk_circular(
|
||
text: str,
|
||
target_tokens: int = 600,
|
||
overlap_tokens: int = 80,
|
||
) -> list[dict]:
|
||
"""Circular: header-aware split, then pack paragraphs up to target size with overlap."""
|
||
text = text.strip()
|
||
if not text:
|
||
return []
|
||
|
||
target_chars = int(target_tokens / _TOKENS_PER_CHAR)
|
||
overlap_chars = int(overlap_tokens / _TOKENS_PER_CHAR)
|
||
|
||
headers = list(_HEADER_RE.finditer(text))
|
||
segments: list[tuple[str, str]] = [] # (heading, body)
|
||
if not headers:
|
||
segments.append(("", text))
|
||
else:
|
||
if headers[0].start() > 0:
|
||
segments.append(("", text[: headers[0].start()].strip()))
|
||
for i, h in enumerate(headers):
|
||
end = headers[i + 1].start() if i + 1 < len(headers) else len(text)
|
||
heading = h.group(0).strip().rstrip(":").lstrip("#").strip()
|
||
body = text[h.end() : end].strip()
|
||
if body:
|
||
segments.append((heading, body))
|
||
|
||
chunks: list[Chunk] = []
|
||
for heading, body in segments:
|
||
if len(body) <= target_chars:
|
||
chunks.append(
|
||
Chunk(
|
||
chunk_index=len(chunks),
|
||
heading_path=heading,
|
||
section_ref=heading,
|
||
content=body,
|
||
token_count=_count_tokens(body),
|
||
)
|
||
)
|
||
continue
|
||
|
||
# Pack paragraphs
|
||
paragraphs = [p.strip() for p in body.split("\n\n") if p.strip()]
|
||
buf = ""
|
||
for p in paragraphs:
|
||
if buf and len(buf) + len(p) + 2 > target_chars:
|
||
chunks.append(
|
||
Chunk(
|
||
chunk_index=len(chunks),
|
||
heading_path=heading,
|
||
section_ref=heading,
|
||
content=buf.strip(),
|
||
token_count=_count_tokens(buf),
|
||
)
|
||
)
|
||
# Overlap: keep last N chars
|
||
buf = (buf[-overlap_chars:] + "\n\n" + p) if overlap_chars > 0 else p
|
||
else:
|
||
buf = f"{buf}\n\n{p}" if buf else p
|
||
if buf.strip():
|
||
chunks.append(
|
||
Chunk(
|
||
chunk_index=len(chunks),
|
||
heading_path=heading,
|
||
section_ref=heading,
|
||
content=buf.strip(),
|
||
token_count=_count_tokens(buf),
|
||
)
|
||
)
|
||
|
||
return _as_dicts(chunks)
|
||
|
||
|
||
def chunk(text: str, kind: Literal["law", "regulation", "circular"]) -> list[dict]:
|
||
if kind in ("law", "regulation"):
|
||
raw = chunk_statute(text)
|
||
else:
|
||
raw = chunk_circular(text)
|
||
# Reconstruct Chunk objects to apply the safety ceiling.
|
||
chunks = [
|
||
Chunk(
|
||
chunk_index=r["chunk_index"],
|
||
heading_path=r["heading_path"] or "",
|
||
section_ref=r["section_ref"] or "",
|
||
content=r["content"],
|
||
token_count=r["token_count"],
|
||
)
|
||
for r in raw
|
||
]
|
||
return _as_dicts(_split_oversized(chunks))
|