from __future__ import annotations
from collections.abc import Mapping
from pathlib import Path
from typing import Any
[docs]
def make_text_unit_record(
*,
schema_version: str,
doc_id: str,
source_uri: str | None,
filename: str,
page_number: int,
unit_index: int,
text: str,
bbox: tuple[float, float, float, float],
extracted_at: str,
unit_type: str = "block",
warning: str | None = None,
) -> dict[str, Any]:
"""Create one flat BigQuery-ready text unit record."""
normalized_text = text.strip()
bbox_x0, bbox_y0, bbox_x1, bbox_y1 = bbox
return {
"schema_version": schema_version,
"doc_id": doc_id,
"source_uri": source_uri,
"filename": Path(filename).name,
"page_number": page_number,
"unit_type": unit_type,
"unit_index": unit_index,
"text": normalized_text,
"bbox_x0": float(bbox_x0),
"bbox_y0": float(bbox_y0),
"bbox_x1": float(bbox_x1),
"bbox_y1": float(bbox_y1),
"char_count": len(normalized_text),
"word_count": len(normalized_text.split()),
"extractor": "pymupdf",
"extracted_at": extracted_at,
"warning": warning,
}
[docs]
def is_text_unit_record(record: Mapping[str, Any]) -> bool:
"""Return whether a mapping has the v0.1 flat record shape."""
expected_fields = {
"schema_version",
"doc_id",
"source_uri",
"filename",
"page_number",
"unit_type",
"unit_index",
"text",
"bbox_x0",
"bbox_y0",
"bbox_x1",
"bbox_y1",
"char_count",
"word_count",
"extractor",
"extracted_at",
"warning",
}
return set(record) == expected_fields