"""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)