indexing¶

def document_title(root: etree._Element) -> str | None:
    """Best-effort document title across the vocabularies opm ships ODDs for."""
    from opm.transform import xpath_select

    for expr in _TITLE_XPATH.values():
        try:
            result = xpath_select(root, expr)
        except Exception:  # noqa: BLE001 — a vocabulary mismatch is expected here
            continue
        if isinstance(result, list):
            result = result[0] if result else None
        if isinstance(result, etree._Element):
            text = _clean(''.join(result.itertext()))
            if text:
                return text
        elif result:
            text = _clean(str(result))
            if text:
                return text
    return None

def build_records(
    document: list,
    *,
    doc_stem: str,
    source: str,
    title: str | None = None,
    anchors: dict[str, str] | None = None,
    chunk_file: str | None = None,
    page_metadata: dict[str, str] | None = None,
    options: IndexOptions | None = None,
) -> list[dict]:
    """Turn one document's (or one chunk's) JSON-mode records into index records.

    *page_metadata* is copied onto every record — values already stripped to
    scalars, typically from ``fragment`` fields evaluated once for the page.
    """
    options = options or IndexOptions()
    anchors = anchors or {}

    units, extracted = _iter_units(document, options)

    records: list[dict] = []
    seen: dict[str, int] = {}
    ids: dict[int, str] = {}
    # Extracted units come last so the passage they were taken from already has
    # an id to point at.
    for unit in units + extracted:
        pieces = _split(unit, options)
        floor = options.min_chars if unit.min_chars is None else unit.min_chars
        kept = [p for p in pieces if len(p[0]) >= floor]
        if not kept:
            continue
        for position, (text, anchor, anchor_path) in enumerate(kept):
            base = _record_id(doc_stem, anchor, anchor_path, scope=chunk_file)
            record_id = base if len(kept) == 1 else f'{base}-{position}'
            # Two units can still land on one id when neither carries an
            # xml:id and a selector produced records off the same path. A
            # vector store overwrites on a repeated id, so disambiguate rather
            # than silently lose a passage.
            if record_id in seen:
                seen[record_id] += 1
                record_id = f'{record_id}~{seen[record_id]}'
            else:
                seen[record_id] = 0
            ids.setdefault(id(unit), record_id)
            metadata: dict[str, Any] = {
                'source': source,
                'doc': doc_stem,
                'xpath': anchor_path or unit.xpath or '',
                'part': position,
                'n_parts': len(kept),
                'chars': len(text),
                'hash': hashlib.sha1(text.encode('utf-8')).hexdigest()[:16],
            }
            if title:
                metadata['title'] = title
            if unit.heading:
                metadata['heading'] = unit.heading
            if anchor:
                metadata['xml_id'] = anchor
            if chunk_file:
                metadata['chunk'] = chunk_file
            href = _href(anchor, unit.entry_anchor, anchors, chunk_file)
            if href:
                metadata['href'] = href
            if unit.kind:
                metadata['kind'] = unit.kind
                parent_id = ids.get(id(unit.parent)) if unit.parent else None
                if parent_id:
                    metadata['parent'] = parent_id
            for name, values in unit.fields.items():
                metadata[name] = list(values)
            if page_metadata:
                metadata.update(page_metadata)
            records.append({
                'id': record_id,
                'document': text,
                'metadata': metadata,
            })
    return records

def index_document(
    xml_path: Path,
    *,
    cfg: ProjectConfig,
    odd: Path | None = None,
    project_root: Path | None = None,
    options: IndexOptions | None = None,
    base_css: str | None = None,
) -> list[dict]:
    """Transform *xml_path* in ``json`` mode and roll the records up for indexing.

    [`opm.project.Project.index`][opm.project.Project.index] runs this over a corpus. *odd* replaces
    ``[transform.json] odd``, *options* the ``[index]`` settings, and
    *base_css* is passed on to the compiler.

    Where the project chunks its output, each chunk is transformed on its own
    and tagged with the file it will be published as. That is what makes a
    retrieval hit citable: a chunk-based href needs no ``xml:id`` at all, which
    matters because chunk selectors may rebuild the region as a detached tree
    whose ids never existed in the source document.
    """
    from opm.odd_cache import resolve_transform_module
    from opm.transform import (
        load_transform_module,
        project_xpath_env,
    )

    project_root = project_root or Path.cwd()
    # The config already carries the rollup tuning and the field declarations;
    # a caller that passes cfg but no options means "use what the project says".
    options = options or IndexOptions(
        max_chars=cfg.index_max_chars,
        min_chars=cfg.index_min_chars,
        overlap=cfg.index_overlap,
        fields=cfg.index_fields,
        units=cfg.index_units,
    )
    resolved_odd = odd if odd is not None else cfg.odd_for_type('json')
    resolved = resolve_transform_module(
        odd=resolved_odd,
        output_mode='json',
        use_packaged_default=resolved_odd is None,
        base_css=base_css,
    )
    module = load_transform_module(resolved.module_path)

    root = etree.parse(str(xml_path)).getroot()
    title = document_title(root)
    parameters = dict(cfg.parameters)

    # Register lookups (e.g. collection($global:register-root)) need the same
    # XPath runtime context as `opm transform`; without collections, ODD models
    # fall back to the TEI surface form instead of the register main name.
    xpath_env = project_xpath_env(cfg, xml_path)
    transform_opts: dict[str, Any] = dict(parameters)

    def transform(node) -> list:
        payload = module.transform(node, dict(transform_opts) or None, xpath_env=xpath_env)[0]
        return json.loads(payload).get('document', [])

    fragment_fields = any(spec.fragment for spec in options.fields)
    chunking = cfg.chunking
    processor_module = resolved.module_path
    webcomponents = False
    if fragment_fields:
        processor_module, chunking = _web_chunking(cfg, odd, base_css)
        webcomponents = bool(cfg.webcomponents_enabled)

    processor = _chunk_processor(
        root,
        processor_module,
        cfg,
        project_root,
        chunking=chunking,
        xpath_env=xpath_env,
        webcomponents=webcomponents,
    )
    cache: dict = {}

    def records_for(node, *, chunk_file: str | None, anchors: dict | None, position: int, context) -> list[dict]:
        return build_records(
            transform(node),
            doc_stem=xml_path.stem,
            source=str(xml_path),
            title=title,
            anchors=anchors,
            chunk_file=chunk_file,
            page_metadata=_page_metadata(
                processor, options.fields, context, position, cache,
            ) if fragment_fields else None,
            options=options,
        )

    if processor is None or not processor.chunks:
        return records_for(
            root, chunk_file=None, anchors=None, position=0, context=root,
        )

    anchors = processor.build_anchor_index()
    records: list[dict] = []
    for position, chunk in enumerate(processor.chunks):
        metadata = processor.generate_chunk_metadata(chunk, position)
        records.extend(
            records_for(
                chunk,
                chunk_file=metadata.file,
                anchors=anchors,
                position=position,
                context=chunk,
            ),
        )
    return records

def write_jsonl(records: list[dict], path: Path) -> None:
    """Write *records* to *path*, one JSON object per line (UTF-8)."""
    with path.open('w', encoding='utf-8') as handle:
        for record in records:
            handle.write(json.dumps(record, ensure_ascii=False))
            handle.write('\n')