Spaces:
Sleeping
Sleeping
| """Metadata helpers for indexed chunks. | |
| The chunk metadata is what makes retrieval *and* citations work later. We | |
| keep the schema small and stable: | |
| { | |
| "doc_id": sha256(file_bytes)[:16] # stable across re-ingests | |
| "source": absolute path of the original file | |
| "filename": basename of the original file | |
| "header_path": "h1 > h2 > h3" (empty for headerless content) | |
| "chunk_index": int — 0-based position in the document | |
| "total_chunks": int — total chunks generated for this doc | |
| "ingested_at": ISO-8601 UTC timestamp | |
| "parser": "docling" | "qwen3-vl" | "passthrough" | |
| "pages": int | None (only set for PDFs) | |
| "extension": ".pdf" | ".docx" | ... (debugging convenience) | |
| } | |
| Anything else can live in a sibling DB later. We deliberately do NOT add | |
| ``access_level``, ``department``, ``tags`` until a feature actually needs them. | |
| """ | |
| from __future__ import annotations | |
| import hashlib | |
| import uuid | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| from typing import Any | |
| # Deterministic UUID namespace used for chunk IDs. uuid5(NAMESPACE, key) | |
| # always returns the same UUID for the same key, which means re-ingesting a | |
| # document upserts (replaces) existing points instead of duplicating them. | |
| CHUNK_ID_NAMESPACE = uuid.UUID("c1a8f6c0-5d2a-4d4c-8a4f-9d8d3e9b1a01") | |
| def compute_doc_id(file_path: str | Path) -> str: | |
| """Stable 16-char hex id derived from the file's content hash.""" | |
| path = Path(file_path) | |
| h = hashlib.sha256() | |
| with path.open("rb") as f: | |
| for block in iter(lambda: f.read(1 << 20), b""): | |
| h.update(block) | |
| return h.hexdigest()[:16] | |
| def chunk_uuid(doc_id: str, chunk_index: int) -> str: | |
| """Deterministic UUID for a (doc_id, chunk_index) pair.""" | |
| return str(uuid.uuid5(CHUNK_ID_NAMESPACE, f"{doc_id}-{chunk_index}")) | |
| def build_chunk_metadata( | |
| *, | |
| doc_id: str, | |
| source_path: str | Path, | |
| parser: str, | |
| pages: int | None = None, | |
| extra: dict[str, Any] | None = None, | |
| ) -> dict[str, Any]: | |
| """Build the per-document portion of metadata. Per-chunk fields | |
| (``chunk_index``, ``total_chunks``, ``header_path``) are added by the | |
| chunker after it knows how many chunks were produced. | |
| """ | |
| path = Path(source_path) | |
| base: dict[str, Any] = { | |
| "doc_id": doc_id, | |
| "source": str(path.absolute()), | |
| "filename": path.name, | |
| "extension": path.suffix.lower(), | |
| "parser": parser, | |
| "pages": pages, | |
| "ingested_at": datetime.now(timezone.utc).isoformat(timespec="seconds"), | |
| } | |
| if extra: | |
| base.update(extra) | |
| return base | |