backend/app/utils/validators.py - develop - zemars/it-dashboard

backend/app/utils/validators.py
113 строк5 KB
Дмитрий
fix(docs): актуализировать risk_log и docstring init_db
26 май 2026, 22:07
26 май 2026, 22:079981d43
# Автор: Дмитрий Жещинский — правила валидации CSV-данных
# при импорте. Реализуют требования V-01..V-06 из спецификации.
#
# Каждая функция возвращает None если строка прошла проверку, либо строку с
# причиной отклонения. Сервис импорта собирает все причины в errors[].

from datetime import datetime
from typing import Iterable, List, Optional


ALLOWED_PRIORITIES = {"P1", "P2", "P3"}
ALLOWED_EVENT_TYPES = {"assigned", "first_response", "resolved"}


def parse_dt(raw: Optional[str]) -> Optional[datetime]:
    """Парсим datetime из CSV. Поддерживаем ISO (с T и пробелом).

    Возвращает None если значение пустое или невалидное.
    """
    if raw is None:
        return None
    s = str(raw).strip()
    if not s or s.lower() in {"nan", "none", "null"}:
        return None
    # Несколько форматов на всякий случай
    for fmt in ("%Y-%m-%dT%H:%M:%S", "%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M", "%Y-%m-%d"):
        try:
            return datetime.strptime(s, fmt)
        except ValueError:
            continue
    return None


# ---- V-01: хронология событий ----
def validate_chronology(
    created_at: datetime,
    first_response_ts: Optional[datetime],
    resolved_ts: Optional[datetime],
    closed_at: Optional[datetime],
) -> Optional[str]:
    """created_at <= first_response_ts <= resolved_ts <= closed_at.

    Если какой-то из ts отсутствует — он не участвует в сравнении.
    """
    chain = [created_at, first_response_ts, resolved_ts, closed_at]
    chain = [x for x in chain if x is not None]
    for i in range(len(chain) - 1):
        if chain[i] > chain[i + 1]:
            return "V-01: нарушена хронология событий (created_at <= first_response <= resolved <= closed_at)"
    return None


# ---- V-02: уникальность событий на тикет ----
def validate_unique_events(events: Iterable[str]) -> Optional[str]:
    """На тикет ровно одно first_response и ровно одно resolved.

    events — последовательность event_type для одного тикета.
    """
    counts: dict[str, int] = {}
    for et in events:
        counts[et] = counts.get(et, 0) + 1
    dupes: List[str] = []
    for et in ("first_response", "resolved"):
        if counts.get(et, 0) > 1:
            dupes.append(et)
    if dupes:
        return f"V-02: дублирующиеся события: {', '.join(dupes)}"
    return None


# ---- V-03: ссылочная целостность события -> тикет ----
def validate_event_ticket_exists(ticket_id: str, known_ticket_ids: set[str]) -> Optional[str]:
    """events.ticket_id должен существовать в tickets."""
    if ticket_id not in known_ticket_ids:
        return f"V-03: событие ссылается на неизвестный тикет {ticket_id!r}"
    return None


# ---- V-04: наличие SLA-норматива для всех приоритетов ----
def validate_sla_targets_present(
    used_priorities: set[str],
    available_priorities: set[str],
) -> Optional[str]:
    """В sla_targets должны быть записи для всех приоритетов из tickets."""
    missing = used_priorities - available_priorities
    if missing:
        return f"V-04: нет SLA-нормативов для приоритетов: {sorted(missing)}"
    return None


# ---- V-05: корректность временных меток ----
def validate_timestamps(created_at: Optional[datetime], closed_at: Optional[datetime]) -> Optional[str]:
    """created_at должен быть валиден; closed_at >= created_at если задан."""
    if created_at is None:
        return "V-05: некорректное значение created_at"
    if closed_at is not None and closed_at < created_at:
        return "V-05: closed_at раньше created_at"
    return None


# ---- V-06: допустимые priority ----
def validate_priority(priority: str) -> Optional[str]:
    """priority может быть только P1, P2 или P3."""
    if priority not in ALLOWED_PRIORITIES:
        return f"V-06: недопустимый priority={priority!r} (ожидалось P1/P2/P3)"
    return None


def validate_event_type(event_type: str) -> Optional[str]:
    """Дополнительная проверка для events: тип события должен быть из enum."""
    if event_type not in ALLOWED_EVENT_TYPES:
        return f"недопустимый event_type={event_type!r}"
    return None