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)