from datetime import datetime, timedelta from api.lib import report_service, message_processor, db_manager from . import request, Blueprint from .lib import api_response import json from chat_session import agent from tools import TimeUtils, now, from_datetime v1 = Blueprint('v1', __name__) # API路由 - 消息相关 @v1.route('/api/messages_by_ids', methods=['POST']) def get_messages_by_ids(): """根据消息ID列表获取消息内容""" try: data = request.get_json() message_ids = data.get('message_ids', []) if not message_ids: return api_response(data={'messages': []}) messages_data = report_service.get_messages_by_ids(message_ids) formatted_messages = [] for msg in messages_data: message = dict(msg) message = message_processor.process_message_content(message) formatted_messages.append(message) return api_response(data={'messages': formatted_messages}) except Exception as e: return api_response(success=False, error=e, status_code=500) @v1.route('/api/messages') def get_messages(): """获取消息列表API""" try: limit = request.args.get('limit', 100, type=int) offset = request.args.get('offset', 0, type=int) group_id = request.args.get('group_id', type=str) messages_data = report_service.get_messages(limit, offset, group_id) formatted_messages = [] for msg in messages_data: message = dict(msg) message = message_processor.process_message_content(message) formatted_messages.append(message) return api_response(data={ 'messages': formatted_messages, 'total': len(formatted_messages) }) except Exception as e: return api_response(success=False, error=e, status_code=500) # API路由 - 用户和群组 @v1.route('/api/users') def get_users(): """获取用户列表API""" try: users_data = report_service.get_all_users() users = [dict(user) for user in users_data] return api_response(data={'users': users}) except Exception as e: return api_response(success=False, error=e, status_code=500) @v1.route('/api/groups') def get_groups(): """获取群组列表API""" try: groups_data = report_service.get_all_groups() groups = [dict(group) for group in groups_data] return api_response(data={'groups': groups}) except Exception as e: return api_response(success=False, error=e, status_code=500) @v1.route('/api/group_messages') def get_group_messages(): """获取指定群组的消息API""" try: group_id = request.args.get('group_id', type=str) limit = request.args.get('limit', 100, type=int) offset = request.args.get('offset', 0, type=int) if not group_id: return api_response(success=False, error='缺少群组ID参数', status_code=400) messages_data = report_service.get_group_messages(group_id, limit, offset) group_messages_info = report_service.get_group_message_count(group_id) formatted_messages = [] for msg in messages_data: message = dict(msg) message = message_processor.process_message_content(message) formatted_messages.append(message) return api_response(data={'messages': formatted_messages, **group_messages_info}) except Exception as e: return api_response(success=False, error=e, status_code=500) # API路由 - 统计信息 @v1.route('/api/statistics') def get_statistics(): """获取统计数据API""" try: statistics = report_service.get_statistics() return api_response(data=statistics) except Exception as e: return api_response(success=False, error=e, status_code=500) # API路由 - 搜索 @v1.route('/api/search_messages') def search_messages(): """搜索消息API""" try: keyword = request.args.get('keyword', '') limit = request.args.get('limit', 100, type=int) search_type = request.args.get('search_type', 'all') # all, question, answer if not keyword: return api_response(success=False, error='请输入搜索关键词', status_code=400) group_id = request.args.get('group_id', type=str) # 智能搜索:支持模糊匹配和精确匹配 messages_data = report_service.search_messages(keyword, limit, group_id) # 根据搜索类型过滤消息 if search_type == 'question': messages_data = [msg for msg in messages_data if msg.get('type') in ['text', 'mixed']] elif search_type == 'answer': messages_data = [msg for msg in messages_data if msg.get('type') in ['text', 'mixed']] formatted_messages = [] for msg in messages_data: message = dict(msg) message = message_processor.process_message_content(message) # 添加搜索关键词高亮 if keyword: content = message.get('content', '') if keyword.lower() in content.lower(): # 简单的关键词高亮(前端可以进一步处理) highlighted_content = content.replace(keyword, f'{keyword}') message['highlighted_content'] = highlighted_content formatted_messages.append(message) return api_response(data={'messages': formatted_messages, 'total': len(formatted_messages)}) except Exception as e: return api_response(success=False, error=e, status_code=500) # API路由 - 搜索建议 @v1.route('/api/search_suggestions') def search_suggestions(): """搜索建议API""" try: keyword = request.args.get('keyword', '') if not keyword: return api_response(data={'suggestions': []}) # 获取热门搜索关键词 suggestions = [] # 1. 从消息内容中提取相关关键词 try: messages = report_service.search_messages(keyword, 10, None) for msg in messages: content = msg.get('content', '') # 简单的关键词提取逻辑 words = content.split() for word in words: if len(word) > 2 and keyword.lower() in word.lower(): if word not in suggestions: suggestions.append(word) if len(suggestions) >= 5: break if len(suggestions) >= 5: break except: pass # 2. 添加常见搜索关键词 common_keywords = ['信息披露', '公司治理', '业务操作', '系统填报', '政策法规', '季度报告', '股东大会'] for kw in common_keywords: if keyword.lower() in kw.lower() and kw not in suggestions: suggestions.append(kw) if len(suggestions) >= 8: break return api_response(data={'suggestions': suggestions[:8]}) except Exception as e: return api_response(success=False, error=e, status_code=500) # API路由 - QA管理 @v1.route('/api/qa') def get_qa_list(): """获取QA列表API""" try: # 获取查询参数 group_id = request.args.get('group_id', '') keyword = request.args.get('keyword', '') start_time = request.args.get('start_time', '') end_time = request.args.get('end_time', '') # 构建查询条件 query_conditions = [] query_params = [] if group_id: query_conditions.append("groupId = ?") query_params.append(group_id) if start_time and end_time: try: # 将时间字符串转换为时间戳(毫秒级) start_timestamp = int(TimeUtils(start_time + ' 00:00:00').to_timestamp() * 1000) end_timestamp = int(TimeUtils(end_time + ' 23:59:59').to_timestamp() * 1000) query_conditions.append("start_time >= ? AND end_time <= ?") query_params.extend([start_timestamp, end_timestamp]) except ValueError: # 时间格式错误时,忽略时间条件 pass # 构建查询语句 base_query = "SELECT * FROM qa" if query_conditions: base_query += " WHERE " + " AND ".join(query_conditions) base_query += " ORDER BY id DESC" # 执行查询 conn = db_manager.get_connection() cursor = conn.cursor() cursor.execute(base_query, query_params) qa_data = cursor.fetchall() conn.close() qa_list = [dict(qa) for qa in qa_data] # 如果有关键词搜索,进行内容过滤 if keyword: filtered_qa_list = [] for qa in qa_list: # 检查群组ID是否匹配 if group_id != qa.get('groupId') and group_id != "": continue # 检查问题内容是否匹配 try: question_ids = json.loads(qa.get('question', '[]')) + json.loads(qa.get('answer', '[]')) if question_ids: # 获取问题消息内容进行匹配 messages = report_service.get_messages_by_ids(question_ids) for msg in messages: if keyword.lower() in (msg['content'] or '').lower(): filtered_qa_list.append(qa) break except Exception as e: print(e) pass qa_list = filtered_qa_list return api_response(data={'qa_list': qa_list, 'total': len(qa_list)}) except Exception as e: return api_response(success=False, error=e, status_code=500) @v1.route('/api/qa/') def get_qa_detail(qa_id): """获取单个QA详情API""" try: qa_data = report_service.get_qa_by_id(qa_id) if not qa_data: return api_response(success=False, error='QA记录不存在', status_code=404) qa_data = dict(qa_data) try: qa_data['question_list'] = json.loads(qa_data['question']) qa_data['answer_list'] = json.loads(qa_data['answer']) except: qa_data['question_list'] = [] qa_data['answer_list'] = [] return api_response(data={'qa': qa_data}) except Exception as e: return api_response(success=False, error=e, status_code=500) @v1.route('/api/qa', methods=['POST']) def create_qa(): """创建QA记录API""" try: data = request.get_json() group_id = data.get('groupId', '') question_ids = data.get('question', []) answer_ids = data.get('answer', []) # 处理时间字符串,转换为时间戳(毫秒级) start_time = data.get('start_time', '') end_time = data.get('end_time', '') try: if start_time: start_time = int(TimeUtils(start_time).to_timestamp() * 1000) if end_time: end_time = int(TimeUtils(end_time).to_timestamp() * 1000) except ValueError: # 时间格式错误时,使用空字符串 start_time = '' end_time = '' conn = db_manager.get_connection() cursor = conn.cursor() index_id = cursor.execute_query(f"SELECT id FROM qa order by id desc limit 1", fetch_all=False)[0] index_id += 1 cursor.execute(''' INSERT INTO qa ("id", "groupId", "question", "answer", "start_time", "end_time") VALUES (?, ?, ?, ?, ?, ?) ''', (index_id, group_id, json.dumps(question_ids), json.dumps(answer_ids), start_time, end_time)) conn.commit() qa_id = cursor.lastrowid conn.close() return api_response(data={ 'id': qa_id, 'message': 'QA记录创建成功' }) except Exception as e: return api_response(success=False, error=e, status_code=500) @v1.route('/api/qa/', methods=['PUT']) def update_qa(qa_id): """更新QA记录API""" try: data = request.get_json() group_id = data.get('groupId', '') question_ids = data.get('question', []) answer_ids = data.get('answer', []) # 处理时间字符串,转换为时间戳(毫秒级) start_time = data.get('start_time', '') end_time = data.get('end_time', '') try: if start_time: start_time = int(TimeUtils(start_time).to_timestamp() * 1000) if end_time: end_time = int(TimeUtils(end_time).to_timestamp() * 1000) except ValueError: # 时间格式错误时,保持原值不更新 pass # 检查记录是否存在 if not report_service.check_qa_exists(qa_id): return api_response(success=False, error='QA记录不存在', status_code=404) conn = db_manager.get_connection() cursor = conn.cursor() # 如果提供了时间参数,则更新时间字段 if start_time and end_time: cursor.execute(''' UPDATE qa SET groupId = ?, question = ?, answer = ?, start_time = ?, end_time = ? WHERE id = ? ''', (group_id, json.dumps(question_ids), json.dumps(answer_ids), start_time, end_time, qa_id)) else: cursor.execute(''' UPDATE qa SET groupId = ?, question = ?, answer = ? WHERE id = ? ''', (group_id, json.dumps(question_ids), json.dumps(answer_ids), qa_id)) conn.commit() conn.close() return api_response(data={'message': 'QA记录更新成功'}) except Exception as e: return api_response(success=False, error=e, status_code=500) @v1.route('/api/qa/', methods=['DELETE']) def delete_qa(qa_id): """删除单个QA记录API""" try: # 检查记录是否存在 if not report_service.check_qa_exists(qa_id): return api_response(success=False, error='QA记录不存在', status_code=404) conn = db_manager.get_connection() cursor = conn.cursor() cursor.execute('DELETE FROM qa WHERE id = ?', (qa_id,)) conn.commit() conn.close() return api_response(data={'message': 'QA记录删除成功'}) except Exception as e: return api_response(success=False, error=e, status_code=500) @v1.route('/api/qa/batch', methods=['DELETE']) def batch_delete_qa(): """批量删除QA记录API""" try: data = request.get_json() qa_ids = data.get('qa_ids', []) if not qa_ids: return api_response(success=False, error='请提供要删除的QA记录ID列表', status_code=400) # 检查所有记录是否存在 for qa_id in qa_ids: if not report_service.check_qa_exists(qa_id): return api_response(success=False, error=f'QA记录 {qa_id} 不存在', status_code=404) conn = db_manager.get_connection() cursor = conn.cursor() # 构建批量删除语句 placeholders = ','.join(['?'] * len(qa_ids)) query = f'DELETE FROM qa WHERE id IN ({placeholders})' cursor.execute(query, qa_ids) conn.commit() conn.close() return api_response(status_code=404) except Exception as e: return api_response(success=False, error=e, status_code=500) # API路由 - 报告生成 @v1.route('/api/reports/daily') def generate_daily_report(): """生成日报API""" try: date_str = request.args.get('date', now().to_string(TimeUtils.FORMAT_DATE)) target_date, start_timestamp, end_timestamp = report_service.get_date_range(date_str) yesterday_start, yesterday_end = report_service.get_previous_day_range(target_date) # 使用ReportService类获取数据 today_message_count = report_service.get_message_count_by_time_range(start_timestamp, end_timestamp) yesterday_message_count = report_service.get_message_count_by_time_range(yesterday_start, yesterday_end) message_change_rate = report_service.calculate_change_rate( today_message_count, yesterday_message_count ) # 获取活跃用户数 today_active_users = report_service.get_active_users_by_time_range(start_timestamp, end_timestamp) yesterday_active_users = report_service.get_active_users_by_time_range(yesterday_start, yesterday_end) user_change_rate = report_service.calculate_change_rate( today_active_users, yesterday_active_users ) # 获取群组和成员数量 group_count = report_service.get_group_count() member_count = report_service.get_member_count() # 获取QA记录数量 today_qa_count = report_service.get_qa_count_by_time_range(start_timestamp, end_timestamp) yesterday_qa_count = report_service.get_qa_count_by_time_range(yesterday_start, yesterday_end) non_qa_count = report_service.get_non_answer_qa_count_by_time_range(start_timestamp, end_timestamp) qa_change_rate = report_service.calculate_change_rate( today_qa_count, yesterday_qa_count ) # 获取群组活跃度 group_activity_data = report_service.get_group_activity_data(start_timestamp, end_timestamp) # 获取热点话题 recent_messages = report_service.get_recent_messages_by_time_range(start_timestamp, end_timestamp, 100) hot_topics = report_service.analyze_hot_topics(recent_messages) # 生成直方图 group_activity_chart = report_service.generate_group_activity_chart(group_activity_data) hourly_activity_chart = report_service.generate_hourly_activity_chart(start_timestamp, end_timestamp) # 生成日报 report_date = from_datetime(target_date).format_custom('%Y年%m月%d日 (%A)') daily_report = f"""# **致远速联高端社群运营日报** **报告周期**:{report_date} **覆盖范围**:{group_count}个群,总计成员 {member_count} 人 ## **1. 核心数据概览** | 指标 | 今日数据 | 昨日数据 | 变化 | | :--- | :--- | :--- | :--- | | 总消息量 | {today_message_count} 条 | {yesterday_message_count} 条 | {'↑' if message_change_rate >= 0 else '↓'}{abs(message_change_rate):.1f}% | | 活跃人数 | {today_active_users} 人 | {yesterday_active_users} 人 | {'↑' if user_change_rate >= 0 else '↓'}{abs(user_change_rate):.1f}% | | 问题新增记录 | {today_qa_count} 条 | {yesterday_qa_count} 条 | {'↑' if qa_change_rate >= 0 else '↓'}{abs(qa_change_rate):.1f}% | ## **2. 分群活跃度对比** ``` {group_activity_chart} ``` ## **3. 今日热点与时段** ### **热点问题TOP 3**: """ for i, topic in enumerate(hot_topics, 1): daily_report += f" {i}. {topic}\n" daily_report += f"""### **活跃时段**:消息高峰分布 ``` {hourly_activity_chart} ``` ## **4. 待办事项** {"有" + str(non_qa_count) + "条未被回答的问题,关注活跃群组的后续讨论,确保所有疑问被解答。" if non_qa_count > 0 else "无未被回答的问题,继续保持活跃群组的讨论。"} 明日重点关注:根据今日热点预测明日可能的问题。 """ return api_response(data={ 'report': daily_report, 'date': date_str, 'statistics': { 'today_message_count': today_message_count, 'yesterday_message_count': yesterday_message_count, 'message_change_rate': message_change_rate, 'today_active_users': today_active_users, 'yesterday_active_users': yesterday_active_users, 'user_change_rate': user_change_rate, 'today_qa_count': today_qa_count, 'yesterday_qa_count': yesterday_qa_count, 'qa_change_rate': qa_change_rate, 'group_count': group_count, 'member_count': member_count } }) except Exception as e: return api_response(success=False, error=e, status_code=500) @v1.route('/api/reports/weekly') def generate_weekly_report(): """生成周报API""" try: week_str = request.args.get('week', now().format_custom('%Y-%W')) # 解析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(from_datetime(start_date).to_timestamp()) end_timestamp = int(from_datetime(end_date).to_timestamp()) # 获取上周同期数据 last_week_start = start_date - timedelta(weeks=1) last_week_end = end_date - timedelta(weeks=1) last_week_start_ts = int(from_datetime(last_week_start).to_timestamp()) last_week_end_ts = int(from_datetime(last_week_end).to_timestamp()) # 使用ReportService类获取数据 week_message_count = report_service.get_message_count_by_time_range(start_timestamp, end_timestamp) last_week_message_count = report_service.get_message_count_by_time_range(last_week_start_ts, last_week_end_ts) message_change_rate = report_service.calculate_change_rate( week_message_count, last_week_message_count ) # 获取活跃用户数 week_active_users = report_service.get_active_users_by_time_range(start_timestamp, end_timestamp) # 获取群组和成员数量 group_count = report_service.get_group_count() member_count = report_service.get_member_count() # 获取QA记录数量 qa_count = report_service.get_qa_count_by_time_range(start_timestamp, end_timestamp) non_answer_qa_count = report_service.get_non_answer_qa_count_by_time_range(start_timestamp, end_timestamp) # 获取本周消息用于分类分析 week_messages = report_service.get_weekly_messages(start_timestamp, end_timestamp, 200) category_breakdown = report_service.analyze_category_breakdown(week_messages) # 生成周报 week_period = f"{from_datetime(start_date).format_custom('%Y年%m月%d日')} - {from_datetime(end_date).format_custom('%Y年%m月%d日')} (第{week_str.split('-')[1]}周)" weekly_report = f"""# **致远速联高端社群运营周报** **报告周期**:{week_period} ## **1. 本周摘要** - 本周社群总消息量 **{week_message_count}条**,{'环比上升' if message_change_rate >= 0 else '环比下降'} **{abs(message_change_rate):.1f}%**。 - 活跃用户数达到 **{week_active_users}人**,社群参与度良好。 - 新增问题解答记录 **{qa_count}条**,知识沉淀效果显著。 - 其中,有 **{non_answer_qa_count}条** 未被回答,值得注意。 ## **2. 深度洞察:上市公司提问分类** """ # 添加问题分类信息 for category, percentage in category_breakdown.items(): weekly_report += f"- {category}:{percentage}%\n" weekly_report += f"""\n**洞察**:问题集中在信披和法规,说明成员对合规性高度重视。 ## **3. 内容价值与知识沉淀** - **本周高价值问答**:系统自动识别并记录了{qa_count}条有价值的问答记录。 - **建议**:将高频问答整理至知识库,便于后续参考。 """ result = agent.chat_deepseek(weekly_report, "根据下面的内容给出总结和建议,格式如下:\n## **4. 总结与建议**\n**总结**:\n**建议**:\n") return api_response(data={ 'report': weekly_report + "\n" + result, 'week': week_str, 'statistics': { 'week_message_count': week_message_count, 'last_week_message_count': last_week_message_count, 'message_change_rate': message_change_rate, 'week_active_users': week_active_users, 'group_count': group_count, 'member_count': member_count, 'qa_count': qa_count, 'category_breakdown': category_breakdown } }) except Exception as e: return api_response(success=False, error=e, status_code=500)