from datetime import datetime from typing import Any, Generic, Literal, TypeVar from pydantic import BaseModel, ConfigDict, Field, model_validator T = TypeVar("T") # --------------------------------------------------------------------------- # Generic API response wrapper # --------------------------------------------------------------------------- class APIResponse(BaseModel, Generic[T]): """Standard envelope for every API response.""" code: int = 0 message: str = "success" data: T | None = None class PaginationParams(BaseModel): """Query parameters for paginated list endpoints.""" page: int = Field(default=1, ge=1, description="Page number (1-indexed)") page_size: int = Field(default=20, ge=1, le=100, description="Items per page") class PaginatedList(BaseModel, Generic[T]): """Paginated result set.""" items: list[T] total: int page: int page_size: int total_pages: int # --------------------------------------------------------------------------- # Device schemas # --------------------------------------------------------------------------- class DeviceBase(BaseModel): imei: str = Field(..., min_length=15, max_length=20, pattern=r"^\d{15,17}$", description="IMEI number (digits only)") device_type: str = Field(..., max_length=10, description="Device type code (e.g. P240, P241)") name: str | None = Field(None, max_length=100, description="Friendly name") timezone: str = Field(default="+8", max_length=30) language: str = Field(default="cn", max_length=10) class DeviceCreate(DeviceBase): pass class DeviceUpdate(BaseModel): name: str | None = Field(None, max_length=100) iccid: str | None = Field(None, max_length=30) imsi: str | None = Field(None, max_length=20) timezone: str | None = Field(None, max_length=30) language: str | None = Field(None, max_length=10) class DeviceResponse(DeviceBase): model_config = ConfigDict(from_attributes=True) id: int status: str battery_level: int | None = None gsm_signal: int | None = None last_heartbeat: datetime | None = None last_login: datetime | None = None iccid: str | None = None imsi: str | None = None created_at: datetime updated_at: datetime | None = None class DeviceListResponse(APIResponse[PaginatedList[DeviceResponse]]): pass class DeviceSingleResponse(APIResponse[DeviceResponse]): pass # --------------------------------------------------------------------------- # Location Record schemas # --------------------------------------------------------------------------- class LocationRecordBase(BaseModel): device_id: int location_type: str = Field(..., max_length=10) latitude: float | None = Field(None, ge=-90, le=90) longitude: float | None = Field(None, ge=-180, le=180) speed: float | None = None course: float | None = None gps_satellites: int | None = None gps_positioned: bool = False mcc: int | None = None mnc: int | None = None lac: int | None = None cell_id: int | None = None rssi: int | None = None neighbor_cells: list[dict[str, Any]] | None = None wifi_data: list[dict[str, Any]] | None = None report_mode: int | None = None is_realtime: bool = True mileage: float | None = None address: str | None = None raw_data: str | None = None recorded_at: datetime class LocationRecordCreate(LocationRecordBase): pass class LocationRecordResponse(LocationRecordBase): model_config = ConfigDict(from_attributes=True) id: int imei: str | None = None created_at: datetime class LocationRecordFilters(BaseModel): device_id: int | None = None location_type: str | None = None start_time: datetime | None = None end_time: datetime | None = None class LocationListResponse(APIResponse[PaginatedList[LocationRecordResponse]]): pass # --------------------------------------------------------------------------- # Alarm Record schemas # --------------------------------------------------------------------------- class AlarmRecordBase(BaseModel): device_id: int alarm_type: str = Field(..., max_length=30) alarm_source: str | None = Field(None, max_length=20) protocol_number: int latitude: float | None = Field(None, ge=-90, le=90) longitude: float | None = Field(None, ge=-180, le=180) speed: float | None = None course: float | None = None mcc: int | None = None mnc: int | None = None lac: int | None = None cell_id: int | None = None battery_level: int | None = None gsm_signal: int | None = None fence_data: dict[str, Any] | None = None wifi_data: list[dict[str, Any]] | None = None address: str | None = None recorded_at: datetime class AlarmRecordCreate(AlarmRecordBase): pass class AlarmRecordResponse(AlarmRecordBase): model_config = ConfigDict(from_attributes=True) id: int imei: str | None = None acknowledged: bool created_at: datetime class AlarmAcknowledge(BaseModel): acknowledged: bool = True class AlarmRecordFilters(BaseModel): device_id: int | None = None alarm_type: str | None = None acknowledged: bool | None = None start_time: datetime | None = None end_time: datetime | None = None class AlarmListResponse(APIResponse[PaginatedList[AlarmRecordResponse]]): pass # --------------------------------------------------------------------------- # Heartbeat Record schemas # --------------------------------------------------------------------------- class HeartbeatRecordBase(BaseModel): device_id: int protocol_number: int terminal_info: int battery_level: int gsm_signal: int extension_data: dict[str, Any] | None = None class HeartbeatRecordCreate(HeartbeatRecordBase): pass class HeartbeatRecordResponse(HeartbeatRecordBase): model_config = ConfigDict(from_attributes=True) id: int imei: str | None = None created_at: datetime class HeartbeatListResponse(APIResponse[PaginatedList[HeartbeatRecordResponse]]): pass # --------------------------------------------------------------------------- # Attendance Record schemas # --------------------------------------------------------------------------- class AttendanceRecordBase(BaseModel): device_id: int attendance_type: str = Field(..., max_length=20) attendance_source: str = Field(default="device", max_length=20) # device, bluetooth, fence protocol_number: int gps_positioned: bool = False latitude: float | None = Field(None, ge=-90, le=90) longitude: float | None = Field(None, ge=-180, le=180) speed: float | None = None course: float | None = None gps_satellites: int | None = None battery_level: int | None = None gsm_signal: int | None = None mcc: int | None = None mnc: int | None = None lac: int | None = None cell_id: int | None = None wifi_data: list[dict[str, Any]] | None = None lbs_data: list[dict[str, Any]] | dict[str, Any] | None = None address: str | None = None recorded_at: datetime class AttendanceRecordCreate(AttendanceRecordBase): pass class AttendanceRecordResponse(AttendanceRecordBase): model_config = ConfigDict(from_attributes=True) id: int imei: str | None = None created_at: datetime class AttendanceRecordFilters(BaseModel): device_id: int | None = None attendance_type: str | None = None start_time: datetime | None = None end_time: datetime | None = None class AttendanceListResponse(APIResponse[PaginatedList[AttendanceRecordResponse]]): pass # --------------------------------------------------------------------------- # Bluetooth Record schemas # --------------------------------------------------------------------------- class BluetoothRecordBase(BaseModel): device_id: int record_type: str = Field(..., max_length=20) protocol_number: int beacon_mac: str | None = None beacon_uuid: str | None = None beacon_major: int | None = None beacon_minor: int | None = None rssi: int | None = None beacon_battery: float | None = None beacon_battery_unit: str | None = None attendance_type: str | None = None bluetooth_data: dict[str, Any] | None = None latitude: float | None = Field(None, ge=-90, le=90) longitude: float | None = Field(None, ge=-180, le=180) recorded_at: datetime class BluetoothRecordCreate(BluetoothRecordBase): pass class BluetoothRecordResponse(BluetoothRecordBase): model_config = ConfigDict(from_attributes=True) id: int imei: str | None = None created_at: datetime class BluetoothRecordFilters(BaseModel): device_id: int | None = None record_type: str | None = None start_time: datetime | None = None end_time: datetime | None = None class BluetoothListResponse(APIResponse[PaginatedList[BluetoothRecordResponse]]): pass # --------------------------------------------------------------------------- # Beacon Config schemas # --------------------------------------------------------------------------- class BeaconConfigBase(BaseModel): beacon_mac: str = Field(..., max_length=20, pattern=r"^([0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2}$", description="信标MAC地址 (AA:BB:CC:DD:EE:FF)") beacon_uuid: str | None = Field(None, max_length=36, description="iBeacon UUID") beacon_major: int | None = Field(None, ge=0, le=65535, description="iBeacon Major") beacon_minor: int | None = Field(None, ge=0, le=65535, description="iBeacon Minor") name: str = Field(..., max_length=100, description="信标名称") floor: str | None = Field(None, max_length=20, description="楼层") area: str | None = Field(None, max_length=100, description="区域") latitude: float | None = Field(None, ge=-90, le=90, description="纬度") longitude: float | None = Field(None, ge=-180, le=180, description="经度") address: str | None = Field(None, description="详细地址") status: Literal["active", "inactive"] = Field(default="active", description="状态") class BeaconConfigCreate(BeaconConfigBase): pass class BeaconConfigUpdate(BaseModel): beacon_uuid: str | None = Field(None, max_length=36) beacon_major: int | None = Field(None, ge=0, le=65535) beacon_minor: int | None = Field(None, ge=0, le=65535) name: str | None = Field(None, max_length=100) floor: str | None = Field(None, max_length=20) area: str | None = Field(None, max_length=100) latitude: float | None = Field(None, ge=-90, le=90) longitude: float | None = Field(None, ge=-180, le=180) address: str | None = None status: Literal["active", "inactive"] | None = None class BeaconConfigResponse(BaseModel): """Response model — no pattern validation on output (existing data may not conform).""" model_config = ConfigDict(from_attributes=True) id: int beacon_mac: str beacon_uuid: str | None = None beacon_major: int | None = None beacon_minor: int | None = None name: str floor: str | None = None area: str | None = None latitude: float | None = None longitude: float | None = None address: str | None = None status: str created_at: datetime updated_at: datetime | None = None # --------------------------------------------------------------------------- # Fence Config schemas # --------------------------------------------------------------------------- class FenceConfigCreate(BaseModel): name: str = Field(..., max_length=100, description="围栏名称") fence_type: Literal["circle", "polygon", "rectangle"] = Field(..., description="围栏类型") center_lat: float | None = Field(None, ge=-90, le=90, description="中心纬度 (WGS-84)") center_lng: float | None = Field(None, ge=-180, le=180, description="中心经度 (WGS-84)") radius: float | None = Field(None, ge=0, description="半径 (米)") points: str | None = Field(None, description="多边形顶点 JSON [[lng,lat],...]") color: str = Field(default="#3b82f6", max_length=20, description="边框颜色") fill_color: str | None = Field(None, max_length=20, description="填充颜色") fill_opacity: float = Field(default=0.2, ge=0, le=1, description="填充透明度") description: str | None = Field(None, description="描述") is_active: bool = Field(default=True, description="是否启用") class FenceConfigUpdate(BaseModel): name: str | None = Field(None, max_length=100) fence_type: Literal["circle", "polygon", "rectangle"] | None = None center_lat: float | None = Field(None, ge=-90, le=90) center_lng: float | None = Field(None, ge=-180, le=180) radius: float | None = Field(None, ge=0) points: str | None = None color: str | None = Field(None, max_length=20) fill_color: str | None = Field(None, max_length=20) fill_opacity: float | None = Field(None, ge=0, le=1) description: str | None = None is_active: bool | None = None class FenceConfigResponse(BaseModel): model_config = ConfigDict(from_attributes=True) id: int name: str fence_type: str center_lat: float | None = None center_lng: float | None = None radius: float | None = None points: str | None = None color: str fill_color: str | None = None fill_opacity: float description: str | None = None is_active: bool created_at: datetime updated_at: datetime | None = None # --------------------------------------------------------------------------- # Device-Fence Binding schemas # --------------------------------------------------------------------------- class DeviceFenceBindRequest(BaseModel): device_ids: list[int] = Field(..., min_length=1, max_length=100, description="设备ID列表") class FenceBindForDeviceRequest(BaseModel): fence_ids: list[int] = Field(..., min_length=1, max_length=100, description="围栏ID列表") class DeviceFenceBindingResponse(BaseModel): model_config = ConfigDict(from_attributes=True) id: int device_id: int fence_id: int created_at: datetime class FenceDeviceDetail(BaseModel): """Binding detail with device info.""" binding_id: int device_id: int device_name: str | None = None imei: str | None = None is_inside: bool = False last_check_at: datetime | None = None class DeviceFenceDetail(BaseModel): """Binding detail with fence info.""" binding_id: int fence_id: int fence_name: str | None = None fence_type: str | None = None is_inside: bool = False last_check_at: datetime | None = None # --------------------------------------------------------------------------- # Command Log schemas # --------------------------------------------------------------------------- class CommandCreate(BaseModel): device_id: int command_type: str = Field(..., max_length=30) command_content: str = Field(..., max_length=500) server_flag: str = Field(..., max_length=20) # --------------------------------------------------------------------------- # Batch operation schemas # --------------------------------------------------------------------------- class BatchDeviceCreateItem(BaseModel): """Single device in a batch create request.""" imei: str = Field(..., min_length=15, max_length=20, pattern=r"^\d{15,17}$", description="IMEI number") device_type: str = Field(default="P241", max_length=10, description="Device type (P240/P241)") name: str | None = Field(None, max_length=100, description="Friendly name") class BatchDeviceCreateRequest(BaseModel): """Batch create devices.""" devices: list[BatchDeviceCreateItem] = Field(..., min_length=1, max_length=500, description="List of devices to create") class BatchDeviceCreateResult(BaseModel): """Result of a single device in batch create.""" imei: str success: bool device_id: int | None = None error: str | None = None class BatchDeviceCreateResponse(BaseModel): """Summary of batch create operation.""" total: int created: int failed: int results: list[BatchDeviceCreateResult] class BatchCommandRequest(BaseModel): """Send the same command to multiple devices.""" device_ids: list[int] | None = Field(default=None, min_length=1, max_length=100, description="Device IDs (provide device_ids or imeis)") imeis: list[str] | None = Field(default=None, min_length=1, max_length=100, description="IMEI list (alternative to device_ids)") command_type: str = Field(..., max_length=30, description="Command type (e.g. online_cmd)") command_content: str = Field(..., max_length=500, description="Command content") @model_validator(mode="after") def check_device_ids_or_imeis(self): if not self.device_ids and not self.imeis: raise ValueError("Must provide device_ids or imeis / 必须提供 device_ids 或 imeis") if self.device_ids and self.imeis: raise ValueError("Provide device_ids or imeis, not both / 不能同时提供 device_ids 和 imeis") return self class BatchCommandResult(BaseModel): """Result of a single command in batch send.""" device_id: int imei: str success: bool command_id: int | None = None error: str | None = None class BatchCommandResponse(BaseModel): """Summary of batch command operation.""" total: int sent: int failed: int results: list[BatchCommandResult] class BatchDeviceDeleteRequest(BaseModel): """Batch delete devices.""" device_ids: list[int] = Field(..., min_length=1, max_length=100, description="Device IDs to delete") class BatchDeviceUpdateRequest(BaseModel): """Batch update devices with the same settings.""" device_ids: list[int] = Field(..., min_length=1, max_length=500, description="Device IDs to update") update: DeviceUpdate = Field(..., description="Fields to update") class CommandUpdate(BaseModel): response_content: str | None = None status: str | None = Field(None, max_length=20) sent_at: datetime | None = None response_at: datetime | None = None class CommandResponse(BaseModel): model_config = ConfigDict(from_attributes=True) id: int device_id: int command_type: str command_content: str server_flag: str response_content: str | None = None status: str sent_at: datetime | None = None response_at: datetime | None = None created_at: datetime class CommandListResponse(APIResponse[PaginatedList[CommandResponse]]): pass # --------------------------------------------------------------------------- # API Key schemas # --------------------------------------------------------------------------- class ApiKeyCreate(BaseModel): name: str = Field(..., min_length=1, max_length=100, description="Key name / 名称") permissions: Literal["read", "write", "admin"] = Field(default="read", description="Permission level") class ApiKeyUpdate(BaseModel): name: str | None = Field(None, max_length=100) permissions: Literal["read", "write", "admin"] | None = None is_active: bool | None = None class ApiKeyResponse(BaseModel): model_config = ConfigDict(from_attributes=True) id: int name: str permissions: str is_active: bool last_used_at: datetime | None = None created_at: datetime class ApiKeyCreateResponse(ApiKeyResponse): """Returned only on creation — includes the plaintext key (shown once).""" key: str