diff --git a/backend/app/api/v1/enhanced_scan.py b/backend/app/api/v1/enhanced_scan.py index 59307f2..9e2d190 100644 --- a/backend/app/api/v1/enhanced_scan.py +++ b/backend/app/api/v1/enhanced_scan.py @@ -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) diff --git a/backend/app/services/enhanced_scan_service.py b/backend/app/services/enhanced_scan_service.py index ee5ef99..27b04bc 100644 --- a/backend/app/services/enhanced_scan_service.py +++ b/backend/app/services/enhanced_scan_service.py @@ -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信息"""