ipam/backend/app/schemas/network.py

from pydantic import BaseModel, Field, field_validator, field_serializer
from typing import Optional, List, Dict, Any
from datetime import datetime
import ipaddress
from app.models.network import IPStatus, TaskStatus, TaskType
from app.schemas._tz_util import to_business_iso, to_utc_iso


# Pydantic v2 字段序列化器：把 datetime 转成业务时区 ISO 字符串
# 让前端拿到 'YYYY-MM-DDTHH:MM:SS+08:00'，可直接当本地时间显示
_datetime_business_serializer = field_serializer(
    'datetime',
    when_used='always',
    check_fields=None,  # 重要：应用到所有 datetime 字段
)(lambda dt: to_business_iso(dt))


# ========== 网段相关 Schemas ==========

class NetworkBase(BaseModel):
    cidr: str = Field(..., description="网段CIDR，如: 192.168.1.0/24")
    name: Optional[str] = Field(None, max_length=100, description="网段名称")
    description: Optional[str] = Field(None, description="描述")
    group_name: Optional[str] = Field(None, max_length=100, description="分组名称")
    gateway: Optional[str] = Field(None, max_length=50, description="网关地址")
    vlan_id: Optional[int] = Field(None, description="VLAN ID")

    @field_validator('cidr')
    @classmethod
    def validate_cidr(cls, v):
        try:
            network = ipaddress.ip_network(v, strict=False)
            return str(network)
        except ValueError:
            raise ValueError(f"无效的CIDR格式: {v}")

    @field_validator('gateway')
    @classmethod
    def validate_gateway(cls, v):
        if v:
            try:
                ipaddress.ip_address(v)
            except ValueError:
                raise ValueError(f"无效的IP地址: {v}")
        return v


class NetworkCreate(NetworkBase):
    pass


class NetworkUpdate(BaseModel):
    name: Optional[str] = None
    description: Optional[str] = None
    group_name: Optional[str] = None
    gateway: Optional[str] = None
    vlan_id: Optional[int] = None


class Network(NetworkBase):
    id: int
    total_ips: int
    used_ips: int
    reserved_ips: int
    created_at: datetime
    updated_at: Optional[datetime]

    # Pydantic v2: 用 model_serializer 在序列化整个 model 时把所有 datetime 转成业务时区
    @field_serializer('created_at', 'updated_at')
    def _tz_serializers(self, dt: Optional[datetime]) -> Optional[str]:
        return to_business_iso(dt)

    class Config:
        from_attributes = True


class NetworkStats(BaseModel):
    """网段统计信息"""
    id: int
    cidr: str
    name: Optional[str]
    total_ips: int
    used_ips: int
    reserved_ips: int
    available_ips: int
    usage_percent: float
    group_name: Optional[str]
    vlan_id: Optional[int]

    class Config:
        from_attributes = True


class NetworkListResponse(BaseModel):
    total: int
    items: List[Network]


# ========== IP地址相关 Schemas ==========

class IPAddressBase(BaseModel):
    ip_address: str
    mac_address: Optional[str] = None
    hostname: Optional[str] = None
    owner: Optional[str] = None
    business_type: Optional[str] = None
    notes: Optional[str] = None
    custom_fields: Dict[str, Any] = Field(default_factory=dict)
    switch_name: Optional[str] = None
    switch_port: Optional[str] = None
    vendor: Optional[str] = None


class IPAddressCreate(IPAddressBase):
    network_id: int
    status: IPStatus = IPStatus.AVAILABLE


class IPAddressUpdate(BaseModel):
    mac_address: Optional[str] = None
    hostname: Optional[str] = None
    status: Optional[IPStatus] = None
    owner: Optional[str] = None
    business_type: Optional[str] = None
    notes: Optional[str] = None
    custom_fields: Optional[Dict[str, Any]] = None


class IPAddress(IPAddressBase):
    id: int
    network_id: int
    status: IPStatus
    last_seen: Optional[datetime]
    first_seen: datetime
    created_at: datetime
    updated_at: Optional[datetime]

    @field_serializer('last_seen', 'first_seen', 'created_at', 'updated_at')
    def _tz_serializers(self, dt: Optional[datetime]) -> Optional[str]:
        return to_business_iso(dt)

    class Config:
        from_attributes = True


class IPAddressListResponse(BaseModel):
    total: int
    items: List[IPAddress]


# ========== 扫描任务相关 Schemas ==========

class ScanTaskBase(BaseModel):
    task_type: TaskType
    network_id: Optional[int] = None


class ScanTaskCreate(ScanTaskBase):
    pass


class ScanTask(BaseModel):
    id: int
    network_id: Optional[int]
    task_type: TaskType
    status: TaskStatus
    progress: int
    celery_task_id: Optional[str]
    total_count: int
    success_count: int
    failed_count: int
    started_at: Optional[datetime]
    completed_at: Optional[datetime]
    error_message: Optional[str]
    created_at: datetime

    @field_serializer(
        'started_at', 'completed_at', 'created_at',
        check_fields=None,
    )
    def _tz_serializers(self, dt: Optional[datetime]) -> Optional[str]:
        return to_business_iso(dt)

    class Config:
        from_attributes = True


class ScanTaskListResponse(BaseModel):
    total: int
    items: List[ScanTask]


class ScanResult(BaseModel):
    """单次扫描结果"""
    ip_address: str
    status: str
    mac_address: Optional[str] = None
    hostname: Optional[str] = None
    response_time: Optional[float] = None