duty-teller/duty_teller/importers/duty_schedule.py

"""Parser for duty-schedule JSON format. No DB access."""

import json
from dataclasses import dataclass
from datetime import date, timedelta

# Символы дежурства в ячейке duty (CSV с разделителем ;)
DUTY_MARKERS = frozenset({"б", "Б", "в", "В"})
UNAVAILABLE_MARKER = "Н"
VACATION_MARKER = "О"

# Limits to avoid abuse and unreasonable input.
MAX_SCHEDULE_ROWS = 500
MAX_FULL_NAME_LENGTH = 200
MAX_DUTY_STRING_LENGTH = 10000


@dataclass
class DutyScheduleEntry:
    """One person's schedule: full_name and three lists of dates by event type."""

    full_name: str
    duty_dates: list[date]
    unavailable_dates: list[date]
    vacation_dates: list[date]


@dataclass
class DutyScheduleResult:
    """Parsed duty schedule: start_date, end_date, and per-person entries."""

    start_date: date
    end_date: date
    entries: list[DutyScheduleEntry]


class DutyScheduleParseError(Exception):
    """Invalid or missing fields in duty-schedule JSON."""

    pass


def parse_duty_schedule(raw_bytes: bytes) -> DutyScheduleResult:
    """Parse duty-schedule JSON into DutyScheduleResult.

    Expects meta.start_date (YYYY-MM-DD) and schedule (array). For each schedule
    item: name (required), duty string with ';' separator; index i = start_date + i days.
    Cell values: в/В/б/Б => duty, Н => unavailable, О => vacation; rest ignored.

    Args:
        raw_bytes: UTF-8 encoded JSON bytes.

    Returns:
        DutyScheduleResult with start_date, end_date, and entries (per-person dates).

    Raises:
        DutyScheduleParseError: On invalid JSON, missing/invalid meta or schedule,
            or invalid item fields.
    """
    try:
        data = json.loads(raw_bytes.decode("utf-8"))
    except (json.JSONDecodeError, UnicodeDecodeError) as e:
        raise DutyScheduleParseError(f"Invalid JSON or encoding: {e}") from e

    meta = data.get("meta")
    if not meta or not isinstance(meta, dict):
        raise DutyScheduleParseError("Missing or invalid 'meta'")

    start_str = meta.get("start_date")
    if not start_str or not isinstance(start_str, str):
        raise DutyScheduleParseError("Missing or invalid meta.start_date")
    try:
        start_date = date.fromisoformat(start_str.strip())
    except ValueError as e:
        raise DutyScheduleParseError(f"Invalid meta.start_date: {start_str}") from e

    # Reject dates outside current year ± 1.
    today = date.today()
    min_year = today.year - 1
    max_year = today.year + 1
    if not (min_year <= start_date.year <= max_year):
        raise DutyScheduleParseError(
            f"meta.start_date year must be between {min_year} and {max_year}"
        )

    schedule = data.get("schedule")
    if not isinstance(schedule, list):
        raise DutyScheduleParseError("Missing or invalid 'schedule' (must be array)")

    if len(schedule) > MAX_SCHEDULE_ROWS:
        raise DutyScheduleParseError(
            f"schedule has too many rows (max {MAX_SCHEDULE_ROWS})"
        )

    max_days = 0
    entries: list[DutyScheduleEntry] = []

    for row in schedule:
        if not isinstance(row, dict):
            raise DutyScheduleParseError("schedule item must be an object")
        name = row.get("name")
        if name is None or not isinstance(name, str):
            raise DutyScheduleParseError("schedule item must have 'name' (string)")
        full_name = name.strip()
        if not full_name:
            raise DutyScheduleParseError("schedule item 'name' cannot be empty")
        if len(full_name) > MAX_FULL_NAME_LENGTH:
            raise DutyScheduleParseError(
                f"schedule item 'name' must not exceed {MAX_FULL_NAME_LENGTH} characters"
            )

        duty_str = row.get("duty")
        if duty_str is None:
            duty_str = ""
        if not isinstance(duty_str, str):
            raise DutyScheduleParseError("schedule item 'duty' must be string")
        if len(duty_str) > MAX_DUTY_STRING_LENGTH:
            raise DutyScheduleParseError(
                f"schedule item 'duty' must not exceed {MAX_DUTY_STRING_LENGTH} characters"
            )

        cells = [c.strip() for c in duty_str.split(";")]
        max_days = max(max_days, len(cells))

        duty_dates: list[date] = []
        unavailable_dates: list[date] = []
        vacation_dates: list[date] = []
        for i, cell in enumerate(cells):
            d = start_date + timedelta(days=i)
            if cell in DUTY_MARKERS:
                duty_dates.append(d)
            elif cell == UNAVAILABLE_MARKER:
                unavailable_dates.append(d)
            elif cell == VACATION_MARKER:
                vacation_dates.append(d)
        entries.append(
            DutyScheduleEntry(
                full_name=full_name,
                duty_dates=duty_dates,
                unavailable_dates=unavailable_dates,
                vacation_dates=vacation_dates,
            )
        )

    if max_days == 0:
        end_date = start_date
    else:
        end_date = start_date + timedelta(days=max_days - 1)

    if not (min_year <= end_date.year <= max_year):
        raise DutyScheduleParseError(
            f"Computed end_date year must be between {min_year} and {max_year}"
        )

    return DutyScheduleResult(start_date=start_date, end_date=end_date, entries=entries)