ubuntu每五分钟执行企业微信机器人推送

# -*- coding: utf-8 -*-
"""
Created on Tue Apr 22 09:05:46 2025

@author: 1
"""

import requests
import time
import schedule
import sched
from datetime import datetime, timedelta
import threading
import pymysql  # 用于连接MySQL数据库
from mysql.connector import Error

# 信息等级定义
INFO_LEVELS = {
    '紧急': {'retry_interval': 60, 'max_retries': 3},
    '重要': {'retry_interval': 60, 'max_retries': 3},
    '一般': {'retry_interval': 60, 'max_retries': 3}
}

def send_message(group, message, info_level):
    headers = {'Content-Type': 'application/json'}
    payload = {
        "msgtype": "text",
        "text": {
            "content": message
        }
    }
    try:
        print(13)
        print(group)
        response = requests.post(group['robot_webhook'], json=payload, headers=headers)
        print(14)
        if response.status_code == 200 and response.json().get('errcode') == 0:
            print(15)
            print(f"[{datetime.now()}] 成功发送到群 {group['group_name']}")
            return True
        else:
            print(16)
            print(f"[{datetime.now()}] 发送到群 {group['group_name']} 失败，状态码: {response.status_code}")
            return False
    except Exception as e:
        print(17)
        print(f"[{datetime.now()}] 发送到群 {group['group_name']} 异常: {e}")
        return False

def retry_send(group, message, info_level, attempt=1):
    max_retries = INFO_LEVELS[info_level]['max_retries']
    retry_interval = INFO_LEVELS[info_level]['retry_interval']
    print(6)
    if attempt > max_retries:
        print(f"[{datetime.now()}] 最终发送失败到群 {group['group_name']}，将在1小时后重试")
        time.sleep(3600)  # 等待1小时后重试
        attempt = 1
    print(7)
    success = send_message(group, message, info_level)
    print(8)
    if not success:
        time.sleep(retry_interval)
        retry_send(group, message, info_level, attempt + 1)

def send_to_superior_groups(group, message, info_level):
    current_group = group
    while current_group['parent_group_id'] is not None:
        superior = find_group_by_id(current_group['parent_group_id'])
        if superior:
            retry_send(superior, message, info_level)
            current_group = superior
        else:
            break

def find_group_by_id(group_id):
    try:
        # 连接数据库
        connection = pymysql.connect(**DB_CONFIG)
        with connection.cursor() as cursor:
            # 查询未发送的紧急消息
            sql = "select id,group_name,level,robot_webhook,parent_group_id from wechat_groups where id = %s"
            cursor.execute(sql, (group_id,))
            results = cursor.fetchall()
            
            for row in results:
                target_group = row['id']  # 根据 upno 找到对应的群组
                if target_group == group_id:
                    return row
            return None
    except Exception as e:
        print(f"[{datetime.now()}] 遍历机器人时出错: {e}")
        return None
    finally:
        if connection:
            connection.close()
            
def find_group_by_name(group_name):
    try:
        # 连接数据库
        connection = pymysql.connect(**DB_CONFIG)
        with connection.cursor() as cursor:
            # 查询未发送的紧急消息
            sql = "select id,group_name,level,robot_webhook,parent_group_id from wechat_groups where group_name = %s"
            cursor.execute(sql, (group_name,))
            results = cursor.fetchall()
            
            for row in results:
                target_group = row['group_name']  # 根据 upno 找到对应的群组
                if target_group == group_name:
                    return row
            return None
    except Exception as e:
        print(f"[{datetime.now()}] 遍历机器人时出错: {e}")
        return None
    finally:
        if connection:
            connection.close()

def schedule_message(group, message, info_level, send_time):
    print(9)
    delay = (send_time - datetime.now()).total_seconds()
    print(10)
    if delay > 0:
        print(11)
        scheduler = sched.scheduler(time.time, time.sleep)
        scheduler.enterabs(send_time.timestamp(), 1, retry_send, argument=(group, message, info_level))
        scheduler.run()
    else:
        print(12)
        retry_send(group, message, info_level)

def send_message_with_scheduling(group, message, info_level, send_time=None):
    if send_time:
        schedule_message(group, message, info_level, send_time)
    else:
        retry_send(group, message, info_level)
    if group['parent_group_id'] is not None:
        send_to_superior_groups(group, message, info_level)

def run_scheduler():
    while True:
        schedule.run_pending()
        time.sleep(1)

# 数据库连接配置
DB_CONFIG = {
    'host': '*********',
    'user': 'root',
    'password': '************',
    'database': '*************',
    'charset': 'utf8mb4',
    'cursorclass': pymysql.cursors.DictCursor
}

def check_and_send_urgent_messages():
    try:
        # 连接数据库
        connection = pymysql.connect(**DB_CONFIG)
        with connection.cursor() as cursor:
            # 查询未发送的紧急消息
            sql = "SELECT id, sqlcmd, result, createdate, role, details, levelname, upno FROM ai_dayairesult WHERE levelname = '紧急' AND isend = 0"
            cursor.execute(sql)
            results = cursor.fetchall()
            print(1)
            
            for row in results:
                message = f"紧急通知: {row['result']}"  # 只发送result字段内容
                target_group = find_group_by_name(row['role'])  # 根据role字段找到对应的群组
                print(2)
                if target_group:
                    print(3)
                    print(target_group)
                    print(31)
                    send_message_with_scheduling(target_group, message, "紧急")
                    print(4)
                    # 更新数据库状态为已发送
                    update_sql = "UPDATE ai_dayairesult SET isend = 1 WHERE id = %s"
                    cursor.execute(update_sql, (row['id'],))
                    print(5)
            # 提交事务
            connection.commit()
    except Exception as e:
        print(f"[{datetime.now()}] 检查并发送紧急消息时出错: {e}")
    finally:
        if connection:
            connection.close()

def schedule_next_check():
    # 设置每5分钟检查一次紧急消息
    schedule.every(5).minutes.do(check_and_send_urgent_messages)

# 示例用法
if __name__ == "__main__":
    # 启动调度器线程
    scheduler_thread = threading.Thread(target=run_scheduler)
    scheduler_thread.daemon = True
    scheduler_thread.start()
    
    # 设置定时检查紧急消息
    schedule_next_check()
    
    # 立即执行一次检查
    check_and_send_urgent_messages()
    
    # 保持主线程运行
    while True:
        schedule.run_pending()
        time.sleep(1)