excel_exporter.py
6.1 KB
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
175
176
177
178
179
180
181
182
183
184
185
186
187
188
"""
Excel导出模块
将发送记录导出为Excel文件
"""
import pandas as pd
from datetime import datetime
from typing import List, Dict, Optional
from .database import Database
class ExcelExporter:
"""Excel导出器"""
def __init__(self, database: Database):
self.database = database
def export_to_excel(self, file_path: str, start_date: str = None, end_date: str = None) -> bool:
"""
导出发送记录到Excel文件
Args:
file_path: 导出文件路径
start_date: 开始日期 (YYYY-MM-DD)
end_date: 结束日期 (YYYY-MM-DD)
Returns:
bool: 导出是否成功
"""
try:
# 获取数据
records = self.database.get_sent_messages_for_export(start_date, end_date)
if not records:
print("没有找到符合条件的记录")
return False
# 转换为DataFrame
df = pd.DataFrame(records)
# 数据处理
if not df.empty:
# 处理时间格式
if '发送时间' in df.columns:
df['发送时间'] = pd.to_datetime(df['发送时间']).dt.strftime('%Y-%m-%d %H:%M:%S')
# 处理标签列(如果是JSON格式)
if '标签' in df.columns:
df['标签'] = df['标签'].apply(self._format_tags)
# 处理触发方式
if '触发方式' in df.columns:
df['触发方式'] = df['触发方式'].apply(
lambda x: '对话触发' if x == 'conversation' else '订单触发' if x == 'order' else x
)
# 导出到Excel
with pd.ExcelWriter(file_path, engine='openpyxl') as writer:
# 主数据表
df.to_excel(writer, sheet_name='发送记录', index=False)
# 统计信息表
stats_df = self._generate_statistics(records)
stats_df.to_excel(writer, sheet_name='统计信息', index=False)
# 格式化Excel
self._format_excel(writer, df)
return True
except Exception as e:
print(f"Excel导出失败: {e}")
return False
def _format_tags(self, tags_json: str) -> str:
"""格式化标签字段"""
if not tags_json:
return ""
try:
import json
tags = json.loads(tags_json)
return ", ".join(tags) if isinstance(tags, list) else str(tags)
except:
return str(tags_json)
def _generate_statistics(self, records: List[Dict]) -> pd.DataFrame:
"""生成统计信息"""
if not records:
return pd.DataFrame()
stats = []
# 总体统计
total_count = len(records)
stats.append({"统计项目": "总发送数量", "数值": total_count})
# 按触发方式统计
conversation_count = sum(1 for r in records if r.get('触发方式') == 'conversation')
order_count = sum(1 for r in records if r.get('触发方式') == 'order')
stats.append({"统计项目": "对话触发数量", "数值": conversation_count})
stats.append({"统计项目": "订单触发数量", "数值": order_count})
# 按日期统计
date_counts = {}
for record in records:
sent_time = record.get('发送时间', '')
if sent_time:
try:
date = sent_time.split()[0] # 提取日期部分
date_counts[date] = date_counts.get(date, 0) + 1
except:
pass
# 添加日期统计
for date, count in sorted(date_counts.items()):
stats.append({"统计项目": f"{date} 发送数量", "数值": count})
return pd.DataFrame(stats)
def _format_excel(self, writer, df: pd.DataFrame):
"""格式化Excel工作表"""
try:
from openpyxl.styles import Font, PatternFill, Alignment
from openpyxl.utils.dataframe import dataframe_to_rows
# 获取工作表
worksheet = writer.sheets['发送记录']
# 设置标题行格式
header_font = Font(bold=True, color="FFFFFF")
header_fill = PatternFill(start_color="366092", end_color="366092", fill_type="solid")
for cell in worksheet[1]:
cell.font = header_font
cell.fill = header_fill
cell.alignment = Alignment(horizontal="center")
# 调整列宽
column_widths = {
'A': 20, # 对话ID
'B': 15, # 客户姓名
'C': 15, # 订单ID
'D': 40, # 发送内容
'E': 20, # 发送时间
'F': 20, # 对话创建时间
'G': 20, # 最后对话时间
'H': 20, # 标签
'I': 12, # 触发方式
}
for col, width in column_widths.items():
worksheet.column_dimensions[col].width = width
# 设置文本换行
for row in worksheet.iter_rows(min_row=2):
for cell in row:
cell.alignment = Alignment(wrap_text=True, vertical="top")
except Exception as e:
print(f"Excel格式化失败: {e}")
def export_simple_csv(self, file_path: str, start_date: str = None, end_date: str = None) -> bool:
"""
导出简单的CSV文件(备用方案)
Args:
file_path: 导出文件路径
start_date: 开始日期
end_date: 结束日期
Returns:
bool: 导出是否成功
"""
try:
records = self.database.get_sent_messages_for_export(start_date, end_date)
if not records:
return False
df = pd.DataFrame(records)
df.to_csv(file_path, index=False, encoding='utf-8-sig')
return True
except Exception as e:
print(f"CSV导出失败: {e}")
return False