spb/uqo-research-db

1# ============================================================2# Project: EDGAR Research Database3# Description: SEC EDGAR Research Infrastructure4#5# Authors:6# - Manel Kammoun — manel.kammoun@uqo.ca7# - Djerrycharli Tandjambianda — djerrycharli.tandjambianda@uqo.ca8# - Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca9#10# Institution:11# Université du Québec en Outaouais (UQO)12#13# License:14# Research / Academic Use15# ============================================================1617"""Word-window chunker (re-created 2026-10-27; the original module was lost).1819Produces the same record layout as the existing ``filing_chunks`` table:20``chunk_id`` (uuid4), ``section_id``, ``chunk_text``, ``chunk_order``,21``word_count``, ``char_count``, ``start_char``, ``end_char``.22Chunks are ``chunk_size`` words long with ``overlap`` words shared between23consecutive chunks; a trailing chunk shorter than ``min_chunk_size`` words is24merged into the previous one.25"""2627from __future__ import annotations2829import uuid303132class TextChunker:33    def __init__(34        self,35        chunk_size: int = 512,36        overlap: int = 64,37        min_chunk_size: int = 50,38    ) -> None:39        if overlap >= chunk_size:40            raise ValueError("overlap must be smaller than chunk_size")41        self.chunk_size = chunk_size42        self.overlap = overlap43        self.min_chunk_size = min_chunk_size4445    def chunk_text(self, text: str, section_id: str | None = None) -> list[dict]:46        words = text.split()47        if not words:48            return []4950        step = self.chunk_size - self.overlap51        windows: list[list[str]] = []52        start = 053        while start < len(words):54            window = words[start:start + self.chunk_size]55            if windows and len(window) < self.min_chunk_size:56                # fold a tiny tail into the previous window57                tail = window[self.overlap:] if len(window) > self.overlap else []58                windows[-1].extend(tail)59                break60            windows.append(window)61            if start + self.chunk_size >= len(words):62                break63            start += step6465        chunks: list[dict] = []66        cursor = 067        for order, window in enumerate(windows):68            chunk_text = " ".join(window)69            chunks.append({70                "chunk_id": str(uuid.uuid4()),71                "section_id": section_id,72                "chunk_text": chunk_text,73                "chunk_order": order,74                "word_count": len(window),75                "char_count": len(chunk_text),76                "start_char": cursor,77                "end_char": cursor + len(chunk_text),78            })79            cursor += len(chunk_text) + 180        return chunks8182    def chunk_section(self, section: dict) -> list[dict]:83        text = section.get("text") or ""84        return self.chunk_text(text, section_id=section.get("section_id"))85