tools/validate.py - develop - zemars/it-dashboard

tools/validate.py
187 строк7 KB
Дмитрий
fix(docs): актуализировать risk_log и docstring init_db
26 май 2026, 22:07
26 май 2026, 22:079981d43
"""Standalone-валидатор CSV-файлов.

Автор: Дмитрий Жещинский — аналитика + валидация.

Применяет правила V-01..V-06 (см. spec) к CSV-файлам и печатает таблицу
ошибок. Используется в CI и при отладке перед загрузкой через UI.

Запуск:
    python3 tools/validate.py data/C19/T1
    python3 tools/validate.py data/C19/invalid --files tickets_bad_priority.csv
"""
from __future__ import annotations

import argparse
import csv
import sys
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Set, Tuple


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


@dataclass
class ValidationError:
    file: str
    row: int  # номер строки в CSV (с учётом заголовка = 2 для первой данных)
    code: str  # V-01..V-06
    message: str


def parse_dt(raw: Optional[str]) -> Optional[datetime]:
    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


def validate_tickets(path: Path) -> List[ValidationError]:
    errors: List[ValidationError] = []
    if not path.exists():
        return errors
    with path.open(encoding="utf-8") as f:
        reader = csv.DictReader(f)
        for idx, row in enumerate(reader, start=2):
            priority = (row.get("priority") or "").strip()
            if priority not in ALLOWED_PRIORITIES:
                errors.append(
                    ValidationError(path.name, idx, "V-06", f"недопустимый priority={priority!r}")
                )

            created_at = parse_dt(row.get("created_at"))
            closed_at = parse_dt(row.get("closed_at"))
            if created_at is None:
                errors.append(
                    ValidationError(path.name, idx, "V-05", "некорректное значение created_at")
                )
            elif closed_at is not None and closed_at < created_at:
                errors.append(
                    ValidationError(path.name, idx, "V-05", "closed_at раньше created_at")
                )
    return errors


def validate_events(path: Path, known_tickets: Set[str]) -> List[ValidationError]:
    errors: List[ValidationError] = []
    if not path.exists():
        return errors

    # Сначала собираем все события по тикетам, чтобы проверить V-02
    per_ticket: Dict[str, List[Tuple[int, str, Optional[datetime]]]] = {}
    with path.open(encoding="utf-8") as f:
        reader = csv.DictReader(f)
        for idx, row in enumerate(reader, start=2):
            tid = (row.get("ticket_id") or "").strip()
            etype = (row.get("event") or row.get("event_type") or "").strip()
            ts = parse_dt(row.get("ts"))

            if etype not in ALLOWED_EVENT_TYPES:
                errors.append(
                    ValidationError(path.name, idx, "V-06", f"недопустимый event={etype!r}")
                )
                continue

            if known_tickets and tid not in known_tickets:
                errors.append(
                    ValidationError(path.name, idx, "V-03", f"событие ссылается на неизвестный ticket_id={tid!r}")
                )
                continue

            per_ticket.setdefault(tid, []).append((idx, etype, ts))

    # V-02: ровно один first_response и один resolved
    for tid, items in per_ticket.items():
        counts: Dict[str, int] = {}
        for _, et, _ in items:
            counts[et] = counts.get(et, 0) + 1
        for et in ("first_response", "resolved"):
            if counts.get(et, 0) > 1:
                # репортим вторую (и далее) запись
                seen = 0
                for ridx, t, _ in items:
                    if t == et:
                        seen += 1
                        if seen > 1:
                            errors.append(
                                ValidationError(
                                    path.name, ridx, "V-02",
                                    f"дубликат события {et!r} для тикета {tid}",
                                )
                            )
    return errors


def validate_dir(directory: Path, files_filter: Optional[List[str]] = None) -> List[ValidationError]:
    """Валидируем все CSV в директории. Возвращает список ошибок."""
    all_errors: List[ValidationError] = []

    tickets_path = directory / "tickets.csv"
    events_path = directory / "events.csv"
    # тестовые файлы invalid/* тоже поддерживаем
    if files_filter:
        files = [directory / name for name in files_filter]
    else:
        files = list(directory.glob("*.csv"))

    # Сначала валидируем все tickets-подобные файлы
    known_ticket_ids: Set[str] = set()
    for p in files:
        if "ticket" in p.name:
            all_errors.extend(validate_tickets(p))
            # собираем ticket_id для проверки V-03 в events
            try:
                with p.open(encoding="utf-8") as f:
                    for row in csv.DictReader(f):
                        tid = (row.get("ticket_id") or "").strip()
                        if tid:
                            known_ticket_ids.add(tid)
            except Exception:
                pass

    for p in files:
        if "event" in p.name:
            all_errors.extend(validate_events(p, known_ticket_ids))

    return all_errors


def print_report(errors: List[ValidationError]) -> None:
    if not errors:
        print("✓ Ошибок не найдено")
        return
    print(f"Найдено ошибок: {len(errors)}\n")
    print(f"{'CODE':<6} {'FILE':<32} {'ROW':<5} MESSAGE")
    print("-" * 90)
    for e in errors:
        print(f"{e.code:<6} {e.file:<32} {e.row:<5} {e.message}")


def main() -> int:
    parser = argparse.ArgumentParser(description="Валидатор CSV-файлов IT Dashboard")
    parser.add_argument("path", type=Path, help="Папка с CSV-файлами")
    parser.add_argument(
        "--files", nargs="+", default=None,
        help="Ограничить валидацию указанными именами файлов",
    )
    args = parser.parse_args()
    if not args.path.is_dir():
        print(f"Ошибка: {args.path} не директория", file=sys.stderr)
        return 2
    errors = validate_dir(args.path, args.files)
    print_report(errors)
    return 1 if errors else 0


if __name__ == "__main__":
    sys.exit(main())