370 lines
13 KiB
Python
370 lines
13 KiB
Python
# 练习方案生成服务
|
||
|
||
import os
|
||
import re
|
||
import requests
|
||
from datetime import datetime
|
||
from app.config import load_api_config
|
||
|
||
|
||
def generate_practice_plan(
|
||
student_name, problems, practice_time="30-60分钟", goals=None
|
||
):
|
||
"""
|
||
根据学员问题和练习时间生成针对性练习方案
|
||
|
||
Args:
|
||
student_name: 学员姓名
|
||
problems: 问题列表 [{problem_id, problem_name, severity, level, content}]
|
||
practice_time: 练习时间描述
|
||
goals: 目标列表 [{goal_id, goal_name, goal_content, status, mastery_level}]，默认为空列表
|
||
|
||
Returns:
|
||
dict: 包含方案内容的字典
|
||
"""
|
||
|
||
# 解析练习时间
|
||
time_mapping = {
|
||
"15分钟": {"total": 15, "basic": 10, "tech": 2, "piece": 3},
|
||
"30分钟": {"total": 30, "basic": 15, "tech": 5, "piece": 10},
|
||
"45分钟": {"total": 45, "basic": 15, "tech": 10, "piece": 20},
|
||
"60分钟": {"total": 60, "basic": 20, "tech": 15, "piece": 25},
|
||
"90分钟": {"total": 90, "basic": 25, "tech": 25, "piece": 40},
|
||
"120分钟": {"total": 120, "basic": 30, "tech": 35, "piece": 55},
|
||
"150分钟以上": {"total": 150, "basic": 35, "tech": 45, "piece": 70},
|
||
}
|
||
time_config = time_mapping.get(practice_time, time_mapping["30分钟"])
|
||
|
||
# 从数据库问题内容构建
|
||
problem_contents = []
|
||
for p in problems:
|
||
content = p.get("content", "") or ""
|
||
problem_contents.append(
|
||
{
|
||
"name": p["problem_name"],
|
||
"severity": p["severity"],
|
||
"level": p.get("level", ""),
|
||
"content": _extract_key_sections(content) if content else {"problem": f"针对{p['problem_name']}的练习"},
|
||
"time_allocation": _calculate_time_allocation(
|
||
p["severity"], time_config
|
||
),
|
||
}
|
||
)
|
||
|
||
# 生成方案
|
||
plan = {
|
||
"student_name": student_name,
|
||
"practice_time": practice_time,
|
||
"total_daily_minutes": time_config["total"],
|
||
"problems": [
|
||
{
|
||
"name": p["name"],
|
||
"severity": p["severity"],
|
||
"level": p.get("level", ""),
|
||
"focus": p["time_allocation"],
|
||
}
|
||
for p in problem_contents
|
||
],
|
||
"generated_at": datetime.now().strftime("%Y-%m-%d"),
|
||
}
|
||
|
||
return plan
|
||
|
||
|
||
def _extract_key_sections(content):
|
||
"""从问题文件中提取关键部分"""
|
||
sections = {}
|
||
|
||
# 提取问题表现
|
||
match = re.search(r"## 问题表现\s*\n(.*?)(?=##|$)", content, re.DOTALL)
|
||
if match:
|
||
sections["problem"] = match.group(1).strip()
|
||
|
||
# 提取原因分析
|
||
match = re.search(r"## 原因分析\s*\n(.*?)(?=##|$)", content, re.DOTALL)
|
||
if match:
|
||
sections["reason"] = match.group(1).strip()
|
||
|
||
# 提取日常基础练习
|
||
match = re.search(
|
||
r"## 针对性练习方案\s*.*?\| 练习名称.*?\n\|[-|\s]+\|.*?\n((?:\|.*?\n)+)",
|
||
content,
|
||
re.DOTALL,
|
||
)
|
||
if match:
|
||
table_lines = match.group(1).strip().split("\n")
|
||
practices = []
|
||
for line in table_lines:
|
||
if "|" in line and line.strip():
|
||
parts = [p.strip() for p in line.split("|")[1:-1]]
|
||
if len(parts) >= 4 and parts[0] != "练习名称":
|
||
practices.append(
|
||
{
|
||
"name": parts[0],
|
||
"duration": parts[1],
|
||
"frequency": parts[2],
|
||
"purpose": parts[3],
|
||
}
|
||
)
|
||
sections["practices"] = practices
|
||
|
||
# 提取具体操作
|
||
match = re.search(r"## 具体操作\s*\n(.*?)(?=##|$)", content, re.DOTALL)
|
||
if match:
|
||
sections["operations"] = match.group(1).strip()
|
||
|
||
# 提取练习提醒
|
||
match = re.search(r"### ⚠️ 禁忌\s*\n(.*?)(?=###|$)", content, re.DOTALL)
|
||
if match:
|
||
sections["taboos"] = match.group(1).strip()
|
||
|
||
match = re.search(r"### ✓ 正确做法\s*\n(.*?)(?=###|$)", content, re.DOTALL)
|
||
if match:
|
||
sections["dos"] = match.group(1).strip()
|
||
|
||
# 提取评估标准
|
||
match = re.search(r"## 评估标准\s*\n(.*?)(?=##|$)", content, re.DOTALL)
|
||
if match:
|
||
sections["standards"] = match.group(1).strip()
|
||
|
||
return sections
|
||
|
||
|
||
def _calculate_time_allocation(severity, time_config):
|
||
"""根据严重程度计算时间分配"""
|
||
multiplier = {"严重": 1.3, "中等": 1.0, "轻微": 0.7}.get(severity, 1.0)
|
||
|
||
return {
|
||
"basic": int(time_config["basic"] * multiplier),
|
||
"tech": int(time_config["tech"] * multiplier),
|
||
"piece": int(time_config["piece"] / multiplier)
|
||
if severity == "严重"
|
||
else time_config["piece"],
|
||
}
|
||
|
||
|
||
def _get_severity_advice(severity):
|
||
"""根据严重程度给出建议"""
|
||
advice = {
|
||
"严重": "需要重点突破，建议每次练习开始时先练这部分，精力最充沛时处理难点。",
|
||
"中等": "保持稳定练习，逐步改善，不要急于求成。",
|
||
"轻微": "作为日常练习的一部分即可，注意保持。",
|
||
}
|
||
return advice.get(severity, "")
|
||
|
||
|
||
# ==================== AI生成报告 ====================
|
||
|
||
|
||
def generate_ai_report(
|
||
student_name, wechat_nickname, problems, practice_time, time_config, template_id=None, dry_run=False, goals=None
|
||
):
|
||
"""
|
||
使用AI生成个性化的练习报告
|
||
|
||
Args:
|
||
student_name: 学员姓名
|
||
wechat_nickname: 微信昵称
|
||
problems: 问题列表 [{problem_id, problem_name, severity, level, content}]
|
||
practice_time: 练习时间描述
|
||
time_config: 时间配置 {total, basic, tech, piece}
|
||
template_id: 可选的AI提示词模板ID
|
||
dry_run: 如果为True，只返回提示词不调用API
|
||
goals: 目标列表 [{goal_id, goal_name, goal_content, status, mastery_level}]
|
||
practice_time: 练习时间描述
|
||
time_config: 时间配置 {total, basic, tech, piece}
|
||
template_id: 可选的AI提示词模板ID
|
||
dry_run: 如果为True，只返回提示词不调用API
|
||
|
||
Returns:
|
||
tuple: (prompt, ai_report, error) - 提示词、AI报告内容、错误信息
|
||
"""
|
||
config = load_api_config()
|
||
api_key = config.get("api_key", "")
|
||
base_url = config.get("base_url", "")
|
||
model = config.get("model", "doubao-seed-2.0-pro")
|
||
temperature = config.get("temperature", 0.7)
|
||
provider = config.get("provider", "volcengine")
|
||
|
||
# 优先从数据库模板获取AI提示词，其次从配置文件
|
||
prompt_template = ""
|
||
try:
|
||
from app.models import Template
|
||
if template_id:
|
||
# 使用指定的模板
|
||
tmpl = Template.query.get(template_id)
|
||
if tmpl and tmpl.type == "ai_prompt":
|
||
prompt_template = tmpl.content
|
||
else:
|
||
# 使用排序后的第一个模板
|
||
tmpl = Template.query.filter_by(type="ai_prompt").order_by(Template.sort_order.asc()).first()
|
||
if tmpl:
|
||
prompt_template = tmpl.content
|
||
except:
|
||
pass
|
||
if not prompt_template:
|
||
prompt_template = config.get("prompt_template", "")
|
||
|
||
# 从问题文件内容中提取关键信息
|
||
def extract_problem_info(content_str):
|
||
if not content_str or not isinstance(content_str, str):
|
||
return {"practices": "针对性练习", "standards": ""}
|
||
import re
|
||
|
||
# 匹配表格中的练习名称
|
||
practice_matches = re.findall(r"\|\s*([^|]+?)\s*\|\s*\d+\s*分钟", content_str)
|
||
practice_names = (
|
||
", ".join([m.strip() for m in practice_matches[:3]])
|
||
if practice_matches
|
||
else "针对性练习"
|
||
)
|
||
|
||
# 提取评估标准
|
||
standards_match = re.search(
|
||
r"## 评估标准\s*\n(.*?)(?=##|$)", content_str, re.DOTALL
|
||
)
|
||
standards = standards_match.group(1).strip() if standards_match else ""
|
||
|
||
return {"practices": practice_names, "standards": standards}
|
||
|
||
# 构建学员问题列表（学员被诊断出的问题）
|
||
student_problems = []
|
||
for p in problems:
|
||
level = p.get("level", "入门")
|
||
severity = p.get("severity", "中等")
|
||
student_problems.append(f"- **{p['problem_name']}**（级别：{level}，程度：{severity}）")
|
||
|
||
# 构建完整问题内容（包含所有详细信息）
|
||
problems_full = []
|
||
for p in problems:
|
||
content = p.get("content", "") or ""
|
||
level = p.get("level", "入门")
|
||
severity = p.get("severity", "中等")
|
||
|
||
problems_full.append(
|
||
f"### 问题：{p['problem_name']}\n"
|
||
f"**学员级别**: {level} | **问题程度**: {severity}\n\n"
|
||
f"**详细内容和练习方法**:\n{content}"
|
||
)
|
||
|
||
# 构建学员未达成目标列表（只包含未完成的目标）
|
||
student_goals = []
|
||
if goals:
|
||
for g in goals:
|
||
# 只处理未完成的目标
|
||
if g.get("status") != "已完成":
|
||
student_goals.append(
|
||
f"- **{g['goal_name']}**\n 内容：{g.get('goal_content', '未提供具体内容')}"
|
||
)
|
||
|
||
# 使用配置的模板，如果为空则使用默认模板
|
||
if prompt_template:
|
||
prompt = prompt_template.format(
|
||
student_name=student_name,
|
||
wechat_nickname=wechat_nickname or "未设置",
|
||
practice_time=practice_time,
|
||
student_problems="\n".join(student_problems) if student_problems else "无",
|
||
problems="\n\n".join(problems_full) if problems_full else "无",
|
||
student_goals="\n".join(student_goals) if student_goals else "无",
|
||
)
|
||
else:
|
||
# 默认模板（向后兼容）
|
||
prompt = f"""你是一位专业的钢琴教师，请为学员生成一份简洁的个性化练习方案报告。
|
||
|
||
## 学员信息
|
||
- 姓名：{student_name}
|
||
- 微信昵称：{wechat_nickname}
|
||
- 每日练习时间：{practice_time}（约{time_config["total"]}分钟）
|
||
- 基础练习：{time_config["basic"]}分钟 | 技术练习：{time_config["tech"]}分钟 | 曲目练习：{time_config["piece"]}分钟
|
||
|
||
## 级别定义
|
||
- 启蒙：刚开始学习，需要建立基本概念
|
||
- 入门：掌握基础姿势和简单曲目
|
||
- 进阶：能独立练习中等难度曲目
|
||
- 熟练：技术稳定，能处理复杂曲目
|
||
- 精通：达到专业演奏水平
|
||
|
||
## 问题详情
|
||
{chr(10).join(problems_summary)}
|
||
|
||
请生成一份简洁的练习方案报告（不超过400字），包含：
|
||
1. 方案概述（一句话）
|
||
2. 每日练习安排（按时间阶段列出）
|
||
3. 针对每个问题的核心练习建议（1-2句话/问题）
|
||
4. 重点注意事项（最多3条）
|
||
|
||
语言要专业、简洁、有鼓励性。使用Markdown格式。"""
|
||
|
||
# 如果是 dry_run（预览模式），直接返回提示词不调用API
|
||
if dry_run:
|
||
# 计算各部分字数
|
||
student_problems_text = "\n".join(student_problems) if student_problems else "无"
|
||
problems_text = "\n\n".join(problems_full) if problems_full else "无"
|
||
student_goals_text = "\n".join(student_goals) if student_goals else "无"
|
||
return prompt, None, None, {
|
||
"student_problems_length": len(student_problems_text),
|
||
"problems_length": len(problems_text),
|
||
"student_goals_length": len(student_goals_text),
|
||
}
|
||
|
||
# API调用需要 Key
|
||
if not api_key:
|
||
return None, None, "未配置API Key，请在设置页面配置", None
|
||
|
||
# 调用API
|
||
headers = {
|
||
"Authorization": f"Bearer {api_key}",
|
||
"Content-Type": "application/json",
|
||
}
|
||
|
||
payload = {
|
||
"model": model,
|
||
"messages": [{"role": "user", "content": prompt}],
|
||
"temperature": temperature,
|
||
"max_tokens": 2000,
|
||
}
|
||
|
||
# 确定 endpoint
|
||
if provider == "minimax":
|
||
endpoint = f"{base_url}/messages"
|
||
else:
|
||
endpoint = f"{base_url}/chat/completions"
|
||
|
||
try:
|
||
response = requests.post(
|
||
endpoint,
|
||
headers=headers,
|
||
json=payload,
|
||
timeout=60,
|
||
)
|
||
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
# MiniMax (Anthropic格式) 返回结构不同
|
||
if provider == "minimax":
|
||
# MiniMax 返回的 content 可能是 [{"type": "thinking", ...}, {"type": "text", "text": "..."}]
|
||
content_list = result.get("content", [])
|
||
content = ""
|
||
for item in content_list:
|
||
if item.get("type") == "text":
|
||
content = item.get("text", "")
|
||
break
|
||
else:
|
||
content = (
|
||
result.get("choices", [{}])[0].get("message", {}).get("content", "")
|
||
)
|
||
return prompt, content, None, None
|
||
else:
|
||
error_msg = f"API错误: {response.status_code}"
|
||
try:
|
||
error_detail = response.json()
|
||
error_msg = error_detail.get("error", {}).get("message", error_msg)
|
||
except:
|
||
pass
|
||
return prompt, None, error_msg, None
|
||
|
||
except requests.exceptions.Timeout:
|
||
return prompt, None, "API请求超时，请稍后重试", None
|
||
except Exception as e:
|
||
return prompt, None, f"调用API失败: {str(e)}", None
|