72 lines
2.6 KiB
Python
72 lines
2.6 KiB
Python
from datetime import date, timedelta, datetime
|
|
import holidays
|
|
from config import Config
|
|
import psycopg2
|
|
import pandas as pd
|
|
import pymysql # 使用 pymysql 替代 mysql.connector
|
|
|
|
|
|
class JCB_efficient_car_pickup:
|
|
def __init__(self):
|
|
# 使用 pymysql 连接数据库
|
|
self.conn = pymysql.connect(
|
|
host=Config.JCB_CONN_host,
|
|
database=Config.JCB_CONN_INFO_database,
|
|
user=Config.JCB_CONN_INFO_user,
|
|
password=Config.JCB_CONN_INFO_password,
|
|
# charset='utf8mb4', # 设置字符集以避免编码问题
|
|
# cursorclass=pymysql.cursors.DictCursor # 返回字典形式的结果
|
|
)
|
|
|
|
def get_jcb_details(self, days_back=1):
|
|
"""
|
|
从固定的数据库中获取前几天的NGV明细。
|
|
参数 `days_back` 表示相对于今天的天数偏移量，默认为1（即前一天）。
|
|
返回包含NGV明细的pandas DataFrame。
|
|
"""
|
|
try:
|
|
# 获得连接并创建游标
|
|
cursor = self.conn.cursor()
|
|
|
|
# 获取指定天数前的日期
|
|
# now_time = datetime.now()
|
|
# target_time = now_time + timedelta(days=-days_back)
|
|
target_date_id = "接车宝" # 获取目标日期
|
|
|
|
# SQL 查询语句
|
|
sql = f"""
|
|
SELECT * FROM jdy_hs_holo_dws_sales_magic_box_ngv_d WHERE 产品名称 = '{target_date_id}';
|
|
"""
|
|
|
|
# 执行查询并获取结果
|
|
cursor.execute(sql)
|
|
rows = cursor.fetchall() # pymysql 的 DictCursor 会返回字典列表
|
|
|
|
# 将结果转换为 DataFrame
|
|
if rows: # 如果有数据
|
|
data_NGV = pd.DataFrame(rows)
|
|
else: # 如果没有数据，返回空 DataFrame
|
|
data_NGV = pd.DataFrame()
|
|
|
|
# 尝试自动解析日期时间字符串
|
|
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()
|
|
|
|
return data_NGV
|
|
|
|
except Exception as e:
|
|
print(f"Error occurred: {e}")
|
|
return None
|
|
|
|
|
|
if __name__ == "__main__":
|
|
start = JCB_efficient_car_pickup()
|
|
result = start.get_jcb_details(days_back=1)
|
|
if result is not None:
|
|
print(result.head()) # 打印前几行数据 |