data_job/job_test.py

from utils.loadconfig import get_path
from utils.log import Log
from utils.date_time import get_variable_parameter
from utils.execute_sql import execute_sql


def read_sql_files(directory, day):
    log = Log().getlog()
    directory = get_path() + directory

    try:
        with open(directory, 'r', encoding='utf-8') as f:
            content = f.read()
            # 遍历字典，将字符串中的占位符替换为对应的值
            for key, value in get_variable_parameter(day).items():
                content = content.replace(key, str(value))
            log.info(f"文件路径: {directory}")

            try:
                returnstr = execute_sql(content)
                log.info(f"执行结果: {returnstr}")
                log.info("-" * 50)
            except Exception as e:
                log.error(f"执行 SQL 文件 {directory} 时出现错误: {e}")

    except UnicodeDecodeError:
        log.error(f"文件 {directory} 编码格式可能不是utf-8，无法正确读取，请检查。")
    except Exception as e:
        log.error(f"读取文件 {directory} 时出现错误: {e}")




if __name__ == '__main__':
    # 遍历300到0
    for day in range(301):
        read_sql_files('/sql/ads/user_retention.sql', day)