""" Оркестратор сканирования: связывает поиск файлов, чтение содержимого и текстовый поиск воедино, ведёт статистику и журнал ошибок. Улучшение по сравнению с оригинальным ps1 (который обрабатывал файлы строго последовательно, один за другим): чтение и разбор файлов распараллелены через `ThreadPoolExecutor`, поскольку узким местом является дисковый I/O и C-расширения (PyMuPDF, zipfile, olefile), которые освобождают GIL — это даёт кратный прирост скорости на больших деревьях каталогов. Аудит v2: * Добавлен таймаут на один файл (через `future.result(timeout=...)`) — защита от зависания на повреждённых/очень больших PDF или DOC, которые могут "повесить" PyMuPDF/olefile. Файл, превысивший таймаут, попадает в журнал как ошибка, но не блокирует остальное сканирование. * Число потоков и таймаут берутся из AppSettings (диалог "Параметры"). * После завершения сканирования освобождаются COM-объекты Word (если использовался метод COM) — закрывает утечку ресурсов. """ from __future__ import annotations import time from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeoutError, as_completed from datetime import datetime from pathlib import Path from threading import Event from typing import Callable, Optional from app.config import ErrorEntry, ScanReport, ScanSettings, ScanStats, SearchResult from app.readers import extract_text from app.readers.doc_reader import _release_word_all from app.scanning.file_finder import iter_files_safe from app.scanning.search_engine import find_matches from app.settings_store import load_settings ProgressCallback = Callable[[int, int, str], None] ResultCallback = Callable[[SearchResult], None] LogCallback = Callable[[str, str], None] # level, message _EXT_STAT_FIELD = { ".docx": "docx_count", ".doc": "doc_count", ".txt": "txt_count", ".pdf": "pdf_count", } class DocumentScanner: """Выполняет полный цикл: поиск -> чтение -> поиск слов -> отчёт.""" def run( self, settings: ScanSettings, cancel_event: Optional[Event] = None, on_progress: Optional[ProgressCallback] = None, on_result: Optional[ResultCallback] = None, on_log: Optional[LogCallback] = None, ) -> ScanReport: start = time.monotonic() app_settings = load_settings() stats = ScanStats() results: list[SearchResult] = [] errors: list[ErrorEntry] = [] cancel_event = cancel_event or Event() def log(level: str, message: str, file_path: str = "") -> None: if level in ("WARNING", "ERROR"): errors.append(ErrorEntry( file_path=file_path, level=level, message=message, timestamp=datetime.now().strftime("%Y-%m-%d %H:%M:%S"), )) if on_log: on_log(level, message) def on_skip(path: str, reason: str) -> None: if reason == "temp": stats.temp_files += 1 stats.skipped += 1 entries = list( iter_files_safe( settings.paths, settings.file_types, settings.max_file_size_bytes, cancel_event=cancel_event, on_skip=on_skip, ) ) total = len(entries) if on_progress: on_progress(0, total, "Найдено файлов: %d" % total) if total == 0: return ScanReport( results=[], errors=errors, stats=stats, elapsed_seconds=time.monotonic() - start, doc_read_method=settings.doc_read_method.value, ) # Число потоков: берём значение из ScanSettings (которое уже # подставлено из AppSettings в main_window._start_scan, если было 0). # Ограничиваем сверху количеством файлов, чтобы не создавать лишние # потоки для маленьких деревьев, и снизу — единицей. max_workers = min(settings.max_workers or app_settings.max_workers, max(1, total)) max_workers = max(1, max_workers) per_file_timeout = app_settings.per_file_timeout processed_count = 0 pool_cancelled = False pool = ThreadPoolExecutor(max_workers=max(1, max_workers)) try: future_map = { pool.submit(self._process_one, entry, settings): entry for entry in entries } for future in as_completed(future_map): if cancel_event.is_set(): for f in future_map: f.cancel() # shutdown(cancel_futures=True) — Python 3.9+, # экстренно останавливает невыполненные задачи pool.shutdown(wait=False, cancel_futures=True) pool_cancelled = True break entry = future_map[future] processed_count += 1 try: file_results, warn, err, locked = future.result(timeout=per_file_timeout) except FuturesTimeoutError: err = f"Таймаут чтения файла ({per_file_timeout} с)" log("WARNING", f"{entry.path.name}: {err}", str(entry.path)) file_results, warn, locked = [], None, False except Exception as exc: err = str(exc) log("ERROR", f"{entry.path.name}: {err}", str(entry.path)) file_results, warn, locked = [], None, False stats.processed += 1 field = _EXT_STAT_FIELD.get(entry.extension) if field: setattr(stats, field, getattr(stats, field) + 1) if locked: stats.locked += 1 elif err: stats.errors += 1 if "Таймаут" not in (err or ""): log("ERROR", f"{entry.path.name}: {err}", str(entry.path)) for res in file_results: results.append(res) stats.found += 1 if on_result: on_result(res) if on_progress: on_progress(processed_count, total, entry.path.name) finally: # При выходе (нормальном или по cancel) — гарантированно завершаем # пул, чтобы потоки не висели до сборки мусора if not pool_cancelled: pool.shutdown(wait=False) # Освобождаем COM-ресурсы Word во всех потоках, где они создавались try: _release_word_all() except Exception: pass return ScanReport( results=results, errors=errors, stats=stats, elapsed_seconds=time.monotonic() - start, doc_read_method=settings.doc_read_method.value, cancelled=cancel_event.is_set(), ) # ------------------------------------------------------------------ # @staticmethod def _process_one(entry, settings: ScanSettings): outcome = extract_text(entry.path, entry.extension, settings) if outcome.locked: return [], None, None, True if outcome.error: return [], None, outcome.error, False if not outcome.text: return [], outcome.warning, None, False matches = find_matches( outcome.text, settings.words, case_sensitive=settings.case_sensitive, whole_word=settings.whole_word, context_chars=settings.context_chars, ) modified_str = datetime.fromtimestamp(entry.modified).strftime("%Y-%m-%d %H:%M:%S") results = [ SearchResult( file_name=entry.path.name, full_path=str(entry.path), word=m.word, context=m.context, file_type=entry.extension.lstrip("."), modified=modified_str, occurrence_index=i + 1, matched_text=m.matched_text, ) for i, m in enumerate(matches) ] return results, outcome.warning, None, False