energy-data-trans/qianan_wave.py

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()