"""Воркер распознавания: запрос `recognition.tabular.request` → ack → S3 → парсер → S3 → `recognition.result`.

Design D-2, D-4, D-5, D-7; spec tabular-recognition-service «Подтверждение запроса до обработки», «Результат по
контракту», «Ошибки транспорта превращаются в результат», «Парсер — заменяемый интерфейс».

- `on_request(msg)` — обработчик подписки NATS (queue group `tabular-parser`): разбор тела и валидация по контракту
  (pydantic `RecognitionRequest`, MAJOR `schema_version` = 1). Невалидный запрос — конверт `error SCHEMA_MISMATCH`
  без ack (если `job_id`, `file_id` и `attempt` читаемы; иначе собрать валидный конверт нельзя — error-лог и сброс).
  Валидный: `job_id` уже в работе или обработан этим процессом — только повторный ack (`ack:<job_id>:<attempt>` этого
  запроса), второй обработки нет; иначе — `job_id` в множество «в работе», ack `processing` **до** чтения S3, задание
  во внутреннюю `asyncio.Queue`.
- Рабочие корутины (`concurrency`): S3 get → `Parser.parse` через `asyncio.to_thread` (R-13) → для `done | needs_check`
  файл результата `results/{file_id}/{job_id}/result.json` → конверт `done | needs_check | error` с
  `metrics.duration_ms` (всегда, ≥ 0) → публикация с `Nats-Msg-Id = result:<job_id>:<attempt>`.
- Таблица ошибок D-5: S3 get упал → `FILE_READ_ERROR`; исключение парсера → `UNKNOWN_ERROR` (тип и текст исключения
  одной строкой, без трассировки); запись файла результата упала → `UNKNOWN_ERROR`; публикация не удалась после
  повторов (`NatsPublishError`) → error-лог, задание дальше не идёт (ядро закроет его по `reply_deadline`).
  Никакое исключение не роняет процесс и рабочие корутины.
- Дедупликация: `job_id` в работе (очередь + обрабатываемые) — множество; завершённые (любой итог, в том числе ошибка и
  неудачная публикация результата) — LRU размера `done_cache_size`.
- Остановка: новые запросы не принимаются (без ack — ядро переотправит другой инстанции), очередь и текущие задания
  дорабатываются не дольше `timeout`, остальное отменяется с warning-логом.
"""

import asyncio
import json
import re
import time
from collections import OrderedDict
from collections.abc import Callable, Mapping
from dataclasses import dataclass
from typing import Any, Protocol

from loguru import logger
from pydantic import ValidationError

from src.contract import (
    CONTRACT_VERSION,
    NATS_MSG_ID_HEADER,
    RESULT_SUBJECT,
    ErrorCode,
    MessageError,
    MessageStatus,
    Metrics,
    RecognitionRequest,
    RecognitionResult,
    RecognitionResultMessage,
    ResultStatus,
    S3Ref,
    ServiceInfo,
    ack_msg_id,
    result_msg_id,
)
from src.core.clients import Storage
from src.core.nats import NatsPublishError, Publisher
from src.parser import ParseOutcome, Parser

SUPPORTED_MAJOR = "1"
STOP_TIMEOUT = 10.0  # секунд на доработку очереди при остановке (lifespan)
ERROR_MESSAGE_LIMIT = 1000  # символов в `error.message` конверта

_VERSION_RE = re.compile(r"(\d+)\.\d+", re.ASCII)
_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)

Clock = Callable[[], float]


class InboundMessage(Protocol):
    """Входящее сообщение NATS (подмножество `nats.aio.msg.Msg`, нужное воркеру)."""

    @property
    def data(self) -> bytes: ...

    @property
    def headers(self) -> Mapping[str, str] | None: ...


def result_key(file_id: object, job_id: object) -> str:
    """Ключ файла результата в бакете запроса (design D-4)."""
    return f"results/{file_id}/{job_id}/result.json"


class InvalidRequestError(Exception):
    """Запрос не прошёл контракт: `SCHEMA_MISMATCH`."""


@dataclass(frozen=True, slots=True)
class _Job:
    request: RecognitionRequest


def _one_line(text: str) -> str:
    """Сообщение ошибки для конверта: одна строка, без трассировки, ограниченной длины."""
    flat = " ".join(text.split())
    return flat if len(flat) <= ERROR_MESSAGE_LIMIT else flat[: ERROR_MESSAGE_LIMIT - 1] + "…"


def _describe(exc: BaseException) -> str:
    return _one_line(f"{type(exc).__name__}: {exc}")


def _validation_summary(exc: ValidationError) -> str:
    parts = []
    for error in exc.errors(include_url=False, include_input=False):
        location = ".".join(str(item) for item in error["loc"]) or "<тело>"
        parts.append(f"{location}: {error['msg']}")
    return "; ".join(parts)


class RecognitionWorker:
    """Обработчик запросов распознавания; зависимости — протоколы (`Storage`, `Publisher`, `Parser`, design D-7)."""

    def __init__(  # noqa: PLR0913 - все параметры, кроме зависимостей, именованные
        self,
        *,
        storage: Storage,
        publisher: Publisher,
        parser: Parser,
        service_name: str,
        service_version: str,
        concurrency: int = 1,
        done_cache_size: int = 1000,
        clock: Clock = time.monotonic,
    ) -> None:
        if concurrency < 1 or done_cache_size < 1:
            raise ValueError("concurrency и done_cache_size должны быть ≥ 1")
        self._storage = storage
        self._publisher = publisher
        self._parser = parser
        self._service = ServiceInfo.model_validate({"name": service_name, "version": service_version})
        self._concurrency = concurrency
        self._done_cache_size = done_cache_size
        self._clock = clock
        self._queue: asyncio.Queue[_Job] = asyncio.Queue()
        self._in_progress: set[str] = set()  # job_id в очереди или в обработке
        self._done: OrderedDict[str, None] = OrderedDict()  # LRU завершённых job_id
        self._tasks: list[asyncio.Task[None]] = []
        self._accepting = True

    # --- жизненный цикл ---------------------------------------------------------------------------------------------
    def start(self) -> None:
        """Запускает рабочие корутины (`concurrency` штук)."""
        if self._tasks:
            return
        self._accepting = True
        self._tasks = [
            asyncio.create_task(self._run(), name=f"recognition-worker-{index}") for index in range(self._concurrency)
        ]
        logger.info("Воркер распознавания запущен. concurrency={concurrency}", concurrency=self._concurrency)

    async def join(self) -> None:
        """Ждёт, пока очередь опустеет и все взятые задания завершатся."""
        await self._queue.join()

    async def stop(self, timeout: float = STOP_TIMEOUT) -> None:
        """Перестаёт принимать запросы, дорабатывает очередь не дольше `timeout` секунд, затем отменяет остальное."""
        self._accepting = False
        if not self._tasks:
            return
        try:
            await asyncio.wait_for(self._queue.join(), timeout)
        except TimeoutError:
            logger.warning(
                "Остановка воркера: задания не завершились за отведённое время и отменены (ack опубликован, ядро "
                "закроет их по reply_deadline). error_category=shutdown, in_flight={in_flight}, pending={pending}",
                in_flight=len(self._in_progress) - self._queue.qsize(),
                pending=self._queue.qsize(),
            )
        tasks, self._tasks = self._tasks, []
        for task in tasks:
            task.cancel()
        await asyncio.gather(*tasks, return_exceptions=True)
        logger.info("Воркер распознавания остановлен")

    # --- приём запроса ----------------------------------------------------------------------------------------------
    async def on_request(self, msg: InboundMessage) -> None:
        """Обработчик подписки на `recognition.tabular.request`; исключения наружу не выходят."""
        if not self._accepting:
            logger.warning("Воркер останавливается: запрос не принят (без ack, ядро переотправит)")
            return
        started = self._clock()
        try:
            request = self._parse_request(msg.data)
        except InvalidRequestError as exc:
            await self._reject(msg, str(exc), started)
            return

        job_id = str(request.job_id)
        duplicate = job_id in self._in_progress or job_id in self._done
        if duplicate:
            if job_id in self._done:
                self._done.move_to_end(job_id)
            logger.info(
                "Повторный запрос: только ack, без повторной обработки. job_id={job_id}, attempt={attempt}",
                job_id=job_id,
                attempt=request.attempt,
            )
        else:
            self._in_progress.add(job_id)
        await self._publish_ack(request)
        if not duplicate:
            self._queue.put_nowait(_Job(request))

    def _parse_request(self, data: bytes) -> RecognitionRequest:
        try:
            raw = json.loads(data)
        except (ValueError, UnicodeDecodeError) as exc:
            raise InvalidRequestError(f"тело запроса — не JSON ({type(exc).__name__})") from None
        if isinstance(raw, dict):
            version = raw.get("schema_version")
            match = _VERSION_RE.fullmatch(version) if isinstance(version, str) else None
            if match is not None and match.group(1) != SUPPORTED_MAJOR:
                raise InvalidRequestError(
                    f"неподдерживаемая версия запроса schema_version={version!r}: сервис поддерживает "
                    f"{SUPPORTED_MAJOR}.x"
                )
        try:
            return RecognitionRequest.model_validate(raw)
        except ValidationError as exc:
            raise InvalidRequestError(
                f"запрос не соответствует recognition-request.schema.json: {_validation_summary(exc)}"
            ) from None

    async def _reject(self, msg: InboundMessage, reason: str, started: float) -> None:
        """`SCHEMA_MISMATCH` без ack. Без читаемых `job_id`/`file_id`/`attempt` конверт не собрать — лог и сброс."""
        raw = self._raw_dict(msg.data)
        job_id = self._readable_uuid(raw, "job_id")
        file_id = self._readable_uuid(raw, "file_id")
        attempt = self._readable_attempt(raw, msg.headers, job_id) if job_id is not None else None
        if job_id is None or file_id is None or attempt is None:
            logger.error(
                "Запрос отброшен: не проходит контракт, а job_id/file_id/attempt не читаемы — результат не "
                "публикуется. error_category=schema_mismatch, reason={reason}, size={size}",
                reason=reason,
                size=len(msg.data),
            )
            return
        logger.error(
            "Запрос не проходит контракт: публикуется SCHEMA_MISMATCH без ack. error_category=schema_mismatch, "
            "job_id={job_id}, reason={reason}",
            job_id=job_id,
            reason=reason,
        )
        envelope = self._envelope(
            job_id=job_id,
            file_id=file_id,
            status=MessageStatus.ERROR,
            error=MessageError(code=ErrorCode.SCHEMA_MISMATCH, message=_one_line(reason)),
            metrics=self._metrics(None, started),
        )
        await self._publish(envelope, result_msg_id(job_id, attempt), job_id)

    @staticmethod
    def _raw_dict(data: bytes) -> dict[str, Any]:
        try:
            raw = json.loads(data)
        except (ValueError, UnicodeDecodeError):
            return {}
        return raw if isinstance(raw, dict) else {}

    @staticmethod
    def _readable_uuid(raw: Mapping[str, Any], name: str) -> str | None:
        value = raw.get(name)
        return value if isinstance(value, str) and _UUID_RE.fullmatch(value) else None

    @staticmethod
    def _readable_attempt(raw: Mapping[str, Any], headers: Mapping[str, str] | None, job_id: str) -> int | None:
        """`attempt` из тела (целое ≥ 1), иначе из заголовка запроса `Nats-Msg-Id = <job_id>:<attempt>`."""
        attempt = raw.get("attempt")
        if isinstance(attempt, int) and not isinstance(attempt, bool) and attempt >= 1:
            return attempt
        header = (headers or {}).get(NATS_MSG_ID_HEADER, "")
        prefix, _, tail = header.rpartition(":")
        if prefix.lower() == job_id.lower() and tail.isdigit() and int(tail) >= 1:
            return int(tail)
        return None

    async def _publish_ack(self, request: RecognitionRequest) -> None:
        envelope = self._envelope(job_id=request.job_id, file_id=request.file_id, status=MessageStatus.PROCESSING)
        await self._publish(envelope, ack_msg_id(request.job_id, request.attempt), str(request.job_id))

    # --- обработка задания ------------------------------------------------------------------------------------------
    async def _run(self) -> None:
        while True:
            job = await self._queue.get()
            job_id = str(job.request.job_id)
            try:
                await self._process(job.request)
            except Exception:
                # Страховка: _process сам превращает ошибки в конверты; сюда попадает только непредвиденное
                logger.exception("Сбой обработки задания. error_category=worker_error, job_id={job_id}", job_id=job_id)
            finally:
                self._finish(job_id)
                self._queue.task_done()

    def _finish(self, job_id: str) -> None:
        self._in_progress.discard(job_id)
        self._done[job_id] = None
        self._done.move_to_end(job_id)
        while len(self._done) > self._done_cache_size:
            self._done.popitem(last=False)

    async def _process(self, request: RecognitionRequest) -> None:
        started = self._clock()
        job_id = str(request.job_id)
        envelope = await self._build_result(request, started)
        logger.info(
            "Задание обработано. job_id={job_id}, status={status}, error_code={error_code}",
            job_id=job_id,
            status=envelope.status.value,
            error_code=envelope.error.code.value if envelope.error else None,
        )
        await self._publish(envelope, result_msg_id(request.job_id, request.attempt), job_id)

    async def _build_result(self, request: RecognitionRequest, started: float) -> RecognitionResultMessage:
        """Конвейер S3 → парсер → S3; любой сбой — конверт `error` по таблице D-5."""
        try:
            data = await self._storage.get_bytes(request.s3.bucket, request.s3.key)
        except Exception as exc:
            logger.warning(
                "Не удалось прочитать исходный файл. error_category=s3_error, job_id={job_id}, key={key}, "
                "error={error}",
                job_id=request.job_id,
                key=request.s3.key,
                error=_describe(exc),
            )
            message = f"не удалось прочитать s3://{request.s3.bucket}/{request.s3.key}: {_describe(exc)}"
            return self._error(request, ErrorCode.FILE_READ_ERROR, message, None, started)

        try:
            outcome: ParseOutcome = await asyncio.to_thread(self._parser.parse, data, request.file_type)
        except Exception as exc:
            logger.opt(exception=exc).error(
                "Исключение парсера. error_category=parser_error, job_id={job_id}", job_id=request.job_id
            )
            return self._error(request, ErrorCode.UNKNOWN_ERROR, f"ошибка разбора: {_describe(exc)}", None, started)

        if outcome.status is ResultStatus.ERROR:
            error = outcome.error or MessageError(
                code=ErrorCode.UNKNOWN_ERROR, message="парсер вернул ошибку без описания"
            )
            return self._envelope(
                job_id=request.job_id,
                file_id=request.file_id,
                status=MessageStatus.ERROR,
                error=error,
                metrics=self._metrics(outcome.metrics, started),
            )

        key = result_key(request.file_id, request.job_id)
        try:
            metrics = self._metrics(outcome.metrics, started)
            result = RecognitionResult.model_validate(
                {
                    "schema_version": CONTRACT_VERSION,
                    "job_id": request.job_id,
                    "file_id": request.file_id,
                    "service": self._service,
                    "status": ResultStatus(outcome.status),
                    "metrics": metrics,
                    "documents": list(outcome.documents),
                }
            )
            await self._storage.put_json(request.s3.bucket, key, result.to_wire())
        except Exception as exc:
            logger.opt(exception=exc).error(
                "Не удалось записать файл результата. error_category=s3_error, job_id={job_id}, key={key}",
                job_id=request.job_id,
                key=key,
            )
            message = f"не удалось записать результат s3://{request.s3.bucket}/{key}: {_describe(exc)}"
            return self._error(request, ErrorCode.UNKNOWN_ERROR, message, outcome.metrics, started)

        return self._envelope(
            job_id=request.job_id,
            file_id=request.file_id,
            status=MessageStatus(outcome.status.value),
            result_s3=S3Ref(bucket=request.s3.bucket, key=key),
            metrics=metrics,
        )

    def _error(
        self, request: RecognitionRequest, code: ErrorCode, message: str, metrics: Metrics | None, started: float
    ) -> RecognitionResultMessage:
        return self._envelope(
            job_id=request.job_id,
            file_id=request.file_id,
            status=MessageStatus.ERROR,
            error=MessageError(code=code, message=_one_line(message)),
            metrics=self._metrics(metrics, started),
        )

    def _metrics(self, metrics: Metrics | None, started: float) -> Metrics:
        """Метрики парсера + `duration_ms` (всегда, ≥ 0)."""
        duration_ms = max(0, round((self._clock() - started) * 1000))
        fields = metrics.model_dump(exclude_unset=True) if metrics is not None else {}
        return Metrics.model_validate({**fields, "duration_ms": duration_ms})

    def _envelope(  # noqa: PLR0913 - поля конверта именованные
        self,
        *,
        job_id: object,
        file_id: object,
        status: MessageStatus,
        error: MessageError | None = None,
        result_s3: S3Ref | None = None,
        metrics: Metrics | None = None,
    ) -> RecognitionResultMessage:
        # Необязательные поля без null в схеме передаются, только если заданы (to_wire выводит заданные)
        fields: dict[str, Any] = {
            "schema_version": CONTRACT_VERSION,
            "job_id": job_id,
            "file_id": file_id,
            "status": status,
            "service": self._service,
        }
        if result_s3 is not None:
            fields["result_s3"] = result_s3
            fields["result_schema_version"] = CONTRACT_VERSION
        if error is not None:
            fields["error"] = error
        if metrics is not None:
            fields["metrics"] = metrics
        return RecognitionResultMessage.model_validate(fields)

    async def _publish(self, envelope: RecognitionResultMessage, msg_id: str, job_id: str) -> None:
        """Публикация конверта; исчерпанные повторы и прочие сбои — error-лог, не исключение (D-5)."""
        payload = json.dumps(envelope.to_wire(), ensure_ascii=False).encode("utf-8")
        try:
            await self._publisher.publish(RESULT_SUBJECT, payload, msg_id)
        except NatsPublishError as exc:
            logger.error(
                "Конверт не опубликован после всех повторов; ядро закроет задание по reply_deadline. "
                "error_category=nats_error, job_id={job_id}, msg_id={msg_id}, status={status}, error={error}",
                job_id=job_id,
                msg_id=msg_id,
                status=envelope.status.value,
                error=str(exc),
            )
        except Exception as exc:
            logger.opt(exception=exc).error(
                "Сбой публикации конверта. error_category=nats_error, job_id={job_id}, msg_id={msg_id}",
                job_id=job_id,
                msg_id=msg_id,
            )