"""Contracts and validation for editable Booking Excel extraction drafts.""" from __future__ import annotations import re import unicodedata from dataclasses import dataclass from datetime import datetime from typing import Optional from .excel import BookingExcelError DRAFT_ID_PATTERN = re.compile(r"^bookingdraft-[0-9a-f]{32}$") MAX_REVIEW_ROOM_TYPE_LENGTH = 128 MAX_REVIEW_DELETE_ITEMS = 50 def validate_draft_id(value: object) -> str: if not isinstance(value, str) or DRAFT_ID_PATTERN.fullmatch(value) is None: raise BookingExcelError( "BOOKING_EXCEL_DRAFT_INVALID", "Excel 提取草稿编号无效", 400, ) return value def validate_review_item_id(value: object) -> int: if isinstance(value, bool) or not isinstance(value, int) or value <= 0: raise BookingExcelError( "BOOKING_EXCEL_REVIEW_ITEM_INVALID", "待确认记录编号无效", 400, ) return value def validate_review_item_ids(value: object) -> tuple[int, ...]: if ( not isinstance(value, (list, tuple)) or not value or len(value) > MAX_REVIEW_DELETE_ITEMS ): raise BookingExcelError( "BOOKING_EXCEL_REVIEW_ITEMS_INVALID", f"请选择 1 至 {MAX_REVIEW_DELETE_ITEMS} 条有效记录", 400, ) item_ids = tuple(validate_review_item_id(item_id) for item_id in value) if len(set(item_ids)) != len(item_ids): raise BookingExcelError( "BOOKING_EXCEL_REVIEW_ITEMS_INVALID", "所选记录不能重复", 400, ) return item_ids def validate_review_room_type(value: object) -> str: if not isinstance(value, str): raise BookingExcelError( "BOOKING_EXCEL_ROOM_TYPE_INVALID", "请填写房型", 400, ) normalized = unicodedata.normalize("NFKC", value).strip() normalized = re.sub(r"\s+", " ", normalized) if ( not normalized or len(normalized) > MAX_REVIEW_ROOM_TYPE_LENGTH or any(ord(char) < 32 or ord(char) == 127 for char in normalized) ): raise BookingExcelError( "BOOKING_EXCEL_ROOM_TYPE_INVALID", "请填写有效房型", 400, ) return normalized def validate_review_quantity(value: object) -> int: if isinstance(value, bool) or not isinstance(value, int) or not 1 <= value <= 9999: raise BookingExcelError( "BOOKING_EXCEL_QUANTITY_INVALID", "房间数量必须是 1 至 9999 的整数", 400, ) return value @dataclass(frozen=True) class BookingDraftSummary: draft_id: str filename: str status: str source_rows: int worksheet_count: int distinct_group_codes: int extracted_items: int confirmed_items: int pending_items: int deleted_items: int confirmed_room_quantity: int created_at: Optional[datetime] updated_at: Optional[datetime] def to_dict(self) -> dict[str, object]: return { "draft_id": self.draft_id, "filename": self.filename, "status": self.status, "source_rows": self.source_rows, "worksheet_count": self.worksheet_count, "distinct_group_codes": self.distinct_group_codes, "extracted_items": self.extracted_items, "confirmed_items": self.confirmed_items, "pending_items": self.pending_items, "deleted_items": self.deleted_items, "confirmed_room_quantity": self.confirmed_room_quantity, "created_at": self.created_at.isoformat() if self.created_at else None, "updated_at": self.updated_at.isoformat() if self.updated_at else None, } @dataclass(frozen=True) class BookingDraftItem: item_id: int draft_id: str worksheet: str row_no: int item_no: int tour_code: str room_type_raw: str room_type_code: Optional[str] quantity: int review_status: str automatic: bool source_fragment: str updated_at: Optional[datetime] def to_dict(self) -> dict[str, object]: return { "item_id": self.item_id, "draft_id": self.draft_id, "worksheet": self.worksheet, "row_no": self.row_no, "item_no": self.item_no, "tour_code": self.tour_code, "room_type_raw": self.room_type_raw, "room_type": self.room_type_code, "quantity": self.quantity, "review_status": self.review_status, "automatic": self.automatic, "source_fragment": self.source_fragment, "updated_at": self.updated_at.isoformat() if self.updated_at else None, } @dataclass(frozen=True) class BookingDraftPage: summary: BookingDraftSummary items: tuple[BookingDraftItem, ...] total: int limit: int offset: int def to_dict(self) -> dict[str, object]: return { "summary": self.summary.to_dict(), "items": [item.to_dict() for item in self.items], "pagination": { "total": self.total, "limit": self.limit, "offset": self.offset, "has_previous": self.offset > 0, "has_next": self.offset + len(self.items) < self.total, }, }