from datetime import datetime from typing import Any, List, Optional from pydantic import BaseModel, ConfigDict, Field, field_validator from app.core.utils import parse_datetime def _coerce_optional_datetime(value: Any) -> Optional[datetime]: """兼容前端 date-picker 的 YYYY-MM-DD 与完整日期时间字符串。""" if value is None or value == "": return None if isinstance(value, datetime): return value if isinstance(value, str): parsed = parse_datetime(value.strip()) if parsed is not None: return parsed raise ValueError("无效的日期格式") return value class CellProductBatchItem(BaseModel): """批量入库时每一份产品的编码与储物格。""" model_config = ConfigDict(populate_by_name=True) product_code: str = Field(..., alias="productCode") storage_slot_id: Optional[str] = Field(None, alias="storageSlotId") passage: Optional[str] = None spec_name: Optional[str] = Field(None, alias="specName") expected_cell_count: Optional[float] = Field(None, alias="expectedCellCount") viability: Optional[float] = None @field_validator("passage") @classmethod def validate_item_passage(cls, v): if v is not None and v != "": valid = {f"P{i}" for i in range(1, 9)} if v.upper() not in valid: raise ValueError("代次必须为 P1~P8") return v.upper() return v @field_validator("viability") @classmethod def validate_item_viability(cls, v): if v is not None and (v < 0 or v > 100): raise ValueError("活性必须在 0~100 之间") return v class CellProductCreate(BaseModel): model_config = ConfigDict(populate_by_name=True) product_code: Optional[str] = Field(None, alias="productCode") product_name: str = Field(..., alias="productName") product_type_id: Optional[str] = Field(None, alias="productTypeId") spec_id: Optional[str] = Field(None, alias="specId") spec_name: Optional[str] = Field(None, alias="specName") batch_no: Optional[str] = Field(None, alias="batchNo") storage_slot_id: Optional[str] = Field(None, alias="storageSlotId") passage: Optional[str] = None freeze_date: Optional[datetime] = Field(None, alias="freezeDate") expected_cell_count: Optional[float] = Field(None, alias="expectedCellCount") viability: Optional[float] = None product_status: Optional[str] = Field("frozen", alias="productStatus") source_type: Optional[str] = Field("manual", alias="sourceType") quantity: Optional[float] = None unit: Optional[str] = None expiry_date: Optional[datetime] = Field(None, alias="expiryDate") product_images: Optional[str] = Field(None, alias="productImages") remark: Optional[str] = None items: Optional[List[CellProductBatchItem]] = Field(None, alias="items") @field_validator("freeze_date", "expiry_date", mode="before") @classmethod def coerce_date_fields(cls, v): return _coerce_optional_datetime(v) @field_validator("viability") @classmethod def validate_viability(cls, v): if v is not None and (v < 0 or v > 100): raise ValueError("活性必须在 0~100 之间") return v @field_validator("passage") @classmethod def validate_passage(cls, v): if v is not None: valid = {f"P{i}" for i in range(1, 9)} if v.upper() not in valid: raise ValueError("代次必须为 P1~P8") return v.upper() return v class CellProductUpdate(BaseModel): model_config = ConfigDict(populate_by_name=True) product_code: Optional[str] = Field(None, alias="productCode") product_name: Optional[str] = Field(None, alias="productName") product_type_id: Optional[str] = Field(None, alias="productTypeId") spec_id: Optional[str] = Field(None, alias="specId") spec_name: Optional[str] = Field(None, alias="specName") batch_no: Optional[str] = Field(None, alias="batchNo") storage_slot_id: Optional[str] = Field(None, alias="storageSlotId") passage: Optional[str] = None freeze_date: Optional[datetime] = Field(None, alias="freezeDate") expected_cell_count: Optional[float] = Field(None, alias="expectedCellCount") viability: Optional[float] = None product_status: Optional[str] = Field(None, alias="productStatus") quantity: Optional[float] = None unit: Optional[str] = None expiry_date: Optional[datetime] = Field(None, alias="expiryDate") product_images: Optional[str] = Field(None, alias="productImages") remark: Optional[str] = None @field_validator("freeze_date", "expiry_date", mode="before") @classmethod def coerce_date_fields(cls, v): return _coerce_optional_datetime(v) @field_validator("viability") @classmethod def validate_viability(cls, v): if v is not None and (v < 0 or v > 100): raise ValueError("活性必须在 0~100 之间") return v @field_validator("passage") @classmethod def validate_passage(cls, v): if v is not None: valid = {f"P{i}" for i in range(1, 9)} if v.upper() not in valid: raise ValueError("代次必须为 P1~P8") return v.upper() return v @field_validator("product_status") @classmethod def validate_status(cls, v): allowed = ("frozen", "sold", "used", "voided", "in_production", "thawed") if v is not None and v not in allowed: raise ValueError(f"状态必须为 {'/'.join(allowed)}") return v class CellProductListQuery(BaseModel): model_config = ConfigDict(populate_by_name=True) page: int = 1 page_size: int = Field(20, alias="pageSize") keyword: Optional[str] = None product_code: Optional[str] = Field(None, alias="productCode") product_name: Optional[str] = Field(None, alias="productName") product_type_id: Optional[str] = Field(None, alias="productTypeId") product_status: Optional[str] = Field(None, alias="productStatus") passage: Optional[str] = None source_type: Optional[str] = Field(None, alias="sourceType")