10 Commits

Author SHA1 Message Date
Your Name 11d830b496 fix(snmp): 修复 SNMP 采集 ARP 表时只补 MAC 不补厂商的问题
问题原因:
- snmp_service.py get_arp_table 第267行条件为
  if ip_addr and not ip_addr.mac_address:
- 只给原本没有 MAC 的 IP 设置 MAC + 厂商
- 如果 IP 已有 MAC(来自 Ping/ARP 扫描)但没有 vendor,
  则永远不会补全厂商

解决方案:
- 拆分为两个 if:
  1. if not ip_addr.mac_address: 写入 MAC
  2. if not ip_addr.vendor and ip_addr.mac_address: 补全厂商
- 确保已有 MAC 但无厂商的 IP 也能被补全
- 已有 backfill_vendor.py 脚本可批量回填历史数据
2026-07-23 15:39:59 +08:00
Your Name 934b604b65 fix(mac-vendor): 修复 OUI 数据库 key 格式不匹配导致永远查不到厂商
问题原因:
- oui.txt 中的 key 格式为 AABBCC(无冒号,6位十六进制)
- 代码中提取的 oui = mac[:8] 是 AA:BB:CC(带冒号,8字符)
- db.get(oui) 永远匹配不到,且回退的遍历逻辑在4万条字典上
  每次查都全扫描,性能极差

解决方案:
- 提取 OUI 后去掉冒号再查询:oui.replace(':', '')
- 直接 db.get() 精确匹配,O(1) 时间复杂度
2026-07-23 15:36:47 +08:00
Your Name 13c457acb3 fix(network): 修复网段 total_ips 包含网络地址和广播地址的问题
问题原因:
- calculate_total_ips 使用 network.num_addresses 计算总IP数量
- 该值包含网络地址(第一个IP)和广播地址(最后一个IP)
- 而 _create_ip_addresses 使用 hosts() 生成IP列表,排除这两个地址
- 导致 total_ips 比实际存储的IP记录多2个
- 例: /24 显示 256 个IP,实际只有 254 条记录

解决方案:
1. calculate_total_ips 改为 num_addresses - 2(排除网络地址和广播地址)
2. 对于 /31 /32 等没有网络/广播地址的网段,特殊处理
3. 提供 fix_network_total_ips.py 迁移脚本修复已有数据
2026-07-23 15:33:13 +08:00
Your Name 05cd71ca1a feat(mac-vendor): 本地 OUI 数据库替代在线查询,修复厂商识别失败
问题原因:
- mac_vendor_lookup 库依赖从 IEEE 官网在线下载 OUI 数据库
- 其他机器无网络或网络不通时,内置精简版数据库不全
- 导致 SNMP 扫描到 MAC 地址后无法识别厂商

解决方案:
1. 从 IEEE OUI 官方数据库下载完整数据 (39782 条记录)
   保存为 backend/app/data/oui.txt 纳入版本管理
2. 修改 EnhancedScanService.get_mac_vendor(),查询顺序:
   1️⃣ 本地 OUI 数据库(免网络、高性能)
   2️⃣ mac_vendor_lookup 在线库(回退)
   3️⃣ 硬编码常用厂商映射(最终回退)
3. start.sh 启动时检测 OUI 文件是否存在,不存在自动下载
4. 支持 AA:BB:CC、AABBCC、aa-bb-cc、001c.14aa.bbcc 多种 MAC 格式
2026-07-23 15:31:54 +08:00
Your Name ee0552f837 fix(frontend): 移除 element-plus alias 防止 CSS 路径解析失败
问题原因:
- 上次提交在 vite.config.js resolve.alias 中将 element-plus
  直接映射到 es/index.mjs 文件路径
- 这导致 import 'element-plus/dist/index.css' 时,
  Vite 拼接路径为 es/index.mjs/dist/index.css,文件不存在
- 报错: Failed to resolve import 'element-plus/dist/index.css'

解决方案:
- 移除 element-plus 的 resolve.alias 映射
- 包名保留为包名,Vite 的 node_modules 解析机制可正确处理
  element-plus 的 JS 入口和 CSS 路径
- start.sh 已有清缓存+自动重试机制兜底
2026-07-23 15:05:35 +08:00
Your Name 094d2cc112 fix(snmp): 修复异步函数同步调用导致'coroutine object is not an iterator'错误
问题原因:
- pysnmp.hlapi.asyncio 的 getCmd/bulkCmd/nextCmd 都是 async def 协程函数
- 返回的是 coroutine 对象,需要用 await 获取结果
- 原代码用 next() 或 for...in 同步迭代协程对象,报错:
  'coroutine' object is not an iterator

解决方案:
1. snmp_service.py: 所有 SNMP 操作改为 async def
   - test_connection(): getCmd 改为 await getCmd()
   - get_arp_table(): bulkCmd 改为 await bulkCmd()
   - get_mac_address_table(): bulkCmd 改为 await bulkCmd()
   - get_interfaces(): bulkCmd 改为 await bulkCmd()
   - poll_device(): 所有子调用加 await
2. api/v1/snmp.py: API 路由改为 async def + await

bulkCmd await 后返回 (errorIndication, errorStatus, errorIndex, varBindTable)
其中 varBindTable 是列表的列表(table 形式),需双层循环解析
2026-07-23 15:01:18 +08:00
Your Name 8b1df674ec fix(frontend): 修复 element-plus 解析失败问题
问题原因:
- element-plus 2.14.x 的 package.json exports 包含 ./*: ./ 通配符
- Vite 5.x 的 import-analysis 在某些场景下无法解析该 exports 配置
- 新机器 git pull + npm install 后 Vite 缓存为空,首次预构建可能失败
- 报错: Failed to resolve entry for package 'element-plus'

解决方案:
1. vite.config.js resolve.alias 添加 element-plus 显式 fallback 路径
   即使 exports 解析失败也能直接定位到 es/index.mjs
2. start.sh 启动前自动清理 .vite 缓存,避免过期缓存问题
3. start.sh 失败自动重试+清缓存机制,提高首次启动成功率
2026-07-23 14:49:54 +08:00
Your Name 6076e174d5 fix(frontend): 撤回 optimizeDeps.include 避免 element-plus 解析失败导致启动崩溃
问题原因:
- 上一次提交添加的 optimizeDeps.include 让 Vite 在启动阶段
  就硬解析 element-plus, 包安装不完整时直接崩 (服务器起不来)
- predev 脚本每次清缓存, 如果源包有问题则重建必然失败

修正:
- 移除 optimizeDeps.include, Vite 自动发现预构建即可
- predev 改为可选的 dev:clean 脚本, 不影响正常启动
- start.sh 加固: 检查 element-plus/es/index.mjs 是否存在
  不存在则自动 npm install

影响范围:
- 其他服务器 git pull 后, start.sh 会自动检测并安装缺失依赖
- 正常 npm run dev 不再每次清缓存, 启动更快
2026-07-23 14:35:17 +08:00
Your Name 8a94f535de fix(frontend): 修复 Vite 缓存过期导致 element-plus 入口解析失败
问题原因:
- Vite dev server 长时间运行(跨天)后 deps 缓存过期
- import-analysis 阶段无法解析 element-plus 包入口
- 报错: Failed to resolve entry for package "element-plus"

解决方案:
- vite.config.js 添加 optimizeDeps.include 显式预构建 element-plus 等核心依赖
- package.json 添加 predev 脚本, npm run dev 前自动清 .vite 缓存
- 防止其他服务器 git pull 后同样踩坑
2026-07-23 14:27:41 +08:00
Your Name ebb04f9ad8 fix(frontend): 修复 vite 命令冲突问题
问题原因:
- 系统 PATH 中有另一个也叫 'vite' 的 Qt GUI 程序
- 优先级高于 npm 的 vite,导致 npm run dev 调用了错误的程序
- 出现 Qt X11 显示相关的错误

解决方案:
- 直接使用 node 调用 vite 的完整路径
- 避免依赖 PATH 解析,确保调用正确的 vite

验证:
- node node_modules/vite/bin/vite.js --version
   vite/5.4.21 linux-x64 node-v23.11.1
2026-07-23 14:16:08 +08:00
8 changed files with 40049 additions and 137 deletions
+6 -6
View File
@@ -437,7 +437,7 @@ def delete_network_device(
# ========== SNMP 操作 ==========
@router.post("/devices/{device_id}/test", summary="测试 SNMP 连接")
def test_snmp_connection(device_id: int, db: Session = Depends(get_db)):
async def test_snmp_connection(device_id: int, db: Session = Depends(get_db)):
"""测试设备的 SNMP 连接(不写审计,避免刷屏)"""
from app.models.snmp import NetworkDevice
@@ -448,7 +448,7 @@ def test_snmp_connection(device_id: int, db: Session = Depends(get_db)):
if not device.snmp_credential:
raise HTTPException(status_code=400, detail="设备未配置 SNMP 凭据")
success, info = SNMPService.test_connection(device, device.snmp_credential)
success, info = await SNMPService.test_connection(device, device.snmp_credential)
return {
"device_id": device_id,
@@ -459,9 +459,9 @@ def test_snmp_connection(device_id: int, db: Session = Depends(get_db)):
@router.post("/devices/{device_id}/poll", summary="立即轮询设备")
def poll_device_now(device_id: int, db: Session = Depends(get_db)):
async def poll_device_now(device_id: int, db: Session = Depends(get_db)):
"""立即轮询设备,获取 ARP 表、MAC 地址表、接口信息(不写审计,常规操作)"""
result = SNMPService.poll_device(db, device_id)
result = await SNMPService.poll_device(db, device_id)
if 'error' in result:
raise HTTPException(status_code=400, detail=result['error'])
@@ -470,7 +470,7 @@ def poll_device_now(device_id: int, db: Session = Depends(get_db)):
@router.post("/devices/{device_id}/arp", summary="获取设备 ARP 表")
def get_device_arp_table(device_id: int, db: Session = Depends(get_db)):
async def get_device_arp_table(device_id: int, db: Session = Depends(get_db)):
"""获取设备的 ARP 表"""
from app.models.snmp import NetworkDevice, ARPEntry
@@ -481,7 +481,7 @@ def get_device_arp_table(device_id: int, db: Session = Depends(get_db)):
if not device.snmp_credential:
raise HTTPException(status_code=400, detail="设备未配置 SNMP 凭据")
entries = SNMPService.get_arp_table(device, device.snmp_credential, db)
entries = await SNMPService.get_arp_table(device, device.snmp_credential, db)
return {
"device_id": device_id,
File diff suppressed because it is too large Load Diff
+49 -4
View File
@@ -7,6 +7,7 @@ import socket
import ipaddress
import re
import logging
import os
try:
from mac_vendor_lookup import MacLookup, VendorNotFoundError
@@ -20,6 +21,40 @@ from app.core.config import settings
logger = logging.getLogger(__name__)
# OUI 数据库文件路径(相对于项目根)
_OUI_DB_FILE = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'data', 'oui.txt')
def _load_oui_database() -> Optional[Dict[str, str]]:
"""从本地文件加载 OUI 数据库,返回 {prefix: vendor} 字典"""
if not os.path.exists(_OUI_DB_FILE):
logger.warning(f"OUI 数据库文件不存在: {_OUI_DB_FILE}")
return None
try:
db = {}
with open(_OUI_DB_FILE, 'r', encoding='utf-8') as f:
for line in f:
line = line.strip()
if ':' in line:
prefix, vendor = line.split(':', 1)
db[prefix] = vendor
logger.info(f"已加载 OUI 数据库: {len(db)} 条记录, 文件: {_OUI_DB_FILE}")
return db
except Exception as e:
logger.error(f"加载 OUI 数据库失败: {e}")
return None
# 全局 OUI 数据库(延迟加载)
_OUI_DB = None
def _get_oui_db() -> Optional[Dict[str, str]]:
global _OUI_DB
if _OUI_DB is None:
_OUI_DB = _load_oui_database()
return _OUI_DB
class EnhancedScanService:
"""增强扫描服务 - 包含ARP、DNS解析、MAC厂商识别、NetBIOS主机名发现"""
@@ -45,13 +80,23 @@ class EnhancedScanService:
if len(mac) < 8:
return None
# 使用 mac_vendor_lookup 库(IEEE OUI 官方数据库
# 提取 OUI(前 6 位十六进制 = AA:BB:CC
oui = mac[:8] # "AA:BB:CC"
# 1️⃣ 优先使用本地 OUI 数据库(免网络、高性能)
db = _get_oui_db()
if db:
# 数据库 key 是 AABBCC 格式(无冒号)
oui_key = oui.replace(':', '')
vendor = db.get(oui_key)
if vendor:
return vendor
# 2️⃣ 回退:mac_vendor_lookup 在线库
if MAC_LOOKUP_AVAILABLE:
lookup = EnhancedScanService._get_mac_lookup()
if lookup is not None:
try:
# 提取 OUI(前 3 字节)
oui = mac[:8] # "AA:BB:CC"
vendor = lookup.lookup(oui)
if vendor:
return vendor
@@ -60,7 +105,7 @@ class EnhancedScanService:
except Exception:
pass
# 回退:硬编码常用厂商映射
# 3️⃣ 回退:硬编码常用厂商映射
return EnhancedScanService._legacy_oui_lookup(mac)
# 硬编码回退映射(mac_vendor_lookup 不可用时使用)
+2 -2
View File
@@ -14,9 +14,9 @@ class NetworkService:
@staticmethod
def calculate_total_ips(cidr: str) -> int:
"""计算网段的总IP数量"""
"""计算网段中实际分配的IP数量(排除网络地址和广播地址)"""
network = ipaddress.ip_network(cidr, strict=False)
return network.num_addresses
return network.num_addresses - 2 if network.num_addresses > 2 else network.num_addresses
@staticmethod
def get_network_addresses(cidr: str) -> List[str]:
+118 -117
View File
@@ -3,6 +3,7 @@ from sqlalchemy.orm import Session
from datetime import datetime
import ipaddress
import logging
import asyncio
from pysnmp.hlapi.asyncio import (
SnmpEngine, CommunityData, UsmUserData,
@@ -117,7 +118,7 @@ class SNMPService:
return None
@staticmethod
def test_connection(device: NetworkDevice, credential: SNMPCredential) -> Tuple[bool, Dict[str, Any]]:
async def test_connection(device: NetworkDevice, credential: SNMPCredential) -> Tuple[bool, Dict[str, Any]]:
"""
测试 SNMP 连接
返回: (成功, 设备信息字典)
@@ -131,11 +132,11 @@ class SNMPService:
retries=credential.retries
)
errorIndication, errorStatus, errorIndex, varBinds = next(
getCmd(SnmpEngine(), auth_data, transport, ContextData(),
ObjectType(ObjectIdentity(SNMPService.OID_SYS_DESCR)),
ObjectType(ObjectIdentity(SNMPService.OID_SYS_NAME)),
ObjectType(ObjectIdentity(SNMPService.OID_SYS_LOCATION)))
errorIndication, errorStatus, errorIndex, varBinds = await getCmd(
SnmpEngine(), auth_data, transport, ContextData(),
ObjectType(ObjectIdentity(SNMPService.OID_SYS_DESCR)),
ObjectType(ObjectIdentity(SNMPService.OID_SYS_NAME)),
ObjectType(ObjectIdentity(SNMPService.OID_SYS_LOCATION))
)
if errorIndication:
@@ -162,7 +163,7 @@ class SNMPService:
return False, {'error': str(e)}
@staticmethod
def get_arp_table(device: NetworkDevice, credential: SNMPCredential, db: Session) -> List[Dict[str, Any]]:
async def get_arp_table(device: NetworkDevice, credential: SNMPCredential, db: Session) -> List[Dict[str, Any]]:
"""
获取 ARP 表 (IP -> MAC 映射)
"""
@@ -178,61 +179,60 @@ class SNMPService:
)
# 使用 bulkCmd 获取 ARP 表
for errorIndication, errorStatus, errorIndex, varBinds in bulkCmd(
errorIndication, errorStatus, errorIndex, varBindTable = await bulkCmd(
SnmpEngine(), auth_data, transport, ContextData(),
0, 50, # nonRepeaters, maxRepetitions
ObjectType(ObjectIdentity(SNMPService.OID_IP_NET_TO_MEDIA_PHYS_ADDRESS)),
ObjectType(ObjectIdentity(SNMPService.OID_IP_NET_TO_MEDIA_NET_ADDRESS)),
ObjectType(ObjectIdentity(SNMPService.OID_IP_NET_TO_MEDIA_IF_INDEX)),
lexicographicMode=False
):
if errorIndication:
logger.error(f"SNMP 错误: {errorIndication}")
break
)
if errorIndication:
logger.error(f"SNMP 错误: {errorIndication}")
elif errorStatus:
logger.error(f"SNMP 错误: {errorStatus}")
else:
# varBindTable 是列表的列表,每行是一条 SNMP 响应记录
for varBinds in varBindTable:
# 解析结果
arp_data = {}
for varBind in varBinds:
oid = str(varBind[0])
value = varBind[1]
if errorStatus:
logger.error(f"SNMP 错误: {errorStatus}")
break
# 提取 IP 地址 (OID 后缀: ifIndex.IP)
parts = oid.split('.')
if len(parts) >= 4:
# 从 OID 中提取 IP 地址
ip_parts = parts[-4:]
ip_address = '.'.join(ip_parts)
# 解析结果
arp_data = {}
for varBind in varBinds:
oid = str(varBind[0])
value = varBind[1]
if SNMPService.OID_IP_NET_TO_MEDIA_PHYS_ADDRESS in oid:
mac = SNMPService._normalize_mac(value)
if mac:
if ip_address not in arp_data:
arp_data[ip_address] = {}
arp_data[ip_address]['mac_address'] = mac
elif SNMPService.OID_IP_NET_TO_MEDIA_IF_INDEX in oid:
if isinstance(value, Integer32):
if_index = int(value)
if ip_address not in arp_data:
arp_data[ip_address] = {}
arp_data[ip_address]['if_index'] = if_index
# 提取 IP 地址 (OID 后缀: ifIndex.IP)
parts = oid.split('.')
if len(parts) >= 4:
# 从 OID 中提取 IP 地址
ip_parts = parts[-4:]
ip_address = '.'.join(ip_parts)
if SNMPService.OID_IP_NET_TO_MEDIA_PHYS_ADDRESS in oid:
mac = SNMPService._normalize_mac(value)
if mac:
if ip_address not in arp_data:
arp_data[ip_address] = {}
arp_data[ip_address]['mac_address'] = mac
elif SNMPService.OID_IP_NET_TO_MEDIA_IF_INDEX in oid:
if isinstance(value, Integer32):
if_index = int(value)
if ip_address not in arp_data:
arp_data[ip_address] = {}
arp_data[ip_address]['if_index'] = if_index
# 转换为列表
now = datetime.utcnow()
for ip, data in arp_data.items():
if 'mac_address' in data:
entry = {
'ip_address': ip,
'mac_address': data.get('mac_address'),
'interface': str(data.get('if_index', '')),
'device_id': device.id,
'last_seen': now.isoformat(),
'discovered_at': now.isoformat()
}
arp_entries.append(entry)
# 转换为列表
now = datetime.utcnow()
for ip, data in arp_data.items():
if 'mac_address' in data:
entry = {
'ip_address': ip,
'mac_address': data.get('mac_address'),
'interface': str(data.get('if_index', '')),
'device_id': device.id,
'last_seen': now.isoformat(),
'discovered_at': now.isoformat()
}
arp_entries.append(entry)
# 保存到数据库
for entry_data in arp_entries:
@@ -258,16 +258,18 @@ class SNMPService:
db.commit()
# 更新 IP 资产台账中的 MAC 地址
# 更新 IP 资产台账中的 MAC 地址和厂商信息
for entry in arp_entries:
from app.models.network import IPAddress as IPAddressModel
ip_addr = db.query(IPAddressModel).filter(
IPAddressModel.ip_address == entry['ip_address']
).first()
if ip_addr and not ip_addr.mac_address:
ip_addr.mac_address = entry['mac_address']
# 识别厂商
ip_addr.vendor = EnhancedScanService.get_mac_vendor(entry['mac_address'])
if ip_addr:
if not ip_addr.mac_address:
ip_addr.mac_address = entry['mac_address']
# 始终尝试补全厂商(有 MAC 但没厂商时也需要)
if not ip_addr.vendor and ip_addr.mac_address:
ip_addr.vendor = EnhancedScanService.get_mac_vendor(ip_addr.mac_address)
db.commit()
device.last_successful_poll = now
@@ -281,7 +283,7 @@ class SNMPService:
return arp_entries
@staticmethod
def get_mac_address_table(device: NetworkDevice, credential: SNMPCredential, db: Session, vlan_id: Optional[int] = None) -> List[Dict[str, Any]]:
async def get_mac_address_table(device: NetworkDevice, credential: SNMPCredential, db: Session, vlan_id: Optional[int] = None) -> List[Dict[str, Any]]:
"""
获取 MAC 地址表 (MAC -> 端口 映射)
"""
@@ -297,39 +299,39 @@ class SNMPService:
)
# 获取 MAC 地址表
for errorIndication, errorStatus, errorIndex, varBinds in bulkCmd(
errorIndication, errorStatus, errorIndex, varBindTable = await bulkCmd(
SnmpEngine(), auth_data, transport, ContextData(),
0, 100,
ObjectType(ObjectIdentity(SNMPService.OID_DOT1D_TP_FDB_PORT)),
ObjectType(ObjectIdentity(SNMPService.OID_DOT1D_TP_FDB_STATUS)),
lexicographicMode=False
):
if errorIndication:
break
)
if errorIndication:
logger.error(f"获取 MAC 地址表失败: {errorIndication}")
elif errorStatus:
logger.error(f"获取 MAC 地址表失败: {errorStatus}")
else:
for varBinds in varBindTable:
for varBind in varBinds:
oid = str(varBind[0])
value = varBind[1]
if errorStatus:
break
# OID 后缀是 MAC 地址: .a.b.c.d.e.f
parts = oid.split('.')
if len(parts) >= 6:
mac_parts = parts[-6:]
mac_address = ':'.join(f'{int(p):02X}' for p in mac_parts)
for varBind in varBinds:
oid = str(varBind[0])
value = varBind[1]
if SNMPService.OID_DOT1D_TP_FDB_PORT in oid:
port = int(value) if isinstance(value, Integer32) else None
# OID 后缀是 MAC 地址: .a.b.c.d.e.f
parts = oid.split('.')
if len(parts) >= 6:
mac_parts = parts[-6:]
mac_address = ':'.join(f'{int(p):02X}' for p in mac_parts)
if SNMPService.OID_DOT1D_TP_FDB_PORT in oid:
port = int(value) if isinstance(value, Integer32) else None
entry = {
'mac_address': mac_address,
'port_number': port,
'device_id': device.id,
'vlan_id': vlan_id
}
mac_entries.append(entry)
entry = {
'mac_address': mac_address,
'port_number': port,
'device_id': device.id,
'vlan_id': vlan_id
}
mac_entries.append(entry)
# 保存到数据库
now = datetime.utcnow()
@@ -361,7 +363,7 @@ class SNMPService:
return mac_entries
@staticmethod
def get_interfaces(device: NetworkDevice, credential: SNMPCredential, db: Session) -> List[Dict[str, Any]]:
async def get_interfaces(device: NetworkDevice, credential: SNMPCredential, db: Session) -> List[Dict[str, Any]]:
"""
获取设备接口信息
"""
@@ -379,7 +381,7 @@ class SNMPService:
# 获取接口信息
interface_data = {}
for errorIndication, errorStatus, errorIndex, varBinds in bulkCmd(
errorIndication, errorStatus, errorIndex, varBindTable = await bulkCmd(
SnmpEngine(), auth_data, transport, ContextData(),
0, 100,
ObjectType(ObjectIdentity(SNMPService.OID_IF_DESCR)),
@@ -390,35 +392,34 @@ class SNMPService:
ObjectType(ObjectIdentity(SNMPService.OID_IF_ADMIN_STATUS)),
ObjectType(ObjectIdentity(SNMPService.OID_IF_OPER_STATUS)),
lexicographicMode=False
):
if errorIndication or errorStatus:
break
)
if not errorIndication and not errorStatus:
for varBinds in varBindTable:
for varBind in varBinds:
oid = str(varBind[0])
value = varBind[1]
parts = oid.split('.')
if_index = parts[-1]
for varBind in varBinds:
oid = str(varBind[0])
value = varBind[1]
parts = oid.split('.')
if_index = parts[-1]
if if_index not in interface_data:
interface_data[if_index] = {'if_index': int(if_index)}
if if_index not in interface_data:
interface_data[if_index] = {'if_index': int(if_index)}
if SNMPService.OID_IF_DESCR in oid:
interface_data[if_index]['if_descr'] = value.prettyPrint()
elif SNMPService.OID_IF_TYPE in oid:
interface_data[if_index]['if_type'] = str(value)
elif SNMPService.OID_IF_MTU in oid:
interface_data[if_index]['if_mtu'] = int(value) if isinstance(value, Integer32) else None
elif SNMPService.OID_IF_SPEED in oid:
interface_data[if_index]['if_speed'] = int(value) if isinstance(value, (Integer32, Gauge32)) else None
elif SNMPService.OID_IF_PHYS_ADDRESS in oid:
interface_data[if_index]['if_phys_address'] = SNMPService._normalize_mac(value)
elif SNMPService.OID_IF_ADMIN_STATUS in oid:
status_map = {1: 'up', 2: 'down', 3: 'testing'}
interface_data[if_index]['if_admin_status'] = status_map.get(int(value), 'unknown')
elif SNMPService.OID_IF_OPER_STATUS in oid:
status_map = {1: 'up', 2: 'down', 3: 'testing'}
interface_data[if_index]['if_oper_status'] = status_map.get(int(value), 'unknown')
if SNMPService.OID_IF_DESCR in oid:
interface_data[if_index]['if_descr'] = value.prettyPrint()
elif SNMPService.OID_IF_TYPE in oid:
interface_data[if_index]['if_type'] = str(value)
elif SNMPService.OID_IF_MTU in oid:
interface_data[if_index]['if_mtu'] = int(value) if isinstance(value, Integer32) else None
elif SNMPService.OID_IF_SPEED in oid:
interface_data[if_index]['if_speed'] = int(value) if isinstance(value, (Integer32, Gauge32)) else None
elif SNMPService.OID_IF_PHYS_ADDRESS in oid:
interface_data[if_index]['if_phys_address'] = SNMPService._normalize_mac(value)
elif SNMPService.OID_IF_ADMIN_STATUS in oid:
status_map = {1: 'up', 2: 'down', 3: 'testing'}
interface_data[if_index]['if_admin_status'] = status_map.get(int(value), 'unknown')
elif SNMPService.OID_IF_OPER_STATUS in oid:
status_map = {1: 'up', 2: 'down', 3: 'testing'}
interface_data[if_index]['if_oper_status'] = status_map.get(int(value), 'unknown')
# 转换为列表并保存
now = datetime.utcnow()
@@ -485,7 +486,7 @@ class SNMPService:
return device
@staticmethod
def poll_device(db: Session, device_id: int) -> Dict[str, Any]:
async def poll_device(db: Session, device_id: int) -> Dict[str, Any]:
"""轮询设备,获取所有信息"""
device = db.query(NetworkDevice).filter(NetworkDevice.id == device_id).first()
if not device:
@@ -505,17 +506,17 @@ class SNMPService:
}
# 获取 ARP 表
arp_entries = SNMPService.get_arp_table(device, credential, db)
arp_entries = await SNMPService.get_arp_table(device, credential, db)
result['arp_entries'] = arp_entries
result['arp_count'] = len(arp_entries)
# 获取 MAC 地址表
mac_entries = SNMPService.get_mac_address_table(device, credential, db)
mac_entries = await SNMPService.get_mac_address_table(device, credential, db)
result['mac_entries'] = mac_entries
result['mac_count'] = len(mac_entries)
# 获取接口信息
interfaces = SNMPService.get_interfaces(device, credential, db)
interfaces = await SNMPService.get_interfaces(device, credential, db)
result['interfaces'] = interfaces
result['interface_count'] = len(interfaces)
+39
View File
@@ -0,0 +1,39 @@
"""
修复已有网段的 total_ips 字段:将 num_addresses 改为 hosts 数量(排除网络地址和广播地址)
用法: cd /root/ipam/backend && source venv/bin/activate && python scripts/fix_network_total_ips.py
"""
import sys
import os
# 确保能找到 app 模块
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from app.core.database import SessionLocal
from app.models.network import Network
from app.services.network_service import NetworkService
def fix_total_ips():
db = SessionLocal()
try:
networks = db.query(Network).all()
fixed = 0
for net in networks:
correct = NetworkService.calculate_total_ips(net.cidr)
actual_hosts = len(list(NetworkService.get_network_addresses(net.cidr)))
if net.total_ips != correct:
print(f" [{net.cidr:20s}] {net.total_ips:>5} -> {correct:>5} (hosts={actual_hosts})")
net.total_ips = correct
fixed += 1
db.commit()
print(f"\n✅ 已修复 {fixed} 个网段")
except Exception as e:
db.rollback()
print(f"❌ 错误: {e}")
finally:
db.close()
if __name__ == "__main__":
fix_total_ips()
+4 -3
View File
@@ -4,9 +4,10 @@
"version": "0.0.0",
"type": "module",
"scripts": {
"dev": "vite",
"build": "vite build",
"preview": "vite preview"
"dev": "node node_modules/vite/bin/vite.js",
"dev:clean": "rm -rf node_modules/.vite && node node_modules/vite/bin/vite.js --force",
"build": "node node_modules/vite/bin/vite.js build",
"preview": "node node_modules/vite/bin/vite.js preview"
},
"dependencies": {
"@element-plus/icons-vue": "^2.3.2",
+46 -5
View File
@@ -138,15 +138,41 @@ start_frontend() {
cd $FRONTEND_DIR
# 检查 node_modules
if [ ! -d "node_modules" ]; then
# 检查 node_modules 及关键依赖是否完整
if [ ! -d "node_modules" ] || [ ! -f "node_modules/element-plus/es/index.mjs" ]; then
echo -e "${YELLOW}安装前端依赖...${NC}"
npm install
fi
# 检查 OUI 数据库文件
if [ ! -f "$BACKEND_DIR/app/data/oui.txt" ]; then
echo -e "${YELLOW}OUI 数据库不存在,尝试从 IEEE 下载...${NC}"
cd $BACKEND_DIR
source venv/bin/activate
python3 -c "
from mac_vendor_lookup import MacLookup
import os
os.makedirs('app/data', exist_ok=True)
lookup = MacLookup()
lookup.update_vendors()
# 复制到项目目录
import shutil
src = os.path.expanduser('~/.cache/mac-vendors.txt')
if os.path.exists(src):
shutil.copy(src, 'app/data/oui.txt')
print('OUI 数据库下载完成')
" 2>/dev/null || echo -e "${YELLOW}OUI 数据库下载失败(网络问题),使用内置硬编码回退${NC}"
fi
# 检查端口并清理
kill_port $FRONTEND_PORT
# 清理 Vite 缓存,避免 element-plus exports 解析失败
if [ -d "node_modules/.vite" ]; then
echo -e "${YELLOW}清理 Vite 缓存...${NC}"
rm -rf node_modules/.vite
fi
# 后台启动前端
nohup npm run dev -- --host 0.0.0.0 --port $FRONTEND_PORT > /tmp/ipam-frontend.log 2>&1 &
FRONTEND_PID=$!
@@ -159,9 +185,24 @@ start_frontend() {
echo " 日志: /tmp/ipam-frontend.log"
return 0
else
echo -e "${RED}❌ 前端服务启动失败${NC}"
echo " 查看日志: tail -50 /tmp/ipam-frontend.log"
return 1
# 如果启动失败,尝试清缓存重试一次
echo -e "${YELLOW}首次启动失败,尝试清理 Vite 缓存重试...${NC}"
kill -9 $FRONTEND_PID 2>/dev/null
sleep 1
rm -rf node_modules/.vite
nohup npm run dev -- --host 0.0.0.0 --port $FRONTEND_PORT > /tmp/ipam-frontend.log 2>&1 &
FRONTEND_PID=$!
sleep 12
if curl -s http://localhost:$FRONTEND_PORT > /dev/null; then
echo -e "${GREEN}✅ 前端服务启动成功 (端口 $FRONTEND_PORT)${NC}"
echo " PID: $FRONTEND_PID"
echo " 日志: /tmp/ipam-frontend.log"
return 0
else
echo -e "${RED}❌ 前端服务启动失败${NC}"
echo " 查看日志: tail -50 /tmp/ipam-frontend.log"
return 1
fi
fi
echo ""
}