from datetime import datetime from typing import Any, Generic, TypeVar from pydantic import BaseModel, ConfigDict, Field 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, description="IMEI number") device_type: str = Field(..., max_length=10, description="Device type code") 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) status: str | None = Field(None, max_length=20) battery_level: int | None = None gsm_signal: int | None = None 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 = None longitude: float | None = None 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 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=10) protocol_number: int latitude: float | None = None longitude: float | None = None 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 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 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) protocol_number: int gps_positioned: bool = False latitude: float | None = None longitude: float | None = None 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]] | None = None address: str | None = None recorded_at: datetime class AttendanceRecordCreate(AttendanceRecordBase): pass class AttendanceRecordResponse(AttendanceRecordBase): model_config = ConfigDict(from_attributes=True) id: int 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 = None longitude: float | None = None recorded_at: datetime class BluetoothRecordCreate(BluetoothRecordBase): pass class BluetoothRecordResponse(BluetoothRecordBase): model_config = ConfigDict(from_attributes=True) id: int 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, description="信标MAC地址") beacon_uuid: str | None = Field(None, max_length=36, description="iBeacon UUID") beacon_major: int | None = Field(None, description="iBeacon Major") beacon_minor: int | None = Field(None, 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, description="纬度") longitude: float | None = Field(None, description="经度") address: str | None = Field(None, description="详细地址") status: str = Field(default="active", max_length=20, description="状态") class BeaconConfigCreate(BeaconConfigBase): pass class BeaconConfigUpdate(BaseModel): beacon_uuid: str | None = Field(None, max_length=36) beacon_major: int | None = None beacon_minor: int | None = None 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 = None longitude: float | None = None address: str | None = None status: str | None = Field(None, max_length=20) class BeaconConfigResponse(BeaconConfigBase): model_config = ConfigDict(from_attributes=True) id: int created_at: datetime updated_at: datetime | None = None # --------------------------------------------------------------------------- # Command Log schemas # --------------------------------------------------------------------------- class CommandCreate(BaseModel): device_id: int command_type: str = Field(..., max_length=30) command_content: str server_flag: str = Field(..., max_length=20) 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