import logging from datetime import datetime, timedelta from pathlib import Path import pymysql import redis # -------------------------- 日志配置 -------------------------- # 创建日志目录(如果不存在) log_dir = Path("/data/logs/wave_data") log_dir.mkdir(exist_ok=True) # 配置日志格式和输出 logging.basicConfig( level=logging.INFO, # 日志级别:DEBUG/INFO/WARNING/ERROR/CRITICAL format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", # 日志格式 handlers=[ # 输出到文件(按日期命名,编码UTF-8避免中文乱码) logging.FileHandler(log_dir / f"{datetime.now().strftime('%Y%m%d')}.log", encoding="utf-8"), # 同时输出到控制台 logging.StreamHandler() ] ) logger = logging.getLogger("WaveData") # 日志器名称,便于区分不同模块 # -------------------------- 配置信息 -------------------------- # Redis 配置 REDIS_CONFIG = { 'host': 'localhost', 'port': 6379, 'db': 10, 'password': '123456', 'decode_responses': True # 自动将返回值解码为字符串,避免字节串处理 } # MySQL 配置 MYSQL_CONFIG = { 'host': '127.0.0.1', 'port': 4000, 'user': 'root', 'password': 'admin123456', 'database': 'energy_data_prod', 'charset': 'utf8mb4' } now = datetime.now() last_hour = now - timedelta(hours=1) FILE_PATH = f"/data/skf-data/National Grid/乾安风电场/{last_hour.year}/{str(last_hour.month).zfill(2)}{str(last_hour.day).zfill(2)}/{str(last_hour.hour).zfill(2)}" # -------------------------- 核心逻辑 -------------------------- def main(): target_key = f"energy:skf:shake:{last_hour.strftime('%Y%m%d%H')}" logger.info(f"开始处理任务,目标 Redis Key:{target_key}") redis_conn = None mysql_conn = None mysql_cursor = None try: # 2. 连接 Redis 并检查 Key 是否存在 redis_conn = redis.Redis(**REDIS_CONFIG) # 先判断 Key 是否存在 if not redis_conn.exists(target_key): logger.warning(f"Redis Key {target_key} 不存在,跳过后续操作") return # 直接退出,不执行后续逻辑 # 3. 获取 Key 的值并判断是否等于 '1' value = redis_conn.get(target_key) logger.info(f"Redis Key {target_key} 的值为:{value}") if value == '1': # 3.1 删除 Redis 中的目标 Key redis_conn.delete(target_key) logger.info(f"已删除 Redis Key:{target_key}") # 3.2 连接 MySQL 并插入数据 mysql_conn = pymysql.connect(**MYSQL_CONFIG) mysql_cursor = mysql_conn.cursor() # 定义插入语句(参数化查询,避免 SQL 注入) insert_sql = """ INSERT INTO `data_transfer` ( `task_name`, `wind_farm_code`, `wind_farm_name`, `transfer_type`, `read_dir`, `data_collector`, `transfer_status`, `trans_sys_status` ) VALUES ( %s, 'WOF043800107', '乾安一期', 'wave', %s, '系统', -1, -1); """ # 构造插入参数 task_name = f"自动-{last_hour.strftime('%Y-%m-%d %H')}" read_dir = FILE_PATH # 执行插入 mysql_cursor.execute(insert_sql, (task_name, read_dir)) mysql_conn.commit() logger.info(f"成功向 MySQL 插入记录:任务名称={task_name},读取目录={read_dir}") else: logger.warning(f"Redis Key {target_key} 的值不是 '1'(当前值:{value}),跳过插入操作") except redis.RedisError as e: logger.error(f"Redis 操作出错:{str(e)}", exc_info=True) # exc_info=True 打印详细异常栈 except pymysql.MySQLError as e: logger.error(f"MySQL 操作出错:{str(e)}", exc_info=True) # 插入失败时回滚 MySQL 事务 if mysql_conn: mysql_conn.rollback() except Exception as e: logger.error(f"未知错误:{str(e)}", exc_info=True) finally: # 确保所有连接关闭 # 关闭 MySQL 游标和连接 if mysql_cursor: mysql_cursor.close() if mysql_conn: mysql_conn.close() # 关闭 Redis 连接 if redis_conn: redis_conn.close() logger.info("任务处理完成,所有连接已关闭") if __name__ == "__main__": main()