"""Structure-aware parsing of the corpus PDFs into numbered articles. Why this file is not three lines of ``page.get_text()`` ------------------------------------------------------ The corpus is real government typesetting. Five separate properties of it would each silently poison every downstream metric if the text were extracted naively. Every one was found by measuring the actual files, and every one is asserted in ``tests/test_parse.py`` against those files. 1. **Two-column layout.** The MOHRE Labour Law is a landscape two-column document. Reading blocks in y-order interleaves the columns, splicing Article 2 into Article 4 mid-sentence. Blocks are assigned to a column and read column-major. 2. **A broken ``Th`` ligature.** The Labour Law embeds ``CronosPro`` with the ``Th`` ligature mapped to the single codepoint ``T``, so the text layer literally reads "Te employer shall" and "Tis Decree-Law". A word list would be a guess ("Ten", "Tree" and "Tan" are real words). The defect is instead detected geometrically: the ligature glyph is drawn twice as wide as a real ``T``. Measured over the whole document the two populations are cleanly separated with no overlap — real ``T`` n=95 advance/font-size 0.52 - 0.62 ("Type of", "Their") ``Th`` ligature n=306 advance/font-size 1.00 - 1.13 ("Te employer") so :data:`WIDE_GLYPH_RATIO` = 0.80 sits inside a 0.38-wide empty gap. Only the wide population is rewritten. ``scripts/glyph_audit.py`` is the tool that found this and re-runs on any new corpus to prove no other glyph is affected. 3. **Contents pages that look exactly like article headings.** Both instruments in the Labour Law PDF open with a table of contents listing ``Article (1) ... Article (74)``. Indexing those would create 113 empty articles. Contents pages are detected by their "Contents" heading and excluded from the body pass — then *reused* as an independent ground-truth index to validate what the body pass produced (see :func:`reconcile_with_contents`). The document checks the parser. 4. **One PDF, several instruments.** The Labour Law file also contains Cabinet Resolution No. 1 of 2022, whose article numbering restarts at 1. "Article 5" is ambiguous across them, so instruments are split into parts and every citation key carries its part. 5. **A missing heading in the source document.** Cabinet Resolution Article (2) has no ``Article (2)`` label on its body page — the official PDF prints only its section title, though its own table of contents lists the article. The gap is recovered from the section-title marker and reported in the diagnostics rather than being papered over. """ from __future__ import annotations import json import logging import re import unicodedata from collections import Counter from dataclasses import dataclass, field from pathlib import Path from typing import Any, Final import fitz from app.core.models import Article, ParsedDocument from app.core.settings import Settings, get_settings logger = logging.getLogger(__name__) # ── glyph-level repair ─────────────────────────────────────────────────────── # A glyph whose advance exceeds this multiple of its font size is drawing more than # one character. Only applied to codepoints listed in LIGATURE_REPAIRS: naturally wide # letters such as m, W and M also exceed this ratio and must not be touched. WIDE_GLYPH_RATIO: Final = 0.80 LIGATURE_REPAIRS: Final[dict[str, str]] = {"T": "Th"} # ── line classification ────────────────────────────────────────────────────── # An article heading is a line consisting of nothing but the article label, optionally # followed by a footnote marker. Anchoring BOTH ends is what stops the in-body # cross-reference "...in accordance with Article (9) of this Law." being read as a # heading. Verified against every `Article ...` line in the corpus: 270 bare labels, # 2 with a trailing asterisk, 1 contents range, 1 in-body cross-reference. ARTICLE_HEADING_RE: Final = re.compile( r"^\s*Article\s*\(?\s*(\d{1,3})\s*\)?\s*[*†‡]?\s*[:.–—-]?\s*$", re.IGNORECASE, ) # Contents rows may cover a span, e.g. "Article (58-64)". CONTENTS_ENTRY_RE: Final = re.compile( r"^\s*Article\s*\(?\s*(\d{1,3})\s*(?:[-–—]\s*(\d{1,3}))?\s*\)?\s*$", re.IGNORECASE, ) # A legal instrument's own title line: the designation and NOTHING else. Anchoring the # end is essential — the recitals of every Dubai law list other instruments # ("Law No. (16) of 2007 Establishing the Real Estate Regulatory Agency;") and those # must not be mistaken for the title of the document being parsed. INSTRUMENT_TITLE_RE: Final = re.compile( r"^\s*(?P(?:Federal\s+)?(?:Decree[-\s]?Law|Cabinet\s+Resolution|Law|Decree|Resolution))" r"\s+No\.?\s*\(?\s*(?P\d{1,4})\s*\)?\s*of\s+(?P\d{4})\s*[*†‡]?\s*\d{0,2}\s*$", re.IGNORECASE, ) # Same designation, but allowed to be followed by descriptive text. Used only to derive # the canonical key of the manifest title, never to detect a title line in the PDF. INSTRUMENT_KEY_RE: Final = re.compile( r"(?P(?:Federal\s+)?(?:Decree[-\s]?Law|Cabinet\s+Resolution|Law|Decree|Resolution))" r"\s+No\.?\s*\(?\s*(?P\d{1,4})\s*\)?\s*of\s+(?P\d{4})", re.IGNORECASE, ) PAGE_FURNITURE_RE: Final = re.compile(r"^\s*(?:page\s+\d+\s+of\s+\d+|\d{1,4})\s*$", re.IGNORECASE) CONTENTS_HEADING_RE: Final = re.compile(r"^\s*contents\s*$", re.IGNORECASE) # The Labour Law's definitions table sets the term, its colon and its definition as # three separate blocks. A lone colon carries no letters, so the generic # "drop punctuation-only lines" rule (which removes bullet glyphs and the Arabic rules # that separate Dubai law headings) would silently delete every separator in the table, # yielding "State United Arab Emirates." instead of "State: United Arab Emirates.". PUNCTUATION_JOINER_RE: Final = re.compile(r"^[:;]$") # Section-title bullet the Labour Law prefixes to every article title. TITLE_MARKER_CHARS: Final = "►▶•●▪»" # A line is page furniture when its font is meaningfully smaller than the body font. # Relative, not absolute: body text is 8pt in the Labour Law and 11pt in the Dubai # laws, while their footnotes are 6.5pt and 8pt respectively. FURNITURE_SIZE_RATIO: Final = 0.90 FULL_WIDTH_RATIO: Final = 0.75 GUTTER_RATIO: Final = 0.02 MAX_TITLE_CHARS: Final = 90 CENTRED_TOLERANCE_RATIO: Final = 0.10 MIN_BLOCKS_FOR_COLUMN_DETECTION: Final = 4 # A contents row needs at least a label and a title to be usable. # A contents row ends in the page it points at, and an entry spanning pages prints a # range: "Article (58-64) | Penalties | 39-40". PAGE_FURNITURE_RE covers a bare page # number but not a range, so the range survived into the title and seven articles were # indexed as "Penalties 39-40". Kept separate from PAGE_FURNITURE_RE deliberately: this # shape is only unambiguous inside a table of contents, where a trailing numeric cell is # always a page reference. In body text it could be a real range. CONTENTS_PAGE_CELL_RE: Final = re.compile(r"^\s*\d{1,4}\s*[-–—]\s*\d{1,4}\s*$") # The same range, welded to the title when the PDF emits both in one block. CONTENTS_TRAILING_PAGE_RE: Final = re.compile(r"[\s.·•]*\b\d{1,4}(?:\s*[-–—]\s*\d{1,4})?\s*$") MIN_CONTENTS_ROW_CELLS: Final = 2 # A page is either one column or two; nothing in this corpus uses three. TWO_COLUMNS: Final = 2 # Baseline tolerance when reassembling table-of-contents rows from stacked blocks. ROW_TOLERANCE_PT: Final = 3.0 # Blocks and lines whose tops fall in the same band of this height are one visual row # and must read left-to-right. Without it, the Labour Law definitions table inverts: # the definition block starts at y=121.6 and its own term at y=121.7, so ordering by # raw y emits "United Arab Emirates." before the word "State" it defines. The band is # far narrower than the tightest line pitch in the corpus (10pt), so genuine rows never # merge. ROW_QUANTUM_PT: Final = 3.0 # Largest article span a single contents row may cover, e.g. "Article (58-64)". MAX_CONTENTS_SPAN: Final = 50 # A line reaching this fraction of the text-column width is a justified body line, # never a title. See the docstring of `_looks_like_title`. JUSTIFIED_LINE_RATIO: Final = 0.90 # When a title is recognised only by being centred, it must also be clearly short. CENTRED_TITLE_WIDTH_RATIO: Final = 0.60 # Percentile of body-line widths used as the text-column width. BODY_WIDTH_PERCENTILE: Final = 0.90 # Whitespace handling after NFKC. Written with escapes so a reviewer can read the exact # ranges rather than trust invisible literals. # Zero-width characters carry no spacing and must be DELETED, not collapsed to a # space: turning U+200B into " " inserts a word break the document never had. _ZERO_WIDTH_RE: Final = re.compile("[\\u200b-\\u200d\\ufeff]") # Everything genuinely space-like collapses to one space. _WHITESPACE_RE: Final = re.compile("[ \\t\\r\\n\\u00a0\\u2000-\\u200a\\u202f\\u205f\\u3000]+") InstrumentKey = tuple[str, int, int] # `ParsedDocument` is defined in app.core.models but is part of this module's public # surface: callers parse a corpus and receive them. Declared explicitly because # `no_implicit_reexport` (correctly) refuses to forward an import by accident. __all__ = [ "ARTICLE_HEADING_RE", "INSTRUMENT_TITLE_RE", "LIGATURE_REPAIRS", "WIDE_GLYPH_RATIO", "Article", "ContentsEntry", "ParseDiagnostics", "ParseError", "ParsedDocument", "load_manifest", "parse_corpus", "parse_pdf", "reconcile_with_contents", ] class ParseError(RuntimeError): """The corpus could not be parsed into a well-formed article structure.""" @dataclass(frozen=True, slots=True) class _Line: """One visual line of text with the geometry needed to classify it.""" text: str page: int x0: float y0: float x1: float size: float bold: bool @property def centre(self) -> float: return (self.x0 + self.x1) / 2.0 @dataclass(frozen=True, slots=True) class _Block: """A layout block: the unit that column ordering operates on.""" lines: tuple[_Line, ...] x0: float y0: float x1: float @property def centre(self) -> float: return (self.x0 + self.x1) / 2.0 @property def width(self) -> float: return self.x1 - self.x0 @dataclass(frozen=True, slots=True) class _Layout: """Document-wide typographic scale, measured once from the file itself. Nothing here is hardcoded per document: `body_size` is the font size carrying the most characters, and `body_line_width` is the 90th-percentile width of lines set in it — i.e. the width of a full justified line in the text column. """ body_size: float body_line_width: float @property def furniture_ceiling(self) -> float: return self.body_size * FURNITURE_SIZE_RATIO @dataclass(frozen=True, slots=True) class ContentsEntry: """One row of a document's own table of contents — independent ground truth.""" article_no: int title: str @dataclass(slots=True) class ParseDiagnostics: """Observable evidence that the parser did what it claims. Printed by ``make index`` and asserted in tests, so a regression in extraction quality fails a build instead of quietly degrading retrieval. """ ligature_repairs: int = 0 contents_pages_skipped: int = 0 furniture_lines_dropped: int = 0 duplicate_article_splits: int = 0 parts_detected: int = 0 inferred_headings: int = 0 contents_entries_seen: int = 0 titles_from_contents: int = 0 contents_missing_in_body: list[str] = field(default_factory=list) contents_extra_in_body: list[str] = field(default_factory=list) def as_dict(self) -> dict[str, Any]: return { "ligature_repairs": self.ligature_repairs, "contents_pages_skipped": self.contents_pages_skipped, "furniture_lines_dropped": self.furniture_lines_dropped, "duplicate_article_splits": self.duplicate_article_splits, "parts_detected": self.parts_detected, "inferred_headings": self.inferred_headings, "contents_entries_seen": self.contents_entries_seen, "titles_from_contents": self.titles_from_contents, "contents_missing_in_body": self.contents_missing_in_body, "contents_extra_in_body": self.contents_extra_in_body, } @property def contents_check_passed(self) -> bool: return not self.contents_missing_in_body and not self.contents_extra_in_body # ───────────────────────────────────────────────────────────────────────────── # Glyph -> text # ───────────────────────────────────────────────────────────────────────────── def _span_text(span: dict[str, Any], diagnostics: ParseDiagnostics) -> str: """Reconstruct a span's text, repairing ligatures that lost characters.""" size = float(span.get("size") or 0.0) out: list[str] = [] for char in span["chars"]: code: str = char["c"] replacement = LIGATURE_REPAIRS.get(code) if replacement is not None and size > 0: bbox = char["bbox"] if (bbox[2] - bbox[0]) / size > WIDE_GLYPH_RATIO: out.append(replacement) diagnostics.ligature_repairs += 1 continue out.append(code) return "".join(out) def _normalise(text: str) -> str: """NFKC-fold compatibility ligatures (fi, ffi, fl) and collapse whitespace. NFKC already maps U+00A0 to a plain space, so no separate pass is needed for it. """ folded = _ZERO_WIDTH_RE.sub("", unicodedata.normalize("NFKC", text)) return _WHITESPACE_RE.sub(" ", folded).strip() def _has_letters_or_digits(text: str) -> bool: return any(ch.isalnum() for ch in text) def _instrument_key(text: str) -> InstrumentKey | None: """Canonical identity of a legal instrument: ``(kind, number, year)``.""" match = INSTRUMENT_KEY_RE.search(text) if match is None: return None kind = re.sub(r"[\s-]+", "", match.group("kind")).lower() return (kind, int(match.group("number")), int(match.group("year"))) # ───────────────────────────────────────────────────────────────────────────── # Page -> ordered lines # ───────────────────────────────────────────────────────────────────────────── def _extract_blocks(page: fitz.Page, page_no: int, diagnostics: ParseDiagnostics) -> list[_Block]: raw: dict[str, Any] = page.get_text("rawdict") blocks: list[_Block] = [] for raw_block in raw["blocks"]: if raw_block.get("type") != 0: continue lines: list[_Line] = [] for raw_line in raw_block.get("lines", []): spans = raw_line["spans"] if not spans: continue text = _normalise("".join(_span_text(sp, diagnostics) for sp in spans)) if not text: continue lx0, ly0, lx1, _ly1 = raw_line["bbox"] lines.append( _Line( text=text, page=page_no, x0=lx0, y0=ly0, x1=lx1, size=max(float(sp["size"]) for sp in spans), bold=any( "bold" in str(sp.get("font", "")).lower() or bool(int(sp.get("flags", 0)) & 16) for sp in spans ), ) ) if not lines: continue # Within a block, horizontal text reads top-down then left-right. The PDF's # internal line order does not guarantee this (the Labour Law emits some # headings after their own subtitle), so it is imposed here. lines.sort(key=lambda ln: (_row_band(ln.y0), ln.x0)) bx0, by0, bx1, _by1 = raw_block["bbox"] blocks.append(_Block(lines=tuple(lines), x0=bx0, y0=by0, x1=bx1)) return blocks def _row_band(y: float) -> int: """Quantise a top coordinate into a visual row band. See ROW_QUANTUM_PT.""" return int(y // ROW_QUANTUM_PT) def _is_two_column(blocks: list[_Block], page_width: float) -> bool: """True when the page is typeset in two columns separated by a clear gutter.""" mid = page_width / 2.0 margin = page_width * GUTTER_RATIO narrow = [b for b in blocks if b.width < page_width * FULL_WIDTH_RATIO] if len(narrow) < MIN_BLOCKS_FOR_COLUMN_DETECTION: return False if any(b.x0 < mid - margin and b.x1 > mid + margin for b in narrow): return False return any(b.x1 <= mid for b in narrow) and any(b.x0 >= mid for b in narrow) def _order_blocks(blocks: list[_Block], page_width: float) -> list[_Block]: """Return blocks in human reading order.""" if not _is_two_column(blocks, page_width): return sorted(blocks, key=lambda b: (_row_band(b.y0), b.x0)) mid = page_width / 2.0 return sorted(blocks, key=lambda b: (0 if b.centre < mid else 1, _row_band(b.y0), b.x0)) def _page_lines(page: fitz.Page, page_no: int, diagnostics: ParseDiagnostics) -> list[_Line]: blocks = _extract_blocks(page, page_no, diagnostics) return [line for block in _order_blocks(blocks, page.rect.width) for line in block.lines] def _is_contents_page(lines: list[_Line], body_size: float) -> bool: return any(CONTENTS_HEADING_RE.match(ln.text) and ln.size >= body_size for ln in lines) def _measure_layout(all_lines: list[_Line]) -> _Layout: """Measure the document's body font size and text-column width.""" weight: Counter[float] = Counter() for line in all_lines: weight[round(line.size, 1)] += len(line.text) if not weight: raise ParseError("document contains no extractable text") body_size = weight.most_common(1)[0][0] widths = sorted(ln.x1 - ln.x0 for ln in all_lines if round(ln.size, 1) == body_size) if not widths: raise ParseError("document contains no body-size lines") index = min(len(widths) - 1, int(len(widths) * BODY_WIDTH_PERCENTILE)) return _Layout(body_size=body_size, body_line_width=widths[index]) # ───────────────────────────────────────────────────────────────────────────── # Contents pages -> independent ground truth # ───────────────────────────────────────────────────────────────────────────── def _parse_contents_page(lines: list[_Line], page_width: float) -> list[ContentsEntry]: """Read a table-of-contents page into ``(article_no, title)`` rows. A contents row reads ``Article (N) | Title | page``, but the PDF does not emit it as one block: the labels are one stacked block, the titles another, the page numbers a third. Rows are therefore reassembled geometrically — lines sharing a baseline within :data:`ROW_TOLERANCE_PT` form a row, and within a row the label is the entry, the trailing number is the page, and whatever lies between them is the title. """ mid = page_width / 2.0 columns: list[list[_Line]] = [[], []] if _is_two_column_lines(lines, page_width) else [[]] two_columns = len(columns) == TWO_COLUMNS for line in lines: columns[1 if two_columns and line.centre >= mid else 0].append(line) entries: list[ContentsEntry] = [] for column in columns: for row in _group_rows(column): entries.extend(_row_to_entries(row)) return entries def _is_two_column_lines(lines: list[_Line], page_width: float) -> bool: mid = page_width / 2.0 margin = page_width * GUTTER_RATIO if any(ln.x0 < mid - margin and ln.x1 > mid + margin for ln in lines): return False return any(ln.x1 <= mid for ln in lines) and any(ln.x0 >= mid for ln in lines) def _group_rows(lines: list[_Line]) -> list[list[_Line]]: """Cluster lines that share a baseline into rows, each sorted left to right.""" rows: list[list[_Line]] = [] for line in sorted(lines, key=lambda ln: (ln.page, ln.y0, ln.x0)): if rows and abs(rows[-1][0].y0 - line.y0) <= ROW_TOLERANCE_PT: rows[-1].append(line) else: rows.append([line]) for row in rows: row.sort(key=lambda ln: ln.x0) return rows def _row_to_entries(row: list[_Line]) -> list[ContentsEntry]: """Turn one contents row into its article entries, expanding any span.""" label_index = next( (i for i, ln in enumerate(row) if CONTENTS_ENTRY_RE.match(ln.text)), None, ) if label_index is None: return [] match = CONTENTS_ENTRY_RE.match(row[label_index].text) if match is None: # pragma: no cover - guarded by the search above return [] first = int(match.group(1)) last = int(match.group(2)) if match.group(2) else first if last < first or last - first > MAX_CONTENTS_SPAN: return [] title_parts = [ ln.text.strip().lstrip(TITLE_MARKER_CHARS).strip() for ln in row[label_index + 1 :] if not PAGE_FURNITURE_RE.match(ln.text.strip()) and not CONTENTS_PAGE_CELL_RE.match(ln.text.strip()) and not CONTENTS_ENTRY_RE.match(ln.text.strip()) ] title = _normalise(" ".join(part for part in title_parts if part)) # Only strip when something with letters survives: better a title carrying a stray # page number than no title at all. stripped = CONTENTS_TRAILING_PAGE_RE.sub("", title).strip() if stripped and any(ch.isalpha() for ch in stripped): title = stripped return [ContentsEntry(article_no=n, title=title) for n in range(first, last + 1)] def reconcile_with_contents( articles: list[Article], contents_by_part: list[list[ContentsEntry]], diagnostics: ParseDiagnostics, law_id: str, ) -> list[Article]: """Cross-check and enrich parsed articles using the document's own contents pages. The k-th contents page in a file describes the k-th instrument in it, which makes it an index produced by the publisher rather than by this parser — genuinely independent evidence. It is used twice: * **Validation.** Any article the contents promises but the body pass did not produce, or vice versa, is recorded by name. A layout change then surfaces as a named discrepancy instead of a quietly smaller index. * **Titles.** The contents carries the official title of every article, already de-hyphenated and never truncated, so it supersedes the title inferred from page geometry (which can clip a title that wraps onto a second line). """ if not contents_by_part: return articles order: list[str] = [] by_part: dict[str, list[Article]] = {} for article in articles: if article.part_id not in by_part: order.append(article.part_id) by_part[article.part_id] = [] by_part[article.part_id].append(article) titles: dict[tuple[str, int], str] = {} for index, part_id in enumerate(order): if index >= len(contents_by_part): break entries = contents_by_part[index] expected = {entry.article_no for entry in entries} diagnostics.contents_entries_seen += len(expected) produced = {article.article_no for article in by_part[part_id]} for missing in sorted(expected - produced): diagnostics.contents_missing_in_body.append(f"{law_id}/{part_id}#art{missing}") for extra in sorted(produced - expected): diagnostics.contents_extra_in_body.append(f"{law_id}/{part_id}#art{extra}") for entry in entries: if entry.title: titles[(part_id, entry.article_no)] = entry.title enriched: list[Article] = [] for article in articles: official = titles.get((article.part_id, article.article_no)) if official and official != article.title: diagnostics.titles_from_contents += 1 enriched.append(article.model_copy(update={"title": official})) else: enriched.append(article) return enriched # ───────────────────────────────────────────────────────────────────────────── # Body pages -> articles # ───────────────────────────────────────────────────────────────────────────── def _looks_like_title(line: _Line, layout: _Layout, page_width: float) -> bool: """A short, heading-ish line that names the article beside it. The geometric width test is load-bearing, not cosmetic. Body text in the Dubai laws is fully justified, so the first line of an article spans the whole text column and its midpoint lands exactly on the page centre — indistinguishable from a centred title by position alone. Without the width test, Decree 43/2013 Article 1 loses its opening line ("When renewing Real Property Lease Contracts...") to the title field: silent corpus corruption that no downstream check would catch. A real title is therefore recognised by a *positive* typographic signal — a bullet marker, bold weight, or a larger font — and the centre-alignment fallback (needed only by Decree 43/2013, which sets titles in the same light face as its body) is additionally required to be narrow. """ text = line.text.strip().lstrip(TITLE_MARKER_CHARS).strip() if not text or len(text) > MAX_TITLE_CHARS: return False if text.endswith((".", ";", ":", ",")): return False if ARTICLE_HEADING_RE.match(text) or PAGE_FURNITURE_RE.match(text): return False if not _has_letters_or_digits(text): return False width = line.x1 - line.x0 if width >= layout.body_line_width * JUSTIFIED_LINE_RATIO: return False if line.bold or line.size > layout.body_size: return True if line.text.startswith(tuple(TITLE_MARKER_CHARS)): return True centred = abs(line.centre - page_width / 2.0) < page_width * CENTRED_TOLERANCE_RATIO return centred and width < layout.body_line_width * CENTRED_TITLE_WIDTH_RATIO def _slugify(text: str) -> str: slug = re.sub(r"[^a-z0-9]+", "-", text.lower()).strip("-") return slug[:60] or "part" @dataclass(slots=True) class _ArticleAccumulator: part_id: str part_title: str article_no: int title: str page_start: int page_end: int body: list[str] = field(default_factory=list) def add(self, line: _Line) -> None: self.body.append(line.text) self.page_end = max(self.page_end, line.page) def text(self) -> str: return re.sub(r"\s{2,}", " ", " ".join(self.body)).strip() @dataclass(slots=True) class _Segmenter: """Walks the ordered line stream and emits articles. Kept as a class purely so the state machine's variables are named and inspectable rather than a dozen ``nonlocal`` bindings. """ law_id: str law_label: str law_title: str layout: _Layout diagnostics: ParseDiagnostics articles: list[Article] = field(default_factory=list) current: _ArticleAccumulator | None = None pending_title: str = "" pending_key: InstrumentKey | None = None open_part_key: InstrumentKey | None = None open_part_id: str = "" open_part_title: str = "" part_ordinal: int = 0 seen_in_part: set[int] = field(default_factory=set) unassigned_chars: int = 0 part_opened: bool = False def __post_init__(self) -> None: self.pending_title = self.law_title self.pending_key = _instrument_key(self.law_title) # ── article lifecycle ──────────────────────────────────────────────── def close(self) -> None: if self.current is None: return body = self.current.text() if body: self.articles.append( Article( law_id=self.law_id, law_label=self.law_label, law_title=self.law_title, part_id=self.current.part_id, part_title=self.current.part_title, article_no=self.current.article_no, title=self.current.title, text=body, page_start=self.current.page_start, page_end=self.current.page_end, ) ) self.current = None def _open_part_if_needed(self, article_no: int) -> None: same_instrument = self.part_opened and self.open_part_key == self.pending_key duplicate = article_no in self.seen_in_part if same_instrument and not duplicate: return if same_instrument and duplicate: # A repeated number inside one instrument means a new structural section # began: the Dubai amendment restates articles of the law it amends, then # closes with its own Article 2. Splitting keeps every citation key unique, # so a chip can never resolve to two different texts. self.diagnostics.duplicate_article_splits += 1 self.part_ordinal += 1 self.open_part_key = self.pending_key self.open_part_title = self.pending_title self.open_part_id = f"{_slugify(self.pending_title)}-{self.part_ordinal}" self.seen_in_part = set() self.part_opened = True self.diagnostics.parts_detected += 1 def start_article(self, article_no: int, title: str, page: int) -> None: self.close() self._open_part_if_needed(article_no) self.seen_in_part.add(article_no) self.current = _ArticleAccumulator( part_id=self.open_part_id, part_title=self.open_part_title, article_no=article_no, title=title, page_start=page, page_end=page, ) # ── stream handling ────────────────────────────────────────────────── def on_instrument_title(self, text: str) -> None: key = _instrument_key(text) if key is None or key == self.pending_key: return # A different instrument begins here; whatever article was open ended with the # previous one. Without this the Cabinet Resolution's recitals are appended to # Labour Law Article 74. self.close() self.pending_key = key self.pending_title = text def attach_punctuation(self, mark: str) -> None: """Glue a standalone separator onto the text it belongs to.""" if self.current is not None and self.current.body: self.current.body[-1] = f"{self.current.body[-1]}{mark}" def on_body(self, line: _Line) -> None: if self.current is None: self.unassigned_chars += len(line.text) return self.current.add(line) def parse_pdf( pdf_path: Path, *, law_id: str, law_label: str, law_title: str, diagnostics: ParseDiagnostics | None = None, ) -> ParsedDocument: """Parse one corpus PDF into its numbered articles. Raises: ParseError: if the file yields no articles, which means the layout changed and the extraction rules no longer describe it. Failing loudly here is deliberate: a silently empty index makes every downstream metric meaningless. """ diag = diagnostics if diagnostics is not None else ParseDiagnostics() document = fitz.open(pdf_path) try: pages = [ (index + 1, document[index].rect.width, _page_lines(document[index], index + 1, diag)) for index in range(document.page_count) ] page_count = document.page_count finally: document.close() layout = _measure_layout([ln for _, _, lines in pages for ln in lines]) contents_by_part: list[list[ContentsEntry]] = [] segmenter = _Segmenter( law_id=law_id, law_label=law_label, law_title=law_title, layout=layout, diagnostics=diag, ) for page_no, page_width, lines in pages: if _is_contents_page(lines, layout.body_size): diag.contents_pages_skipped += 1 contents_by_part.append(_parse_contents_page(lines, page_width)) continue for position, line in enumerate(lines): stripped = line.text.strip() if line.size < layout.furniture_ceiling or PAGE_FURNITURE_RE.match(stripped): diag.furniture_lines_dropped += 1 continue if PUNCTUATION_JOINER_RE.match(stripped): segmenter.attach_punctuation(stripped) continue if not _has_letters_or_digits(stripped): diag.furniture_lines_dropped += 1 continue heading = ARTICLE_HEADING_RE.match(stripped) if heading is not None: segmenter.start_article( int(heading.group(1)), _find_article_title(lines, position, layout, page_width), page_no, ) continue if INSTRUMENT_TITLE_RE.match(stripped) and line.size >= layout.body_size: segmenter.on_instrument_title(stripped) continue if _is_consumed_title(lines, position, segmenter.current): continue inferred = _infer_missing_heading(lines, position, segmenter, layout, page_width) if inferred is not None: diag.inferred_headings += 1 segmenter.start_article(inferred[0], inferred[1], page_no) continue segmenter.on_body(line) segmenter.close() if not segmenter.articles: raise ParseError( f"{pdf_path.name}: no articles were extracted. The document layout no longer " "matches the extraction rules in app/rag/parse.py." ) articles = _propagate_sections( reconcile_with_contents(segmenter.articles, contents_by_part, diag, law_id) ) return ParsedDocument( law_id=law_id, law_label=law_label, law_title=law_title, filename=pdf_path.name, page_count=page_count, articles=tuple(articles), unassigned_chars=segmenter.unassigned_chars, ) def _propagate_sections(articles: list[Article]) -> list[Article]: """Give every article the nearest preceding heading inside its own instrument. Dubai Law 26/2007 prints chapter headings ("Term of Lease Contract", "Eviction Cases") above the first article of each chapter only, so Articles 6, 7 and 8 carry no heading of their own while plainly belonging to one. Carrying the chapter forward gives those articles topical context in the embedded text, which is precisely the signal a bare "Where the term of a Lease Contract expires..." otherwise lacks. The carry resets at every part boundary so a chapter can never leak from one instrument into another. """ out: list[Article] = [] current_part = "" section = "" for article in articles: if article.part_id != current_part: current_part = article.part_id section = "" if article.title: section = article.title out.append(article.model_copy(update={"section": section})) return out def _find_article_title( lines: list[_Line], position: int, layout: _Layout, page_width: float ) -> str: """Locate the human title beside an article heading. The corpus uses both conventions: the Labour Law emits ``Article (6)`` and ``> Recruitment and Employment of Workers`` inside one block in either order, while the Dubai laws centre the title on the line above the number. """ for neighbour in (position + 1, position - 1): if not 0 <= neighbour < len(lines): continue candidate = lines[neighbour] if candidate.size < layout.furniture_ceiling: continue if _looks_like_title(candidate, layout, page_width): return candidate.text.strip().lstrip(TITLE_MARKER_CHARS).strip() return "" def _is_consumed_title( lines: list[_Line], position: int, current: _ArticleAccumulator | None ) -> bool: """True when this line was already taken as the current article's title.""" if current is None or not current.title or current.body: return False text = lines[position].text.strip().lstrip(TITLE_MARKER_CHARS).strip() return text == current.title def _infer_missing_heading( lines: list[_Line], position: int, segmenter: _Segmenter, layout: _Layout, page_width: float, ) -> tuple[int, str] | None: """Recover an article whose ``Article (N)`` label is absent from the source PDF. Cabinet Resolution No. 1 of 2022 prints the section title of its Article 2 but not the label, while its own table of contents lists the article. A bare section-title marker appearing mid-body, not attached to any label, is therefore treated as the start of the next article — but only when that number is unused and the next explicit label in the stream is further ahead, so a title that simply precedes its own label is never mistaken for a missing one. """ line = lines[position] if segmenter.current is None or not segmenter.current.body: return None if not line.text.startswith(tuple(TITLE_MARKER_CHARS)): return None if not _looks_like_title(line, layout, page_width): return None candidate_no = segmenter.current.article_no + 1 if candidate_no in segmenter.seen_in_part: return None for following in lines[position + 1 :]: heading = ARTICLE_HEADING_RE.match(following.text.strip()) if heading is None: continue # The very next label is the one this title belongs to, or it re-uses the # number we would have inferred. Either way, do not invent an article. return None if int(heading.group(1)) <= candidate_no else _titled(candidate_no, line) return _titled(candidate_no, line) def _titled(article_no: int, line: _Line) -> tuple[int, str]: return article_no, line.text.strip().lstrip(TITLE_MARKER_CHARS).strip() # ───────────────────────────────────────────────────────────────────────────── # Corpus-level entry point # ───────────────────────────────────────────────────────────────────────────── def load_manifest(settings: Settings | None = None) -> list[dict[str, Any]]: cfg = settings or get_settings() if not cfg.manifest_path.exists(): raise ParseError( f"{cfg.manifest_path} not found. Run `make corpus` first — it downloads the " "official law PDFs and writes the provenance manifest." ) payload: dict[str, Any] = json.loads(cfg.manifest_path.read_text(encoding="utf-8")) documents: list[dict[str, Any]] = payload["documents"] return documents def parse_corpus( settings: Settings | None = None, diagnostics: ParseDiagnostics | None = None, ) -> list[ParsedDocument]: """Parse every document named in the corpus manifest.""" cfg = settings or get_settings() diag = diagnostics if diagnostics is not None else ParseDiagnostics() parsed: list[ParsedDocument] = [] for entry in load_manifest(cfg): pdf_path = cfg.pdf_dir / entry["filename"] if not pdf_path.exists(): raise ParseError(f"{pdf_path} is missing. Run `make corpus`.") document = parse_pdf( pdf_path, law_id=entry["law_id"], law_label=entry["label"], law_title=entry["title"], diagnostics=diag, ) logger.info( "parsed %s: %d articles across %d pages", document.law_id, len(document.articles), document.page_count, ) parsed.append(document) return parsed