feat: 初始提交 v1.2.0 - 钢琴练习方案生成系统

This commit is contained in:
hmo
2026-04-21 20:00:33 +08:00
commit fd593bddf4
44 changed files with 10936 additions and 0 deletions
+315
View File
@@ -0,0 +1,315 @@
# PDF生成服务 - 支持中文
import os
import re
from reportlab.lib.pagesizes import A4
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.lib.units import mm
from reportlab.lib import colors
from reportlab.platypus import (
SimpleDocTemplate,
Paragraph,
Spacer,
Table,
TableStyle,
PageBreak,
)
from reportlab.pdfbase import pdfmetrics
from reportlab.pdfbase.ttfonts import TTFont
from reportlab.lib.enums import TA_CENTER, TA_LEFT, TA_RIGHT
def register_chinese_font():
"""注册中文字体"""
# 尝试注册系统自带的中文字体
font_paths = [
# Windows常用字体路径
r"C:\Windows\Fonts\simsun.ttc", # 宋体
r"C:\Windows\Fonts\msyh.ttc", # 微软雅黑
r"C:\Windows\Fonts\STSONG.TTF", # 华文宋体
r"C:\Windows\Fonts\simhei.ttf", # 黑体
# Linux容器字体路径
"/usr/share/fonts/truetype/wqy/wqy-microhei.ttc",
"/usr/share/fonts/wqy-microhei/wqy-microhei.ttc",
"/usr/share/fonts/noto/NotoSansCJK-Regular.ttc",
"/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
]
# 尝试注册一个可用的字体
for font_path in font_paths:
if os.path.exists(font_path):
try:
pdfmetrics.registerFont(TTFont("Chinese", font_path))
return True
except Exception as e:
continue
return False
# 尝试注册中文字体
_has_chinese_font = register_chinese_font()
def clean_text(text):
"""清理文本,移除不支持的字符"""
if not text:
return ""
# 移除多余空白
text = re.sub(r"\s+", " ", text)
return text.strip()
class PianoPDF:
def __init__(self):
self.elements = []
self.styles = getSampleStyleSheet()
# 创建中文字体样式
if _has_chinese_font:
self.base_font = "Chinese"
else:
self.base_font = "Helvetica"
# 标题样式
self.title_style = ParagraphStyle(
"CustomTitle",
parent=self.styles["Heading1"],
fontName=self.base_font,
fontSize=18,
spaceAfter=10 * mm,
alignment=TA_CENTER,
)
# 副标题样式
self.heading_style = ParagraphStyle(
"CustomHeading",
parent=self.styles["Heading2"],
fontName=self.base_font,
fontSize=14,
spaceAfter=6 * mm,
spaceBefore=6 * mm,
textColor=colors.HexColor("#2c3e50"),
)
# 正文样式
self.body_style = ParagraphStyle(
"CustomBody",
parent=self.styles["Normal"],
fontName=self.base_font,
fontSize=10,
spaceAfter=4 * mm,
leading=14,
)
# 表格样式
self.table_style = TableStyle(
[
("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#3498db")),
("TEXTCOLOR", (0, 0), (-1, 0), colors.whitesmoke),
("ALIGN", (0, 0), (-1, -1), "LEFT"),
("FONTNAME", (0, 0), (-1, 0), self.base_font),
("FONTSIZE", (0, 0), (-1, 0), 10),
("BOTTOMPADDING", (0, 0), (-1, 0), 8),
("BACKGROUND", (0, 1), (-1, -1), colors.white),
("FONTNAME", (0, 1), (-1, -1), self.base_font),
("FONTSIZE", (0, 1), (-1, -1), 9),
("GRID", (0, 0), (-1, -1), 0.5, colors.grey),
(
"ROWBACKGROUNDS",
(0, 1),
(-1, -1),
[colors.white, colors.HexColor("#f8f9fa")],
),
]
)
def add_title(self, text):
"""添加标题"""
self.elements.append(Paragraph(clean_text(text), self.title_style))
self.elements.append(Spacer(1, 5 * mm))
def add_heading(self, text):
"""添加副标题"""
self.elements.append(Paragraph(clean_text(text), self.heading_style))
def add_paragraph(self, text):
"""添加段落"""
if text:
self.elements.append(Paragraph(clean_text(text), self.body_style))
self.elements.append(Spacer(1, 3 * mm))
def add_list(self, items):
"""添加列表"""
for item in items:
if item:
self.elements.append(
Paragraph(f"{clean_text(item)}", self.body_style)
)
self.elements.append(Spacer(1, 3 * mm))
def add_table(self, data):
"""添加表格"""
if not data or len(data) < 2:
return
# 确保所有数据都是字符串
table_data = []
for row in data:
table_data.append([clean_text(str(cell)) for cell in row])
if table_data:
table = Table(table_data)
table.setStyle(self.table_style)
self.elements.append(table)
self.elements.append(Spacer(1, 5 * mm))
def add_spacer(self, height=5 * mm):
"""添加间距"""
self.elements.append(Spacer(1, height))
def generate_pdf(plan_id, student_name, content, output_dir, rendered_report=None):
"""
使用reportlab生成PDF文件
Args:
plan_id: 方案ID
student_name: 学员姓名
content: 方案内容字典
output_dir: 输出目录
rendered_report: 可选的预渲染报告(Markdown格式),如果提供则使用模板渲染整个报告
Returns:
str: PDF文件路径
"""
os.makedirs(output_dir, exist_ok=True)
output_path = os.path.join(output_dir, f"plan_{plan_id}.pdf")
# 创建PDF文档
doc = SimpleDocTemplate(
output_path,
pagesize=A4,
rightMargin=20 * mm,
leftMargin=20 * mm,
topMargin=20 * mm,
bottomMargin=20 * mm,
)
pdf = PianoPDF()
# 如果提供了预渲染的报告,使用模板渲染整个报告
if rendered_report:
# 解析Markdown并添加到PDF
lines = rendered_report.split('\n')
in_table = False
table_data = []
for line in lines:
line = line.strip()
if not line or line.startswith('---'):
if in_table and table_data:
pdf.add_table(table_data)
table_data = []
in_table = False
continue
if line.startswith('# ') and not line.startswith('## '):
# 一级标题作为主标题
title = line.replace('# ', '').replace('*', '')
pdf.add_title(title)
elif line.startswith('## '):
# 二级标题
if in_table and table_data:
pdf.add_table(table_data)
table_data = []
in_table = False
heading = line.replace('## ', '').replace('**', '')
pdf.add_heading(heading)
elif line.startswith('|') and '|' in line[1:]:
# 表格行
in_table = True
cells = [c.strip() for c in line.split('|')[1:-1]]
if cells and any(cells):
# 跳过表头分隔行
if not all(c.strip().startswith('-') for c in cells if c.strip()):
table_data.append(cells)
elif line.startswith('- '):
# 列表项
pdf.add_paragraph(line)
elif line and not in_table:
# 普通段落
pdf.add_paragraph(line.replace('**', '').replace('*', ''))
# 表格收尾
if in_table and table_data:
pdf.add_table(table_data)
else:
# 使用原有结构化方式生成PDF
# 标题
pdf.add_title(f"钢琴练习方案 - {student_name}")
# 学员信息
pdf.add_heading("📋 学员信息")
pdf.add_paragraph(f"学员姓名:{student_name}")
pdf.add_paragraph(f"每日练习时间:{content.get('practice_time', 'N/A')}")
pdf.add_paragraph(f"生成时间:{content.get('generated_at', '')}")
# 问题诊断
pdf.add_heading("🔍 问题诊断")
for problem in content.get("problems", []):
name = problem.get("name", "")
severity = problem.get("severity", "")
focus = problem.get("focus", {})
pdf.add_paragraph(
f"{name} ({severity}) - 基础练习: {focus.get('basic', 0)}分钟, 技术练习: {focus.get('tech', 0)}分钟"
)
# AI报告(如果有)
if content.get("ai_report"):
pdf.add_heading("📝 AI个性化练习报告")
# 将AI报告分段添加
for line in content["ai_report"].split("\n"):
line = line.strip()
if line.startswith("## "):
pdf.add_heading(line.replace("## ", ""))
elif line.startswith("# "):
pdf.add_title(line.replace("# ", ""))
elif line.startswith("- "):
pdf.add_paragraph(line)
elif line:
pdf.add_paragraph(line)
# 每日练习计划
pdf.add_heading("📅 每日练习计划")
total_minutes = content.get("total_daily_minutes", 0)
pdf.add_paragraph(f"总时长:{total_minutes} 分钟")
# 表格
table_data = [["阶段", "时长", "内容", "目的"]]
for item in content.get("daily_schedule", []):
table_data.append(
[
item.get("phase", ""),
item.get("duration", ""),
item.get("content", ""),
item.get("purpose", ""),
]
)
pdf.add_table(table_data)
# 练习建议
pdf.add_heading("💡 练习建议")
tips = [
"每天在固定时间练习,养成良好习惯",
"练习前先热身,放松手指",
"先难后易,先练习重点部分",
"录音回听,发现问题",
"保持耐心,进步需要时间",
]
pdf.add_list(tips)
# 生成PDF
doc.build(pdf.elements)
return output_path
+443
View File
@@ -0,0 +1,443 @@
# 练习方案生成服务
import os
import re
import requests
from app.config import load_api_config
def generate_practice_plan(
student_name, problems, problems_dir, practice_time="30-60分钟"
):
"""
根据学员问题和练习时间生成针对性练习方案
Args:
student_name: 学员姓名
problems: 问题列表 [{problem_id, problem_name, severity}]
problems_dir: 问题文件所在目录
practice_time: 练习时间描述
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:
problem_file = os.path.join(problems_dir, f"{p['problem_id']}.md")
if os.path.exists(problem_file):
with open(problem_file, "r", encoding="utf-8") as f:
content = f.read()
# 提取关键部分
problem_contents.append(
{
"name": p["problem_name"],
"severity": p["severity"],
"content": _extract_key_sections(content),
"time_allocation": _calculate_time_allocation(
p["severity"], time_config
),
}
)
else:
# 问题文件不存在时,使用默认内容
problem_contents.append(
{
"name": p["problem_name"],
"severity": p["severity"],
"content": {
"problem": f"针对{p['problem_name']}的练习",
"suggestion": "建议每天进行针对性练习",
},
"time_allocation": _calculate_time_allocation(
p["severity"], time_config
),
}
)
# 生成每日练习计划
daily_plan = _generate_daily_schedule(time_config, problem_contents)
# 生成方案
plan = {
"student_name": student_name,
"practice_time": practice_time,
"total_daily_minutes": time_config["total"],
"problems": [
{
"name": p["name"],
"severity": p["severity"],
"focus": p["time_allocation"],
}
for p in problem_contents
],
"daily_schedule": daily_plan,
"generated_at": "2026-04-17",
}
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 _generate_daily_schedule(time_config, problem_contents):
"""生成每日练习计划"""
total = time_config["total"]
# 基础练习(哈农+音阶)
basic_time = time_config["basic"]
# 技术练习(针对问题)
tech_time = time_config["tech"]
# 曲目练习
piece_time = time_config["piece"]
schedule = []
# 1. 热身
schedule.append(
{
"phase": "热身",
"duration": "3分钟",
"content": "手部放松操 + 呼吸调节",
"purpose": "放松肌肉,进入状态",
}
)
# 2. 基础练习
if basic_time > 0:
schedule.append(
{
"phase": "基础练习",
"duration": f"{basic_time}分钟",
"content": "哈农练习曲 + 音阶练习",
"purpose": "建立基本功",
}
)
# 3. 技术练习(针对问题)
if tech_time > 0 and problem_contents:
tech_items = []
for p in problem_contents:
practices = p.get("content", {}).get("practices", [])
if practices:
# 取前2个练习
for practice in practices[:2]:
tech_items.append(practice["name"])
schedule.append(
{
"phase": "技术练习",
"duration": f"{tech_time}分钟",
"content": "".join(tech_items[:3]) if tech_items else "针对性练习",
"purpose": "解决具体问题",
}
)
# 4. 曲目练习
if piece_time > 0:
schedule.append(
{
"phase": "曲目练习",
"duration": f"{piece_time}分钟",
"content": "复习所学曲目 + 预习新曲目",
"purpose": "提升演奏能力",
}
)
# 5. 总结
schedule.append(
{
"phase": "总结",
"duration": "3分钟",
"content": "练习记录 + 问题检查 + 明日计划",
"purpose": "巩固学习成果",
}
)
return schedule
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
):
"""
使用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
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", "")
if not api_key:
return None, "未配置API Key,请在设置页面配置"
# 从问题文件内容中提取关键信息
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}"
)
# 使用配置的模板,如果为空则使用默认模板
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),
problems="\n\n".join(problems_full),
)
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:
return prompt, None, 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
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
except requests.exceptions.Timeout:
return prompt, None, "API请求超时,请稍后重试"
except Exception as e:
return prompt, None, f"调用API失败: {str(e)}"