# ''' # Description: 监听 todolist 表变化并运行相关脚本 # Author: Manda & Trae AI # Version: 1.1 # Date: 2024-07-26 # ''' import time import subprocess import logging from datetime import datetime import os import sys import io import mysql.connector # 导入 mysql.connector from dataBaseConnecter import DatabaseConnector # 导入 DatabaseConnector 以获取配置和连接 sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8') # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('notify_main.log', encoding='utf-8'), logging.StreamHandler() ] ) # --- 新增数据库连接和监听逻辑 --- def get_db_conn(): """获取数据库连接""" try: # 使用 DatabaseConnector 来获取连接，它会处理配置加载和SSL connector = DatabaseConnector() # 创建实例以加载配置 conn = connector.connect_db() # 使用其连接方法 if conn: logging.info("✅ 数据库连接成功") return conn else: logging.error("❌ 数据库连接失败 (来自 connect_db)") return None except Exception as e: logging.error(f"❌ 获取数据库连接时发生异常: {e}", exc_info=True) return None def get_latest_update_time(conn): """获取 todolist 表的最新更新时间 (假设有更新时间戳字段) 注意：需要确认 todolist 表是否有合适的更新时间字段。 这里暂时假设存在一个名为 'updated_at' 的字段。 如果不存在，需要修改为查询 MAX(id) 或其他能代表最新记录的方式。 """ with conn.cursor() as cursor: try: # !! 重要：请根据实际 todolist 表结构修改这里的查询 !! # 假设存在 updated_at 字段 # cursor.execute("SELECT MAX(updated_at) FROM todolist") # 根据用户提供的表结构，使用 last_modified 字段 cursor.execute("SELECT MAX(last_modified) FROM todolist") result = cursor.fetchone() # 如果表为空，result[0] 可能是 None return result[0] if result else None except mysql.connector.Error as err: logging.error(f"查询最新更新时间失败: {err}") # 可以考虑返回一个特殊值或重新抛出异常，取决于错误处理策略 return None # 或者 raise err # --- 结束新增数据库连接和监听逻辑 --- def run_script(script_name: str): """运行指定的Python脚本""" try: logging.info(f"开始运行脚本: {script_name}") # 捕获原始字节输出，不立即进行文本解码 result = subprocess.run(['python', script_name], capture_output=True, check=False) # 使用 check=False 避免在非零退出码时抛出异常 # 手动解码 stdout 和 stderr，并处理解码错误 stdout_decoded = result.stdout.decode('utf-8', errors='ignore') stderr_decoded = result.stderr.decode('utf-8', errors='ignore') if result.returncode == 0: logging.info(f"脚本 {script_name} 运行成功") if stdout_decoded: logging.info(f"输出: {stdout_decoded}") else: logging.error(f"脚本 {script_name} 运行失败 (返回码: {result.returncode})") if stderr_decoded: logging.error(f"错误: {stderr_decoded}") # 即使失败，也记录 stdout 以便调试 if stdout_decoded: logging.info(f"失败时的输出: {stdout_decoded}") except Exception as e: logging.error(f"运行脚本 {script_name} 时发生错误: {str(e)}") def run_all_scripts(): """运行所有脚本""" logging.info("检测到更新，开始执行所有脚本") # 获取当前脚本所在的目录 current_dir = os.path.dirname(os.path.abspath(__file__)) # 确保脚本顺序正确，如果它们之间有依赖关系 scripts = ['db2txt.py', 'usrSpareTime.py', 'compareDb2txt.py'] for script in scripts: # 使用完整的文件路径 script_path = os.path.join(current_dir, script) if os.path.exists(script_path): run_script(script_path) # 根据脚本执行时间调整等待时间，如果需要的话 # time.sleep(1) # 短暂等待，避免文件系统或进程间竞争 else: logging.error(f"脚本文件不存在: {script_path}") logging.info("所有脚本执行完成") # --- 修改 main 函数为监听模式 --- def monitor_todolist_and_run(interval=5): """监听 todolist 表并联动执行脚本""" conn = None last_update = None # 初始连接和获取状态 while conn is None: conn = get_db_conn() if conn: last_update = get_latest_update_time(conn) logging.info(f"初始 todolist 表状态 (last_modified): {last_update}") break else: logging.warning("数据库连接失败，5秒后重试...") time.sleep(5) while True: time.sleep(interval) try: # 检查连接是否仍然有效，如果无效则尝试重连 if not conn or not conn.is_connected(): logging.warning("数据库连接丢失，尝试重新连接...") conn = get_db_conn() if not conn: logging.error("重新连接失败，跳过本次检查。") continue # 跳过这次循环 else: # 重连成功后，重新获取一次最新状态，避免因断连期间的更新而误判 last_update = get_latest_update_time(conn) logging.info(f"重新连接成功，当前 todolist 表状态 (last_modified): {last_update}") current_update = get_latest_update_time(conn) # 检查 current_update 是否为 None (查询失败或表为空) if current_update is None and last_update is None: logging.debug("数据库查询失败或表仍为空，继续监听...") elif current_update != last_update: logging.info(f"检测到 todolist 表有更新 (last_modified 从 {last_update} 变为 {current_update}), 开始联动执行...") run_all_scripts() # 触发脚本执行 last_update = current_update # 更新状态 else: logging.debug(f"无更新 (last_modified: {current_update})，继续监听...") except Exception as e: logging.error(f"监听或执行过程中发生错误: {e}", exc_info=True) # 发生未知错误时，尝试关闭可能存在的旧连接，并在下次循环重连 if conn and conn.is_connected(): try: conn.close() logging.info("因异常关闭数据库连接") except Exception as close_err: logging.error(f"关闭连接时出错: {close_err}") conn = None # 强制下次循环重连 time.sleep(interval) # 避免错误循环过快 if __name__ == "__main__": logging.info("启动 todolist 监听与脚本执行程序") monitor_todolist_and_run(interval=5) # 设置监听间隔为5秒 # --- 移除旧的 schedule 相关代码 --- # def main(): # logging.info("启动定时任务程序") # # # 设置每小时运行一次 # schedule.every().hour.at(":43").do(run_all_scripts) # # # 立即运行一次 # run_all_scripts() # # # 持续运行 # while True: # try: # schedule.run_pending() # time.sleep(60) # except Exception as e: # logging.error(f"运行时发生错误: {str(e)}") # time.sleep(60) # 发生错误时等待一分钟后继续 # # if __name__ == "__main__": # main()