91 lines
3.0 KiB
Python
91 lines
3.0 KiB
Python
from datetime import datetime, timedelta
|
||
import pandas as pd
|
||
import pymysql
|
||
from sqlalchemy import create_engine
|
||
import requests
|
||
import json
|
||
|
||
|
||
# 1. 获取当前系统时间[1,3,6](@ref)
|
||
def get_current_time():
|
||
"""获取格式化的当前系统时间"""
|
||
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
|
||
# 2. 数据库查询（核心函数）
|
||
def fetch_activation_data():
|
||
"""获取激活车辆数据"""
|
||
connection_uri = f"mysql+pymysql://{DB_CONFIG['user']}:{DB_CONFIG['password']}@{DB_CONFIG['host']}:{DB_CONFIG['port']}/{DB_CONFIG['database']}"
|
||
engine = create_engine(connection_uri)
|
||
|
||
sql_query = """
|
||
SELECT
|
||
v_sim_info.ACTIVATION_DATE as active_date
|
||
FROM v_vehicle_info
|
||
JOIN v_tbox_info USING (VIN)
|
||
JOIN v_sim_info USING (SN)
|
||
WHERE v_sim_info.STATE = 2; -- 只查询激活状态车辆
|
||
"""
|
||
return pd.read_sql(sql_query, engine)
|
||
|
||
|
||
# 3. 统计计算
|
||
def calculate_stats(df):
|
||
"""计算多维统计指标"""
|
||
# 基础数据处理[1](@ref)
|
||
df['active_date'] = pd.to_datetime(df['active_date']).dt.date
|
||
|
||
# 本周统计（06-02至06-05）
|
||
start_week = datetime(2025, 6, 2).date()
|
||
end_week = datetime(2025, 6, 5).date()
|
||
weekly_data = df[(df['active_date'] >= start_week) & (df['active_date'] <= end_week)]
|
||
|
||
# 近三月统计（03-05至06-05）
|
||
start_3month = (datetime.now() - timedelta(days=90)).date()
|
||
monthly_data = df[df['active_date'] >= start_3month]
|
||
|
||
return {
|
||
"本周激活总数": len(weekly_data),
|
||
"本周每日激活": weekly_data.groupby('active_date').size().to_dict(),
|
||
"近三月激活总数": len(monthly_data),
|
||
"累计激活总数": len(df)
|
||
}
|
||
|
||
|
||
# 4. 飞书机器人消息发送
|
||
def send_feishu_report(stats):
|
||
"""发送统计报告"""
|
||
message = f"🚗 **车辆激活统计报告**（{get_current_time()}）\n\n"
|
||
message += f"🔹 **本周激活总数**: {stats['本周激活总数']}辆\n"
|
||
message += "🔹 **本周每日激活**:\n"
|
||
for date, count in stats['本周每日激活'].items():
|
||
message += f" ▸ {date}: {count}辆\n"
|
||
message += f"\n🔹 **近三月激活总数**: {stats['近三月激活总数']}辆\n"
|
||
message += f"🔹 **累计激活总数**: {stats['累计激活总数']}辆"
|
||
|
||
requests.post(FEISHU_WEBHOOK_URL, json={
|
||
"msg_type": "text",
|
||
"content": {"text": message}
|
||
})
|
||
|
||
|
||
# 主流程
|
||
if __name__ == "__main__":
|
||
# 配置参数
|
||
FEISHU_WEBHOOK_URL = "https://open.feishu.cn/open-apis/bot/v2/hook/ff3726f6-d4e3-455b-ae65-bca471305733"
|
||
# 数据库连接
|
||
DB_CONFIG = {
|
||
'host': 'eutsp-prod.mysql.germany.rds.aliyuncs.com',
|
||
'port': 3306, # MySQL默认端口
|
||
'user': 'international_tsp_eu_r',
|
||
'password': 'ZXhBgo1TB2XbF3kP',
|
||
'database': 'chery_international_tsp_eu'
|
||
}
|
||
|
||
# 执行统计
|
||
df = fetch_activation_data()
|
||
stats = calculate_stats(df)
|
||
|
||
# 发送报告
|
||
send_feishu_report(stats)
|
||
print("✅ 统计报告已发送至飞书") |