CoolBotDataSys/scripts/crawler_monitor.py

#!/usr/bin/env python3
"""
Crawler 健康监控 - 检查所有服务器爬虫是否正常运行
每10分钟运行一次，检查最近30分钟内各实例是否有记录
"""
import sys
import os
sys.path.insert(0, '/root/coolbot-data')

import psycopg2
import requests
from datetime import datetime, timedelta

# ========== 配置 ==========
# 各服务器实例ID（对应crawl_logs.instance_id）
INSTANCES = {
    'server_1': '1号服务器-每00/30分',
    'server_2': '2号服务器-每10/40分',
    'server_3': '3号服务器-每20/50分',
}

DB_CONFIG = {
    'host': os.environ.get('DB_HOST', 'pgm-bp1t1008h019ez6c.pg.rds.aliyuncs.com'),
    'port': int(os.environ.get('DB_PORT', 5432)),
    'database': os.environ.get('DB_NAME', 'coolbot_data'),
    'user': os.environ.get('DB_USER', 'coolbot'),
    'password': os.environ.get('DB_PASSWORD', 'Coolbot123'),
}

# 飞书 Webhook
FEISHU_WEBHOOK = os.environ.get('FEISHU_WEBHOOK_URL', '')
FEISHU_SECRET = os.environ.get('FEISHU_APP_SECRET', '')

# 检查时间窗口（分钟）
WINDOW_MINUTES = 35  # 宽松5分钟容错

# ==========================

def get_db():
    return psycopg2.connect(**DB_CONFIG)

def check_instance_status(conn, instance_id):
    """检查某实例最近是否在窗口内运行过"""
    cur = conn.cursor()
    cur.execute("""
        SELECT COUNT(*), MAX(finished_at) 
        FROM crawl_logs 
        WHERE instance_id = %s 
          AND finished_at >= NOW() - INTERVAL '%s minutes'
          AND status = 'success'
    """, (instance_id, WINDOW_MINUTES))
    r = cur.fetchone()
    cur.close()
    return {'count': r[0], 'last_run': r[1]}

def send_feishu_alert(message):
    """发送飞书告警"""
    if not FEISHU_WEBHOOK:
        print("⚠️ 未配置飞书Webhook，跳过告警")
        return
    
    now = datetime.now().strftime('%H:%M')
    card = {
        "msg_type": "interactive",
        "card": {
            "header": {
                "title": {"tag": "plain_text", "content": f"⚠️ 爬虫监控告警 {now}"},
                "template": "red"
            },
            "elements": [
                {"tag": "div", "text": {"tag": "lark_md", "content": message}}
            ]
        }
    }
    try:
        r = requests.post(FEISHU_WEBHOOK, json=card, timeout=10)
        print(f"飞书告警响应: {r.status_code}")
    except Exception as e:
        print(f"飞书发送失败: {e}")

def send_feishu_healthy():
    """发送健康状态（仅关键时段发送，避免骚扰）"""
    hour = datetime.now().hour
    if hour not in [8, 9, 17, 18, 19]:  # 只在工作时段发送
        return
    
    if not FEISHU_WEBHOOK:
        return
    
    card = {
        "msg_type": "interactive",
        "card": {
            "header": {
                "title": {"tag": "plain_text", "content": f"✅ 爬虫监控正常 {datetime.now().strftime('%H:%M')}"},
                "template": "green"
            },
            "elements": [
                {"tag": "div", "text": {"tag": "lark_md", "content": "所有服务器爬虫运行正常"}}
            ]
        }
    }
    try:
        requests.post(FEISHU_WEBHOOK, json=card, timeout=10)
    except:
        pass

def main():
    print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 爬虫健康检查...")
    
    conn = get_db()
    
    all_ok = True
    details = []
    
    for instance_id, name in INSTANCES.items():
        status = check_instance_status(conn, instance_id)
        last_run_str = status['last_run'].strftime('%H:%M') if status['last_run'] else '从未运行'
        
        if status['count'] > 0:
            details.append(f"✅ {name}: 最近 {last_run_str} (共{status['count']}次)")
        else:
            details.append(f"❌ {name}: ⚠️ 超过{WINDOW_MINUTES}分钟无记录！")
            all_ok = False
    
    conn.close()
    
    print('\n'.join(details))
    
    if all_ok:
        print("✅ 所有服务器正常")
        send_feishu_healthy()
    else:
        msg = "**爬虫告警：部分服务器超过30分钟无爬取记录**\n\n" + '\n'.join(details)
        print(f"\n❌ 告警: {msg}")
        send_feishu_alert(msg)
    
    return 0 if all_ok else 1

if __name__ == '__main__':
    sys.exit(main())