import schedule
import time

def sync_data():
    print("開(kāi)始同步數(shù)據(jù)...")
    # 模擬一些工作
    time.sleep(10)
    print("數(shù)據(jù)同步完成！")

# 每5分鐘執(zhí)行一次
schedule.every(5).minutes.do(sync_data)

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

import threading
import schedule
import time
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

def run_threaded(job_func):
    """將任務(wù)函數(shù)放入線程中執(zhí)行"""
    job_thread = threading.Thread(target=job_func)
    job_thread.start()
    # 注意：這里我們不調(diào)用 job_thread.join()

schedule.every(5).minutes.do(run_threaded, sync_data)

def sync_data():
    """數(shù)據(jù)同步任務(wù)"""
    try:
        logger.info("數(shù)據(jù)同步任務(wù)開(kāi)始。")
        # 這里是你的核心業(yè)務(wù)邏輯，比如調(diào)用API、讀寫(xiě)數(shù)據(jù)庫(kù)等
        time.sleep(10)  # 模擬耗時(shí)操作
        # 模擬一個(gè)可能的錯(cuò)誤
        # 1 / 0
        logger.info("數(shù)據(jù)同步任務(wù)成功完成。")
    except Exception as e:
        # 捕獲所有異常，記錄錯(cuò)誤日志，避免線程異常導(dǎo)致整個(gè)程序退出
        logger.error(f"數(shù)據(jù)同步任務(wù)發(fā)生嚴(yán)重錯(cuò)誤: {e}", exc_info=True)

def run_scheduler():
    """運(yùn)行調(diào)度器的循環(huán)"""
    logger.info("定時(shí)任務(wù)調(diào)度器開(kāi)始運(yùn)行...")
    while True:
        try:
            schedule.run_pending()
            time.sleep(1)  # 避免CPU空轉(zhuǎn)
        except KeyboardInterrupt:
            logger.info("接收到中斷信號(hào)，調(diào)度器正在停止...")
            break
        except Exception as e:
            # 捕獲調(diào)度器本身的未知異常，記錄并繼續(xù)運(yùn)行
            logger.critical(f"調(diào)度器主循環(huán)發(fā)生未知異常: {e}", exc_info=True)
            time.sleep(60)  # 發(fā)生嚴(yán)重錯(cuò)誤時(shí)，等待一分鐘再重試循環(huán)

# 創(chuàng)建一個(gè)全局鎖，用于保護(hù)共享資源
data_file_lock = threading.Lock()

def sync_data():
    try:
        logger.info("數(shù)據(jù)同步任務(wù)開(kāi)始。")
        # 在操作共享文件前獲取鎖
        with data_file_lock:
            logger.info("獲取到文件鎖，開(kāi)始寫(xiě)入數(shù)據(jù)...")
            time.sleep(5)  # 模擬文件操作
            logger.info("數(shù)據(jù)寫(xiě)入完成，釋放文件鎖。")
        logger.info("數(shù)據(jù)同步任務(wù)成功完成。")
    except Exception as e:
        logger.error(f"數(shù)據(jù)同步任務(wù)發(fā)生錯(cuò)誤: {e}", exc_info=True)

"""
一個(gè)健壯的、基于 schedule 庫(kù)的 Python 定時(shí)任務(wù)執(zhí)行器。
適用于需要 7x24 小時(shí)運(yùn)行的后臺(tái)服務(wù)。
"""
import schedule
import time
import threading
import logging
from datetime import datetime

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)
logger = logging.getLogger(__name__)

# 示例：用于保護(hù)共享資源的鎖
shared_resource_lock = threading.Lock()

def run_threaded(job_func):
    """
    將任務(wù)函數(shù)放入獨(dú)立線程中執(zhí)行，避免阻塞主調(diào)度循環(huán)。
    Args:
        job_func: 要執(zhí)行的任務(wù)函數(shù)
    """
    job_thread = threading.Thread(target=job_func, name=job_func.__name__)
    job_thread.start()
    # 重要：不在此處join，防止阻塞調(diào)度器。

def sync_data():
    """示例任務(wù)1：模擬數(shù)據(jù)同步"""
    task_name = "數(shù)據(jù)同步"
    try:
        logger.info(f"[{task_name}] 任務(wù)開(kāi)始。")
        # 模擬一些網(wǎng)絡(luò)I/O或數(shù)據(jù)庫(kù)操作
        time.sleep(12)  # 假設(shè)本次執(zhí)行需要12秒
        # 模擬操作共享資源
        with shared_resource_lock:
            logger.info(f"[{task_name}] 獲取鎖，操作共享資源...")
            time.sleep(1)
        # 模擬一個(gè)偶然的錯(cuò)誤（取消注釋來(lái)測(cè)試）
        # if datetime.now().second % 10 == 0:
        #     raise ConnectionError("模擬網(wǎng)絡(luò)突然斷開(kāi)")
        logger.info(f"[{task_name}] 任務(wù)成功完成。")
    except Exception as e:
        logger.error(f"[{task_name}] 執(zhí)行失敗: {e}", exc_info=True)

def cleanup_temp():
    """示例任務(wù)2：模擬清理臨時(shí)文件"""
    task_name = "清理臨時(shí)文件"
    try:
        logger.info(f"[{task_name}] 任務(wù)開(kāi)始。")
        time.sleep(3)  # 模擬清理工作
        logger.info(f"[{task_name}] 任務(wù)完成。")
    except Exception as e:
        logger.error(f"[{task_name}] 執(zhí)行失敗: {e}", exc_info=True)

def send_heartbeat():
    """示例任務(wù)3：發(fā)送心跳信號(hào)，證明調(diào)度器還活著"""
    try:
        # 這是一個(gè)非?？斓娜蝿?wù)，不需要放線程也行，這里為了統(tǒng)一也放了。
        logger.info("心跳檢測(cè)：調(diào)度器運(yùn)行正常。")
    except Exception as e:
        logger.error(f"心跳任務(wù)失敗: {e}", exc_info=True)

def run_scheduler():
    """調(diào)度器主循環(huán)"""
    logger.info("=== 定時(shí)任務(wù)調(diào)度器啟動(dòng) ===")

    # 1. 定義任務(wù)調(diào)度規(guī)則
    # 每5分鐘執(zhí)行一次數(shù)據(jù)同步（使用線程）
    schedule.every(5).minutes.do(run_threaded, sync_data)
    # 每小時(shí)執(zhí)行一次清理
    schedule.every().hour.do(run_threaded, cleanup_temp)
    # 每30秒發(fā)送一次心跳
    schedule.every(30).seconds.do(run_threaded, send_heartbeat)

    # 2. 啟動(dòng)后立即運(yùn)行一次某些任務(wù)（可選）
    logger.info("啟動(dòng)后立即執(zhí)行一次清理和心跳...")
    run_threaded(cleanup_temp)
    run_threaded(send_heartbeat)

    # 3. 堅(jiān)不可摧的主循環(huán)
    while True:
        try:
            schedule.run_pending()
            # 睡眠1秒，平衡CPU使用率和調(diào)度精度
            time.sleep(1)
        except KeyboardInterrupt:
            logger.info("接收到鍵盤(pán)中斷信號(hào)，正在優(yōu)雅停止調(diào)度器...")
            break
        except Exception as e:
            # 兜底：捕獲調(diào)度循環(huán)自身的任何異常，記錄并嘗試恢復(fù)
            logger.critical(f"調(diào)度器主循環(huán)發(fā)生未預(yù)期異常，將在一分鐘后重試: {e}", exc_info=True)
            time.sleep(60)

    logger.info("=== 定時(shí)任務(wù)調(diào)度器已停止 ===")

if __name__ == "__main__":
    run_scheduler()