53 lines
1.9 KiB
Python
53 lines
1.9 KiB
Python
from datetime import date, timedelta, datetime
|
|
import holidays
|
|
from config import Config
|
|
import psycopg2
|
|
import pandas as pd
|
|
|
|
class GetProvateMiniProgram(Config):
|
|
def get_private_mini_program_details(self, days_back=1):
|
|
"""
|
|
从固定的数据库中获取前几天的NGV明细。
|
|
参数 `days_back` 表示相对于今天的天数偏移量，默认为1（即前一天）。
|
|
返回包含NGV明细的pandas DataFrame。
|
|
"""
|
|
try:
|
|
# 获得连接
|
|
conn = psycopg2.connect(**self.conn)
|
|
cursor = conn.cursor()
|
|
|
|
# 获取指定天数前的日期
|
|
now_time = datetime.now()
|
|
target_time = now_time + timedelta(days=-days_back)
|
|
target_date_id = int(target_time.strftime('%Y%m%d')) # 获取目标日期
|
|
|
|
# sql语句查询
|
|
sql = f"""
|
|
SELECT * FROM "public"."holo_ads_report_saas_profile_ngv_detail_d" WHERE "date_id" = '{target_date_id}' ;
|
|
"""
|
|
|
|
# 执行语句并获取结果集
|
|
cursor.execute(sql)
|
|
rows = cursor.fetchall()
|
|
all_fields = cursor.description
|
|
|
|
# 执行结果转化为dataframe
|
|
col = [i[0] for i in all_fields]
|
|
data_NGV = pd.DataFrame(rows, columns=col)
|
|
|
|
# 尝试自动解析日期时间字符串
|
|
time_format = "%Y-%m-%d %H:%M:%S"
|
|
if 'saas_create_time' in data_NGV.columns:
|
|
data_NGV['saas_create_time'] = pd.to_datetime(data_NGV['saas_create_time'], format=time_format,
|
|
errors='coerce')
|
|
data_NGV['saas_create_time'] = data_NGV['saas_create_time'].dt.strftime('%Y-%m-%d')
|
|
|
|
# 关闭游标和连接
|
|
cursor.close()
|
|
conn.close()
|
|
|
|
return data_NGV
|
|
|
|
except Exception as e:
|
|
print(f"Error occurred: {e}")
|
|
return None |