fix(scan): 修复单IP扫描500错误并清除MySQL不兼容SQL

1. enhanced_scan.py: comprehensive_scan_ip 误用 schema 的 IPAddress 做
   db.query 导致 ArgumentError(500)。改用 ORM 模型 IPAddressModel,与
   文件内其他函数一致。
2. enhanced_scan_service.py: update_ip_from_scan_result 使用 PostgreSQL
   的 @> 包含操作符,MySQL 上报 SQL语法错误(1064)。改为纯 Python
   ipaddress 遍历匹配网段,新增 _find_network_for_ip 辅助方法。
This commit is contained in:
Your Name
2026-08-12 10:47:32 +08:00
parent 11d830b496
commit b7551af688
2 changed files with 28 additions and 12 deletions
+2 -1
View File
@@ -113,7 +113,8 @@ def comprehensive_scan_ip(
EnhancedScanService.update_ip_from_scan_result(db, result)
# 刷新所属网段统计字段(used_ips/reserved_ips),保证网段列表利用率能即时反映
db_ip = db.query(IPAddress).filter(IPAddress.ip_address == ip_address).first()
from app.models.network import IPAddress as IPAddressModel
db_ip = db.query(IPAddressModel).filter(IPAddressModel.ip_address == ip_address).first()
if db_ip:
NetworkService.get_stats(db, db_ip.network_id)
+26 -11
View File
@@ -429,17 +429,7 @@ class EnhancedScanService:
# 如果IP不存在,尝试自动创建
if not db_ip:
network = db.query(Network).filter(
Network.cidr.op('@>')((ip_address + '/32').encode('utf-8'))
).first()
if not network:
for net in db.query(Network).all():
try:
if ip_address in ipaddress.ip_network(net.cidr, strict=False):
network = net
break
except Exception:
pass
network = EnhancedScanService._find_network_for_ip(db, ip_address)
if not network:
logger.warning(f"无法为扫描到的IP {ip_address} 找到所属网段,跳过创建")
return
@@ -481,6 +471,31 @@ class EnhancedScanService:
return db_ip
@staticmethod
def _find_network_for_ip(db: Session, ip_address: str) -> Optional[Network]:
"""
查找 IP 所属的网段(兼容 MySQL,不使用 PostgreSQL 的 @> 操作符)。
优先按网段越大越精确匹配,无法精确匹配时返回包含该 IP 的最小/最合适网段。
"""
try:
target = ipaddress.ip_address(ip_address)
except ValueError:
return None
best = None
best_prefix = -1
for net in db.query(Network).all():
try:
cidr = str(net.cidr)
net_obj = ipaddress.ip_network(cidr, strict=False)
if target in net_obj:
# 取前缀最长的(网段最小、最精确)
if net_obj.prefixlen > best_prefix:
best = net
best_prefix = net_obj.prefixlen
except Exception:
continue
return best
@staticmethod
def bulk_update_ips_from_scan(db: Session, scan_results: List[Dict[str, Any]]):
"""批量更新IP信息"""