#!/usr/bin/env python # -*- coding: utf-8 -*- from flask import Flask, request, make_response import hashlib import time import xml.etree.ElementTree as ET import os import json from openai import OpenAI from dotenv import load_dotenv from duckduckgo_search import DDGS import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart # 加载环境变量 load_dotenv() app = Flask(__name__) # 配置 TOKEN = os.getenv('TOKEN') API_KEY = os.getenv("API_KEY") BASE_URL = os.getenv("OPENAI_BASE_URL") EMAIL_KEY = os.getenv("EMAIL_KEY") client = OpenAI(api_key=API_KEY, base_url=BASE_URL) # 存储用户会话信息 user_sessions = {} # 定义函数列表 FUNCTIONS = [ { "name": "search_duckduckgo", "description": "使用DuckDuckGo搜索引擎查询信息。可以搜索最新新闻、文章、博客等内容。", "parameters": { "type": "object", "properties": { "keywords": { "type": "array", "items": {"type": "string"}, "description": "搜索的关键词列表。例如：['Python', '机器学习', '最新进展']。" } }, "required": ["keywords"] } }, { "name": "send_email", "description": "发送电子邮件。", "parameters": { "type": "object", "properties": { "to": { "type": "string", "description": "收件人邮箱地址" }, "subject": { "type": "string", "description": "邮件主题" }, "content": { "type": "string", "description": "邮件内容" } }, "required": ["to", "subject", "content"] } } ] def verify_wechat(request): # 获取微信服务器发送过来的参数 data = request.args signature = data.get('signature') timestamp = data.get('timestamp') nonce = data.get('nonce') echostr = data.get('echostr') # 对参数进行字典排序，拼接字符串 temp = [timestamp, nonce, TOKEN] temp.sort() temp = ''.join(temp) # 加密 if (hashlib.sha1(temp.encode('utf8')).hexdigest() == signature): return echostr else: return 'error', 403 def getUserMessageContentFromXML(xml_content): # 解析XML字符串 root = ET.fromstring(xml_content) # 提取数据 content = root.find('Content').text from_user_name = root.find('FromUserName').text to_user_name = root.find('ToUserName').text return content, from_user_name, to_user_name def generate_response_xml(from_user_name, to_user_name, output_content): output_xml = ''' %s ''' response = make_response(output_xml % (from_user_name, to_user_name, str(int(time.time())), output_content)) response.content_type = 'application/xml' return response def get_openai_response(messages, functions=None, function_call=None): try: response = client.chat.completions.create( model="gpt-4o-mini", messages=messages, functions=functions, function_call=function_call ) return response.choices[0].message except Exception as e: print(f"调用OpenAI API时出错: {str(e)}") return None def split_message(message, max_length=500): return [message[i:i+max_length] for i in range(0, len(message), max_length)] def search_duckduckgo(keywords): search_term = " ".join(keywords) with DDGS() as ddgs: return list(ddgs.text(keywords=search_term, region="cn-zh", safesearch="on", max_results=5)) def send_email(to, subject, content): try: with smtplib.SMTP('106.15.184.28', 8025) as smtp: smtp.login("jwt", EMAIL_KEY) message = MIMEMultipart() message['From'] = "Me " message['To'] = to message['Subject'] = subject message.attach(MIMEText(content, 'html')) smtp.sendmail("aixiao@aixiao.xyz", to, message.as_string()) return True except Exception as e: print(f"发送邮件时出错: {str(e)}") return False def process_function_call(response_message): function_name = response_message.function_call.name function_args = json.loads(response_message.function_call.arguments) print(f"\n模型选择调用函数: {function_name}") if function_name == "search_duckduckgo": keywords = function_args.get('keywords', []) if not keywords: print("错误：模型没有提供搜索关键词") return None print(f"关键词: {', '.join(keywords)}") return search_duckduckgo(keywords) elif function_name == "send_email": to = function_args.get('to') subject = function_args.get('subject') content = function_args.get('content') if send_email(to, subject, content): return "邮件发送成功" else: return "邮件发送失败" else: print(f"未知的函数名称: {function_name}") return None @app.route('/api/wx', methods=['GET', 'POST']) def wechatai(): if request.method == 'GET': return verify_wechat(request) else: # 处理POST请求 print("user request data: ", request.data) user_message_content, from_user_name, to_user_name = getUserMessageContentFromXML(request.data) print("user message content: ", user_message_content) if from_user_name not in user_sessions: user_sessions[from_user_name] = {'messages': [], 'pending_response': []} session = user_sessions[from_user_name] if user_message_content.lower() == '继续': if session['pending_response']: response_content = session['pending_response'].pop(0) if session['pending_response']: response_content += '\n\n回复"继续"获取下一部分。' else: response_content += '\n\n回复结束。' else: response_content = "没有待发送的消息。" else: session['messages'].append({"role": "user", "content": user_message_content}) response_message = get_openai_response(session['messages'], functions=FUNCTIONS, function_call="auto") if response_message.function_call: function_response = process_function_call(response_message) if function_response: session['messages'].extend([ response_message.model_dump(), { "role": "function", "name": response_message.function_call.name, "content": json.dumps(function_response, ensure_ascii=False) } ]) final_response = get_openai_response(session['messages']) if final_response: gpt_response = final_response.content else: gpt_response = "抱歉，我遇到了一些问题，无法回答您的问题。" else: gpt_response = "抱歉，我在执行任务时遇到了问题。" else: gpt_response = response_message.content session['messages'].append({"role": "assistant", "content": gpt_response}) response_parts = split_message(gpt_response) if len(response_parts) > 1: response_content = response_parts[0] + '\n\n回复"继续"获取下一部分。' session['pending_response'] = response_parts[1:] else: response_content = response_parts[0] return generate_response_xml(from_user_name, to_user_name, response_content) if __name__ == '__main__': app.run(host='0.0.0.0', port=7860, debug=True)