rssKeeper/backend/health_checker.py

"""RSS 源健康度检测"""
from datetime import datetime, timedelta
from typing import List, Dict
from sqlalchemy import func
from sqlalchemy.orm import Session
from models import Feed, FetchLog


def get_feed_health(db: Session, feed_id: int = None) -> List[Dict]:
    """获取 RSS 源健康度信息
    返回每个源的健康状态详情
    """
    now = datetime.utcnow()
    query = db.query(Feed)
    if feed_id:
        query = query.filter(Feed.id == feed_id)

    feeds = query.all()
    results = []

    for feed in feeds:
        total = feed.success_count + feed.fail_count
        success_rate = round(feed.success_count / total * 100, 1) if total > 0 else 0

        days_since_fetch = None
        if feed.last_fetch_at:
            days_since_fetch = (now - feed.last_fetch_at).days

        # 获取最近 7 天抓取记录
        week_ago = now - timedelta(days=7)
        recent_logs = db.query(FetchLog).filter(
            FetchLog.feed_id == feed.id,
            FetchLog.created_at >= week_ago
        ).order_by(FetchLog.created_at.desc()).limit(10).all()

        health = feed.health_status(now=now)

        results.append({
            "id": feed.id,
            "title": feed.title or feed.url,
            "url": feed.url,
            "is_active": feed.is_active,
            "health_status": health,
            "health_label": _health_label(health),
            "success_rate": success_rate,
            "success_count": feed.success_count,
            "fail_count": feed.fail_count,
            "total_fetches": total,
            "last_fetch_at": feed.last_fetch_at.isoformat() if feed.last_fetch_at else None,
            "days_since_fetch": days_since_fetch,
            "article_count": feed.article_count,
            "last_error": feed.last_error,
            "recent_logs": [
                {
                    "status": log.status,
                    "articles_fetched": log.articles_fetched,
                    "response_time_ms": log.response_time_ms,
                    "created_at": log.created_at.isoformat(),
                    "error_message": log.error_message if log.status == "fail" else None,
                }
                for log in recent_logs
            ],
        })

    return results


def _health_label(status: str) -> str:
    labels = {
        "healthy": "健康",
        "warning": "警告",
        "unhealthy": "异常",
        "unknown": "未知",
    }
    return labels.get(status, "未知")


def get_overall_stats(db: Session) -> Dict:
    """获取整体统计信息"""
    total_feeds = db.query(Feed).count()
    active_feeds = db.query(Feed).filter(Feed.is_active == True).count()
    total_articles_count = db.query(func.sum(Feed.article_count)).scalar() or 0

    # 健康源统计
    feeds = db.query(Feed).all()
    healthy = warning = unhealthy = 0
    now = datetime.utcnow()
    for feed in feeds:
        status = feed.health_status(now=now)
        if status == "healthy":
            healthy += 1
        elif status == "warning":
            warning += 1
        elif status == "unhealthy":
            unhealthy += 1

    # 今日抓取
    today = now.replace(hour=0, minute=0, second=0, microsecond=0)
    today_fetches = db.query(FetchLog).filter(FetchLog.created_at >= today).count()
    today_success = db.query(FetchLog).filter(
        FetchLog.created_at >= today, FetchLog.status == "success"
    ).count()

    return {
        "total_feeds": total_feeds,
        "active_feeds": active_feeds,
        "total_articles": total_articles_count,
        "healthy_feeds": healthy,
        "warning_feeds": warning,
        "unhealthy_feeds": unhealthy,
        "today_fetches": today_fetches,
        "today_success": today_success,
        "today_success_rate": round(today_success / today_fetches * 100, 1) if today_fetches > 0 else 0,
    }