Method initialize_queue

ocrflux/work_queue.py:251–278

Load the work queue from the local index file and initialize it for processing. Removes already completed work items and randomizes the order.

(self)

            await asyncio.to_thread(upload_zstd_csv_local, self._index_path, combined_lines)

    async def initialize_queue(self) -> int:
        """
        Load the work queue from the local index file and initialize it for processing.
        Removes already completed work items and randomizes the order.
        """
        # 1) Read the index
        work_queue_lines = await asyncio.to_thread(download_zstd_csv_local, self._index_path)
        work_queue = {parts[0]: parts[1:] for line in work_queue_lines if (parts := line.strip().split(",")) and line.strip()}

        # 2) Determine which items are completed by scanning local results/*.jsonl
        if not os.path.isdir(self._results_dir):
            os.makedirs(self._results_dir, exist_ok=True)
        done_work_items = [f for f in os.listdir(self._results_dir) if f.startswith("output_") and f.endswith(".jsonl")]
        done_work_hashes = {fn[len("output_") : -len(".jsonl")] for fn in done_work_items}

        # 3) Filter out completed items
        remaining_work_hashes = set(work_queue) - done_work_hashes
        remaining_items = [WorkItem(hash=hash_, work_paths=work_queue[hash_]) for hash_ in remaining_work_hashes]
        random.shuffle(remaining_items)

        # 4) Initialize our in-memory queue
        self._queue = asyncio.Queue()
        for item in remaining_items:
            await self._queue.put(item)

        logger.info(f"Initialized local queue with {self._queue.qsize()} work items")

        return self._queue.qsize()

    async def is_completed(self, work_hash: str) -> bool:
        """