tech-blog/app/services/visit_service.py

"""访问统计服务"""
from datetime import datetime, date, timedelta
from app import db
from app.models import VisitLog, DailyStats, Article


class VisitService:
    """访问统计业务逻辑"""
    
    @staticmethod
    def log_visit(request, article_id=None):
        """记录访问日志"""
        from flask import request as flask_request
        
        visit = VisitLog(
            ip_address=VisitService.get_client_ip(flask_request),
            user_agent=flask_request.user_agent.string if flask_request.user_agent else None,
            referer=flask_request.referrer,
            path=flask_request.path,
            method=flask_request.method,
            article_id=article_id
        )
        
        db.session.add(visit)
        db.session.commit()
        
        # 更新每日统计
        VisitService.update_daily_stats()
        
        return visit
    
    @staticmethod
    def get_client_ip(request):
        """获取客户端真实IP"""
        # 检查代理头
        if request.headers.get('X-Forwarded-For'):
            return request.headers.get('X-Forwarded-For').split(',')[0].strip()
        elif request.headers.get('X-Real-IP'):
            return request.headers.get('X-Real-IP')
        else:
            return request.remote_addr or '0.0.0.0'
    
    @staticmethod
    def update_daily_stats():
        """更新每日统计"""
        today = date.today()
        stats = DailyStats.query.filter_by(date=today).first()
        
        if not stats:
            stats = DailyStats(date=today)
            db.session.add(stats)
        
        # 计算今日数据
        today_start = datetime.combine(today, datetime.min.time())
        today_end = datetime.combine(today, datetime.max.time())
        
        stats.total_visits = VisitLog.query.filter(
            VisitLog.visited_at >= today_start,
            VisitLog.visited_at <= today_end
        ).count()
        
        stats.unique_ips = db.session.query(
            db.func.count(db.func.distinct(VisitLog.ip_address))
        ).filter(
            VisitLog.visited_at >= today_start,
            VisitLog.visited_at <= today_end
        ).scalar()
        
        stats.article_views = VisitLog.query.filter(
            VisitLog.visited_at >= today_start,
            VisitLog.visited_at <= today_end,
            VisitLog.article_id.isnot(None)
        ).count()
        
        # 找出今日最热门文章
        top_article_result = db.session.query(
            VisitLog.article_id,
            db.func.count(VisitLog.id).label('views')
        ).filter(
            VisitLog.visited_at >= today_start,
            VisitLog.visited_at <= today_end,
            VisitLog.article_id.isnot(None)
        ).group_by(VisitLog.article_id).order_by(db.desc('views')).first()
        
        if top_article_result:
            stats.top_article_id = top_article_result[0]
            stats.top_article_views = top_article_result[1]
        
        db.session.commit()
        return stats
    
    @staticmethod
    def get_stats_summary():
        """获取统计摘要"""
        return {
            'total_visits': VisitLog.get_total_count(),
            'today_visits': VisitLog.get_today_count(),
            'unique_ips': VisitLog.get_unique_ip_count(),
            'daily_stats': DailyStats.get_stats_list(30)
        }
    
    @staticmethod
    def get_visit_list(page=1, per_page=50, path_filter=None):
        """获取访问日志列表"""
        query = VisitLog.query
        
        if path_filter:
            query = query.filter(VisitLog.path.like(f'%{path_filter}%'))
        
        return query.order_by(VisitLog.visited_at.desc()).paginate(
            page=page, per_page=per_page, error_out=False
        )
    
    @staticmethod
    def get_top_ips(limit=10):
        """获取访问最多的IP"""
        from sqlalchemy import func
        
        result = db.session.query(
            VisitLog.ip_address,
            func.count(VisitLog.id).label('visit_count'),
            func.min(VisitLog.visited_at).label('first_visit'),
            func.max(VisitLog.visited_at).label('last_visit')
        ).group_by(VisitLog.ip_address).order_by(db.desc('visit_count')).limit(limit).all()
        
        return result
    
    @staticmethod
    def get_top_articles(days=7, limit=10):
        """获取热门文章"""
        start_date = date.today() - timedelta(days=days)
        start_datetime = datetime.combine(start_date, datetime.min.time())
        
        result = db.session.query(
            Article,
            db.func.count(VisitLog.id).label('views')
        ).join(VisitLog).filter(
            VisitLog.visited_at >= start_datetime,
            VisitLog.article_id.isnot(None)
        ).group_by(Article.id).order_by(db.desc('views')).limit(limit).all()
        
        return result
    
    @staticmethod
    def get_top_paths(limit=20):
        """获取访问最多的路径"""
        from sqlalchemy import func
        
        result = db.session.query(
            VisitLog.path,
            func.count(VisitLog.id).label('visit_count')
        ).group_by(VisitLog.path).order_by(db.desc('visit_count')).limit(limit).all()
        
        return result
    
    @staticmethod
    def clear_old_logs(days=90):
        """清理旧日志"""
        cutoff_date = datetime.now() - timedelta(days=days)
        deleted = VisitLog.query.filter(VisitLog.visited_at < cutoff_date).delete()
        db.session.commit()
        return deleted