finance_data_process/car/car-yue-jie-missing/car_yue_jie_missing.py

"""睿行车辆越界报警日统计缺失数据日期
"""

import calendar  # 导入日历模块，用于处理日期相关操作
from datetime import datetime  # 导入日期时间模块，用于处理日期和时间
import pandas as pd  # 导入pandas库，用于数据处理与分析
import psycopg  # 导入psycopg库，用于连接PostgreSQL数据库
from dateutil.relativedelta import relativedelta  # 导入relativedelta模块，用于日期的相对计算
from loguru import logger  # 导入loguru库，用于日志记录


# 定义一个函数，获取指定日期所在月份的最后一天
def get_last_day_of_month(t):
    _, last_day = calendar.monthrange(t.year, t.month)  # 获取指定年月的最后一天
    return datetime(t.year, t.month, last_day)  # 返回该月最后一天的日期对象


# 获取当前日期
today = datetime.today()
# 计算上一个月的第一天
start_date = today - relativedelta(months=1, day=1)
# 计算上一个月的最后一天
end_date = get_last_day_of_month(start_date)

# 初始化一个空列表，用于存储查询结果
a = []

# 使用psycopg连接到PostgreSQL数据库
with psycopg.connect(
        conninfo="host='172.16.107.5' port=5432 user='finance' password='Finance@unicom23' dbname='financialdb'",
        row_factory=psycopg.rows.dict_row  # 设置行工厂为字典格式，方便后续操作
) as conn:
    with conn.cursor() as curs:  # 创建游标对象，用于执行SQL语句
        # 构造SQL查询语句
        sql = f"""
        with
        t1 as (
            -- 生成从start_date到end_date之间的所有日期序列
            select generate_series('{start_date.strftime('%Y-%m-%d')}'::date, '{end_date.strftime('%Y-%m-%d')}'::date, '1 day'::interval)::date as data_date
        ),
        t2 as (
            -- 从car.car_yue_jie表中提取指定年月的所有唯一日期
            select distinct data_date 
            from car.car_yue_jie 
            where year_month = {start_date.strftime('%Y%m')} 
            order by data_date
        )
        -- 查询在t1中存在但在t2中不存在的日期（即缺失的日期）
        select * 
        from t1 
        where not exists (select 1 from t2 where t2.data_date = t1.data_date)
        """
        # 打印SQL语句，便于调试
        logger.info(f"sql: {sql}")
        # 执行SQL语句
        curs.execute(sql)
        # 获取查询结果
        list = curs.fetchall()
        # 遍历查询结果，将data_date字段的值添加到列表a中
        for x in list:
            a.append(x.get('data_date'))

# 将列表a转换为pandas DataFrame
df = pd.DataFrame(a)
# 打印DataFrame的基本信息，便于调试
print(df.info())
# 将DataFrame保存为Excel文件，文件名为“睿行车辆越界报警日统计缺失数据日期.xlsx”
df.to_excel(
    f"{start_date.strftime('%Y%m')}睿行车辆越界报警日统计缺失数据日期.xlsx",
    index=False,  # 不保存索引
    header=False  # 不保存列名
)