breakpointsoftware/document-parser

1from __future__ import annotations2 3import shutil4from dataclasses import dataclass5from pathlib import Path6 7from dotenv import load_dotenv8from firebase_processed_files import FirebaseProcessedFilesTracker, check_files_already_processed9from google_drive_service import (10    DriveDocument,11    GoogleDriveConfigError,12    load_drive_folder_ids_from_env,13    scan_drive_supported_documents,14)15 16 17load_dotenv()18 19 20@dataclass21class DriveScanResult:22    files_to_process: list[str]23    skipped_files: list[tuple[str, str]]24    discovered_count: int25    temp_dir: str26    warning: str | None = None27 28 29def scan_google_drive_unprocessed_files(30    tracker: FirebaseProcessedFilesTracker | None,31    include_subfolders: bool = True,32) -> DriveScanResult:33    folder_ids = load_drive_folder_ids_from_env()34    scan_result = scan_drive_supported_documents(folder_ids=folder_ids, include_subfolders=include_subfolders)35 36    downloaded_paths = [document.local_path for document in scan_result.documents]37 38    files_to_process, skipped_files, warning = check_files_already_processed(tracker, downloaded_paths)39 40    skipped_set = set(path for path, _ in skipped_files)41    for downloaded in downloaded_paths:42        path = Path(downloaded)43        if path.name in skipped_set:44            try:45                path.unlink(missing_ok=True)46            except Exception:47                pass48 49    return DriveScanResult(50        files_to_process=files_to_process,51        skipped_files=skipped_files,52        discovered_count=scan_result.discovered_count,53        temp_dir=scan_result.temp_dir,54        warning=warning,55    )56 57 58def scan_google_drive_supported_documents(include_subfolders: bool = True) -> tuple[list[DriveDocument], str]:59    """Scans configured Drive folders and downloads supported documents with metadata."""60    folder_ids = load_drive_folder_ids_from_env()61    scan_result = scan_drive_supported_documents(folder_ids=folder_ids, include_subfolders=include_subfolders)62    return scan_result.documents, scan_result.temp_dir63 64 65def cleanup_drive_scan_result(scan_result: DriveScanResult | None) -> None:66    if scan_result is None:67        return68 69    if scan_result.temp_dir:70        shutil.rmtree(scan_result.temp_dir, ignore_errors=True)