689 lines
23 KiB
Python
689 lines
23 KiB
Python
from flask import Blueprint, render_template, request, redirect, url_for, flash, jsonify, current_app
|
||
from openai import OpenAI
|
||
from flask_prompt_master import db
|
||
from flask_prompt_master.models import User, Prompt, Feedback, PromptTemplate, WxUser
|
||
from flask_prompt_master.forms import PromptForm, FeedbackForm
|
||
from config import Config
|
||
import pymysql
|
||
from datetime import datetime
|
||
import requests
|
||
import hashlib
|
||
import time
|
||
import json
|
||
|
||
main_bp = Blueprint('main', __name__)
|
||
|
||
client = OpenAI(api_key=Config.LLM_API_KEY, base_url=Config.LLM_API_URL)
|
||
|
||
# 从配置中获取微信小程序配置
|
||
WX_APPID = Config.WX_APPID
|
||
WX_SECRET = Config.WX_SECRET
|
||
|
||
def get_system_prompt(template_id=None):
|
||
"""获取系统提示词模板"""
|
||
if template_id:
|
||
template = PromptTemplate.query.get(template_id)
|
||
if template:
|
||
return template.system_prompt
|
||
|
||
# 如果没有指定模板ID或模板不存在，返回默认模板
|
||
default_template = PromptTemplate.query.filter_by(is_default=True).first()
|
||
if default_template:
|
||
return default_template.system_prompt
|
||
|
||
# 如果数据库中没有模板，返回硬编码的默认模板
|
||
return """你是一个专业的提示词工程师，擅长将普通的描述转换为结构化、专业的 Prompt。
|
||
|
||
你需要：
|
||
1. 分析用户的需求和意图
|
||
2. 将其转换为清晰、详细的提示词
|
||
3. 添加必要的上下文和约束条件
|
||
4. 使用专业的术语和格式
|
||
5. 确保生成的提示词能够获得最佳的 AI 响应
|
||
|
||
请直接返回优化后的提示词，不要添加任何解释或其他内容。"""
|
||
|
||
def generate_with_llm(input_text, template_id=None):
|
||
"""调用大模型API生成提示词"""
|
||
try:
|
||
system_prompt = get_system_prompt(template_id)
|
||
|
||
# 打印参数
|
||
print("\n=== API 调用参数 ===")
|
||
print(f"模板ID: {template_id}")
|
||
print(f"输入文本: {input_text}")
|
||
print(f"系统提示: {system_prompt}")
|
||
print("==================\n")
|
||
|
||
response = client.chat.completions.create(
|
||
model="deepseek-chat",
|
||
messages=[
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": input_text}
|
||
],
|
||
temperature=0.7,
|
||
max_tokens=500
|
||
)
|
||
|
||
# 打印响应
|
||
generated_text = response.choices[0].message.content.strip()
|
||
print("\n=== API 响应结果 ===")
|
||
print(f"生成的提示词: {generated_text}")
|
||
print("==================\n")
|
||
|
||
return generated_text
|
||
except Exception as e:
|
||
current_app.logger.error(f'LLM API调用失败: {str(e)}')
|
||
return "提示词生成失败，请稍后重试"
|
||
|
||
def get_template_icon(category):
|
||
"""根据分类返回对应的Font Awesome图标类名"""
|
||
icons = {
|
||
'通用工具': 'fa-magic',
|
||
'内容创作': 'fa-pen-fancy',
|
||
'设计创意': 'fa-palette',
|
||
'技术研发': 'fa-code',
|
||
'商业营销': 'fa-chart-line',
|
||
'专业服务': 'fa-briefcase',
|
||
'教育培训': 'fa-graduation-cap',
|
||
'智慧城市': 'fa-city',
|
||
'工业制造': 'fa-industry',
|
||
'生活服务': 'fa-concierge-bell'
|
||
}
|
||
return icons.get(category, 'fa-star') # 默认返回星星图标
|
||
|
||
@main_bp.route('/', methods=['GET', 'POST'])
|
||
def index():
|
||
form = PromptForm()
|
||
templates = PromptTemplate.query.all()
|
||
|
||
# 获取所有可用的分类选项
|
||
industries = sorted(set(t.industry for t in templates if t.industry))
|
||
professions = sorted(set(t.profession for t in templates if t.profession))
|
||
categories = sorted(set(t.category for t in templates if t.category))
|
||
sub_categories = sorted(set(t.sub_category for t in templates if t.sub_category))
|
||
|
||
if form.validate_on_submit():
|
||
template_id = request.form.get('template_id')
|
||
generated_text = generate_with_llm(form.input_text.data, template_id)
|
||
|
||
# 获取默认用户的 uid
|
||
try:
|
||
conn = pymysql.connect(
|
||
host='localhost',
|
||
user='root',
|
||
password='123456',
|
||
database='food_db',
|
||
charset='utf8mb4'
|
||
)
|
||
cursor = conn.cursor()
|
||
cursor.execute("SELECT uid FROM user WHERE login_name = 'admin' LIMIT 1")
|
||
result = cursor.fetchone()
|
||
if result:
|
||
user_id = result[0]
|
||
else:
|
||
user_id = 1 # 如果没有找到用户，使用默认值
|
||
cursor.close()
|
||
conn.close()
|
||
except Exception as e:
|
||
print(f"获取用户ID失败: {str(e)}")
|
||
user_id = 1 # 如果查询失败，使用默认值
|
||
|
||
prompt = Prompt(
|
||
input_text=form.input_text.data,
|
||
generated_text=generated_text,
|
||
user_id=user_id # 使用查询到的用户ID
|
||
)
|
||
db.session.add(prompt)
|
||
db.session.commit()
|
||
return render_template('generate.html', form=form, prompt=prompt, templates=templates,
|
||
get_template_icon=get_template_icon, industries=industries,
|
||
professions=professions, categories=categories,
|
||
sub_categories=sub_categories)
|
||
return render_template('generate.html', form=form, prompt=None, templates=templates,
|
||
get_template_icon=get_template_icon, industries=industries,
|
||
professions=professions, categories=categories,
|
||
sub_categories=sub_categories)
|
||
|
||
@main_bp.route('/prompt/<int:prompt_id>')
|
||
def show_prompt(prompt_id):
|
||
prompt = Prompt.query.get_or_404(prompt_id)
|
||
return render_template('prompt.html', prompt=prompt)
|
||
|
||
@main_bp.route('/feedback/<int:prompt_id>', methods=['GET', 'POST'])
|
||
def submit_feedback(prompt_id):
|
||
prompt = Prompt.query.get_or_404(prompt_id)
|
||
form = FeedbackForm()
|
||
if form.validate_on_submit():
|
||
feedback = Feedback(
|
||
rating=form.rating.data,
|
||
comment=form.comment.data,
|
||
user_id=1, # 临时用户ID
|
||
prompt_id=prompt.id
|
||
)
|
||
db.session.add(feedback)
|
||
db.session.commit()
|
||
flash('感谢您的反馈！')
|
||
return redirect(url_for('main.show_prompt', prompt_id=prompt.id))
|
||
return render_template('feedback.html', form=form, prompt=prompt)
|
||
|
||
# 添加一个API端点来获取模板详情
|
||
@main_bp.route('/api/template/<int:template_id>')
|
||
def get_template_details(template_id):
|
||
template = PromptTemplate.query.get_or_404(template_id)
|
||
return jsonify({
|
||
'id': template.id,
|
||
'name': template.name,
|
||
'description': template.description,
|
||
'system_prompt': template.system_prompt
|
||
})
|
||
|
||
# 添加删除模板的API路由
|
||
@main_bp.route('/api/templates/<int:template_id>', methods=['DELETE'])
|
||
def delete_template(template_id):
|
||
try:
|
||
# 查找模板
|
||
template = PromptTemplate.query.get_or_404(template_id)
|
||
|
||
# 检查是否是默认模板
|
||
if template.is_default:
|
||
return jsonify({
|
||
'success': False,
|
||
'message': '默认模板不能删除'
|
||
}), 403
|
||
|
||
# 检查权限（可选：如果需要检查用户是否有权限删除）
|
||
# if template.user_id != current_user.id:
|
||
# return jsonify({
|
||
# 'success': False,
|
||
# 'message': '没有权限删除此模板'
|
||
# }), 403
|
||
|
||
# 删除模板
|
||
db.session.delete(template)
|
||
db.session.commit()
|
||
|
||
return jsonify({
|
||
'success': True,
|
||
'message': '模板删除成功'
|
||
})
|
||
|
||
except Exception as e:
|
||
# 回滚事务
|
||
db.session.rollback()
|
||
# 记录错误
|
||
current_app.logger.error(f'删除模板失败: {str(e)}')
|
||
return jsonify({
|
||
'success': False,
|
||
'message': '删除模板失败，请稍后重试'
|
||
}), 500
|
||
|
||
# 添加微信小程序API路由
|
||
@main_bp.route('/api/wx/generate', methods=['POST'])
|
||
def wx_generate_prompt():
|
||
"""微信小程序生成提示词接口"""
|
||
try:
|
||
data = request.get_json()
|
||
if not data or 'input_text' not in data or 'uid' not in data:
|
||
return jsonify({
|
||
'code': 400,
|
||
'message': '缺少必要参数',
|
||
'data': None
|
||
})
|
||
|
||
input_text = data.get('input_text')
|
||
template_id = data.get('template_id')
|
||
wx_user_id = data.get('uid') # 使用uid
|
||
|
||
# 调用大模型生成提示词
|
||
generated_text = generate_with_llm(input_text, template_id)
|
||
|
||
# 保存到数据库
|
||
prompt = Prompt(
|
||
input_text=input_text,
|
||
generated_text=generated_text,
|
||
wx_user_id=wx_user_id, # 使用wx_user_id
|
||
created_at=datetime.utcnow()
|
||
)
|
||
db.session.add(prompt)
|
||
db.session.commit()
|
||
|
||
return jsonify({
|
||
'code': 200,
|
||
'message': 'success',
|
||
'data': {
|
||
'prompt_id': prompt.id,
|
||
'input_text': prompt.input_text,
|
||
'generated_text': prompt.generated_text,
|
||
'created_at': prompt.created_at.strftime('%Y-%m-%d %H:%M:%S')
|
||
}
|
||
})
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"生成提示词失败: {str(e)}")
|
||
return jsonify({
|
||
'code': 500,
|
||
'message': str(e),
|
||
'data': None
|
||
})
|
||
|
||
@main_bp.route('/api/wx/templates', methods=['GET'])
|
||
def wx_get_templates():
|
||
"""获取提示词模板列表"""
|
||
try:
|
||
# 获取筛选参数
|
||
industry = request.args.get('industry')
|
||
profession = request.args.get('profession')
|
||
category = request.args.get('category')
|
||
sub_category = request.args.get('sub_category')
|
||
|
||
# 构建查询
|
||
query = PromptTemplate.query
|
||
|
||
if industry:
|
||
query = query.filter_by(industry=industry)
|
||
if profession:
|
||
query = query.filter_by(profession=profession)
|
||
if category:
|
||
query = query.filter_by(category=category)
|
||
if sub_category:
|
||
query = query.filter_by(sub_category=sub_category)
|
||
|
||
templates = query.all()
|
||
|
||
# 返回模板列表
|
||
return jsonify({
|
||
'code': 200,
|
||
'message': 'success',
|
||
'data': [{
|
||
'id': t.id,
|
||
'name': t.name,
|
||
'description': t.description,
|
||
'category': t.category,
|
||
'industry': t.industry,
|
||
'profession': t.profession,
|
||
'sub_category': t.sub_category,
|
||
'is_default': t.is_default
|
||
} for t in templates]
|
||
})
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"获取模板列表失败: {str(e)}")
|
||
return jsonify({
|
||
'code': 500,
|
||
'message': str(e),
|
||
'data': None
|
||
})
|
||
|
||
@main_bp.route('/api/wx/template/<int:template_id>', methods=['GET'])
|
||
def wx_get_template_detail(template_id):
|
||
"""获取模板详情"""
|
||
try:
|
||
template = PromptTemplate.query.get_or_404(template_id)
|
||
|
||
return jsonify({
|
||
'code': 200,
|
||
'message': 'success',
|
||
'data': {
|
||
'id': template.id,
|
||
'name': template.name,
|
||
'description': template.description,
|
||
'category': template.category,
|
||
'industry': template.industry,
|
||
'profession': template.profession,
|
||
'sub_category': template.sub_category,
|
||
'system_prompt': template.system_prompt,
|
||
'is_default': template.is_default
|
||
}
|
||
})
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"获取模板详情失败: {str(e)}")
|
||
return jsonify({
|
||
'code': 500,
|
||
'message': str(e),
|
||
'data': None
|
||
})
|
||
|
||
@main_bp.route('/api/wx/login', methods=['POST'])
|
||
def wx_login():
|
||
"""微信小程序登录接口"""
|
||
try:
|
||
# 添加调试日志
|
||
print("\n=== 微信登录配置 ===")
|
||
print(f"APPID: {WX_APPID}")
|
||
print(f"SECRET: {WX_SECRET}")
|
||
print("==================\n")
|
||
|
||
data = request.get_json()
|
||
if not data or 'code' not in data:
|
||
return jsonify({
|
||
'code': 400,
|
||
'message': '缺少code参数',
|
||
'data': None
|
||
})
|
||
|
||
code = data.get('code')
|
||
print(f"收到的code: {code}")
|
||
|
||
# 请求微信接口
|
||
wx_url = 'https://api.weixin.qq.com/sns/jscode2session'
|
||
params = {
|
||
'appid': WX_APPID,
|
||
'secret': WX_SECRET,
|
||
'js_code': code,
|
||
'grant_type': 'authorization_code'
|
||
}
|
||
|
||
response = requests.get(wx_url, params=params)
|
||
wx_data = response.json()
|
||
print(f"微信返回数据: {wx_data}")
|
||
|
||
if 'errcode' in wx_data:
|
||
return jsonify({
|
||
'code': 500,
|
||
'message': f"微信登录失败：{wx_data.get('errmsg')}",
|
||
'data': None
|
||
})
|
||
|
||
openid = wx_data.get('openid')
|
||
session_key = wx_data.get('session_key')
|
||
|
||
# 查找或创建微信用户
|
||
wx_user = WxUser.query.filter_by(openid=openid).first()
|
||
if not wx_user:
|
||
wx_user = WxUser(
|
||
openid=openid,
|
||
session_key=session_key
|
||
)
|
||
db.session.add(wx_user)
|
||
db.session.commit()
|
||
else:
|
||
wx_user.session_key = session_key
|
||
wx_user.last_login = datetime.utcnow()
|
||
db.session.commit()
|
||
|
||
# 生成登录态token
|
||
token = hashlib.md5(f'{openid}{int(time.time())}'.encode()).hexdigest()
|
||
|
||
return jsonify({
|
||
'code': 200,
|
||
'message': 'success',
|
||
'data': {
|
||
'token': token,
|
||
'openid': openid,
|
||
'uid': wx_user.id,
|
||
'user_info': {
|
||
'id': wx_user.id,
|
||
'nickname': wx_user.nickname,
|
||
'avatar_url': wx_user.avatar_url,
|
||
'gender': wx_user.gender,
|
||
'phone': wx_user.phone
|
||
}
|
||
}
|
||
})
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"微信登录失败: {str(e)}")
|
||
return jsonify({
|
||
'code': 500,
|
||
'message': str(e),
|
||
'data': None
|
||
})
|
||
|
||
@main_bp.route('/api/wx/update_userinfo', methods=['POST'])
|
||
def wx_update_userinfo():
|
||
"""更新微信用户信息"""
|
||
try:
|
||
data = request.get_json()
|
||
if not data or 'openid' not in data:
|
||
return jsonify({
|
||
'code': 400,
|
||
'message': '缺少必要参数',
|
||
'data': None
|
||
})
|
||
|
||
openid = data.get('openid')
|
||
wx_user = WxUser.query.filter_by(openid=openid).first()
|
||
|
||
if not wx_user:
|
||
return jsonify({
|
||
'code': 404,
|
||
'message': '用户不存在',
|
||
'data': None
|
||
})
|
||
|
||
# 更新用户信息
|
||
if 'nickName' in data:
|
||
wx_user.nickname = data['nickName']
|
||
if 'avatarUrl' in data:
|
||
wx_user.avatar_url = data['avatarUrl']
|
||
if 'gender' in data:
|
||
wx_user.gender = data['gender']
|
||
if 'country' in data:
|
||
wx_user.country = data['country']
|
||
if 'province' in data:
|
||
wx_user.province = data['province']
|
||
if 'city' in data:
|
||
wx_user.city = data['city']
|
||
if 'language' in data:
|
||
wx_user.language = data['language']
|
||
|
||
wx_user.updated_at = datetime.utcnow()
|
||
db.session.commit()
|
||
|
||
return jsonify({
|
||
'code': 200,
|
||
'message': 'success',
|
||
'data': {
|
||
'id': wx_user.id,
|
||
'nickname': wx_user.nickname,
|
||
'avatar_url': wx_user.avatar_url,
|
||
'gender': wx_user.gender,
|
||
'country': wx_user.country,
|
||
'province': wx_user.province,
|
||
'city': wx_user.city,
|
||
'language': wx_user.language,
|
||
'phone': wx_user.phone
|
||
}
|
||
})
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"更新用户信息失败: {str(e)}")
|
||
return jsonify({
|
||
'code': 500,
|
||
'message': str(e),
|
||
'data': None
|
||
})
|
||
|
||
@main_bp.route('/api/wx/prompts', methods=['GET'])
|
||
def wx_get_prompts():
|
||
"""获取用户的提示词历史记录"""
|
||
try:
|
||
# 获取参数
|
||
uid = request.args.get('uid')
|
||
page = request.args.get('page', 1, type=int)
|
||
per_page = request.args.get('per_page', 10, type=int)
|
||
|
||
if not uid:
|
||
return jsonify({
|
||
'code': 400,
|
||
'message': '缺少用户ID',
|
||
'data': None
|
||
})
|
||
|
||
# 查询该用户的所有提示词记录
|
||
query = Prompt.query.filter_by(wx_user_id=uid)\
|
||
.order_by(Prompt.created_at.desc())
|
||
|
||
# 分页
|
||
pagination = query.paginate(page=page, per_page=per_page, error_out=False)
|
||
prompts = pagination.items
|
||
|
||
# 返回数据
|
||
return jsonify({
|
||
'code': 200,
|
||
'message': 'success',
|
||
'data': {
|
||
'prompts': [{
|
||
'id': p.id,
|
||
'input_text': p.input_text,
|
||
'generated_text': p.generated_text,
|
||
'created_at': p.created_at.strftime('%Y-%m-%d %H:%M:%S')
|
||
} for p in prompts],
|
||
'pagination': {
|
||
'total': pagination.total, # 总记录数
|
||
'pages': pagination.pages, # 总页数
|
||
'current_page': page, # 当前页
|
||
'per_page': per_page, # 每页记录数
|
||
'has_next': pagination.has_next, # 是否有下一页
|
||
'has_prev': pagination.has_prev # 是否有上一页
|
||
}
|
||
}
|
||
})
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"获取提示词历史失败: {str(e)}")
|
||
return jsonify({
|
||
'code': 500,
|
||
'message': str(e),
|
||
'data': None
|
||
})
|
||
|
||
@main_bp.route('/api/wx/prompt/<int:prompt_id>', methods=['GET'])
|
||
def wx_get_prompt_detail(prompt_id):
|
||
"""获取提示词详情"""
|
||
try:
|
||
prompt = Prompt.query.get_or_404(prompt_id)
|
||
|
||
# 可以选择性地验证用户身份
|
||
# uid = request.args.get('uid')
|
||
# if str(prompt.wx_user_id) != str(uid):
|
||
# return jsonify({
|
||
# 'code': 403,
|
||
# 'message': '无权访问此记录',
|
||
# 'data': None
|
||
# })
|
||
|
||
return jsonify({
|
||
'code': 200,
|
||
'message': 'success',
|
||
'data': {
|
||
'id': prompt.id,
|
||
'input_text': prompt.input_text,
|
||
'generated_text': prompt.generated_text,
|
||
'created_at': prompt.created_at.strftime('%Y-%m-%d %H:%M:%S'),
|
||
'feedbacks': [{
|
||
'id': f.id,
|
||
'rating': f.rating,
|
||
'comment': f.comment,
|
||
'created_at': f.created_at.strftime('%Y-%m-%d %H:%M:%S')
|
||
} for f in prompt.feedbacks]
|
||
}
|
||
})
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"获取提示词详情失败: {str(e)}")
|
||
return jsonify({
|
||
'code': 500,
|
||
'message': str(e),
|
||
'data': None
|
||
})
|
||
|
||
@main_bp.route('/api/wx/prompts/count', methods=['GET'])
|
||
def wx_get_prompts_count():
|
||
"""获取用户的提示词历史记录数量"""
|
||
try:
|
||
# 获取用户ID
|
||
uid = request.args.get('uid')
|
||
|
||
if not uid:
|
||
return jsonify({
|
||
'code': 400,
|
||
'message': '缺少用户ID',
|
||
'data': None
|
||
})
|
||
|
||
# 查询该用户的提示词记录数量
|
||
count = Prompt.query.filter_by(wx_user_id=uid).count()
|
||
|
||
# 获取今日记录数量
|
||
today = datetime.now().date()
|
||
today_count = Prompt.query.filter_by(wx_user_id=uid)\
|
||
.filter(db.func.date(Prompt.created_at) == today)\
|
||
.count()
|
||
|
||
# 获取本月记录数量
|
||
this_month = today.replace(day=1)
|
||
month_count = Prompt.query.filter_by(wx_user_id=uid)\
|
||
.filter(db.func.date(Prompt.created_at) >= this_month)\
|
||
.count()
|
||
|
||
return jsonify({
|
||
'code': 200,
|
||
'message': 'success',
|
||
'data': {
|
||
'total_count': count, # 总记录数
|
||
'today_count': today_count, # 今日记录数
|
||
'month_count': month_count, # 本月记录数
|
||
'uid': uid
|
||
}
|
||
})
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"获取提示词历史数量失败: {str(e)}")
|
||
return jsonify({
|
||
'code': 500,
|
||
'message': str(e),
|
||
'data': None
|
||
})
|
||
|
||
@main_bp.route('/api/wx/prompt/<int:prompt_id>', methods=['DELETE'])
|
||
def wx_delete_prompt(prompt_id):
|
||
"""删除提示词记录"""
|
||
try:
|
||
# 从 URL 参数或请求体中获取用户ID
|
||
uid = request.args.get('uid') or request.get_json().get('uid')
|
||
if not uid:
|
||
return jsonify({
|
||
'code': 400,
|
||
'message': '缺少用户ID',
|
||
'data': None
|
||
})
|
||
|
||
# 查找记录
|
||
prompt = Prompt.query.get_or_404(prompt_id)
|
||
|
||
# 验证是否是用户自己的记录
|
||
if str(prompt.wx_user_id) != str(uid):
|
||
return jsonify({
|
||
'code': 403,
|
||
'message': '无权删除此记录',
|
||
'data': None
|
||
})
|
||
|
||
# 删除相关的反馈
|
||
Feedback.query.filter_by(prompt_id=prompt_id).delete()
|
||
|
||
# 删除提示词记录
|
||
db.session.delete(prompt)
|
||
db.session.commit()
|
||
|
||
return jsonify({
|
||
'code': 200,
|
||
'message': '删除成功',
|
||
'data': {
|
||
'id': prompt_id
|
||
}
|
||
})
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"删除提示词记录失败: {str(e)}")
|
||
db.session.rollback()
|
||
return jsonify({
|
||
'code': 500,
|
||
'message': str(e),
|
||
'data': None
|
||
})
|
||
|
||
# ... 其他路由保持不变，但要把 @app 改成 @main_bp ... |