1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
| """
每日销售报告自动化:
1. 从数据库拉取数据
2. 用 pandas 分析
3. 生成 Excel + 图表
4. 生成 PDF 报告
5. 邮件发送给管理层
"""
import pandas as pd
import pymysql
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.application import MIMEApplication
from datetime import datetime, timedelta
import matplotlib.pyplot as plt
class DailyReport:
def __init__(self):
self.db_config = {
'host': 'localhost',
'user': 'report_user',
'password': 'password',
'database': 'sales'
}
self.email_config = {
'smtp': 'smtp.gmail.com',
'port': 587,
'user': 'report@company.com',
'password': 'app_password'
}
def fetch_data(self):
"""从数据库拉取昨日销售数据"""
yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
conn = pymysql.connect(**self.db_config)
query = f"""
SELECT order_id, product_name, category, amount, customer_id, created_at
FROM orders
WHERE DATE(created_at) = '{yesterday}'
"""
df = pd.read_sql(query, conn)
conn.close()
return df, yesterday
def analyze(self, df):
"""数据分析"""
analysis = {
'总订单数': len(df),
'总销售额': df['amount'].sum(),
'平均订单金额': df['amount'].mean(),
'独立客户数': df['customer_id'].nunique(),
}
category_stats = df.groupby('category')['amount'].agg(['sum', 'count', 'mean'])
top_products = df.groupby('product_name')['amount'].sum().nlargest(10)
return analysis, category_stats, top_products
def generate_charts(self, df, date):
"""生成图表"""
fig, axes = plt.subplots(2, 2, figsize=(14, 10))
category_sum = df.groupby('category')['amount'].sum()
axes[0, 0].pie(category_sum.values, labels=category_sum.index, autopct='%1.1f%%')
axes[0, 0].set_title('各类别销售额占比')
df['hour'] = pd.to_datetime(df['created_at']).dt.hour
hourly = df.groupby('hour')['amount'].sum()
axes[0, 1].bar(hourly.index, hourly.values)
axes[0, 1].set_title('时段销售分布')
axes[0, 1].set_xlabel('小时')
axes[1, 0].hist(df['amount'], bins=30, edgecolor='black')
axes[1, 0].set_title('订单金额分布')
axes[1, 0].set_xlabel('金额')
top = df.groupby('product_name')['amount'].sum().nlargest(10)
axes[1, 1].barh(top.index, top.values)
axes[1, 1].set_title('TOP 10 商品')
plt.tight_layout()
chart_path = f'charts_{date}.png'
plt.savefig(chart_path, dpi=100, bbox_inches='tight')
plt.close()
return chart_path
def generate_excel(self, df, analysis, category_stats, top_products, date):
"""生成 Excel 报告"""
excel_path = f'report_{date}.xlsx'
with pd.ExcelWriter(excel_path, engine='openpyxl') as writer:
pd.DataFrame([analysis]).T.to_excel(
writer, sheet_name='概览', header=['数值'])
df.to_excel(writer, sheet_name='订单明细', index=False)
category_stats.to_excel(writer, sheet_name='分类统计')
top_products.to_excel(writer, sheet_name='TOP 商品')
return excel_path
def send_email(self, date, attachments, recipients):
"""发送邮件"""
msg = MIMEMultipart()
msg['Subject'] = f'每日销售报告 - {date}'
html = f"""
<h2>每日销售报告</h2>
<p><strong>报告日期:</strong> {date}</p>
<p>详细数据请查看附件。</p>
<p><small>本报告由系统自动生成,请勿直接回复。</small></p>
"""
msg.attach(MIMEText(html, 'html'))
for file_path in attachments:
with open(file_path, 'rb') as f:
attach = MIMEApplication(f.read())
attach.add_header('Content-Disposition', 'attachment',
filename=file_path.split('/')[-1])
msg.attach(attach)
with smtplib.SMTP(self.email_config['smtp'], self.email_config['port']) as server:
server.starttls()
server.login(self.email_config['user'], self.email_config['password'])
for recipient in recipients:
msg['To'] = recipient
server.send_message(msg)
del msg['To']
print(f"报告已发送给 {len(recipients)} 位收件人")
def run(self):
"""执行完整流程"""
print("开始生成每日报告...")
df, date = self.fetch_data()
print(f" ✓ 拉取 {len(df)} 条订单数据")
analysis, category_stats, top_products = self.analyze(df)
print(f" ✓ 总销售额: ¥{analysis['总销售额']:.2f}")
chart_path = self.generate_charts(df, date)
print(f" ✓ 图表已生成: {chart_path}")
excel_path = self.generate_excel(df, analysis, category_stats, top_products, date)
print(f" ✓ Excel 已生成: {excel_path}")
recipients = ['manager@company.com', 'ceo@company.com']
self.send_email(date, [excel_path, chart_path], recipients)
print("✓ 报告流程完成")
if __name__ == '__main__':
report = DailyReport()
report.run()
|
