H3ConuMS-v2/backend/app/tasks/check_tasks.py

"""状态检查任务"""
import time
import traceback
import redis as redis_lib
from datetime import datetime, timedelta
from sqlalchemy import func, case
from app.core.celery_app import celery_app
from app.core.database import SessionLocal
from app.core.config import settings
from app.services.check_service import CheckService

_LAST_RUN_KEY = "check_all_devices:last_run"
_RUNNING_KEY = "check_all_devices:running"
_INTERVAL_REDIS_KEY = "system:check_interval_seconds"


def _get_redis():
    return redis_lib.from_url(settings.REDIS_URL, decode_responses=True)


def _get_check_interval(r) -> int:
    """从 Redis 读取配置间隔，回退到 DB，再回退到默认值"""
    cached = r.get(_INTERVAL_REDIS_KEY)
    if cached:
        return int(cached)
    # 从 DB 读取并缓存
    db = SessionLocal()
    try:
        from app.models.setting import SystemSetting
        setting = db.query(SystemSetting).filter_by(key='check_interval_seconds').first()
        interval = int(setting.value) if setting else settings.CHECK_INTERVAL
        r.set(_INTERVAL_REDIS_KEY, str(interval))
        return interval
    finally:
        db.close()


@celery_app.task(bind=True)
def check_all_devices(self):
    """检查所有设备状态（支持可配置间隔，最小5分钟）"""
    r = _get_redis()
    interval = _get_check_interval(r)

    # Redis 节流：检查距上次运行是否已超过配置间隔
    last_run = r.get(_LAST_RUN_KEY)
    now = time.time()
    if last_run and (now - float(last_run)) < interval:
        remaining = int(interval - (now - float(last_run)))
        return {'skipped': True, 'reason': f'间隔未到，还需等待 {remaining} 秒', 'interval': interval}

    # 标记正在运行（TTL 10分钟防止异常时永久卡住）
    r.set(_RUNNING_KEY, '1', ex=600)

    db = SessionLocal()
    self.update_state(state='PROGRESS', meta={'current': 0, 'total': 0, 'status': '获取OLT列表...'})
    try:
        service = CheckService(db)
        from app.models.device import OLTDevice
        olts = db.query(OLTDevice).all()

        total = len(olts)
        self.update_state(state='PROGRESS', meta={'current': 0, 'total': total, 'status': f'准备检查 {total} 个OLT...'})

        results = []
        errors = []
        total_online = 0
        total_offline = 0

        for idx, olt in enumerate(olts):
            self.update_state(state='PROGRESS', meta={
                'current': idx, 'total': total,
                'status': f'检查 OLT: {olt.location or olt.ip_address}...'
            })
            try:
                result = service.update_status_only(olt.id)
                total_online += result.get('online', 0)
                total_offline += result.get('offline', 0)
                results.append({
                    'olt_id': olt.id,
                    'olt_name': olt.location or olt.ip_address,
                    'online': result.get('online', 0),
                    'offline': result.get('offline', 0),
                    'success': True
                })
            except Exception as e:
                errors.append({
                    'olt_id': olt.id,
                    'olt_name': olt.location or olt.ip_address,
                    'error': str(e)
                })
                results.append({
                    'olt_id': olt.id,
                    'olt_name': olt.location or olt.ip_address,
                    'success': False,
                    'error': str(e)
                })

        self.update_state(state='PROGRESS', meta={'current': total, 'total': total, 'status': '检查完成'})

        # 通知 WebSocket 客户端状态已更新
        try:
            import json as _json
            r.publish("h3c_onu:status_updates", _json.dumps({
                "type": "check_complete", "total_online": total_online,
                "total_offline": total_offline, "total_olts": total
            }))
        except Exception:
            pass

        return {
            'success': True,
            'total_olts': total,
            'total_online': total_online,
            'total_offline': total_offline,
            'results': results,
            'errors': errors
        }
    except Exception as e:
        return {
            'success': False,
            'error': str(e),
            'traceback': traceback.format_exc()
        }
    finally:
        # 任务完成后记录时间、清除运行标记
        r.set(_LAST_RUN_KEY, str(time.time()))
        r.delete(_RUNNING_KEY)
        db.close()


@celery_app.task
def aggregate_daily_snapshot():
    """聚合昨日设备状态快照（每天凌晨执行）"""
    from app.models.device import DeviceStatusHistory, DeviceDailySnapshot, ONUDevice
    db = SessionLocal()
    try:
        yesterday = (datetime.utcnow() - timedelta(days=1)).date()
        date_str = yesterday.strftime('%Y-%m-%d')

        # 如果已存在则跳过（幂等）
        exists = db.query(DeviceDailySnapshot).filter_by(snapshot_date=date_str).first()
        if exists:
            return {'skipped': True, 'date': date_str}

        # 昨天每台设备的最后一次检查状态
        day_start = datetime.combine(yesterday, datetime.min.time())
        day_end = datetime.combine(yesterday, datetime.max.time())

        daily_latest_subq = (
            db.query(
                DeviceStatusHistory.onu_device_id,
                func.max(DeviceStatusHistory.checked_at).label("max_checked_at"),
            )
            .filter(DeviceStatusHistory.checked_at.between(day_start, day_end))
            .group_by(DeviceStatusHistory.onu_device_id)
            .subquery()
        )

        row = (
            db.query(
                func.count().label("total"),
                func.sum(case((DeviceStatusHistory.status == 'online', 1), else_=0)).label("online"),
                func.sum(case((DeviceStatusHistory.status == 'offline', 1), else_=0)).label("offline"),
            )
            .join(
                daily_latest_subq,
                (DeviceStatusHistory.onu_device_id == daily_latest_subq.c.onu_device_id) &
                (DeviceStatusHistory.checked_at == daily_latest_subq.c.max_checked_at)
            )
            .one()
        )

        snapshot = DeviceDailySnapshot(
            snapshot_date=date_str,
            total=int(row.total or 0),
            online=int(row.online or 0),
            offline=int(row.offline or 0),
        )
        db.add(snapshot)
        db.commit()
        return {'success': True, 'date': date_str, 'total': snapshot.total, 'online': snapshot.online}
    except Exception as e:
        db.rollback()
        return {'success': False, 'error': str(e), 'traceback': traceback.format_exc()}
    finally:
        db.close()


@celery_app.task
def cleanup_status_history():
    """清理 30 天前的设备状态历史记录（每天凌晨执行）"""
    from app.models.device import DeviceStatusHistory
    db = SessionLocal()
    try:
        cutoff = datetime.utcnow() - timedelta(days=30)
        deleted = db.query(DeviceStatusHistory).filter(
            DeviceStatusHistory.checked_at < cutoff
        ).delete(synchronize_session=False)
        db.commit()
        return {"deleted": deleted}
    finally:
        db.close()