b489f6e0b3
process_once(root, fixtures, delay): новые <uuid>_<имя> -> ответ из фикстуры (точное имя, затем префикс, file_uid подставляется) либо INVALID_DOCUMENT, атомарная запись ответа, затем переименование в .done. Цикл python -m src.scan_emulator с корректной остановкой по SIGINT/SIGTERM. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
221 lines
11 KiB
Python
221 lines
11 KiB
Python
"""Эмулятор ИС Сканирование (Content AI) для разработки и тестов (D-6, R-11, spec scan-emulator).
|
||
|
||
Воспроизводит протокол горячих папок: для нового файла `<uuid>_<name>.<ext>` во входящем каталоге через задержку кладёт
|
||
`<output>/<uuid>/<uuid>.json` с эталонным ответом из `SCAN_FIXTURES_DIR` и переименовывает входящий файл в `….done`.
|
||
Эмулятор читает собственные переменные окружения (`EmulatorSettings`), а не настройки шлюза: в его контейнере нет
|
||
переменных MinIO/NATS. С каталогами работает через `LocalHotFolder`, тем самым тестируя и его.
|
||
"""
|
||
|
||
import json
|
||
import math
|
||
import os
|
||
import re
|
||
import threading
|
||
import time
|
||
from collections.abc import Mapping
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from dotenv import dotenv_values
|
||
from loguru import logger
|
||
|
||
from src.gateway.hotfolder import LocalHotFolder
|
||
|
||
__all__ = ["EmulatorSettings", "process_once", "run"]
|
||
|
||
# `<uuid>_<name>`: канонический UUID (36 символов) и непустое имя после `_`
|
||
_INBOX_NAME = re.compile(
|
||
r"^(?P<uuid>[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12})_(?P<name>.+)$"
|
||
)
|
||
_DONE_SUFFIX = ".done"
|
||
_ERROR_CODE = "INVALID_DOCUMENT"
|
||
|
||
|
||
def _read_float(env: Mapping[str, str], name: str, default: float, *, positive: bool) -> float:
|
||
raw = env.get(name)
|
||
if raw is None or raw.strip() == "":
|
||
return default
|
||
try:
|
||
value = float(raw)
|
||
except ValueError:
|
||
raise ValueError(f"{name}: ожидается число, получено {raw!r}") from None
|
||
if not math.isfinite(value) or value < 0 or (positive and value == 0):
|
||
raise ValueError(
|
||
f"{name}: ожидается {'положительное' if positive else 'неотрицательное'} число, получено {raw!r}"
|
||
)
|
||
return value
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class EmulatorSettings:
|
||
"""Настройки эмулятора из переменных окружения (имена совпадают с `.env.example`)."""
|
||
|
||
hotfolder_path: Path
|
||
input_dir: str = "Input"
|
||
output_dir: str = "Output"
|
||
fixtures_dir: Path = Path("../../fixtures/ocr")
|
||
delay: float = 2.0
|
||
poll_interval: float = 1.0
|
||
|
||
@classmethod
|
||
def from_env(cls, environ: Mapping[str, str] | None = None) -> "EmulatorSettings":
|
||
"""Разбор окружения. Без аргумента: `os.environ` поверх файла `ENV_FILE` (по умолчанию `.env`, если он есть).
|
||
|
||
`SCAN_HOTFOLDER_PATH` обязателен (`ValueError`); `SCAN_INPUT_DIR`=Input, `SCAN_OUTPUT_DIR`=Output,
|
||
`SCAN_FIXTURES_DIR`=../../fixtures/ocr, `SCAN_EMULATOR_DELAY`=2 (с, >= 0),
|
||
`SCAN_EMULATOR_POLL_SECONDS`=1 (с, > 0).
|
||
"""
|
||
if environ is None:
|
||
env_file = os.environ.get("ENV_FILE", ".env")
|
||
file_values = {k: v for k, v in dotenv_values(env_file).items() if v is not None}
|
||
env: Mapping[str, str] = {**file_values, **os.environ}
|
||
else:
|
||
env = environ
|
||
path = env.get("SCAN_HOTFOLDER_PATH", "").strip()
|
||
if not path:
|
||
raise ValueError("SCAN_HOTFOLDER_PATH не задан: укажите корень горячей папки (каталог с Input/ и Output/)")
|
||
return cls(
|
||
hotfolder_path=Path(path),
|
||
input_dir=env.get("SCAN_INPUT_DIR") or "Input",
|
||
output_dir=env.get("SCAN_OUTPUT_DIR") or "Output",
|
||
fixtures_dir=Path(env.get("SCAN_FIXTURES_DIR") or "../../fixtures/ocr"),
|
||
delay=_read_float(env, "SCAN_EMULATOR_DELAY", 2.0, positive=False),
|
||
poll_interval=_read_float(env, "SCAN_EMULATOR_POLL_SECONDS", 1.0, positive=True),
|
||
)
|
||
|
||
|
||
def _error_answer(job_id: str, message: str) -> dict[str, Any]:
|
||
return {"file_uid": job_id, "error": {"code": _ERROR_CODE, "message": message}, "data_ocr": []}
|
||
|
||
|
||
def _find_fixture(fixtures: Path, stem: str) -> Path | None:
|
||
"""`<stem>.json`, затем `<stem>*.json` (первый по алфавиту). Имя сравнивается буквально, без glob."""
|
||
try:
|
||
names = sorted(p.name for p in fixtures.iterdir() if p.name.lower().endswith(".json") and p.is_file())
|
||
except OSError:
|
||
return None
|
||
if f"{stem}.json" in names:
|
||
return fixtures / f"{stem}.json"
|
||
for name in names:
|
||
if name.startswith(stem):
|
||
return fixtures / name
|
||
return None
|
||
|
||
|
||
def _build_answer(job_id: str, name: str, fixtures: Path) -> dict[str, Any]:
|
||
"""Ответ ИС на файл `name` (без префикса uuid): фикстура с подставленным `file_uid` либо ошибка Content AI."""
|
||
fixture = _find_fixture(fixtures, Path(name).stem)
|
||
if fixture is None:
|
||
return _error_answer(job_id, f"{name}: нет эталонного ответа в фикстурах")
|
||
try:
|
||
data = json.loads(fixture.read_text(encoding="utf-8"))
|
||
if not isinstance(data, dict):
|
||
raise ValueError("корень не объект JSON")
|
||
except (OSError, ValueError) as exc: # json.JSONDecodeError и UnicodeDecodeError — подклассы ValueError
|
||
logger.error("Эмулятор: фикстура {} повреждена ({}); для {} отдаём {}", fixture, exc, name, _ERROR_CODE)
|
||
return _error_answer(job_id, f"{name}: эталонный ответ {fixture.name} повреждён и не разобран")
|
||
return {"file_uid": job_id, **{key: value for key, value in data.items() if key != "file_uid"}}
|
||
|
||
|
||
def _write_answer(hotfolder: LocalHotFolder, job_id: str, answer: dict[str, Any]) -> None:
|
||
"""Атомарная запись ответа: `.part` в том же каталоге, затем `os.replace`."""
|
||
target = hotfolder.answer_path(job_id)
|
||
target.parent.mkdir(parents=True, exist_ok=True)
|
||
part = target.parent / f".{target.name}.part"
|
||
try:
|
||
part.write_text(json.dumps(answer, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
os.replace(part, target)
|
||
except BaseException:
|
||
part.unlink(missing_ok=True)
|
||
raise
|
||
|
||
|
||
def process_once(
|
||
root: Path | str,
|
||
fixtures: Path | str,
|
||
delay: float,
|
||
*,
|
||
input_dir: str = "Input",
|
||
output_dir: str = "Output",
|
||
) -> list[str]:
|
||
"""Один проход по входящему каталогу; возвращает uuid обработанных файлов.
|
||
|
||
Берёт новые `<uuid>_<name>.<ext>` (не dot-файлы, не `*.done`, не каталоги), при наличии работы один раз за проход
|
||
ждёт `delay` секунд, для каждого файла пишет `<output>/<uuid>/<uuid>.json` и только затем переименовывает входящий
|
||
файл в `….done`. Имена не по шаблону `<uuid>_` игнорируются (одно предупреждение за проход). Нет входящего
|
||
каталога — пустой результат.
|
||
"""
|
||
hotfolder = LocalHotFolder(root, input_dir, output_dir)
|
||
fixtures_dir = Path(fixtures)
|
||
try:
|
||
entries = sorted(hotfolder.input_dir.iterdir(), key=lambda p: p.name)
|
||
except FileNotFoundError:
|
||
return []
|
||
pending: list[tuple[Path, str, str]] = []
|
||
foreign: list[str] = []
|
||
for entry in entries:
|
||
if entry.name.startswith(".") or entry.name.endswith(_DONE_SUFFIX) or not entry.is_file():
|
||
continue
|
||
match = _INBOX_NAME.match(entry.name)
|
||
if match is None:
|
||
foreign.append(entry.name)
|
||
continue
|
||
pending.append((entry, match["uuid"], match["name"]))
|
||
if foreign:
|
||
logger.warning("Эмулятор: файлы не по шаблону <uuid>_<имя> пропущены: {}", ", ".join(foreign))
|
||
if not pending:
|
||
return []
|
||
if delay > 0:
|
||
time.sleep(delay)
|
||
processed: list[str] = []
|
||
for entry, job_id, name in pending:
|
||
try:
|
||
_write_answer(hotfolder, job_id, _build_answer(job_id, name, fixtures_dir))
|
||
os.replace(entry, entry.with_name(entry.name + _DONE_SUFFIX))
|
||
except FileNotFoundError:
|
||
logger.warning("Эмулятор: файл {} исчез во время обработки", entry.name)
|
||
continue
|
||
processed.append(job_id)
|
||
logger.info("Эмулятор: {} -> {}", entry.name, hotfolder.answer_path(job_id))
|
||
return processed
|
||
|
||
|
||
def _count_fixtures(fixtures: Path) -> int:
|
||
return sum(1 for p in fixtures.iterdir() if p.name.lower().endswith(".json") and p.is_file())
|
||
|
||
|
||
def run(settings: EmulatorSettings, stop: threading.Event, *, max_iterations: int | None = None) -> None:
|
||
"""Цикл эмулятора: `process_once` каждые `poll_interval` секунд, пока не взведён `stop` (или не выполнено
|
||
`max_iterations` проходов — для тестов). Сбой прохода логируется и не останавливает цикл."""
|
||
hotfolder = LocalHotFolder(settings.hotfolder_path, settings.input_dir, settings.output_dir)
|
||
logger.info(
|
||
"Эмулятор ИС Сканирование: корень {}, входящий каталог {}, исходящий каталог {}, задержка {} с, опрос {} с",
|
||
hotfolder.root,
|
||
hotfolder.input_dir,
|
||
hotfolder.output_dir,
|
||
settings.delay,
|
||
settings.poll_interval,
|
||
)
|
||
if settings.fixtures_dir.is_dir():
|
||
logger.info("Фикстуры: {}, фикстур: {}", settings.fixtures_dir, _count_fixtures(settings.fixtures_dir))
|
||
else:
|
||
logger.warning(
|
||
"Каталог фикстур {} не найден: на все файлы будет отвечать {}", settings.fixtures_dir, _ERROR_CODE
|
||
)
|
||
iterations = 0
|
||
while not stop.is_set() and (max_iterations is None or iterations < max_iterations):
|
||
try:
|
||
process_once(
|
||
settings.hotfolder_path,
|
||
settings.fixtures_dir,
|
||
settings.delay,
|
||
input_dir=settings.input_dir,
|
||
output_dir=settings.output_dir,
|
||
)
|
||
except Exception: # noqa: BLE001 — эмулятор живёт, пока его не остановят
|
||
logger.exception("Эмулятор: сбой прохода по входящему каталогу")
|
||
iterations += 1
|
||
stop.wait(settings.poll_interval)
|
||
logger.info("Эмулятор остановлен")
|