Coverage for src/lilbee/data/extract/document.py: 100%
260 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-08 09:20 +0000
1"""Document extraction: one xberg pass that natively extracts text and OCRs
2scanned pages/images through the registered backend; chunk + embed the result."""
4from __future__ import annotations
6import contextvars
7import logging
8import time
9from collections.abc import AsyncGenerator, Generator, Sequence
10from contextlib import asynccontextmanager, contextmanager
11from dataclasses import replace
12from pathlib import Path
13from typing import TYPE_CHECKING, Any, Protocol
15from lilbee.app.services import get_services
16from lilbee.core.config import active_config
17from lilbee.data.offload import to_ingest_thread
18from lilbee.data.store import ChunkType, PageTextRecord, SourceMeta
19from lilbee.data.title import derive_title, source_meta_from_extraction
20from lilbee.data.types import (
21 IMAGE_CONTENT_TYPE,
22 MARKDOWN_OUTPUT,
23 PDF_CONTENT_TYPE,
24 ChunkRecord,
25 ExtractMode,
26 MemberRecords,
27 OcrBackendName,
28)
29from lilbee.providers.base import aux_options
30from lilbee.runtime.progress import (
31 DetailedProgressCallback,
32 EventType,
33 ExtractEvent,
34 noop_callback,
35)
37from .backends.vision_ocr import backend_options_for, ocr_request
38from .batch import active_extract_batcher
39from .chunk import ChunkLimitError, build_chunking_config, chunk_text, enforce_chunk_limit
40from .trace import ExtractionTrace, trace_extraction, trace_log
42if TYPE_CHECKING:
43 from xberg import (
44 ExtractedDocument,
45 ExtractionConfig,
46 LayoutDetectionConfig,
47 OcrConfig,
48 PdfConfig,
49 )
51 from .batch import ExtractBatcher
53log = logging.getLogger(__name__)
56class _ExtractedTable(Protocol):
57 """The table fields lilbee indexes as dedicated chunks.
59 Structural, so it is satisfied by both xberg's public ``Table`` and the native
60 type that ``ExtractedDocument.tables`` actually yields.
61 """
63 @property
64 def markdown(self) -> str: ...
66 @property
67 def page_number(self) -> int: ...
70def content_type_to_mode(content_type: str) -> ExtractMode:
71 """Map a content_type to the extraction mode (paginated for PDFs and images)."""
72 if content_type in (PDF_CONTENT_TYPE, IMAGE_CONTENT_TYPE):
73 return ExtractMode.PAGINATED
74 return ExtractMode.MARKDOWN
77def _page_text_record(source: str, page: int, text: str, content_type: str) -> PageTextRecord:
78 """Build one per-page text row for the export dataset."""
79 return PageTextRecord(source=source, page=page, text=text, content_type=content_type)
82_ocr_enable_override: contextvars.ContextVar[bool | None] = contextvars.ContextVar(
83 "lilbee_ocr_enable_override", default=None
84)
85_ocr_timeout_override: contextvars.ContextVar[float | None] = contextvars.ContextVar(
86 "lilbee_ocr_timeout_override", default=None
87)
90# Per-file title for embedding input; a ContextVar (like the OCR overrides) so the
91# call chains need no signature threading and concurrent ingests don't leak titles.
92_embed_title: contextvars.ContextVar[str] = contextvars.ContextVar("lilbee_embed_title", default="")
95@contextmanager
96def _title_scope(title: str) -> Generator[None, None, None]:
97 token = _embed_title.set(title or "")
98 try:
99 yield
100 finally:
101 _embed_title.reset(token)
104def _embed_inputs(texts: list[str], title: str | None = None) -> list[str]:
105 """Embedding inputs, title-prefixed when ``cfg.embed_titles`` is on.
107 Only the vector sees the title; the stored chunk text is unchanged.
108 ``None`` falls back to the scoped per-file title (the OCR chains).
109 """
110 effective = title if title is not None else _embed_title.get()
111 if not effective or not active_config().embed_titles:
112 return texts
113 return [f"{effective}\n{text}" for text in texts]
116# Contextual enrichment: characters of document head shown to the model, and
117# the reply budget for the one situating sentence.
118_ENRICH_HEAD_CHARS = 2000
119_ENRICH_CHUNK_CHARS = 2000
120_ENRICH_MAX_TOKENS = 60
121_ENRICH_PROMPT = (
122 "Document beginning:\n{head}\n\nChunk from the same document:\n{chunk}\n\n"
123 "Write one short sentence situating this chunk within the document, to "
124 "improve search retrieval of the chunk. Answer with only the sentence."
125)
128def _enrich_texts(texts: list[str], doc_head: str, source_name: str) -> list[str]:
129 """Embedding inputs with one LLM-written situating sentence per chunk.
131 Anthropic-style contextual retrieval, opt-in (``cfg.contextual_enrichment``):
132 one generation per chunk, so ingest slows accordingly. Only the vector sees
133 the sentence; stored chunk text and citations stay verbatim. Any failure
134 keeps that chunk's bare text.
135 """
136 if not active_config().contextual_enrichment or not texts:
137 return texts
138 from lilbee.retrieval.reasoning import strip_reasoning
140 provider = get_services().provider
141 head = doc_head[:_ENRICH_HEAD_CHARS]
142 enriched: list[str] = []
143 failed = 0
144 for text in texts:
145 prompt = _ENRICH_PROMPT.format(head=head, chunk=text[:_ENRICH_CHUNK_CHARS])
146 try:
147 response = provider.chat(
148 [{"role": "user", "content": prompt}],
149 stream=False,
150 options=aux_options(_ENRICH_MAX_TOKENS),
151 )
152 lines = strip_reasoning(response.text).strip().splitlines()
153 sentence = lines[0].strip() if lines else ""
154 except Exception:
155 failed += 1
156 sentence = ""
157 enriched.append(f"{sentence}\n{text}" if sentence else text)
158 if failed:
159 log.warning(
160 "Contextual enrichment failed for %d of %d chunks in %s; those embed bare",
161 failed,
162 len(texts),
163 source_name,
164 )
165 return enriched
168def _effective_enable_ocr() -> bool | None:
169 """``cfg.enable_ocr`` unless a per-request OCR override is active.
171 The override is a ContextVar, not a global cfg mutation, so concurrent
172 ingests on the shared HTTP daemon each see their own setting.
173 """
174 override = _ocr_enable_override.get()
175 return active_config().enable_ocr if override is None else override
178def _extraction_timeout_secs() -> int | None:
179 """``cfg.extraction_timeout`` as xberg's per-file cap; None when uncapped.
181 xberg defaults this to 600s. Passing lilbee's own value on every call keeps
182 the cap something a user can see and raise instead of an inherited default.
183 """
184 timeout = active_config().extraction_timeout
185 return timeout if timeout > 0 else None
188def _effective_ocr_timeout() -> float:
189 """``cfg.ocr_timeout`` unless a per-request OCR timeout override is active."""
190 override = _ocr_timeout_override.get()
191 return active_config().ocr_timeout if override is None else override
194@contextmanager
195def ocr_override(
196 enable_ocr: bool | None = None, ocr_timeout: float | None = None
197) -> Generator[None, None, None]:
198 """Scope per-request OCR settings without mutating the global cfg.
200 A ``None`` argument leaves that setting at its cfg default. Each override is
201 isolated to the entering context, so overlapping ingests never clobber one
202 another's OCR config.
203 """
204 tokens: list[tuple[contextvars.ContextVar[Any], contextvars.Token[Any]]] = []
205 try:
206 if enable_ocr is not None:
207 tokens.append((_ocr_enable_override, _ocr_enable_override.set(enable_ocr)))
208 if ocr_timeout is not None:
209 tokens.append((_ocr_timeout_override, _ocr_timeout_override.set(ocr_timeout)))
210 yield
211 finally:
212 for var, token in reversed(tokens):
213 var.reset(token)
216def _ocr_config(ocr_token: str | None) -> OcrConfig:
217 """Pick the OCR backend for this extraction.
219 Mirrors the prior fallback policy: OCR off when ``enable_ocr`` is False; lilbee's
220 vision backend when a vision model is configured; otherwise xberg's tesseract.
221 xberg auto-OCRs only the pages that lack a text layer.
222 """
223 from xberg import OcrConfig
225 config = active_config()
226 if _effective_enable_ocr() is False:
227 return OcrConfig(enabled=False)
228 if config.vision_model:
229 options = backend_options_for(ocr_token) if ocr_token else None
230 return OcrConfig(
231 backend=OcrBackendName.LILBEE_VISION,
232 backend_options=options,
233 )
234 # xberg requires a non-empty language list (4.x defaulted to English;
235 # xberg 1.0 errors on an empty one). cfg.ocr_language is validated non-empty.
236 return OcrConfig(
237 backend=OcrBackendName.TESSERACT,
238 language=list(config.ocr_language),
239 )
242def _ocr_force_requested() -> bool:
243 """Whether LILBEE_OCR_FORCE forces vision OCR on every page (targeted re-ingest lever)."""
244 import os
246 return os.environ.get("LILBEE_OCR_FORCE", "").strip().lower() in {"1", "true", "yes"}
249# Header/footer band stripped when layout detection is on: outermost 5%.
250_TOP_MARGIN_FRACTION = 0.05
251_BOTTOM_MARGIN_FRACTION = 0.05
254def _pdf_options() -> PdfConfig | None:
255 """PdfConfig for the enabled opt-in features (tables, layout), or None when all off."""
256 config = active_config()
257 if not (config.table_extraction or config.layout_detection):
258 return None
259 from xberg import PdfConfig
261 kwargs: dict[str, Any] = {}
262 if config.table_extraction:
263 kwargs["extract_tables"] = True
264 if config.layout_detection:
265 kwargs.update(
266 reading_order=True,
267 top_margin_fraction=_TOP_MARGIN_FRACTION,
268 bottom_margin_fraction=_BOTTOM_MARGIN_FRACTION,
269 )
270 return PdfConfig(**kwargs)
273def warn_if_table_model_ignored() -> None:
274 """Warn when table extraction runs with layout_detection off: xberg only
275 applies table_model inside layout detection, so the model is silently ignored.
276 """
277 config = active_config()
278 if config.table_extraction and not config.layout_detection:
279 log.warning(
280 "table_model=%s is ignored while layout_detection is off: tables use "
281 "the native extractor, not the structure model. Enable layout_detection "
282 "to apply the table model.",
283 config.table_model.value,
284 )
287def _layout_config() -> LayoutDetectionConfig | None:
288 """AUTO-strategy layout config when enabled, else None."""
289 config = active_config()
290 if not config.layout_detection:
291 return None
292 from xberg import LayoutDetectionConfig, LayoutStrategy
294 return LayoutDetectionConfig(strategy=LayoutStrategy.AUTO, table_model=config.table_model)
297def extraction_config(mode: ExtractMode, *, ocr_token: str | None = None) -> ExtractionConfig:
298 """Build ExtractionConfig for the given extraction mode."""
299 from xberg import ExtractionConfig, PageConfig
301 # Files are extracted one per call; xberg parallelizes OCR across a document's
302 # pages internally, and cross-file concurrency is the pipeline's semaphore.
303 chunking = build_chunking_config()
304 ocr = _ocr_config(ocr_token)
305 # Defeats xberg's text-layer short-circuit; vision path only (GPU re-OCR lever).
306 force_ocr = _ocr_force_requested() and ocr.backend == OcrBackendName.LILBEE_VISION
307 if mode is ExtractMode.PAGINATED:
308 paginated = ExtractionConfig(
309 chunking=chunking,
310 pages=PageConfig(extract_pages=True, insert_page_markers=False),
311 ocr=ocr,
312 force_ocr=force_ocr,
313 pdf_options=_pdf_options(),
314 extraction_timeout_secs=_extraction_timeout_secs(),
315 )
316 # The layout fields keep xberg's defaults when layout detection is off.
317 layout = _layout_config()
318 if layout is None:
319 return paginated
320 return replace(paginated, layout=layout, use_layout_for_markdown=True)
321 return ExtractionConfig(
322 chunking=chunking,
323 output_format=MARKDOWN_OUTPUT,
324 ocr=ocr,
325 force_ocr=force_ocr,
326 extraction_timeout_secs=_extraction_timeout_secs(),
327 )
330def make_extract_batcher() -> ExtractBatcher | None:
331 """The extraction batcher for this ingest run, or None when batching is off."""
332 config = active_config()
333 if not config.batch_extraction:
334 return None
335 from .batch import ExtractBatcher
336 from .xberg import aextract_batch
338 return ExtractBatcher(
339 size=config.batch_extraction_size,
340 config_fn=extraction_config,
341 ocr_fn=_ocr_config,
342 batch_fn=aextract_batch,
343 )
346@asynccontextmanager
347async def extract_batching() -> AsyncGenerator[None]:
348 """Activate extraction batching for the enclosed ingest, when the toggle is on.
350 The batcher is set before the block runs so the ingest tasks created inside it
351 inherit it in their copied context; off (the default) is a no-op.
352 """
353 batcher = make_extract_batcher()
354 if batcher is None:
355 yield
356 return
357 from .batch import reset_active_batcher, set_active_batcher
359 token = set_active_batcher(batcher)
360 try:
361 yield
362 finally:
363 await batcher.close()
364 reset_active_batcher(token)
367def _chunk_pages(page_texts: Sequence[tuple[int, str]]) -> list[tuple[int, str]]:
368 """Chunk each page's text. Semantic chunking is off: a single page rarely spans
369 multiple topics, so the semantic round-trip is not worth it."""
370 return [
371 (page_num, chunk)
372 for page_num, text in page_texts
373 for chunk in chunk_text(text, use_semantic=False)
374 ]
377async def chunk_and_embed_pages(
378 page_texts: Sequence[tuple[int, str]],
379 source_name: str,
380 content_type: str,
381 on_progress: DetailedProgressCallback,
382) -> list[ChunkRecord]:
383 """Chunk per-page text and embed every chunk. Used by the dataset import path."""
384 if not page_texts:
385 return []
387 # chunk_text runs xberg's synchronous extractor; offload it so a long
388 # document does not stall sibling files sharing this event loop.
389 all_chunks = await to_ingest_thread(_chunk_pages, page_texts)
390 if not all_chunks:
391 return []
392 texts = [c for _, c in all_chunks]
393 embed_texts = await to_ingest_thread(_enrich_texts, texts, page_texts[0][1], source_name)
394 vectors = await to_ingest_thread(
395 get_services().embedder.embed_batch,
396 _embed_inputs(embed_texts),
397 source=source_name,
398 on_progress=on_progress,
399 )
400 return [
401 ChunkRecord(
402 source=source_name,
403 content_type=content_type,
404 chunk_type=ChunkType.RAW,
405 page_start=page_num,
406 page_end=page_num,
407 line_start=0,
408 line_end=0,
409 chunk=text,
410 chunk_index=i,
411 vector=vec,
412 )
413 for i, ((page_num, text), vec) in enumerate(zip(all_chunks, vectors, strict=True))
414 ]
417def _capture_result_page_texts(
418 doc: ExtractedDocument,
419 source_name: str,
420 content_type: str,
421 page_texts_out: list[PageTextRecord] | None,
422) -> None:
423 """Append an extraction's page texts to the export accumulator.
425 Paginated documents yield one row per ``doc.pages`` entry; others have no
426 page split, so the full ``doc.content`` is recorded as page 0.
427 """
428 if page_texts_out is None:
429 return
430 if doc.pages:
431 page_texts_out.extend(
432 _page_text_record(source_name, page.page_number, page.content, content_type)
433 for page in doc.pages
434 )
435 elif doc.content.strip():
436 page_texts_out.append(_page_text_record(source_name, 0, doc.content, content_type))
439def _document_tables(doc: ExtractedDocument) -> list[_ExtractedTable]:
440 """The result tables to index as dedicated chunks, when table extraction is on.
442 Each table becomes its own chunk carrying xberg's markdown serialization and
443 page metadata. The table's flattened text also stays inside the content
444 chunks: stripping it there would tear holes in reading-order prose and the
445 page-text export, so both are indexed deliberately. The dedicated table
446 chunk adds the structured serialization for targeted retrieval.
447 """
448 if not active_config().table_extraction:
449 return []
450 return [t for t in (doc.tables or []) if t.markdown and t.markdown.strip()]
453def _warn_empty_ocr(source_name: str, media: str) -> None:
454 """Warn that extraction yielded no text and point to the vision-model remedy."""
455 log.warning(
456 "Skipped %s: text extraction produced no usable text. "
457 "For better results on %s, configure a vision model "
458 "via PUT /api/models/vision or set LILBEE_ENABLE_OCR=true.",
459 source_name,
460 media,
461 )
464async def ingest_document(
465 path: Path,
466 source_name: str,
467 content_type: str,
468 *,
469 quiet: bool = False,
470 on_progress: DetailedProgressCallback = noop_callback,
471 page_texts_out: list[PageTextRecord] | None = None,
472) -> tuple[list[ChunkRecord], SourceMeta]:
473 """Extract, chunk, and embed a document in a single xberg pass, with its metadata.
475 xberg extracts native text and, where a page has none, OCRs it through the
476 registered backend (lilbee's vision model, or tesseract). Per-page OCR progress
477 is streamed as a running count via ``ocr_request``. ``quiet`` is accepted for
478 pipeline call compatibility. The returned metadata carries the document's
479 extraction title/authors/date and is derived even when extraction yields nothing.
480 """
481 del quiet
482 doc = await _extract_document(
483 path, source_name, content_type, content_type_to_mode(content_type), on_progress
484 )
485 return await _records_from_document(
486 doc, source_name, content_type, on_progress=on_progress, page_texts_out=page_texts_out
487 )
490async def ingest_archive(
491 path: Path,
492 source_name: str,
493 content_type: str,
494 *,
495 on_progress: DetailedProgressCallback = noop_callback,
496) -> list[MemberRecords]:
497 """Extract an archive once and build records for every member, nested archives included.
499 The archive itself contributes no chunks. Each member is its own source named
500 ``<archive>/<member path>``. xberg unpacks to ``max_archive_depth`` under its
501 zip-bomb limits, so depth and size are enforced before this runs.
502 """
503 doc = await _extract_document(
504 path, source_name, content_type, ExtractMode.PAGINATED, on_progress
505 )
506 members: list[MemberRecords] = []
507 await _collect_members(doc, source_name, members, on_progress)
508 return members
511async def _collect_members(
512 doc: ExtractedDocument,
513 prefix: str,
514 members: list[MemberRecords],
515 on_progress: DetailedProgressCallback,
516) -> None:
517 from lilbee.data.ingest.discovery import archive_content_types, member_content_type
519 for entry in doc.children or []:
520 name = f"{prefix}/{entry.path}"
521 content_type = member_content_type(entry.path, entry.mime_type)
522 if content_type in archive_content_types():
523 await _collect_members(entry.result, name, members, on_progress)
524 continue
525 page_texts: list[PageTextRecord] = []
526 try:
527 records, meta = await _records_from_document(
528 entry.result, name, content_type, on_progress=on_progress, page_texts_out=page_texts
529 )
530 except ChunkLimitError as exc:
531 raise ChunkLimitError(exc.count, exc.limit, member=name) from None
532 members.append(MemberRecords(name, content_type, records, page_texts, meta))
535async def _extract_document(
536 path: Path,
537 source_name: str,
538 content_type: str,
539 mode: ExtractMode,
540 on_progress: DetailedProgressCallback,
541) -> ExtractedDocument:
542 """Run one xberg pass over *path*, with per-page OCR progress and the extraction trace."""
543 from .xberg import aextract_document
545 page_seen = 0
547 def _tick() -> None:
548 nonlocal page_seen
549 page_seen += 1
550 on_progress(
551 EventType.EXTRACT,
552 ExtractEvent(file=source_name, page=page_seen, total_pages=0),
553 )
555 trace_log.debug("extract-start source=%r type=%s", source_name, content_type)
556 started = time.perf_counter()
557 with ocr_request(on_page=_tick, timeout=_effective_ocr_timeout()) as token:
558 batcher = active_extract_batcher()
559 if batcher is not None:
560 doc = await batcher.submit(mode, path.read_bytes(), path.name, token)
561 else:
562 config = extraction_config(mode, ocr_token=token)
563 # xberg's extract is async; awaiting it keeps the OCR page loop off this thread.
564 doc = await aextract_document(path.read_bytes(), filename=path.name, config=config)
565 elapsed = time.perf_counter() - started
567 # One trace line per extraction (filename, timing, counts, OCR pages), plus a
568 # vision line for scanned files. Emitted for empty results too (a slow file
569 # that yields nothing is worth surfacing).
570 trace_extraction(
571 ExtractionTrace(
572 source=source_name,
573 content_type=content_type,
574 elapsed_s=elapsed,
575 page_count=len(doc.pages or []) or len(doc.chunks or []),
576 chunk_count=len(doc.chunks or []),
577 ocr_pages=page_seen,
578 vision_configured=bool(active_config().vision_model),
579 )
580 )
582 return doc
585async def _records_from_document(
586 doc: ExtractedDocument,
587 source_name: str,
588 content_type: str,
589 *,
590 on_progress: DetailedProgressCallback,
591 page_texts_out: list[PageTextRecord] | None,
592) -> tuple[list[ChunkRecord], SourceMeta]:
593 """Chunk-cap, page-capture, and embed one extracted document into its records."""
594 # Derived before the empty-result return so a scan's title/authors survive zero chunks.
595 meta = source_meta_from_extraction(doc.metadata, source_name)
597 tables = _document_tables(doc)
598 if not doc.chunks and not tables:
599 if content_type in (PDF_CONTENT_TYPE, IMAGE_CONTENT_TYPE):
600 _warn_empty_ocr(source_name, "scanned documents")
601 return [], meta
603 enforce_chunk_limit(len(doc.chunks or []) + len(tables))
604 _capture_result_page_texts(doc, source_name, content_type, page_texts_out)
606 # One EXTRACT event per file so progress subscribers show "extracted N pages"
607 # before embedding; result.pages, or the chunk count for non-paginated docs.
608 page_count = len(doc.pages or []) or len(doc.chunks or [])
609 on_progress(
610 EventType.EXTRACT,
611 ExtractEvent(file=source_name, page=page_count, total_pages=page_count),
612 )
614 # Content chunks and table serializations share one embed batch; the vector
615 # list is split back apart below by position.
616 texts = [chunk.content for chunk in doc.chunks or []]
617 table_texts = [table.markdown for table in tables]
618 embed_texts = await to_ingest_thread(
619 _enrich_texts, texts + table_texts, texts[0] if texts else "", source_name
620 )
621 vectors = await to_ingest_thread(
622 get_services().embedder.embed_batch,
623 _embed_inputs(embed_texts, meta.title),
624 source=source_name,
625 on_progress=on_progress,
626 )
627 records = [
628 ChunkRecord(
629 source=source_name,
630 content_type=content_type,
631 chunk_type=ChunkType.RAW,
632 page_start=chunk.metadata.first_page or 0,
633 page_end=chunk.metadata.last_page or 0,
634 line_start=0,
635 line_end=0,
636 chunk=text,
637 chunk_index=chunk.metadata.chunk_index,
638 vector=vec,
639 )
640 for chunk, text, vec in zip(doc.chunks or [], texts, vectors[: len(texts)], strict=True)
641 ]
642 # Table chunk indices continue after the content chunks so a source's
643 # (source, chunk_index) pairs stay unique.
644 records.extend(
645 ChunkRecord(
646 source=source_name,
647 content_type=content_type,
648 chunk_type=ChunkType.TABLE,
649 page_start=table.page_number,
650 page_end=table.page_number,
651 line_start=0,
652 line_end=0,
653 chunk=text,
654 chunk_index=len(texts) + i,
655 vector=vec,
656 )
657 for i, (table, text, vec) in enumerate(
658 zip(tables, table_texts, vectors[len(texts) :], strict=True)
659 )
660 )
661 return records, meta
664def _markdown_h1(text: str) -> str | None:
665 """The document's leading ``# Heading``, the best title a note carries.
667 Only a top-level ATX heading counts; ``##`` and deeper are sections, not the
668 document title. None when the note opens without one.
669 """
670 for line in text.splitlines():
671 stripped = line.strip()
672 if not stripped:
673 continue
674 if stripped.startswith("# "):
675 return stripped[2:].strip() or None
676 return None
677 return None
680async def ingest_markdown(
681 path: Path,
682 source_name: str,
683 on_progress: DetailedProgressCallback = noop_callback,
684 page_texts_out: list[PageTextRecord] | None = None,
685) -> tuple[list[ChunkRecord], SourceMeta]:
686 """Chunk a markdown file with heading context prepended to each chunk.
688 Each chunk gets the heading hierarchy path (e.g. "# Setup > ## Install")
689 prepended for better retrieval context. When ``page_texts_out`` is given,
690 the full text is appended as page 0 for export. The returned metadata's
691 title is the note's leading ``# Heading`` when it has one, else the stem.
692 """
693 raw_text = await to_ingest_thread(path.read_text, encoding="utf-8", errors="replace")
694 meta = SourceMeta(title=derive_title(source_name, _markdown_h1(raw_text)))
695 if not raw_text.strip():
696 return [], meta
698 # chunk_text runs xberg's synchronous extractor; offload it so a large
699 # markdown doc does not stall sibling files sharing this event loop.
700 texts = await to_ingest_thread(
701 chunk_text, raw_text, mime_type="text/markdown", heading_context=True
702 )
703 if not texts:
704 return [], meta
706 enforce_chunk_limit(len(texts))
707 if page_texts_out is not None:
708 page_texts_out.append(_page_text_record(source_name, 0, raw_text, "text"))
710 embed_texts = await to_ingest_thread(_enrich_texts, texts, raw_text, source_name)
711 vectors = await to_ingest_thread(
712 get_services().embedder.embed_batch,
713 _embed_inputs(embed_texts, meta.title),
714 source=source_name,
715 on_progress=on_progress,
716 )
717 records = [
718 ChunkRecord(
719 source=source_name,
720 content_type="text",
721 chunk_type=ChunkType.RAW,
722 page_start=0,
723 page_end=0,
724 line_start=0,
725 line_end=0,
726 chunk=t,
727 chunk_index=idx,
728 vector=vec,
729 )
730 for idx, (t, vec) in enumerate(zip(texts, vectors, strict=True))
731 ]
732 return records, meta