diff --git a/backend/app/tasks/celery_app.py b/backend/app/tasks/celery_app.py index c0e7350..aa03af4 100644 --- a/backend/app/tasks/celery_app.py +++ b/backend/app/tasks/celery_app.py @@ -32,6 +32,7 @@ scan_single_ip_task = _tasks['scan_single_ip_task'] full_network_scan = _tasks['full_network_scan'] update_all_statistics = _tasks['update_all_statistics'] quick_status_update = _tasks['quick_status_update'] +poll_snmp_devices = _tasks['poll_snmp_devices'] # 定时任务配置 celery_app.conf.beat_schedule = { @@ -52,4 +53,9 @@ celery_app.conf.beat_schedule = { 'task': 'app.tasks.scan_tasks.update_all_statistics', 'schedule': 900.0, # 15分钟 }, + # 每60秒检查一次 SNMP 设备轮询(各设备按自己的 poll_interval 判断是否到期) + 'poll-snmp-devices-every-60s': { + 'task': 'app.tasks.scan_tasks.poll_snmp_devices', + 'schedule': 60.0, + }, } diff --git a/backend/app/tasks/scan_tasks.py b/backend/app/tasks/scan_tasks.py index ae2f273..12c67cc 100644 --- a/backend/app/tasks/scan_tasks.py +++ b/backend/app/tasks/scan_tasks.py @@ -212,11 +212,75 @@ def register_tasks(celery_app): finally: db.close() + @celery_app.task(bind=True) + def poll_snmp_devices(self): + """ + 定时任务:自动轮询所有活跃且配置了 SNMP 凭据的网络设备。 + + 每个设备按自己的 arp/mac/interface_poll_interval(秒)判断是否到达 + 轮询间隔,到期才执行 SNMPService.poll_device(采集 ARP 表、MAC 地址 + 表、接口信息)。由 Celery Beat 每 60 秒触发一次检查。 + """ + import asyncio + from datetime import datetime + + from app.models.snmp import NetworkDevice + from app.services.snmp_service import SNMPService + + db = SessionLocal() + try: + devices = db.query(NetworkDevice).filter( + NetworkDevice.is_active == True, + NetworkDevice.snmp_credential_id.isnot(None), + ).all() + + now = datetime.utcnow() + + def _as_naive(dt): + """把 aware/naive datetime 统一转 naive(本地比较用)""" + if dt is None: + return None + if dt.tzinfo is not None: + return dt.replace(tzinfo=None) + return dt + + polled, skipped = 0, 0 + for dev in devices: + # 取三种采集间隔的最小值作为该设备的轮询周期 + intervals = [ + dev.arp_poll_interval, + dev.mac_poll_interval, + dev.interface_poll_interval, + ] + interval = min([i for i in intervals if i], default=300) + + last = _as_naive(dev.last_polled_at) + if last is not None: + elapsed = (now - last).total_seconds() + if elapsed < interval: + skipped += 1 + continue + + try: + # poll_device 是 async 函数,用 asyncio.run 在同步任务中执行 + asyncio.run(SNMPService.poll_device(db, dev.id)) + polled += 1 + logger.info(f"SNMP 自动轮询完成: {dev.name} ({dev.ip_address})") + except Exception as e: + logger.error(f"SNMP 自动轮询失败 {dev.name} ({dev.ip_address}): {e}", exc_info=True) + + logger.info(f"SNMP 自动轮询检查完成: 设备总数 {len(devices)}, 本轮轮询 {polled}, 未到期跳过 {skipped}") + return {'status': 'completed', 'total': len(devices), 'polled': polled, 'skipped': skipped} + finally: + db.close() + # 返回任务函数供外部使用 return { 'scan_network_task': scan_network_task, 'scan_single_ip_task': scan_single_ip_task, 'full_network_scan': full_network_scan, 'update_all_statistics': update_all_statistics, - 'quick_status_update': quick_status_update + 'quick_status_update': quick_status_update, + 'poll_snmp_devices': poll_snmp_devices, } +