chunking/validators.py

import json

from src.chunking.models import Chunk


def validate_chunks(chunks: list[Chunk]) -> None:
    for expected_index, chunk in enumerate(chunks, start=1):
        if not chunk.chunk_id:
            raise ValueError("chunk_id is required")
        if not chunk.doc_id:
            raise ValueError("doc_id is required")
        if not chunk.source_file:
            raise ValueError("source_file is required")
        if chunk.chunk_index != expected_index:
            raise ValueError("chunk_index order is not preserved")
        expected_suffix = f"::chunk_{expected_index:04d}"
        if not (
            chunk.chunk_id == f"{chunk.doc_id}{expected_suffix}"
            or (
                chunk.chunk_id.startswith(f"{chunk.doc_id}::")
                and chunk.chunk_id.endswith(expected_suffix)
            )
        ):
            raise ValueError(
                "chunk_id must be deterministic and scoped to the document or parent"
            )
        if not chunk.content.strip():
            raise ValueError("content is required")
        if not chunk.content_for_embedding.strip():
            raise ValueError("content_for_embedding is required")
        if chunk.metadata.get("token_count", 0) <= 0:
            raise ValueError("token_count must be greater than 0")


def validate_jsonl(path: str) -> None:
    with open(path, encoding="utf-8") as file:
        for line in file:
            if line.strip():
                json.loads(line)