"""Горячие папки ИС Сканирование: протокол `HotFolder` и реализация в обычном каталоге (D-1, R-4, R-5). Файл кладётся во входящий каталог как `_`: сначала временный `.<имя>.part` в служебном каталоге `Tmp` (на той же ФС/шаре), затем атомарный rename во входящий (OQ-79). Ответ ИС — набор файлов каталога `//`: ИС кладёт туда все распознанные документы задания, по одному `*.json` на документ (OQ-78). `poll` отдаёт все `*.json` по возрастанию имени, когда набор готов: есть хотя бы один `*.json`, а набор имён и размеров не менялся `settle_seconds` (`SCAN_SETTLE_SECONDS`, `SettleTracker`). Имена каталогов (`Input`/`Output`/`Tmp`) настраиваются, а не зашиты в код. """ import contextlib import os import threading import time from collections import OrderedDict from collections.abc import Callable from dataclasses import dataclass from pathlib import Path from typing import Protocol _PATH_SEPARATORS = ("/", "\\") # Недопустимые в именах файлов Windows/SMB символы (кроме разделителей пути) и управляющие 0x00-0x1F (с NUL) _ILLEGAL_CHARS = frozenset(':<>"|?*') | frozenset(chr(code) for code in range(0x20)) _FORBIDDEN_CHARS = _ILLEGAL_CHARS | frozenset(_PATH_SEPARATORS) # Предел длины временного имени `._<имя>.part` в байтах UTF-8: ext4 — 255 байт, SMB/NTFS — 255 единиц UTF-16 # (их не больше, чем байт UTF-8); 15 байт — запас MAX_PART_NAME_BYTES = 240 # Суффикс после последней точки считается расширением и сохраняется при укорачивании, если он не длиннее этого _MAX_EXTENSION_CHARS = 16 def validate_name(name: str) -> None: """Проверка имени входящего файла для `put` (общая для `LocalHotFolder` и `SmbHotFolder`). Имя — один компонент пути: не пустое, не начинается с `.` (dot-файлы — служебные `.part`, их игнорирует эмулятор), без разделителей `/` и `\\`, без символов, недопустимых в Windows/SMB (`: < > " | ? *`), и без управляющих символов. Пробелы и кириллица допустимы. Нарушение — `ValueError`. """ if name == "" or name.startswith(".") or any(char in _FORBIDDEN_CHARS for char in name): raise ValueError(f"недопустимое имя файла: {name!r}") def validate_job_id(job_id: str) -> None: """Проверка `job_id` перед построением пути ответа: защита от обхода каталога. UUID проходит; пусто, `.`, `..`, разделители, недопустимые в Windows/SMB и управляющие символы — `ValueError`.""" if job_id in ("", ".", "..") or any(char in _FORBIDDEN_CHARS for char in job_id): raise ValueError(f"недопустимый job_id: {job_id!r}") def _split_basename(name: str) -> str: """Последний компонент пути с любыми разделителями (и POSIX, и Windows); пустая строка, если его нет.""" return name.replace("\\", "/").rsplit("/", 1)[-1] def inbox_name(job_id: str, original_name: str) -> str: """Имя входящего файла `_`. `original_name` приходит из недоверенного запроса («как загрузил пользователь», контракт требует только непустую строку): берётся последний компонент пути (без разделителей и `..`), символы, недопустимые в имени файла Windows/SMB, и управляющие символы заменяются на `_`, завершающие точки и пробелы (их отбрасывает Windows) убираются. Пробелы внутри имени и кириллица сохраняются. Результат всегда проходит `validate_name`, если `job_id` допустим. `ValueError` — только если имени не осталось (пусто, `.`, `..`, одни точки/пробелы). Длинное имя укорачивается (T7.3): основа обрезается по символам с конца, пока `.<результат>.part` не уложится в `MAX_PART_NAME_BYTES` байт UTF-8; расширение (суффикс после последней точки до `_MAX_EXTENSION_CHARS` символов) сохраняется. Короткое имя не меняется. """ base = _split_basename(original_name) if base in (".", ".."): raise ValueError(f"недопустимое имя файла: {original_name!r}") base = "".join("_" if char in _ILLEGAL_CHARS else char for char in base).rstrip(". ") if not base: raise ValueError(f"недопустимое имя файла: {original_name!r}") return f"{job_id}_{_fit_name(job_id, base)}" def _utf8_len(text: str) -> int: # surrogatepass: одиночный суррогат (схема запроса их не пропускает) не роняет подсчёт длины return len(text.encode("utf-8", "surrogatepass")) def _fit_name(job_id: str, base: str) -> str: """Основа имени, при которой `._<основа>.part` занимает не больше `MAX_PART_NAME_BYTES` байт UTF-8.""" budget = MAX_PART_NAME_BYTES - _utf8_len(f".{job_id}_.part") if _utf8_len(base) <= budget: return base dot = base.rfind(".") extension = base[dot:] if 0 < dot and len(base) - dot - 1 <= _MAX_EXTENSION_CHARS else "" stem = base[: len(base) - len(extension)] limit = budget - _utf8_len(extension) used = 0 kept = 0 for char in stem: # по code points: многобайтный символ не разрезается used += _utf8_len(char) if used > limit: break kept += 1 stem = stem[:kept] if not extension: # без расширения имя заканчивается основой: завершающие точки и пробелы отбрасывает Windows stem = stem.rstrip(". ") return (stem or "_") + extension @dataclass(frozen=True) class AnswerFile: """Один файл ответа ИС: имя в каталоге `//` и байты как есть.""" name: str data: bytes # Снимок набора ответа: (имя, размер) всех `*.json` каталога задания по возрастанию имени Snapshot = tuple[tuple[str, int], ...] MonotonicFn = Callable[[], float] # Предел числа заданий, для которых помнится ненастоявшийся набор (задание ушло по таймауту — состояние вытесняется) MAX_PENDING_JOBS = 1024 def is_answer_name(name: str) -> bool: """Файл ответа ИС — `*.json`; расширение без учёта регистра (маски Windows/SMB регистронезависимы).""" return name.lower().endswith(".json") class SettleTracker: """Готовность набора ответа по стабилизации (settle): набор считается готовым, когда он не пуст и одинаковый снимок (имена и размеры) наблюдается не меньше `settle_seconds` по часам `clock`. Состояние — по `job_id`: снимок и момент, с которого он не менялся. Оно сбрасывается, когда набор признан готовым или опустел (каталога/json нет), а число заданий ограничено `max_jobs` (самые давние вытесняются: у задания, ушедшего по таймауту, состояние иначе осталось бы навсегда; вытесненное активное задание настаивается заново). `settle_seconds == 0` — готов любой непустой набор, состояние не хранится. Потокобезопасно: `poll` вызывается через `asyncio.to_thread`. """ def __init__( self, settle_seconds: float, clock: MonotonicFn = time.monotonic, *, max_jobs: int = MAX_PENDING_JOBS ) -> None: if settle_seconds < 0: raise ValueError(f"settle_seconds не может быть отрицательным: {settle_seconds!r}") self.settle_seconds = settle_seconds self._clock = clock self._max_jobs = max_jobs self._lock = threading.Lock() self._seen: OrderedDict[str, tuple[Snapshot, float]] = OrderedDict() def ready(self, job_id: str, snapshot: Snapshot) -> bool: """Набор `snapshot` задания готов. Готовый набор забывается: следующий вызов настаивает набор заново.""" if not snapshot or self.settle_seconds == 0: self.forget(job_id) return bool(snapshot) now = self._clock() with self._lock: seen = self._seen.get(job_id) if seen is not None and seen[0] == snapshot: if now - seen[1] >= self.settle_seconds: del self._seen[job_id] return True return False self._seen[job_id] = (snapshot, now) self._seen.move_to_end(job_id) while len(self._seen) > self._max_jobs: self._seen.popitem(last=False) return False def forget(self, job_id: str) -> None: with self._lock: self._seen.pop(job_id, None) def __len__(self) -> int: with self._lock: return len(self._seen) class HotFolder(Protocol): """Обмен файлами с ИС Сканирование через пару каталогов. Методы синхронные: файловые операции и SMB блокирующие, воркер (asyncio) вызывает их через `asyncio.to_thread`. """ def put(self, name: str, data: bytes) -> None: """Атомарно положить файл `name` (простое имя, без пути) во входящий каталог: временный файл — в служебном каталоге `Tmp`, затем rename. Имя собирает вызывающий (`inbox_name`). Если горячая папка недоступна — исключение (воркер превращает его в `RECOGNITION_FAILED`).""" ... def poll(self, job_id: str) -> list[AnswerFile] | None: """Все `*.json` каталога `//` по возрастанию имени, когда набор готов (не пуст и не менялся `SCAN_SETTLE_SECONDS`); `None`, пока ответа нет или набор ещё пишется. Вызывается повторно с тем же `job_id`: реализация помнит снимок набора между вызовами.""" ... def healthy(self) -> bool: """Горячая папка доступна. Не бросает исключений.""" ... class LocalHotFolder: """Горячая папка в обычном каталоге: dev, compose-том, unit-тесты. `root`, `input_dir`, `output_dir`, `tmp_dir`, `answer_dir` и `answer_path` открыты намеренно (R-4): эмулятор ИС Сканирование работает с ними обычными операциями файловой системы (листинг входящих, `.done`, запись ответа). `settle_seconds` и `clock` (монотонные часы) — готовность набора ответа (`SettleTracker`). """ def __init__( # noqa: PLR0913 — имена трёх каталогов + параметры готовности набора self, root: Path | str, input_dir: str = "Input", output_dir: str = "Output", tmp_dir: str = "Tmp", *, settle_seconds: float = 2, clock: MonotonicFn = time.monotonic, ) -> None: self.root: Path = Path(root) self.input_dir: Path = self.root / input_dir self.output_dir: Path = self.root / output_dir self.tmp_dir: Path = self.root / tmp_dir self._settle = SettleTracker(settle_seconds, clock) def answer_dir(self, job_id: str) -> Path: """Каталог ответа ИС: `/`.""" validate_job_id(job_id) return self.output_dir / job_id def answer_path(self, job_id: str) -> Path: """Путь основного файла ответа `//.json` (так его называет ИС и эмулятор).""" return self.answer_dir(job_id) / f"{job_id}.json" def put(self, name: str, data: bytes) -> None: """Атомарная запись: `..part` в служебном каталоге `tmp_dir`, затем `os.replace` в `input_dir/` (существующий файл заменяется; оба каталога на одной ФС — rename атомарен). Корень не создаётся (нет корня — `FileNotFoundError`), служебный и входящий каталоги создаются без родителей. При сбое временный файл удаляется (ошибка удаления не маскирует исходную). """ validate_name(name) self.tmp_dir.mkdir(exist_ok=True) self.input_dir.mkdir(exist_ok=True) part = self.tmp_dir / f".{name}.part" try: part.write_bytes(data) os.replace(part, self.input_dir / name) except BaseException: # ошибка удаления временного файла не должна маскировать исходную (T7.5) with contextlib.suppress(OSError): part.unlink(missing_ok=True) raise def poll(self, job_id: str) -> list[AnswerFile] | None: """Готовый набор `*.json` каталога ответа или `None` (нет корня/каталога/json-файлов, набор ещё настаивается, файл исчез или изменился между листингом и чтением, набор изменился за время чтения). Другие ошибки ФС поднимаются.""" folder = self.answer_dir(job_id) snapshot = self._snapshot(folder) if not self._settle.ready(job_id, snapshot): return None files: list[AnswerFile] = [] for name, size in snapshot: try: data = (folder / name).read_bytes() except FileNotFoundError: return None if len(data) != size: # файл изменился после листинга — набор настаивается заново return None files.append(AnswerFile(name, data)) # T2.1: за время чтения мог появиться или пропасть файл — такой набор не отдаётся, он настаивается заново # (состояние settle уже сброшено в момент готовности) if self._snapshot(folder) != snapshot: return None return files @property def settle_seconds(self) -> float: """`SCAN_SETTLE_SECONDS`: сколько набор ответа должен не меняться, чтобы считаться готовым.""" return self._settle.settle_seconds def pending_jobs(self) -> int: """Число заданий с ненастоявшимся набором (для тестов и диагностики).""" return len(self._settle) def healthy(self) -> bool: """Корень существует и является каталогом.""" try: return self.root.is_dir() except (OSError, ValueError): return False @staticmethod def _snapshot(folder: Path) -> Snapshot: try: with os.scandir(folder) as entries: found = [] for entry in entries: if not is_answer_name(entry.name): continue try: if entry.is_file(): found.append((entry.name, entry.stat().st_size)) except FileNotFoundError: continue except (FileNotFoundError, NotADirectoryError): return () return tuple(sorted(found))