465 lines
18 KiB
Python
465 lines
18 KiB
Python
"""
|
||
周报生成器
|
||
基于需求文档中的周报样式，结合DeepSeek API生成客户群运营与销售工作周报
|
||
"""
|
||
|
||
import json
|
||
import re
|
||
from datetime import datetime, timedelta
|
||
from typing import Dict, List, Any, Optional
|
||
from tools import TimeUtils, DatabaseManager
|
||
from chat_session import agent
|
||
|
||
|
||
class WeeklyReportGenerator:
|
||
"""周报生成器类"""
|
||
|
||
def __init__(self, db_manager: DatabaseManager):
|
||
self.db = db_manager
|
||
self.execute_query = self.db.execute_query
|
||
|
||
def get_week_time_range(self, week_str: str) -> tuple:
|
||
"""根据周字符串获取时间范围"""
|
||
try:
|
||
# 解析week_str参数，格式为YYYY-WW
|
||
year, week_num = map(int, week_str.split('-'))
|
||
|
||
# 根据周数计算时间范围
|
||
first_day_of_year = datetime(year, 1, 1)
|
||
# 计算第一周的周一（ISO周从周一开始）
|
||
first_monday = first_day_of_year + timedelta(days=(7 - first_day_of_year.weekday()) % 7)
|
||
if first_day_of_year.weekday() <= 3: # 如果1月1日是周一到周四，则属于第一周
|
||
first_monday = first_day_of_year - timedelta(days=first_day_of_year.weekday())
|
||
|
||
# 计算指定周的周一
|
||
week_monday = first_monday + timedelta(weeks=week_num - 1)
|
||
# 计算指定周的周日
|
||
week_sunday = week_monday + timedelta(days=6)
|
||
|
||
start_date = week_monday.replace(hour=0, minute=0, second=0, microsecond=0)
|
||
end_date = week_sunday.replace(hour=23, minute=59, second=59, microsecond=999999)
|
||
|
||
start_timestamp = int(TimeUtils(start_date).to_timestamp())
|
||
end_timestamp = int(TimeUtils(end_date).to_timestamp())
|
||
|
||
return start_timestamp, end_timestamp, start_date, end_date
|
||
except Exception as e:
|
||
raise ValueError(f"周次格式错误: {week_str}, 正确格式: YYYY-WW")
|
||
|
||
def get_group_interaction_data(self, start_timestamp: int, end_timestamp: int) -> List[Dict]:
|
||
"""获取各区域群互动数据明细"""
|
||
query = '''
|
||
SELECT g.id, g.name,
|
||
COUNT(m.id) as message_count,
|
||
COUNT(DISTINCT m.sender) as active_users
|
||
FROM "subscribe" g
|
||
inner JOIN message m ON g.id = m.receiver
|
||
AND m.time >= ? AND m.time <= ?
|
||
GROUP BY g.id, g.name
|
||
ORDER BY message_count DESC
|
||
'''
|
||
|
||
result = self.execute_query(query, (f"{start_timestamp}000", f"{end_timestamp}999"))
|
||
|
||
groups_data = []
|
||
for row in result:
|
||
group_data = dict(row)
|
||
|
||
# 获取核心互动时段
|
||
peak_hours = self.get_peak_interaction_hours(group_data['id'], start_timestamp, end_timestamp)
|
||
|
||
# 计算平均每日消息量
|
||
days_in_week = 7
|
||
avg_daily_messages = group_data['message_count'] / days_in_week if days_in_week > 0 else 0
|
||
|
||
groups_data.append({
|
||
'name': group_data['name'],
|
||
'message_count': group_data['message_count'],
|
||
'active_users': group_data['active_users'],
|
||
'peak_hours': peak_hours,
|
||
'avg_daily_messages': round(avg_daily_messages, 1)
|
||
})
|
||
|
||
return groups_data
|
||
|
||
def get_peak_interaction_hours(self, group_id: str, start_timestamp: int, end_timestamp: int) -> str:
|
||
"""获取群组的核心互动时段"""
|
||
query = '''
|
||
SELECT strftime('%H', datetime(substr(time, 0, 11), 'unixepoch', '+8 hours')) as hour,
|
||
COUNT(*) as count
|
||
FROM message
|
||
WHERE receiver = ? AND time >= ? AND time <= ?
|
||
GROUP BY hour
|
||
ORDER BY count DESC
|
||
LIMIT 3
|
||
'''
|
||
|
||
result = self.execute_query(query, (group_id, f"{start_timestamp}000", f"{end_timestamp}999"))
|
||
|
||
peak_hours = []
|
||
for row in result:
|
||
hour_data = dict(row)
|
||
hour = int(hour_data['hour'])
|
||
peak_hours.append(f"{hour:02d}:00-{hour+1:02d}:00")
|
||
|
||
return ", ".join(peak_hours) if peak_hours else "无数据"
|
||
|
||
def get_hot_topics(self, start_timestamp: int, end_timestamp: int, limit: int = 5) -> List[Dict]:
|
||
"""获取本周热点话题"""
|
||
# 获取本周消息用于话题分析
|
||
query = '''
|
||
SELECT content
|
||
FROM message
|
||
WHERE time >= ? AND time <= ?
|
||
AND type IN ('text', 'mixed')
|
||
ORDER BY time DESC
|
||
LIMIT ?
|
||
'''
|
||
|
||
result = self.execute_query(query, (f"{start_timestamp}000", f"{end_timestamp}999", limit * 10))
|
||
|
||
messages = [dict(row)['content'] for row in result]
|
||
|
||
# 简单的关键词提取和话题分析
|
||
topics = self.analyze_topics(messages, limit)
|
||
|
||
return topics
|
||
|
||
def analyze_topics(self, messages: List[str], limit: int) -> List[Dict]:
|
||
"""分析消息内容，提取热点话题"""
|
||
# 常见关键词和话题
|
||
common_keywords = {
|
||
'数据对接': ['API接口', '响应速度', '数据格式', '同步', '对接'],
|
||
'合规申报': ['申报指引', '材料清单', '审核时限', '合规', '申报'],
|
||
'系统操作': ['操作步骤', '功能使用', '系统填报', '操作指南'],
|
||
'政策法规': ['政策解读', '法规要求', '法律咨询', '政策变化'],
|
||
'业务咨询': ['业务操作', '流程咨询', '业务指导', '问题解答']
|
||
}
|
||
|
||
topics = []
|
||
for topic_name, keywords in common_keywords.items():
|
||
frequency = 0
|
||
related_groups = set()
|
||
|
||
for message in messages:
|
||
if any(keyword in message for keyword in keywords):
|
||
frequency += 1
|
||
# 这里可以进一步分析涉及群聊，简化处理
|
||
related_groups.add("相关群聊")
|
||
|
||
if frequency > 0:
|
||
topics.append({
|
||
'name': topic_name,
|
||
'keywords': keywords[:3], # 取前3个关键词
|
||
'related_groups': list(related_groups)[:2], # 取前2个群聊
|
||
'frequency': frequency,
|
||
'description': f"客户关注{topic_name}相关问题的讨论"
|
||
})
|
||
|
||
# 按频率排序并限制数量
|
||
topics.sort(key=lambda x: x['frequency'], reverse=True)
|
||
return topics[:limit]
|
||
|
||
def get_mapping_messages(self, start_timestamp: int, end_timestamp: int) -> Dict:
|
||
"""获取消息ID到内容的映射"""
|
||
query = '''
|
||
SELECT id, content, time
|
||
FROM message
|
||
WHERE time >= ? AND time <= ?
|
||
'''
|
||
|
||
result = self.execute_query(query, ( f"{start_timestamp}000", f"{end_timestamp}999"))
|
||
|
||
messages = {}
|
||
for row in result:
|
||
message_data = dict(row)
|
||
messages[message_data['id']] = {'content': message_data['content'],
|
||
'timestamp': int(message_data['time'])}
|
||
|
||
return messages
|
||
|
||
def get_customer_issues(self, start_timestamp: int, end_timestamp: int) -> List[Dict]:
|
||
"""获取客户问题汇总与分类"""
|
||
# 获取QA记录中的问题
|
||
query = '''
|
||
SELECT q.groupId, q.question, q.answer, g.name as group_name
|
||
FROM qa q
|
||
INNER JOIN "subscribe" g ON q.groupId = g.id
|
||
WHERE q.start_time >= ? AND q.end_time <= ?
|
||
ORDER BY q.start_time DESC
|
||
LIMIT 50
|
||
'''
|
||
|
||
result = self.execute_query(query, (f"{start_timestamp}000", f"{end_timestamp}999"))
|
||
message_mapping = self.get_mapping_messages( start_timestamp=start_timestamp, end_timestamp=end_timestamp)
|
||
|
||
issues = []
|
||
for row in result:
|
||
qa_data = dict(row)
|
||
|
||
try:
|
||
questions = json.loads(qa_data['question'])
|
||
answers = json.loads(qa_data['answer'])
|
||
|
||
if questions:
|
||
# 提取第一个问题作为代表
|
||
question_text = [message_mapping.get(qid, {}).get('content', '').replace('\n', '<br>') for qid in questions]
|
||
question_text = '<br>'.join(question_text)
|
||
|
||
# 分类问题
|
||
category = self.classify_issue(question_text)
|
||
|
||
issues.append({
|
||
'description': question_text,
|
||
'category': category,
|
||
'frequency': 1, # 简化处理，每个问题计1次
|
||
'group': qa_data.get('group_name', '未知群聊'),
|
||
'status': '已回复' if answers and len(answers) > 0 else '提及'
|
||
})
|
||
except:
|
||
continue
|
||
|
||
return issues
|
||
|
||
def classify_issue(self, question: str) -> str:
|
||
"""分类客户问题"""
|
||
categories = {
|
||
'产品使用': ['操作', '功能', '使用', '步骤', '设置'],
|
||
'服务流程': ['流程', '服务', '响应', 'SLA', '时限'],
|
||
'需求反馈': ['需求', '建议', '改进', '优化', '增加'],
|
||
'其他': []
|
||
}
|
||
|
||
for category, keywords in categories.items():
|
||
if any(keyword in question for keyword in keywords):
|
||
return category
|
||
|
||
return '其他'
|
||
|
||
def generate_weekly_report(self, week_str: str, use_deepseek: bool = True) -> Dict[str, Any]:
|
||
"""生成周报"""
|
||
try:
|
||
# 获取时间范围
|
||
start_ts, end_ts, start_date, end_date = self.get_week_time_range(week_str)
|
||
|
||
# 获取基础统计数据
|
||
total_messages = self.get_message_count_by_time_range(start_ts, end_ts)
|
||
total_groups = self.get_group_count()
|
||
total_active_users = self.get_active_users_by_time_range(start_ts, end_ts)
|
||
|
||
# 获取详细数据
|
||
group_interaction_data = self.get_group_interaction_data(start_ts, end_ts)
|
||
hot_topics = self.get_hot_topics(start_ts, end_ts)
|
||
customer_issues = self.get_customer_issues(start_ts, end_ts)
|
||
topic_message_distribution = self.get_topic_message_distribution(start_ts, end_ts)
|
||
|
||
# 生成报告周期字符串
|
||
week_period = f"{TimeUtils(start_date).format_custom('%Y年%m月%d日')} - {TimeUtils(end_date).format_custom('%Y年%m月%d日')}"
|
||
|
||
# 构建基础周报内容
|
||
report_content = self.build_report_content(
|
||
week_period=week_period,
|
||
total_groups=total_groups,
|
||
total_messages=total_messages,
|
||
total_active_users=total_active_users,
|
||
group_interaction_data=group_interaction_data,
|
||
hot_topics=hot_topics,
|
||
customer_issues=customer_issues,
|
||
topic_message_distribution=topic_message_distribution
|
||
)
|
||
|
||
# 使用DeepSeek API增强报告内容
|
||
if use_deepseek:
|
||
enhanced_content = self.enhance_with_deepseek(report_content)
|
||
report_content += "\n" + enhanced_content
|
||
|
||
return {
|
||
'success': True,
|
||
'report': report_content,
|
||
'week': week_str,
|
||
'statistics': {
|
||
'total_messages': total_messages,
|
||
'total_groups': total_groups,
|
||
'total_active_users': total_active_users,
|
||
'group_interaction_data': group_interaction_data,
|
||
'hot_topics': hot_topics,
|
||
'customer_issues': customer_issues,
|
||
'topic_message_distribution': topic_message_distribution
|
||
}
|
||
}
|
||
|
||
except Exception as e:
|
||
return {
|
||
'success': False,
|
||
'error': str(e)
|
||
}
|
||
|
||
def build_report_content(self, **kwargs) -> str:
|
||
"""构建周报内容"""
|
||
report = f"""# 客户群运营与销售工作周报
|
||
周期： {kwargs['week_period']}
|
||
生成时间： {datetime.now().strftime('%Y年%m月%d日')}
|
||
|
||
## 第一部分：客户群聊天记录分析与洞察
|
||
### 1. 互动概况
|
||
|
||
统计周期： 本周
|
||
|
||
汇总维度：
|
||
|
||
覆盖群聊总数： {kwargs['total_groups']} 个
|
||
|
||
总消息量： {kwargs['total_messages']} 条
|
||
|
||
总活跃人数： {kwargs['total_active_users']} 人
|
||
|
||
各区域群互动数据明细（表格）：
|
||
|
||
|客户群名称（区域）|消息总量（条）|活跃人数（人）|核心互动时段|平均每日消息量|
|
||
|---|---|---|---|---|
|
||
"""
|
||
|
||
# 添加群组数据表格
|
||
for group_data in kwargs['group_interaction_data']:
|
||
report += f"|{group_data['name']}|{group_data['message_count']}|{group_data['active_users']}|{group_data['peak_hours']}|{group_data['avg_daily_messages']}|\n"
|
||
|
||
report += f"""|总计/平均值|{kwargs['total_messages']}|{kwargs['total_active_users']}|-|{round(kwargs['total_messages']/7, 1) if kwargs['total_messages'] > 0 else 0}|
|
||
|
||
### 2. 本周热点话题 TOP 5
|
||
（按讨论频次与参与度综合排序）
|
||
"""
|
||
|
||
# 添加热点话题
|
||
for i, topic in enumerate(kwargs['hot_topics'], 1):
|
||
report += f"""
|
||
话题{i}： {topic['name']}
|
||
|
||
核心关键词： {', '.join(topic['keywords'])}
|
||
|
||
涉及主要群聊： {', '.join(topic['related_groups'])}
|
||
|
||
简要描述： {topic['description']}
|
||
"""
|
||
|
||
report += f"""
|
||
### 3. 客户问题汇总与分类
|
||
（分类：产品使用 / 服务流程 / 需求反馈 / 其他）
|
||
|
||
|问题类别|具体问题描述|涉及群聊|当前状态|
|
||
|---|---|---|---|
|
||
"""
|
||
|
||
# 添加客户问题
|
||
for issue in kwargs['customer_issues']:
|
||
report += f"|{issue['category']}|{issue['description']}|{issue['group']}|{issue['status']}|\n"
|
||
|
||
report += f"""
|
||
### 4. 主题消息发送情况表
|
||
（致远速联发送的重复消息分布情况）
|
||
|
||
|消息内容|涉及群|最新发送时间|
|
||
|---|---|---|
|
||
"""
|
||
|
||
# 添加主题消息发送情况表
|
||
for topic_data in kwargs['topic_message_distribution']: # 只显示前10条
|
||
# 截断过长的消息内容
|
||
content = topic_data['content']
|
||
if len(content) > 50:
|
||
content = content[:47] + "..."
|
||
content = content.replace("\n", "<br>")
|
||
|
||
report += f"|{content}|{topic_data['group']}|{topic_data['time']}|\n"
|
||
|
||
report += f"""
|
||
|
||
### 5. 需求洞察与行动建议
|
||
|
||
（基于热点话题与问题提炼的核心需求分析）
|
||
"""
|
||
|
||
return report
|
||
|
||
def enhance_with_deepseek(self, report_content: str) -> str:
|
||
"""使用DeepSeek API增强报告内容"""
|
||
try:
|
||
prompt = """请基于以下周报数据，提供专业的洞察分析和行动建议。
|
||
要求：
|
||
1. 分析客户需求的核心趋势
|
||
2. 提供具体的行动建议
|
||
3. 格式要求：
|
||
|
||
洞察一： [基于热点话题与问题提炼的核心需求]
|
||
建议： [可落地的建议]
|
||
|
||
洞察二： [基于数据分析的发现]
|
||
建议： [具体的改进建议]
|
||
|
||
请用中文回复，保持专业和简洁。"""
|
||
|
||
enhanced_content = agent.chat_deepseek(report_content, prompt)
|
||
return enhanced_content
|
||
except Exception as e:
|
||
return f"\n## AI分析增强\n由于API调用失败，无法提供AI增强分析。错误：{str(e)}"
|
||
|
||
# 辅助方法 - 从现有代码中复制
|
||
def get_message_count_by_time_range(self, start_timestamp, end_timestamp):
|
||
"""获取指定时间范围内的消息数量"""
|
||
result = self.execute_query('''
|
||
SELECT COUNT(*) as count
|
||
FROM message
|
||
WHERE time >= ?
|
||
AND time <= ?
|
||
''', (f"{start_timestamp}000", f"{end_timestamp}999"), fetch_all=False)
|
||
return result["count"]
|
||
|
||
def get_active_users_by_time_range(self, start_timestamp, end_timestamp):
|
||
"""获取指定时间范围内的活跃用户数量"""
|
||
result = self.execute_query('''
|
||
SELECT COUNT(DISTINCT sender) as count
|
||
FROM message
|
||
WHERE time >= ?
|
||
AND time <= ?
|
||
''', (f"{start_timestamp}000", f"{end_timestamp}999"), fetch_all=False)
|
||
return result["count"]
|
||
|
||
def get_group_count(self):
|
||
"""获取群组总数"""
|
||
result = self.execute_query('SELECT COUNT(*) as count FROM "subscribe"', fetch_all=False)
|
||
return result["count"]
|
||
|
||
def get_topic_message_distribution(self, start_timestamp: int, end_timestamp: int) -> List[Dict]:
|
||
"""获取主题消息发送情况表"""
|
||
query = '''
|
||
SELECT m.content as content,
|
||
s.name as "group",
|
||
strftime('%Y-%m-%d %H:%M:%S', datetime(substr(m.time, 0, 11), 'unixepoch', '+8 hours')) as time
|
||
FROM message as m
|
||
INNER JOIN (
|
||
SELECT m.content, count(1) as count
|
||
FROM message as m
|
||
INNER JOIN subscribe as s ON m.receiver = s.id
|
||
WHERE m.sender = 'ZhiYuanSuLian'
|
||
AND length(m.content) > 20
|
||
AND m.type = 'text'
|
||
AND m.time >= ? AND m.time <= ?
|
||
GROUP BY m.content
|
||
HAVING count(1) > 1
|
||
) as p ON m.content = p.content
|
||
INNER JOIN subscribe as s ON m.receiver = s.id
|
||
ORDER BY m.content, time
|
||
'''
|
||
|
||
result = self.execute_query(query, (f"{start_timestamp}000", f"{end_timestamp}999"))
|
||
|
||
# 按消息内容分组
|
||
topics_list = []
|
||
for row in result:
|
||
row_data = dict(row)
|
||
topics_list.append(row_data)
|
||
|
||
|
||
# 按发送次数降序排序
|
||
topics_list.sort(key=lambda x: x['time'], reverse=True)
|
||
|
||
return topics_list |