663 lines
34 KiB
Python
663 lines
34 KiB
Python
"""Воркер scan-gateway: запрос NATS → ack → S3 → горячая папка → набор ответов ИС → склейка → S3 `raw/` →
|
||
нормализатор → конверт.
|
||
|
||
Spec image-recognition-gateway; design D-2, D-3, D-4, D-5, D-7; решения R-14..R-20, R-24, R-25, R-30.
|
||
|
||
- `on_request(msg)` — колбэк подписки NATS. Только разбирает и проверяет тело, регистрирует `job_id` и запускает
|
||
задачу задания; ничего не ждёт (nats-py вызывает колбэки одной подписки последовательно, T11.4).
|
||
Тело не по схеме / MAJOR ≠ 1 / `pipeline` ≠ `image` → `SCHEMA_MISMATCH` без ack; конверт публикуется, только если
|
||
из тела извлекаются `job_id` и `file_id` (R-16), иначе запрос записывается в лог и отбрасывается.
|
||
- Задача задания: (1) ack `processing` — до чтения S3 и до ожидания слота; (2) слот семафора `SCAN_MAX_INFLIGHT`;
|
||
(3) `handle(request)` — обработка и публикация итогового конверта (R-20).
|
||
- Повторный запрос с `job_id`, который в работе или в LRU обработанных (`DONE_CACHE_SIZE`), получает повторный ack с
|
||
`attempt` повторного запроса; файл в ИС второй раз не отправляется (D-2, R-17).
|
||
- Ошибки — результат, не молчание (таблица D-7): каждое задание заканчивается опубликованным конвертом; исключение
|
||
задания наружу не выходит (кроме отмены при остановке); неудачная публикация — лог error (ядро закроет задание по
|
||
`reply_deadline`).
|
||
- Методы горячей папки синхронные (R-9) — вызываются через `asyncio.to_thread`; нормализатор — тоже в потоке.
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import re
|
||
import uuid
|
||
from collections import OrderedDict
|
||
from collections.abc import Awaitable, Callable, Coroutine
|
||
from dataclasses import dataclass
|
||
from datetime import UTC, datetime, timedelta
|
||
from typing import Any, Protocol
|
||
|
||
from loguru import logger
|
||
from pydantic import ValidationError
|
||
|
||
from src.contract import (
|
||
CONTRACT_VERSION,
|
||
ErrorCode,
|
||
Metrics,
|
||
RecognitionRequest,
|
||
RecognitionResult,
|
||
RecognitionResultMessage,
|
||
ResultStatus,
|
||
S3Ref,
|
||
ServiceInfo,
|
||
build_ack_message,
|
||
build_done_message,
|
||
build_error_message,
|
||
is_supported_major,
|
||
)
|
||
from src.core.clients import JSON_CONTENT_TYPE, OCTET_STREAM
|
||
from src.core.config import Settings
|
||
from src.core.nats import PublishFailed
|
||
from src.gateway.hotfolder import AnswerFile, HotFolder, inbox_name
|
||
from src.gateway.merge import dump_merged, merge_answers
|
||
from src.normalizer import NormalizeOutcome, Normalizer
|
||
|
||
__all__ = [
|
||
"MAX_ERROR_MESSAGE_CHARS",
|
||
"MAX_WARNINGS",
|
||
"MAX_WARNING_CHARS",
|
||
"IncomingMessage",
|
||
"ObjectStorage",
|
||
"ResultPublisher",
|
||
"Worker",
|
||
"raw_key",
|
||
"result_key",
|
||
"utc_now",
|
||
]
|
||
|
||
# Предел текста `error.message` в конверте: конверт остаётся далеко меньше лимита NATS (1 МБ) даже при 4 байтах на
|
||
# символ (T10.1). Длинное сообщение обрезается с многоточием.
|
||
MAX_ERROR_MESSAGE_CHARS = 4000
|
||
# Предел `metrics.warnings` конверта (R-30): не больше MAX_WARNINGS строк по MAX_WARNING_CHARS символов — конверт
|
||
# остаётся далеко меньше лимита NATS (1 МБ) при любом числе ошибочных файлов; лишние строки сводятся в последнюю
|
||
MAX_WARNINGS = 100
|
||
MAX_WARNING_CHARS = 1000
|
||
# Предел фрагментов `repr()` (исключения, недоверенные значения) внутри сообщений и логов
|
||
MAX_FRAGMENT_CHARS = 300
|
||
_ELLIPSIS = "…"
|
||
_MAX_ATTEMPT = 2**31 - 1 # attempt из тела, отвергнутого схемой, берётся только из разумного диапазона
|
||
_UUID_RE = re.compile(r"[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}", re.ASCII)
|
||
_IMAGE_PIPELINE = "image"
|
||
|
||
SleepFn = Callable[[float], Awaitable[None]]
|
||
ClockFn = Callable[[], datetime]
|
||
|
||
|
||
class IncomingMessage(Protocol):
|
||
"""Сообщение NATS в колбэке подписки: воркеру нужно только тело (`nats.aio.msg.Msg` подходит)."""
|
||
|
||
data: bytes
|
||
|
||
|
||
class ObjectStorage(Protocol):
|
||
"""Часть `S3Storage`, нужная воркеру; ошибки — классы `src.core.clients`."""
|
||
|
||
bucket: str
|
||
|
||
async def get_bytes(self, key: str, *, bucket: str | None = None) -> bytes: ...
|
||
|
||
async def put_bytes(self, key: str, data: bytes, content_type: str = OCTET_STREAM) -> None: ...
|
||
|
||
async def put_json(self, key: str, obj: Any) -> None: ...
|
||
|
||
|
||
class ResultPublisher(Protocol):
|
||
"""Часть `NatsGateway`: публикация конверта в `recognition.result` с `Nats-Msg-Id` и повторами."""
|
||
|
||
async def publish_result(self, message: RecognitionResultMessage, attempt: int) -> None: ...
|
||
|
||
|
||
def utc_now() -> datetime:
|
||
return datetime.now(UTC)
|
||
|
||
|
||
def raw_key(job_id: object) -> str:
|
||
"""Ключ сырого ответа ИС в S3 (D-4)."""
|
||
return f"raw/{job_id}.json"
|
||
|
||
|
||
def result_key(file_id: object, job_id: object) -> str:
|
||
"""Ключ файла результата в S3 (контракт, «Схема ключей S3»)."""
|
||
return f"results/{file_id}/{job_id}/result.json"
|
||
|
||
|
||
def clean_text(text: str, limit: int = MAX_ERROR_MESSAGE_CHARS) -> str:
|
||
"""Текст, безопасный для конверта: одиночные суррогаты (из `\\udXXX` в JSON) заменены на `?` — иначе публикация
|
||
падает на кодировании в UTF-8 (T10.2); длина ограничена `limit` с многоточием в конце (T10.1)."""
|
||
safe = text.encode("utf-8", "replace").decode("utf-8")
|
||
if len(safe) > limit:
|
||
safe = safe[: limit - len(_ELLIPSIS)] + _ELLIPSIS
|
||
return safe
|
||
|
||
|
||
def _fragment(value: object) -> str:
|
||
"""Короткое безопасное представление недоверенного значения для сообщения или лога."""
|
||
try:
|
||
text = repr(value)
|
||
except Exception: # repr не должен ронять обработку ошибки
|
||
text = f"<{type(value).__name__}>"
|
||
return clean_text(text, MAX_FRAGMENT_CHARS)
|
||
|
||
|
||
def _describe(exc: BaseException) -> str:
|
||
return f"{type(exc).__name__}: {clean_text(str(exc), MAX_FRAGMENT_CHARS)}"
|
||
|
||
|
||
def _parse_json(data: bytes) -> tuple[object, str | None]:
|
||
"""(разобранный JSON, None) или (None, описание ошибки разбора)."""
|
||
try:
|
||
return json.loads(data), None
|
||
except (ValueError, RecursionError) as exc: # UnicodeDecodeError — подкласс ValueError
|
||
return None, _describe(exc)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class _Rejection:
|
||
"""Тело запроса, не прошедшее проверку: причина и (если тело — JSON-объект) исходный словарь."""
|
||
|
||
reason: str
|
||
raw: dict[str, Any] | None
|
||
|
||
|
||
@dataclass
|
||
class _JobContext:
|
||
"""Состояние задания, нужное для конверта при непредвиденной ошибке: сырой ответ уже в S3."""
|
||
|
||
raw_s3_key: str | None = None
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class _WaitResult:
|
||
answer: list[AnswerFile] | None
|
||
polls: int
|
||
failures: int
|
||
last_error: BaseException | None
|
||
|
||
|
||
class Worker:
|
||
"""Обработка запросов `recognition.image.request` (см. докстринг модуля).
|
||
|
||
Зависимости внедряются: `storage` (S3), `publisher` (NATS), `hotfolder`, `normalizer`; `clock` — часы UTC для
|
||
дедлайна (R-20), `sleep` — пауза между опросами (подмены для тестов).
|
||
Запросы принимаются между `start()` и `stop()`.
|
||
"""
|
||
|
||
def __init__( # noqa: PLR0913 - зависимости воркера; всё, кроме настроек, — только именованные
|
||
self,
|
||
settings: Settings,
|
||
*,
|
||
storage: ObjectStorage,
|
||
publisher: ResultPublisher,
|
||
hotfolder: HotFolder,
|
||
normalizer: Normalizer,
|
||
clock: ClockFn = utc_now,
|
||
sleep: SleepFn = asyncio.sleep,
|
||
) -> None:
|
||
self._settings = settings
|
||
self._storage = storage
|
||
self._publisher = publisher
|
||
self._hotfolder = hotfolder
|
||
self._normalizer = normalizer
|
||
self._clock = clock
|
||
self._sleep = sleep
|
||
self.service = ServiceInfo.model_validate({"name": settings.SERVICE_NAME, "version": settings.VERSION})
|
||
self._semaphore = asyncio.Semaphore(settings.SCAN_MAX_INFLIGHT)
|
||
self._active: set[str] = set() # job_id в работе (ack, ожидание слота, обработка) — не вытесняются
|
||
self._done: OrderedDict[str, None] = OrderedDict() # LRU обработанных job_id
|
||
self._tasks: set[asyncio.Task[None]] = set()
|
||
self._accepting = False
|
||
|
||
# --- жизненный цикл ---------------------------------------------------------------------------------------
|
||
@property
|
||
def active_jobs(self) -> int:
|
||
"""Задания в работе (включая ожидающие слот)."""
|
||
return len(self._active)
|
||
|
||
def start(self) -> None:
|
||
"""Начать приём запросов (колбэк `on_request` подписывается после этого)."""
|
||
self._accepting = True
|
||
logger.info(
|
||
"Воркер scan-gateway запущен. max_inflight={max_inflight}, poll={poll}, timeout={timeout}",
|
||
max_inflight=self._settings.SCAN_MAX_INFLIGHT,
|
||
poll=self._settings.SCAN_POLL_SECONDS,
|
||
timeout=self._settings.SCAN_TIMEOUT_SECONDS,
|
||
)
|
||
|
||
async def stop(self) -> None:
|
||
"""Прекратить приём, отменить задания в работе и дождаться их завершения. Повторный вызов безопасен.
|
||
|
||
Отменённые задания конверта не публикуют: ядро закроет их по `reply_deadline` (D-2).
|
||
"""
|
||
self._accepting = False
|
||
tasks = list(self._tasks)
|
||
for task in tasks:
|
||
task.cancel()
|
||
if tasks:
|
||
await asyncio.gather(*tasks, return_exceptions=True)
|
||
logger.info("Воркер остановлен, отменено заданий: {count}", count=len(tasks))
|
||
|
||
async def wait_idle(self) -> None:
|
||
"""Дождаться завершения всех запущенных задач (ack, задания, публикации отказов) — для тестов и остановки."""
|
||
while self._tasks:
|
||
await asyncio.gather(*list(self._tasks), return_exceptions=True)
|
||
|
||
# --- приём запроса ----------------------------------------------------------------------------------------
|
||
async def on_request(self, msg: IncomingMessage) -> None:
|
||
"""Колбэк подписки: проверка тела, дедупликация по `job_id`, запуск задачи. Не ждёт ни сети, ни ИС."""
|
||
if not self._accepting:
|
||
logger.warning("Воркер не принимает запросы (остановлен): сообщение отброшено, ядро повторит запрос")
|
||
return
|
||
parsed = self._parse_request(msg.data)
|
||
if isinstance(parsed, _Rejection):
|
||
self._reject(parsed)
|
||
return
|
||
request = parsed
|
||
job_id = str(request.job_id)
|
||
if self._seen(job_id):
|
||
logger.info(
|
||
"Повторный запрос: повторный ack без второй отправки в ИС. job_id={job_id}, attempt={attempt}",
|
||
job_id=job_id,
|
||
attempt=request.attempt,
|
||
)
|
||
self._spawn(self._publish(self._ack(request), request.attempt), f"ack-repeat:{job_id}")
|
||
return
|
||
self._active.add(job_id)
|
||
logger.info(
|
||
"Запрос принят. job_id={job_id}, file_id={file_id}, attempt={attempt}, file_type={file_type}",
|
||
job_id=job_id,
|
||
file_id=request.file_id,
|
||
attempt=request.attempt,
|
||
file_type=request.file_type,
|
||
)
|
||
self._spawn(self._run_job(request), f"job:{job_id}")
|
||
|
||
def _parse_request(self, data: bytes) -> RecognitionRequest | _Rejection:
|
||
raw, error = _parse_json(data)
|
||
if error is not None:
|
||
return _Rejection(f"тело запроса не разбирается как JSON ({error})", None)
|
||
if not isinstance(raw, dict):
|
||
return _Rejection(f"тело запроса — не JSON-объект ({type(raw).__name__})", None)
|
||
version = raw.get("schema_version")
|
||
if not is_supported_major(version):
|
||
return _Rejection(f"неподдерживаемая версия контракта {_fragment(version)}: ожидается MAJOR 1", raw)
|
||
pipeline = raw.get("pipeline")
|
||
if pipeline != _IMAGE_PIPELINE:
|
||
return _Rejection(f"pipeline {_fragment(pipeline)} не обслуживается scan-gateway (ожидается image)", raw)
|
||
try:
|
||
return RecognitionRequest.model_validate(raw)
|
||
except ValidationError as exc:
|
||
details = "; ".join(
|
||
f"{'.'.join(str(part) for part in item['loc']) or '<тело>'}: {item['msg']}"
|
||
for item in exc.errors(include_url=False, include_input=False)[:5]
|
||
)
|
||
return _Rejection(f"запрос не соответствует схеме recognition-request: {details}", raw)
|
||
|
||
def _reject(self, rejection: _Rejection) -> None:
|
||
"""`SCHEMA_MISMATCH` без ack (R-16): конверт — только если в теле есть корректные `job_id` и `file_id`.
|
||
|
||
Для `job_id` в работе или уже обработанного конверт не публикуется (T12.1): его `Nats-Msg-Id`
|
||
`result:<job_id>:<attempt>` совпал бы с настоящим результатом, и ядро отбросило бы настоящий как дубль.
|
||
"""
|
||
raw = rejection.raw or {}
|
||
job_id, file_id = raw.get("job_id"), raw.get("file_id")
|
||
if not (_is_uuid(job_id) and _is_uuid(file_id)):
|
||
logger.warning(
|
||
"SCHEMA_MISMATCH: запрос отброшен без ответа (нет корректных job_id/file_id). reason={reason}",
|
||
reason=clean_text(rejection.reason, MAX_FRAGMENT_CHARS),
|
||
)
|
||
return
|
||
# `_active`/`_done` хранят канонический вид (нижний регистр), а UUID в теле может прийти в верхнем
|
||
key = str(uuid.UUID(str(job_id)))
|
||
if self._seen(key):
|
||
logger.warning(
|
||
"SCHEMA_MISMATCH для задания в работе/обработанного: конверт не публикуется. job_id={job_id}, "
|
||
"reason={reason}",
|
||
job_id=key,
|
||
reason=clean_text(rejection.reason, MAX_FRAGMENT_CHARS),
|
||
)
|
||
return
|
||
attempt = raw.get("attempt")
|
||
if not (type(attempt) is int and 1 <= attempt <= _MAX_ATTEMPT):
|
||
attempt = 1
|
||
logger.warning(
|
||
"SCHEMA_MISMATCH: запрос отвергнут, ack не публикуется. job_id={job_id}, reason={reason}",
|
||
job_id=job_id,
|
||
reason=clean_text(rejection.reason, MAX_FRAGMENT_CHARS),
|
||
)
|
||
message = build_error_message(
|
||
str(job_id), str(file_id), self.service, ErrorCode.SCHEMA_MISMATCH, clean_text(rejection.reason)
|
||
)
|
||
self._spawn(self._publish(message, attempt), f"schema-mismatch:{job_id}")
|
||
|
||
# --- дедупликация (D-2, R-17) -----------------------------------------------------------------------------
|
||
def _seen(self, job_id: str) -> bool:
|
||
if job_id in self._active:
|
||
return True
|
||
if job_id in self._done:
|
||
self._done.move_to_end(job_id)
|
||
return True
|
||
return False
|
||
|
||
def _finish(self, job_id: str) -> None:
|
||
self._active.discard(job_id)
|
||
self._done[job_id] = None
|
||
self._done.move_to_end(job_id)
|
||
while len(self._done) > self._settings.DONE_CACHE_SIZE:
|
||
self._done.popitem(last=False)
|
||
|
||
# --- задачи -----------------------------------------------------------------------------------------------
|
||
def _spawn(self, coro: Coroutine[Any, Any, None], name: str) -> None:
|
||
task = asyncio.create_task(coro, name=name)
|
||
self._tasks.add(task)
|
||
task.add_done_callback(self._tasks.discard)
|
||
|
||
async def _run_job(self, request: RecognitionRequest) -> None:
|
||
job_id = str(request.job_id)
|
||
try:
|
||
await self._publish(self._ack(request), request.attempt) # до чтения S3 и до ожидания слота
|
||
async with self._semaphore:
|
||
await self.handle(request)
|
||
except asyncio.CancelledError:
|
||
logger.warning("Задание отменено при остановке воркера. job_id={job_id}", job_id=job_id)
|
||
raise
|
||
except Exception: # handle не бросает; страховка, чтобы исключение не осталось в задаче
|
||
logger.exception("Непредвиденная ошибка задания. job_id={job_id}", job_id=job_id)
|
||
finally:
|
||
self._finish(job_id)
|
||
|
||
def _ack(self, request: RecognitionRequest) -> RecognitionResultMessage:
|
||
return build_ack_message(request.job_id, request.file_id, self.service)
|
||
|
||
async def _publish(self, message: RecognitionResultMessage, attempt: int) -> None:
|
||
"""Публикация с повторами `NatsGateway`; неудача — лог error, исключение не выходит (D-7)."""
|
||
try:
|
||
await self._publisher.publish_result(message, attempt)
|
||
except PublishFailed as exc:
|
||
logger.error(
|
||
"Конверт не опубликован, ядро закроет задание по reply_deadline. job_id={job_id}, status={status}, "
|
||
"msg_id={msg_id}, attempts={attempts}, cause={cause}",
|
||
job_id=message.job_id,
|
||
status=message.status.value,
|
||
msg_id=exc.msg_id,
|
||
attempts=exc.attempts,
|
||
cause=_fragment(exc.__cause__),
|
||
)
|
||
except Exception as exc:
|
||
logger.error(
|
||
"Конверт не опубликован: непредвиденная ошибка. job_id={job_id}, status={status}, error={error}",
|
||
job_id=message.job_id,
|
||
status=message.status.value,
|
||
error=_describe(exc),
|
||
)
|
||
|
||
# --- обработка задания ------------------------------------------------------------------------------------
|
||
async def handle(self, request: RecognitionRequest) -> RecognitionResultMessage:
|
||
"""Обработать задание и опубликовать итоговый конверт; вернуть его.
|
||
|
||
Не бросает исключений (кроме `asyncio.CancelledError`): любой сбой становится конвертом `error` по D-7.
|
||
Ack и семафор — забота вызывающего (`on_request`); здесь — S3 → горячая папка → ответ → `raw/` →
|
||
нормализатор → `results/` → конверт.
|
||
"""
|
||
context = _JobContext()
|
||
try:
|
||
message = await self._process(request, context)
|
||
except asyncio.CancelledError:
|
||
raise
|
||
except Exception as exc:
|
||
logger.exception("Непредвиденная ошибка обработки. job_id={job_id}", job_id=request.job_id)
|
||
message = self._error(
|
||
request,
|
||
ErrorCode.UNKNOWN_ERROR,
|
||
f"непредвиденная ошибка scan-gateway: {_describe(exc)}",
|
||
raw_s3_key=context.raw_s3_key,
|
||
)
|
||
logger.info(
|
||
"Задание обработано. job_id={job_id}, status={status}, error_code={code}",
|
||
job_id=request.job_id,
|
||
status=message.status.value,
|
||
code=message.error.code.value if message.error else None,
|
||
)
|
||
await self._publish(message, request.attempt)
|
||
return message
|
||
|
||
def _error(
|
||
self,
|
||
request: RecognitionRequest,
|
||
code: ErrorCode,
|
||
text: str,
|
||
*,
|
||
raw_s3_key: str | None = None,
|
||
metrics: Metrics | None = None,
|
||
) -> RecognitionResultMessage:
|
||
return build_error_message(
|
||
request.job_id,
|
||
request.file_id,
|
||
self.service,
|
||
code,
|
||
clean_text(text),
|
||
raw_s3_key=raw_s3_key,
|
||
metrics=metrics,
|
||
)
|
||
|
||
async def _process(self, request: RecognitionRequest, context: _JobContext) -> RecognitionResultMessage:
|
||
"""Шаги задания; каждая неудача — конверт `error` по строке таблицы D-7."""
|
||
sent = await self._send(request)
|
||
if isinstance(sent, RecognitionResultMessage):
|
||
return sent
|
||
name, started = sent
|
||
job_id = str(request.job_id)
|
||
# 3. Опрос исходящего каталога до min(reply_deadline, старт + SCAN_TIMEOUT_SECONDS) (D-3, R-20)
|
||
deadline = min(request.reply_deadline, started + timedelta(seconds=self._settings.SCAN_TIMEOUT_SECONDS))
|
||
waited = await self._wait_answer(job_id, deadline)
|
||
if waited.answer is None:
|
||
if waited.failures == waited.polls and waited.last_error is not None:
|
||
return self._error(
|
||
request,
|
||
ErrorCode.RECOGNITION_FAILED,
|
||
f"горячая папка недоступна: исходящий каталог не прочитан ни разу за {waited.polls} опросов "
|
||
f"({_describe(waited.last_error)})",
|
||
)
|
||
logger.warning(
|
||
"ИС Сканирование не ответила до дедлайна. job_id={job_id}, deadline={deadline}, polls={polls}",
|
||
job_id=job_id,
|
||
deadline=deadline.isoformat(),
|
||
polls=waited.polls,
|
||
)
|
||
# R-39: при settle > 0 устойчивая ошибка чтения чередуется с пустыми опросами — код остаётся TIMEOUT,
|
||
# но ошибку опроса видно в тексте, чтобы конверт не вёл к ИС, когда не читается горячая папка
|
||
poll_failure = (
|
||
f"; ошибки опроса исходящего каталога: {waited.failures} из {waited.polls}, "
|
||
f"последняя: {_describe(waited.last_error)}"
|
||
if waited.failures > 0 and waited.last_error is not None
|
||
else ""
|
||
)
|
||
return self._error(
|
||
request,
|
||
ErrorCode.TIMEOUT,
|
||
f"ИС Сканирование не ответила до {deadline.isoformat()} (опросов: {waited.polls}{poll_failure}); "
|
||
f"входящий файл {name} остаётся в каталоге, повторной отправки нет",
|
||
)
|
||
return await self._accept_answer(request, waited.answer, context)
|
||
|
||
async def _send(self, request: RecognitionRequest) -> tuple[str, datetime] | RecognitionResultMessage: # noqa: PLR0911 - каждая ветка — строка таблицы D-7
|
||
"""Срок, исходник из S3, запись во входящий каталог: (имя файла, момент записи) или конверт `error`."""
|
||
job_id = str(request.job_id)
|
||
if request.reply_deadline <= self._clock():
|
||
# задание дождалось слота после срока ответа: ядро уже закрывает его, файл в ИС не отправляется
|
||
return self._error(
|
||
request,
|
||
ErrorCode.TIMEOUT,
|
||
f"срок ответа {request.reply_deadline.isoformat()} истёк до отправки файла в ИС Сканирование",
|
||
)
|
||
|
||
# 1. Исходник из S3 (один бакет сервиса, R-19)
|
||
source = request.s3
|
||
if source.bucket != self._storage.bucket:
|
||
return self._error(
|
||
request,
|
||
ErrorCode.FILE_READ_ERROR,
|
||
f"бакет {_fragment(source.bucket)} не обслуживается: scan-gateway работает с одним бакетом "
|
||
f"{_fragment(self._storage.bucket)}",
|
||
)
|
||
try:
|
||
data = await self._storage.get_bytes(source.key, bucket=source.bucket)
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"Исходник не прочитан из S3. job_id={job_id}, error={error}", job_id=job_id, error=_describe(exc)
|
||
)
|
||
return self._error(
|
||
request,
|
||
ErrorCode.FILE_READ_ERROR,
|
||
f"исходный файл {source.bucket}/{source.key} не прочитан из S3: {_describe(exc)}",
|
||
)
|
||
|
||
# 2. Во входящий каталог как <job_id>_<original_name> (R-5, R-15)
|
||
try:
|
||
name = inbox_name(job_id, request.original_name)
|
||
except ValueError:
|
||
return self._error(
|
||
request,
|
||
ErrorCode.RECOGNITION_FAILED,
|
||
f"недопустимое имя файла {_fragment(request.original_name)}: файл не передан в ИС Сканирование",
|
||
)
|
||
try:
|
||
await asyncio.to_thread(self._hotfolder.put, name, data)
|
||
except Exception as exc:
|
||
logger.error(
|
||
"Горячая папка недоступна: файл не записан. job_id={job_id}, error={error}",
|
||
job_id=job_id,
|
||
error=_describe(exc),
|
||
)
|
||
return self._error(
|
||
request,
|
||
ErrorCode.RECOGNITION_FAILED,
|
||
f"горячая папка недоступна: файл не записан во входящий каталог ({_describe(exc)})",
|
||
)
|
||
return name, self._clock()
|
||
|
||
async def _accept_answer(
|
||
self, request: RecognitionRequest, answer: list[AnswerFile], context: _JobContext
|
||
) -> RecognitionResultMessage:
|
||
"""Набор ответов ИС: склейка, сырой в S3, нормализатор, файл результата; конверт итога."""
|
||
job_id = str(request.job_id)
|
||
|
||
# 4. Склейка всех файлов набора (не JSON — parse_error файла, R-24) и сырой ответ в S3 до нормализации (D-4)
|
||
# (склейка и сериализация — CPU-работа над произвольным JSON: в потоке, как нормализатор; сбой сериализации,
|
||
# например RecursionError на очень глубокой вложенности, — такой же сбой записи сырого ответа)
|
||
merged = await asyncio.to_thread(merge_answers, job_id, answer)
|
||
key = raw_key(job_id)
|
||
try:
|
||
payload = await asyncio.to_thread(dump_merged, merged)
|
||
await self._storage.put_bytes(key, payload, JSON_CONTENT_TYPE)
|
||
except Exception as exc:
|
||
logger.error(
|
||
"Сырой ответ ИС не сохранён. job_id={job_id}, error={error}", job_id=job_id, error=_describe(exc)
|
||
)
|
||
return self._error(
|
||
request, ErrorCode.UNKNOWN_ERROR, f"сырой ответ ИС не сохранён в S3 ({key}): {_describe(exc)}"
|
||
)
|
||
context.raw_s3_key = key
|
||
|
||
# 5. Нормализатор получает склеенный ответ (D-5); его исключение — UNKNOWN_ERROR с сохранённым сырым ответом
|
||
try:
|
||
outcome = await asyncio.to_thread(self._normalizer.normalize, merged)
|
||
except Exception as exc:
|
||
logger.exception("Нормализатор ответа ИС упал. job_id={job_id}", job_id=job_id)
|
||
return self._error(
|
||
request, ErrorCode.UNKNOWN_ERROR, f"нормализатор ответа ИС упал: {_describe(exc)}", raw_s3_key=key
|
||
)
|
||
metrics = _outcome_metrics(outcome, key)
|
||
if outcome.status is ResultStatus.ERROR:
|
||
error = outcome.error
|
||
if error is None: # исключено инвариантом NormalizeOutcome
|
||
raise ValueError("NormalizeOutcome со status=error без error")
|
||
return self._error(request, error.code, error.message, raw_s3_key=key, metrics=metrics)
|
||
|
||
# 6. Успех: файл результата в S3, затем конверт с result_s3
|
||
result = RecognitionResult.model_validate(
|
||
{
|
||
"schema_version": CONTRACT_VERSION,
|
||
"job_id": request.job_id,
|
||
"file_id": request.file_id,
|
||
"service": self.service,
|
||
"status": outcome.status,
|
||
"raw_s3_key": key,
|
||
"documents": list(outcome.documents),
|
||
**({"metrics": metrics} if metrics is not None else {}),
|
||
}
|
||
)
|
||
target = result_key(request.file_id, request.job_id)
|
||
try:
|
||
await self._storage.put_json(target, result.to_wire())
|
||
except Exception as exc:
|
||
logger.error(
|
||
"Файл результата не записан. job_id={job_id}, error={error}", job_id=job_id, error=_describe(exc)
|
||
)
|
||
return self._error(
|
||
request,
|
||
ErrorCode.UNKNOWN_ERROR,
|
||
f"файл результата не записан в S3 ({target}): {_describe(exc)}",
|
||
raw_s3_key=key,
|
||
)
|
||
return build_done_message(
|
||
request.job_id,
|
||
request.file_id,
|
||
self.service,
|
||
S3Ref(bucket=self._storage.bucket, key=target),
|
||
needs_check=outcome.status is ResultStatus.NEEDS_CHECK,
|
||
raw_s3_key=key,
|
||
metrics=metrics,
|
||
)
|
||
|
||
async def _wait_answer(self, job_id: str, deadline: datetime) -> _WaitResult:
|
||
"""Опрос `poll(job_id)` каждые `SCAN_POLL_SECONDS` до дедлайна (последний опрос — в момент дедлайна).
|
||
|
||
Исключения `poll` (SMB и прочие) — warning, опрос продолжается (R-14). Первый готовый набор (`poll` отдаёт его
|
||
только после настаивания, D-1) принимается сразу: файл, не разобранный как JSON, не перечитывается — он
|
||
станет `parse_error` в склейке (R-24). Ненастоявшийся к дедлайну набор — `None` (таймаут).
|
||
"""
|
||
polls = failures = 0
|
||
last_error: BaseException | None = None
|
||
while True:
|
||
polls += 1
|
||
answer: list[AnswerFile] | None
|
||
try:
|
||
answer = await asyncio.to_thread(self._hotfolder.poll, job_id)
|
||
except Exception as exc:
|
||
failures += 1
|
||
last_error = exc
|
||
answer = None
|
||
logger.warning(
|
||
"Опрос исходящего каталога не удался, опрос продолжается. job_id={job_id}, error={error}",
|
||
job_id=job_id,
|
||
error=_describe(exc),
|
||
)
|
||
if answer:
|
||
return _WaitResult(answer, polls, failures, last_error)
|
||
now = self._clock()
|
||
if now >= deadline:
|
||
return _WaitResult(None, polls, failures, last_error)
|
||
await self._sleep(min(self._settings.SCAN_POLL_SECONDS, (deadline - now).total_seconds()))
|
||
|
||
|
||
def _outcome_metrics(outcome: NormalizeOutcome, raw_s3_key: str) -> Metrics | None:
|
||
"""`metrics` конверта: метрики исхода + `warnings` исхода (R-25) в конце `metrics.warnings`.
|
||
|
||
Пустые `warnings` опускаются. Строки безопасны для конверта (`clean_text`), их не больше `MAX_WARNINGS` по
|
||
`MAX_WARNING_CHARS` символов; лишние сводятся в последнюю строку со ссылкой на сырой ответ (R-30).
|
||
"""
|
||
metrics = outcome.metrics
|
||
lines = [*(metrics.warnings or [] if metrics is not None else []), *outcome.warnings]
|
||
if not lines:
|
||
return metrics
|
||
if len(lines) > MAX_WARNINGS:
|
||
hidden = len(lines) - (MAX_WARNINGS - 1)
|
||
lines = [*lines[: MAX_WARNINGS - 1], f"ещё предупреждений: {hidden}; все ошибки файлов — в {raw_s3_key}"]
|
||
warnings = [clean_text(line, MAX_WARNING_CHARS) for line in lines]
|
||
base = metrics.model_dump(exclude_none=True) if metrics is not None else {}
|
||
return Metrics.model_validate({**base, "warnings": warnings})
|
||
|
||
|
||
def _is_uuid(value: object) -> bool:
|
||
return isinstance(value, str) and _UUID_RE.fullmatch(value) is not None
|