ses-monitor/ses_monitor/utils.py

"""
Вспомогательные утилиты: логирование, работа с JSON-путями, тайминг.
"""

from __future__ import annotations

import logging
import re
import time
from contextlib import contextmanager
from typing import Any, Dict, Generator, List, Optional, Tuple, Union

from ses_monitor.config import LoggingConfig


def setup_logging(cfg: LoggingConfig) -> logging.Logger:
    """Настраивает и возвращает корневой логгер проекта.

    Args:
        cfg: Конфигурация логирования.

    Returns:
        Настроенный логгер 'ses_monitor'.
    """
    logger = logging.getLogger("ses_monitor")
    logger.setLevel(getattr(logging, cfg.level.upper(), logging.INFO))

    # Убираем существующие обработчики, чтобы не дублировать
    logger.handlers.clear()

    formatter = logging.Formatter(cfg.format, datefmt="%Y-%m-%d %H:%M:%S")

    # Всегда пишем в stdout для systemd
    console = logging.StreamHandler()
    console.setFormatter(formatter)
    logger.addHandler(console)

    # Опционально: файловый лог
    if cfg.file:
        from logging.handlers import RotatingFileHandler

        file_handler = RotatingFileHandler(
            cfg.file, maxBytes=10 * 1024 * 1024, backupCount=5,
        )
        file_handler.setFormatter(formatter)
        logger.addHandler(file_handler)

    return logger


def get_json_value(data: Union[Dict[str, Any], List[Any]], path: str) -> Any:
    """Извлекает значение из вложенной JSON-структуры по пути.

    Args:
        data: Словарь или список с данными.
        path: Путь в формате "key.subkey[0].field" или "items[0].name".

    Returns:
        Значение по указанному пути, или None если путь не существует.

    Examples:
        >>> get_json_value({"a": {"b": [1, 2]}}, "a.b[0]")
        1
        >>> get_json_value({"answer": [{"text": "да"}]}, "answer[0].text")
        'да'
    """
    try:
        # Разбираем путь: key, [0], .subkey
        tokens = re.findall(r"(?:^|\.)(\w+)|\[(\d+)\]", path)
        current: Any = data

        for key, index in tokens:
            if key:
                if isinstance(current, dict):
                    current = current.get(key)
                else:
                    return None
            elif index is not None:
                idx = int(index)
                if isinstance(current, list) and idx < len(current):
                    current = current[idx]
                else:
                    return None

        return current
    except (KeyError, IndexError, TypeError, AttributeError):
        return None


def check_expectation(
    data: Dict[str, Any],
    path: str,
    condition: str,
    value: Any = None,
) -> Tuple[bool, str]:
    """Проверяет одно условие ожидания для JSON-ответа.

    Args:
        data: JSON-ответ от SES API.
        path: Путь к полю в JSON (формат "answer[0].messages[0]").
        condition: Тип проверки:
            - "equals": поле == value
            - "contains": поле содержит value
            - "not_contains": поле не содержит value
            - "not_empty": поле не пустое
            - "regex": поле соответствует регулярному выражению в value
            - "exists": поле существует и не None
        value: Значение для сравнения (для equals/contains/not_contains/regex).

    Returns:
        (успех, сообщение_об_ошибке). При успехе сообщение пустое.
    """
    actual = get_json_value(data, path)

    if condition == "exists":
        if actual is not None:
            return True, ""
        return False, f"Поле '{path}' не найдено в ответе"

    if condition == "not_empty":
        if actual is not None and actual != "" and actual != [] and actual != {}:
            return True, ""
        return False, f"Поле '{path}' пустое: {actual!r}"

    if actual is None:
        return False, f"Поле '{path}' не найдено в ответе"

    if isinstance(actual, str) and value is not None:
        actual_str = actual.lower()
        if not isinstance(value, str):
            value = str(value)
        value_str = value.lower()

        if condition == "contains":
            if value_str in actual_str:
                return True, ""
            return False, (
                f"Поле '{path}' = {actual_str[:100]!r} "
                f"не содержит {value_str[:100]!r}"
            )

        if condition == "not_contains":
            if value_str not in actual_str:
                return True, ""
            return False, (
                f"Поле '{path}' = {actual_str[:100]!r} "
                f"содержит запрещённое {value_str[:100]!r}"
            )

        if condition == "regex":
            try:
                if re.search(str(value), actual):
                    return True, ""
                return False, (
                    f"Поле '{path}' = {actual[:100]!r} "
                    f"не соответствует regex {value!r}"
                )
            except re.error as e:
                return False, f"Ошибка regex '{value}': {e}"

    if condition == "equals":
        # Приводим к строке для сравнения
        actual_conv = actual
        value_conv = value
        if not isinstance(actual_conv, type(value_conv)):
            # Пробуем сравнить через строковое представление
            if str(actual_conv) == str(value_conv):
                return True, ""
            return False, (
                f"Поле '{path}' = {actual_conv!r} "
                f"не равно {value_conv!r}"
            )
        if actual_conv == value_conv:
            return True, ""
        return False, (
            f"Поле '{path}' = {actual_conv!r} "
            f"не равно {value_conv!r}"
        )

    return False, f"Неизвестное условие проверки: {condition}"


def check_all_expectations(
    data: Dict[str, Any],
    expectations: List[Dict[str, Any]],
) -> Tuple[bool, List[str]]:
    """Проверяет все ожидания для одного шага сценария.

    Args:
        data: JSON-ответ от SES API.
        expectations: Список условий — каждый dict с ключами
            path, condition_type_1, condition_type_2, ...

    Returns:
        (все_прошли, список_сообщений_об_ошибках).
    """
    errors: List[str] = []

    for exp in expectations:
        path = exp.get("path", "")
        if not path:
            errors.append("Ожидание без указания path")
            continue

        for condition, value in exp.items():
            if condition == "path":
                continue  # path — не условие

            ok, msg = check_expectation(data, path, condition, value)
            if not ok:
                errors.append(msg)

    return len(errors) == 0, errors


@contextmanager
def timer() -> Generator[Any, None, None]:
    """Контекстный менеджер для измерения времени выполнения блока.

    Yields:
        Словарь, в который будет записано прошедшее время (в секундах)
        по ключам 'start' и 'elapsed'.

    Usage:
        >>> with timer() as t:
        ...     time.sleep(0.1)
        >>> print(f"{t['elapsed']:.2f}s")
    """
    result: Dict[str, float] = {"start": time.monotonic()}
    try:
        yield result
    finally:
        result["elapsed"] = time.monotonic() - result["start"]