flightinfoV3_sever/scheduleCenter.py

import datetime
import json
import time
import hashlib
import os
import win32api
import hmac
from concurrent.futures import ThreadPoolExecutor as ConcurrentThreadPoolExecutor
import requests
from apscheduler.events import EVENT_JOB_EXECUTED , EVENT_JOB_ERROR
from apscheduler.executors.pool import ThreadPoolExecutor , ProcessPoolExecutor
from apscheduler.schedulers.background import BlockingScheduler
from unitls.settings import softVersion
oldVersion,nowVersion = softVersion()
current_path = os.getcwd()
if nowVersion:
    path = os.path.join(current_path,nowVersion)
else:
    path = None
#oldVersion="ServerWithJWT.exe"
#nowVersion="ServerWithJWT.exe"
#path=os.path.join("D:\\flightinfoV3_sever\dist\ServerWithJWT.exe")
print(path)
# 配置执行器
executors = {
    'default': ThreadPoolExecutor(20),  # 默认线程池，20个线程
    'processpool': ProcessPoolExecutor(5)  # 进程池，5个进程
}

# 配置任务默认参数
job_defaults = {
    'coalesce': False,  # 任务积压时是否合并执行
    'max_instances': 3,  # 最大并发实例数
    'misfire_grace_time': 60  # 任务误点容忍时间（秒）
}

# 创建调度器并配置执行器
scheduler = BlockingScheduler(
    executors=executors,
    job_defaults=job_defaults,
    timezone='Asia/Shanghai'
)
SIGNATURE_KEY=b"secret-key-neverout"  # 用于请求签名的密钥
RestartSts=False

def restart():
    global RestartSts
    if not RestartSts:
        RestartSts=True
        print(f"【{datetime.datetime.now()}】尝试关闭服务器")
        try:
            if oldVersion:
                os.system("taskkill /F /IM %s" % oldVersion)
                print(f"【{datetime.datetime.now()}】旧版关闭服务器成功")
            else:
                print(f"【{datetime.datetime.now()}】旧版服务器未设置，暂不关闭")
        except:
            print(f"【{datetime.datetime.now()}】尝试关闭旧版服务器失败")
        try:
            if nowVersion:
                os.system("taskkill /F /IM %s" % nowVersion)
                print(f"【{datetime.datetime.now()}】新版关闭服务器成功")
            else:
                print(f"【{datetime.datetime.now()}】新版服务器未设置，暂不关闭")
        except:
            print(f"【{datetime.datetime.now()}】尝试关闭新版服务器失败")
        print(f"【{datetime.datetime.now()}】尝试重启服务器")
        try:
            if path:
                win32api.ShellExecute(0 , 'open' , path , '' , '' , 1)
                print(f"【{datetime.datetime.now()}】新版启动成功")
            else:
                print(f"【{datetime.datetime.now()}】请先设置服务器启动版本")
        except:
            print(f"【{datetime.datetime.now()}】新版服务器启动失败")
        time.sleep(30)
        RestartSts=False
        print(f"【{datetime.datetime.now()}】重启环节结束")
def generate_signature() :

    # 1. 获取当前Unix时间戳（秒级）
    timestamp = str(int(time.time()))
    data = { "timestamp" : timestamp }
    # 2. 对数据进行JSON序列化，确保排序一致
    # sort_keys=True 保证键的顺序一致，否则签名会不匹配
    data_str = json.dumps(data , sort_keys = True).encode('utf-8')

    # 3. 组合数据：数据字符串 + | + 时间戳
    signature_base = f"{data_str.decode('utf-8')}|{timestamp}".encode('utf-8')

    # 4. 使用HMAC-SHA256算法生成签名
    signature = hmac.new(
        SIGNATURE_KEY ,
        signature_base ,
        hashlib.sha256
    ).hexdigest()

    return timestamp , signature
def my_listener(event):
    if event.exception:
        print(f"【{datetime.datetime.now()}】多线程异常,错误信息{event.exception}")
    else:
        print(f"【{datetime.datetime.now()}】多线程任务执行正常：{event}")



url = 'http://localhost:5001'
headers = {'Content-Type': 'application/json'}
num = 1
# 添加任务到不同执行器

# 定义A和B请求的处理函数（单独抽离）
def handle_flight_data_request(req_type):
    try:

        timestamp , signature = generate_signature()
        header = { 'Content-Type' : 'application/json' ,
                   'X-Timestamp' : timestamp ,
                   'X-Signature' : signature ,
                   }
        response = requests.post(url + '/getFlightData2',data=json.dumps({'type': req_type}),headers=header)
        response_json = response.json()
        print(f"【{datetime.datetime.now()}】{req_type}刷新;结果:{response_json}")
    except Exception as e:
        print(f"【{datetime.datetime.now()}】{req_type} 请求执行异常: {str(e)}")

def getflightdata():
    try:
        #print('getflightdata time:{}'.format(datetime.datetime.now()))
        global num
        """IO密集型任务，使用默认线程池"""

        timestamp , signature = generate_signature()
        header = {'Content-Type': 'application/json',
                    'X-Timestamp': timestamp,
                    'X-Signature': signature,
                    }
        response = requests.get(url+'/checkLogin', headers=header)
        print(f"【{datetime.datetime.now()}】检查账号登录情况;结果为:{response.json()['code']}{response.json()['msg']}")
        if response.json()['code'] == "0":
            print(f"【{datetime.datetime.now()}】AMRO未登录")
        else:
            with ConcurrentThreadPoolExecutor(max_workers = 2) as executor :
                if num > 2 :
                    # 异步执行A请求
                    executor.submit(handle_flight_data_request, 'A')
                    print(f"【{datetime.datetime.now()}】A刷新请求;")
                    num = 1
                else:
                    # 异步执行B请求
                    executor.submit(handle_flight_data_request, 'B')
                    print(f"【{datetime.datetime.now()}】B刷新请求;")
                    num += 1
            '''
            if num >2:
                getFlightDataA = requests.post(url+'/getFlightData2', data=json.dumps({'type':'A'}),headers=headers)
                # result = getFlightDataA.json()
                print(f"【{datetime.datetime.now()}】A 请求刷新;结果:{getFlightDataA.json()}")
                num = 1
            else:
                getFlightDataB = requests.post(url+'/getFlightData2', data=json.dumps({'type':'B'}),headers=headers)
                print(f"【{datetime.datetime.now()}】B 请求刷新;结果:{getFlightDataB.json()}")
                num += 1
            '''
    except:
         pass
def getTable():
    try:
        timestamp , signature = generate_signature()
        header = {'Content-Type': 'application/json',
                    'X-Timestamp': timestamp,
                    'X-Signature': signature,
                    }
        response = requests.get(url+'/table', headers=header)
        print(f"【{datetime.datetime.now()}】最近一次刷新为：{response.json()['data'][-1] if response.json()['data'] else response.json()['msg'] }")
        #print(result)
    except:
        pass
def postgresqlCheck():
    try:
        timestamp , signature = generate_signature()
        header = {'Content-Type': 'application/json',
                    'X-Timestamp': timestamp,
                    'X-Signature': signature,
                    }
        #print('postgresqlCheck time:{}'.format(datetime.datetime.now()))
        response = requests.get(url+'/postgresql', headers=header)
        print(f"【{datetime.datetime.now()}】云端计算请求：{response.json()}")
        #print(result)
    except:
        pass

def backupdata():
    try:
        timestamp , signature = generate_signature()
        header = {'Content-Type': 'application/json',
                    'X-Timestamp': timestamp,
                    'X-Signature': signature,
                    }
        #print('backupdata time:{}'.format(datetime.datetime.now()))
        response = requests.get(url+'/bakupdata', headers=header)
        print(f"【{datetime.datetime.now()}】5min备份请求：{response.json()}")
    except:
        pass
def health_check():

    try:
        timestamp , signature = generate_signature()
        header = {'Content-Type': 'application/json',
                    'X-Timestamp': timestamp,
                    'X-Signature': signature,
                    }
        #print('backupdata time:{}'.format(datetime.datetime.now()))
        response = requests.get(url+'/static/health', headers=header)
        print(f"【{datetime.datetime.now()}】监控请求：{response.json()}")
        if response.status_code != 200:
            restart()
        else:
            print(f"【{datetime.datetime.now()}】服务正常")
    except Exception as e:
        print(f"【{datetime.datetime.now()}】服务异常:{e}")
        restart()



def cpu_bound_job():
    """CPU密集型任务，使用进程池"""
    pass

scheduler.add_job(getflightdata, 'interval', seconds=30)
scheduler.add_job(getTable, 'interval', seconds=60)
scheduler.add_job(postgresqlCheck, 'interval', seconds=60)
scheduler.add_job(backupdata, 'interval', minutes=5)
scheduler.add_job(health_check, 'interval', seconds=10)
# scheduler.add_job(cpu_bound_job, 'interval', seconds=30, executor='processpool')
print('开启定时任务')
scheduler.add_listener(my_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)
scheduler.start()