Your Name
1e1e5957e8
- Ips.vue: el-select 添加 width: 280px，修复网段下拉框过窄 - Ips.vue: query 参数 networkId → network_id（与 FastAPI snake_case 一致） - Scan.vue: el-select 用 network.id 代替整个 network 对象作为 value，修复切换网段不生效 - enhanced_scan_service.py: 替换失效的 scapy ARP 扫描为 arp-scan 命令；扫描结果自动创建缺失 IP 记录
144 lines
4.7 KiB
Python
144 lines
4.7 KiB
Python
from typing import List, Optional, Dict, Any
|
|
from sqlalchemy.orm import Session
|
|
from datetime import datetime
|
|
|
|
from app.models.network import IPAddress as IPAddressModel
|
|
from app.models.network import Network as NetworkModel
|
|
from app.models.network import IPStatus
|
|
from app.schemas.network import IPAddressUpdate, IPAddressCreate
|
|
|
|
|
|
class IPService:
|
|
"""IP地址管理服务"""
|
|
|
|
@staticmethod
|
|
def get_by_id(db: Session, ip_id: int) -> Optional[IPAddressModel]:
|
|
"""根据ID获取IP"""
|
|
return db.query(IPAddressModel).filter(IPAddressModel.id == ip_id).first()
|
|
|
|
@staticmethod
|
|
def get_by_ip(db: Session, ip_address: str) -> Optional[IPAddressModel]:
|
|
"""根据IP地址获取"""
|
|
return db.query(IPAddressModel).filter(IPAddressModel.ip_address == ip_address).first()
|
|
|
|
@staticmethod
|
|
def get_list(
|
|
db: Session,
|
|
network_id: Optional[int] = None,
|
|
status: Optional[IPStatus] = None,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
search: Optional[str] = None
|
|
) -> tuple[int, List[IPAddressModel]]:
|
|
"""获取IP列表"""
|
|
query = db.query(IPAddressModel)
|
|
|
|
if network_id:
|
|
query = query.filter(IPAddressModel.network_id == network_id)
|
|
|
|
if status:
|
|
query = query.filter(IPAddressModel.status == status)
|
|
|
|
if search:
|
|
search_pattern = f"%{search}%"
|
|
query = query.filter(
|
|
(IPAddressModel.ip_address.like(search_pattern)) |
|
|
(IPAddressModel.mac_address.like(search_pattern)) |
|
|
(IPAddressModel.hostname.like(search_pattern)) |
|
|
(IPAddressModel.owner.like(search_pattern))
|
|
)
|
|
|
|
total = query.count()
|
|
items = query.order_by(IPAddressModel.ip_address).offset(skip).limit(limit).all()
|
|
|
|
return total, items
|
|
|
|
@staticmethod
|
|
def create(db: Session, ip_in: IPAddressCreate) -> IPAddressModel:
|
|
"""手动创建IP（通常由网段创建自动创建）"""
|
|
db_ip = IPAddressModel(**ip_in.model_dump())
|
|
db.add(db_ip)
|
|
db.commit()
|
|
db.refresh(db_ip)
|
|
return db_ip
|
|
|
|
@staticmethod
|
|
def update(db: Session, ip_id: int, ip_in: IPAddressUpdate) -> Optional[IPAddressModel]:
|
|
"""更新IP信息"""
|
|
db_ip = IPService.get_by_id(db, ip_id)
|
|
if not db_ip:
|
|
return None
|
|
|
|
update_data = ip_in.model_dump(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(db_ip, field, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_ip)
|
|
return db_ip
|
|
|
|
@staticmethod
|
|
def update_status(db: Session, ip_address: str, status: IPStatus,
|
|
mac_address: Optional[str] = None,
|
|
hostname: Optional[str] = None) -> Optional[IPAddressModel]:
|
|
"""更新IP状态和扫描信息"""
|
|
db_ip = IPService.get_by_ip(db, ip_address)
|
|
if not db_ip:
|
|
return None
|
|
|
|
db_ip.status = status
|
|
db_ip.last_seen = datetime.utcnow()
|
|
|
|
if mac_address:
|
|
db_ip.mac_address = mac_address
|
|
|
|
if hostname:
|
|
db_ip.hostname = hostname
|
|
|
|
db.commit()
|
|
db.refresh(db_ip)
|
|
return db_ip
|
|
|
|
@staticmethod
|
|
def set_reserved(db: Session, ip_id: int, reserved: bool = True) -> Optional[IPAddressModel]:
|
|
"""设置/取消保留IP"""
|
|
db_ip = IPService.get_by_id(db, ip_id)
|
|
if not db_ip:
|
|
return None
|
|
|
|
db_ip.status = IPStatus.RESERVED if reserved else IPStatus.AVAILABLE
|
|
db.commit()
|
|
db.refresh(db_ip)
|
|
return db_ip
|
|
|
|
@staticmethod
|
|
def bulk_update_status(db: Session, results: List[Dict[str, Any]]):
|
|
"""批量更新IP状态
|
|
|
|
Args:
|
|
results: 扫描结果列表，每个元素包含 ip_address, status, mac_address, hostname
|
|
"""
|
|
for result in results:
|
|
ip_address = result.get('ip_address')
|
|
status = result.get('status')
|
|
mac_address = result.get('mac_address')
|
|
hostname = result.get('hostname')
|
|
|
|
db_ip = IPService.get_by_ip(db, ip_address)
|
|
if db_ip:
|
|
db_ip.status = status
|
|
db_ip.last_seen = datetime.utcnow()
|
|
if mac_address:
|
|
db_ip.mac_address = mac_address
|
|
if hostname:
|
|
db_ip.hostname = hostname
|
|
|
|
db.commit()
|
|
|
|
@staticmethod
|
|
def get_by_network(db: Session, network_id: int) -> List[IPAddressModel]:
|
|
"""获取指定网段下的所有IP"""
|
|
return db.query(IPAddressModel).filter(
|
|
IPAddressModel.network_id == network_id
|
|
).order_by(IPAddressModel.ip_address).all()
|