"""监控与统计服务。""" import logging import os import psutil from collections import deque from datetime import datetime, timezone, timedelta from models import TrafficSnapshot, Connection, AuditLog, Instance, User from database import db log = logging.getLogger("socks.stats") # 实时指标缓冲区（内存） _metrics_buffer = deque(maxlen=600) # 10分钟 @ 10秒间隔 _metrics_lock = __import__("threading").Lock() def _record_metrics(): """记录系统级指标。""" try: cpu = psutil.cpu_percent(interval=0.5) mem = psutil.virtual_memory() net = psutil.net_io_counters() # 计算活跃连接 with db.app.app_context(): active = db.session.query(Connection).filter_by( state="active" ).count() item = { "ts": datetime.now(timezone.utc).isoformat(), "cpu": cpu, "mem_percent": mem.percent, "mem_used_mb": mem.used // 1024 // 1024, "mem_total_mb": mem.total // 1024 // 1024, "net_bytes_sent": net.bytes_sent, "net_bytes_recv": net.bytes_recv, "active_connections": active, } with _metrics_lock: _metrics_buffer.append(item) except Exception as e: log.error("记录指标失败: %s", e) def get_realtime_metrics(): """获取实时系统指标。""" _record_metrics() with _metrics_lock: items = list(_metrics_buffer) if not items: _record_metrics() items = list(_metrics_buffer) return { "latest": items[-1] if items else None, "history": items[-60:], # 最近60个点（10分钟） } def get_traffic_trend(days=30): """获取流量趋势（每日快照）。""" with db.app.app_context(): start = datetime.now(timezone.utc).date() - timedelta(days=days) snapshots = db.session.query( TrafficSnapshot.date, db.func.sum(TrafficSnapshot.bytes_in).label("total_in"), db.func.sum(TrafficSnapshot.bytes_out).label("total_out"), db.func.sum(TrafficSnapshot.connections).label("total_conn"), ).filter( TrafficSnapshot.date >= start ).group_by(TrafficSnapshot.date).order_by( TrafficSnapshot.date ).all() return [{ "date": s.date.isoformat(), "bytes_in_mb": round((s.total_in or 0) / 1024 / 1024, 1), "bytes_out_mb": round((s.total_out or 0) / 1024 / 1024, 1), "connections": s.total_conn or 0, } for s in snapshots] def get_user_stats(): """获取各用户流量排名。""" from sqlalchemy import func with db.app.app_context(): users = db.session.query( User.username, User.bytes_in, User.bytes_out, User.max_concurrent, User.total_traffic_mb, (User.bytes_in + User.bytes_out).label("total_bytes"), ).filter_by(enabled=True).order_by( (User.bytes_in + User.bytes_out).desc() ).limit(20).all() return [{ "username": u.username, "bytes_in_mb": round(u.bytes_in / 1024 / 1024, 1), "bytes_out_mb": round(u.bytes_out / 1024 / 1024, 1), "total_mb": round(u.total_bytes / 1024 / 1024, 1), "limit_mb": u.total_traffic_mb, "usage_pct": round( u.total_bytes / 1024 / 1024 / u.total_traffic_mb * 100, 1 ) if u.total_traffic_mb else None, } for u in users] def get_instance_stats(): """获取各实例流量统计。""" with db.app.app_context(): instances = Instance.query.all() result = [] for inst in instances: conns = db.session.query( db.func.sum(Connection.bytes_in).label("total_in"), db.func.sum(Connection.bytes_out).label("total_out"), db.func.count(Connection.id).label("total_conn"), ).filter( Connection.instance_id == inst.id ).first() result.append({ "id": inst.id, "name": inst.name, "host": inst.listen_host, "port": inst.listen_port, "bytes_in_mb": round((conns.total_in or 0) / 1024 / 1024, 1), "bytes_out_mb": round((conns.total_out or 0) / 1024 / 1024, 1), "total_connections": conns.total_conn or 0, }) return result def get_active_connections(): """获取当前活跃连接列表。""" with db.app.app_context(): conns = db.session.query(Connection).filter_by( state="active" ).order_by(Connection.started_at.desc()).limit(50).all() return [{ "id": c.id, "instance": db.session.query(Instance).get(c.instance_id).name if c.instance_id else "", "username": c.username, "src": f"{c.src_ip}:{c.src_port}" if c.src_ip else "", "dst": f"{c.dst_addr}:{c.dst_port}" if c.dst_addr else "", "protocol": c.protocol, "started": c.started_at.isoformat() if c.started_at else "", "bytes_in": c.bytes_in, "bytes_out": c.bytes_out, } for c in conns] def get_dashboard_summary(): """仪表盘汇总。""" metrics = get_realtime_metrics() with db.app.app_context(): total_instances = db.session.query(Instance).count() running_instances = db.session.query(Instance).filter_by( enabled=True ).count() total_users = db.session.query(User).count() active_users = db.session.query(User).filter_by( enabled=True, banned=False ).count() active_conns = db.session.query(Connection).filter_by( state="active" ).count() today_logs = db.session.query(AuditLog).filter( AuditLog.timestamp >= datetime.now(timezone.utc).replace( hour=0, minute=0, second=0 ) ).count() return { "system": metrics["latest"], "total_instances": total_instances, "running_instances": running_instances, "total_users": total_users, "active_users": active_users, "active_connections": active_conns, "today_logs": today_logs, } def prune_old_data(): """清理旧数据（运行在后台）。""" with db.app.app_context(): # 清理 30 天前的连接记录（closed的） cutoff = datetime.now(timezone.utc) - timedelta(days=30) db.session.query(Connection).filter( Connection.state == "closed", Connection.started_at < cutoff ).delete(synchronize_session=False) # 清理 90 天前的审计日志 cutoff2 = datetime.now(timezone.utc) - timedelta(days=90) db.session.query(AuditLog).filter(AuditLog.timestamp < cutoff2).delete( synchronize_session=False ) db.session.commit() log.info("数据清理完成")