"""睿行车辆越界报警日统计缺失数据日期 """ 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 # 不保存列名 )