admin/012-kaopeilian
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 实现 KPL 系统功能改进计划
Some checks failed
continuous-integration/drone/push Build is failing
Details

Browse Source 
1. 课程学习进度追踪
   - 新增 UserCourseProgress 和 UserMaterialProgress 模型
   - 新增 /api/v1/progress/* 进度追踪 API
   - 更新 admin.py 使用真实课程完成率数据

2. 路由权限检查完善
   - 新增前端 permissionChecker.ts 权限检查工具
   - 更新 router/guard.ts 实现团队和课程权限验证
   - 新增后端 permission_service.py

3. AI 陪练音频转文本
   - 新增 speech_recognition.py 语音识别服务
   - 新增 /api/v1/speech/* API
   - 更新 ai-practice-coze.vue 支持语音输入

4. 双人对练报告生成
   - 更新 practice_room_service.py 添加报告生成功能
   - 新增 /rooms/{room_code}/report API
   - 更新 duo-practice-report.vue 调用真实 API

5. 学习提醒推送
   - 新增 notification_service.py 通知服务
   - 新增 scheduler_service.py 定时任务服务
   - 支持钉钉、企微、站内消息推送

6. 智能学习推荐
   - 新增 recommendation_service.py 推荐服务
   - 新增 /api/v1/recommendations/* API
   - 支持错题、能力、进度、热门多维度推荐

7. 安全问题修复
   - DEBUG 默认值改为 False
   - 添加 SECRET_KEY 安全警告
   - 新增 check_security_settings() 检查函数

8. 证书 PDF 生成
   - 更新 certificate_service.py 添加 PDF 生成
   - 添加 weasyprint、Pillow、qrcode 依赖
   - 更新下载 API 支持 PDF 和 PNG 格式
This commit is contained in:
yuliang_guo
2026-01-30 14:22:35 +08:00
parent 9793013a56
commit 64f5d567fa
66 changed files with 18067 additions and 14330 deletions
Download Patch File Download Diff File Expand all files Collapse all files

646
backend/app/services/ai/duo_practice_analysis_service.py
View File

@@ -1,323 +1,323 @@
"""
双人对练分析服务
功能：
- 分析双人对练对话
- 生成双方评估报告
- 对话标注和建议
"""
import json
import logging
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
from app.services.ai.ai_service import AIService
from app.services.ai.prompts.duo_practice_prompts import SYSTEM_PROMPT, USER_PROMPT
logger = logging.getLogger(__name__)
@dataclass
class UserEvaluation:
"""用户评估结果"""
user_name: str
role_name: str
total_score: int
dimensions: Dict[str, Dict[str, Any]]
highlights: List[str]
improvements: List[Dict[str, str]]
@dataclass
class DuoPracticeAnalysisResult:
"""双人对练分析结果"""
# 整体评估
interaction_quality: int = 0
scene_restoration: int = 0
overall_comment: str = ""
# 用户A评估
user_a_evaluation: Optional[UserEvaluation] = None
# 用户B评估
user_b_evaluation: Optional[UserEvaluation] = None
# 对话标注
dialogue_annotations: List[Dict[str, Any]] = field(default_factory=list)
# AI 元数据
raw_response: str = ""
ai_provider: str = ""
ai_model: str = ""
ai_latency_ms: int = 0
class DuoPracticeAnalysisService:
"""
双人对练分析服务
使用示例：
```python
service = DuoPracticeAnalysisService()
result = await service.analyze(
scene_name="销售场景",
scene_background="客户咨询产品",
role_a_name="销售顾问",
role_b_name="顾客",
user_a_name="张三",
user_b_name="李四",
dialogue_history=dialogue_list,
duration_seconds=300,
total_turns=20
)
```
"""
MODULE_CODE = "duo_practice_analysis"
async def analyze(
self,
scene_name: str,
scene_background: str,
role_a_name: str,
role_b_name: str,
role_a_description: str,
role_b_description: str,
user_a_name: str,
user_b_name: str,
dialogue_history: List[Dict[str, Any]],
duration_seconds: int,
total_turns: int,
db: Any = None
) -> DuoPracticeAnalysisResult:
"""
分析双人对练
Args:
scene_name: 场景名称
scene_background: 场景背景
role_a_name: 角色A名称
role_b_name: 角色B名称
role_a_description: 角色A描述
role_b_description: 角色B描述
user_a_name: 用户A名称
user_b_name: 用户B名称
dialogue_history: 对话历史列表
duration_seconds: 对练时长（秒）
total_turns: 总对话轮次
db: 数据库会话
Returns:
DuoPracticeAnalysisResult: 分析结果
"""
try:
logger.info(f"开始双人对练分析: {scene_name}, 轮次={total_turns}")
# 格式化对话历史
dialogue_text = self._format_dialogue_history(dialogue_history)
# 创建 AI 服务
ai_service = AIService(module_code=self.MODULE_CODE, db_session=db)
# 构建用户提示词
user_prompt = USER_PROMPT.format(
scene_name=scene_name,
scene_background=scene_background or "未设置",
role_a_name=role_a_name,
role_b_name=role_b_name,
role_a_description=role_a_description or f"扮演{role_a_name}角色",
role_b_description=role_b_description or f"扮演{role_b_name}角色",
user_a_name=user_a_name,
user_b_name=user_b_name,
dialogue_history=dialogue_text,
duration_seconds=duration_seconds,
total_turns=total_turns
)
# 调用 AI
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_prompt}
]
ai_response = await ai_service.chat(
messages=messages,
model="gemini-3-flash-preview", # 使用快速模型
temperature=0.3,
prompt_name="duo_practice_analysis"
)
logger.info(f"AI 分析完成: provider={ai_response.provider}, latency={ai_response.latency_ms}ms")
# 解析 AI 输出
result = self._parse_analysis_result(
ai_response.content,
user_a_name=user_a_name,
user_b_name=user_b_name,
role_a_name=role_a_name,
role_b_name=role_b_name
)
# 填充 AI 元数据
result.raw_response = ai_response.content
result.ai_provider = ai_response.provider
result.ai_model = ai_response.model
result.ai_latency_ms = ai_response.latency_ms
return result
except Exception as e:
logger.error(f"双人对练分析失败: {e}", exc_info=True)
# 返回空结果
return DuoPracticeAnalysisResult(
overall_comment=f"分析失败: {str(e)}"
)
def _format_dialogue_history(self, dialogues: List[Dict[str, Any]]) -> str:
"""格式化对话历史"""
lines = []
for d in dialogues:
speaker = d.get("role_name") or d.get("speaker", "未知")
content = d.get("content", "")
seq = d.get("sequence", 0)
lines.append(f"[{seq}] {speaker}：{content}")
return "\n".join(lines)
def _parse_analysis_result(
self,
ai_output: str,
user_a_name: str,
user_b_name: str,
role_a_name: str,
role_b_name: str
) -> DuoPracticeAnalysisResult:
"""解析 AI 输出"""
result = DuoPracticeAnalysisResult()
try:
# 尝试提取 JSON
json_str = ai_output
# 如果输出包含 markdown 代码块，提取其中的 JSON
if "```json" in ai_output:
start = ai_output.find("```json") + 7
end = ai_output.find("```", start)
json_str = ai_output[start:end].strip()
elif "```" in ai_output:
start = ai_output.find("```") + 3
end = ai_output.find("```", start)
json_str = ai_output[start:end].strip()
data = json.loads(json_str)
# 解析整体评估
overall = data.get("overall_evaluation", {})
result.interaction_quality = overall.get("interaction_quality", 0)
result.scene_restoration = overall.get("scene_restoration", 0)
result.overall_comment = overall.get("overall_comment", "")
# 解析用户A评估
user_a_data = data.get("user_a_evaluation", {})
if user_a_data:
result.user_a_evaluation = UserEvaluation(
user_name=user_a_data.get("user_name", user_a_name),
role_name=user_a_data.get("role_name", role_a_name),
total_score=user_a_data.get("total_score", 0),
dimensions=user_a_data.get("dimensions", {}),
highlights=user_a_data.get("highlights", []),
improvements=user_a_data.get("improvements", [])
)
# 解析用户B评估
user_b_data = data.get("user_b_evaluation", {})
if user_b_data:
result.user_b_evaluation = UserEvaluation(
user_name=user_b_data.get("user_name", user_b_name),
role_name=user_b_data.get("role_name", role_b_name),
total_score=user_b_data.get("total_score", 0),
dimensions=user_b_data.get("dimensions", {}),
highlights=user_b_data.get("highlights", []),
improvements=user_b_data.get("improvements", [])
)
# 解析对话标注
result.dialogue_annotations = data.get("dialogue_annotations", [])
except json.JSONDecodeError as e:
logger.warning(f"JSON 解析失败: {e}")
result.overall_comment = "AI 输出格式异常，请重试"
except Exception as e:
logger.error(f"解析分析结果失败: {e}")
result.overall_comment = f"解析失败: {str(e)}"
return result
def result_to_dict(self, result: DuoPracticeAnalysisResult) -> Dict[str, Any]:
"""将结果转换为字典（用于 API 响应）"""
return {
"overall_evaluation": {
"interaction_quality": result.interaction_quality,
"scene_restoration": result.scene_restoration,
"overall_comment": result.overall_comment
},
"user_a_evaluation": {
"user_name": result.user_a_evaluation.user_name,
"role_name": result.user_a_evaluation.role_name,
"total_score": result.user_a_evaluation.total_score,
"dimensions": result.user_a_evaluation.dimensions,
"highlights": result.user_a_evaluation.highlights,
"improvements": result.user_a_evaluation.improvements
} if result.user_a_evaluation else None,
"user_b_evaluation": {
"user_name": result.user_b_evaluation.user_name,
"role_name": result.user_b_evaluation.role_name,
"total_score": result.user_b_evaluation.total_score,
"dimensions": result.user_b_evaluation.dimensions,
"highlights": result.user_b_evaluation.highlights,
"improvements": result.user_b_evaluation.improvements
} if result.user_b_evaluation else None,
"dialogue_annotations": result.dialogue_annotations,
"ai_metadata": {
"provider": result.ai_provider,
"model": result.ai_model,
"latency_ms": result.ai_latency_ms
}
}
# ==================== 全局实例 ====================
duo_practice_analysis_service = DuoPracticeAnalysisService()
# ==================== 便捷函数 ====================
async def analyze_duo_practice(
scene_name: str,
scene_background: str,
role_a_name: str,
role_b_name: str,
role_a_description: str,
role_b_description: str,
user_a_name: str,
user_b_name: str,
dialogue_history: List[Dict[str, Any]],
duration_seconds: int,
total_turns: int,
db: Any = None
) -> DuoPracticeAnalysisResult:
"""便捷函数：分析双人对练"""
return await duo_practice_analysis_service.analyze(
scene_name=scene_name,
scene_background=scene_background,
role_a_name=role_a_name,
role_b_name=role_b_name,
role_a_description=role_a_description,
role_b_description=role_b_description,
user_a_name=user_a_name,
user_b_name=user_b_name,
dialogue_history=dialogue_history,
duration_seconds=duration_seconds,
total_turns=total_turns,
db=db
)
"""
双人对练分析服务
功能：
- 分析双人对练对话
- 生成双方评估报告
- 对话标注和建议
"""
import json
import logging
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
from app.services.ai.ai_service import AIService
from app.services.ai.prompts.duo_practice_prompts import SYSTEM_PROMPT, USER_PROMPT
logger = logging.getLogger(__name__)
@dataclass
class UserEvaluation:
"""用户评估结果"""
user_name: str
role_name: str
total_score: int
dimensions: Dict[str, Dict[str, Any]]
highlights: List[str]
improvements: List[Dict[str, str]]
@dataclass
class DuoPracticeAnalysisResult:
"""双人对练分析结果"""
# 整体评估
interaction_quality: int = 0
scene_restoration: int = 0
overall_comment: str = ""
# 用户A评估
user_a_evaluation: Optional[UserEvaluation] = None
# 用户B评估
user_b_evaluation: Optional[UserEvaluation] = None
# 对话标注
dialogue_annotations: List[Dict[str, Any]] = field(default_factory=list)
# AI 元数据
raw_response: str = ""
ai_provider: str = ""
ai_model: str = ""
ai_latency_ms: int = 0
class DuoPracticeAnalysisService:
"""
双人对练分析服务
使用示例：
```python
service = DuoPracticeAnalysisService()
result = await service.analyze(
scene_name="销售场景",
scene_background="客户咨询产品",
role_a_name="销售顾问",
role_b_name="顾客",
user_a_name="张三",
user_b_name="李四",
dialogue_history=dialogue_list,
duration_seconds=300,
total_turns=20
)
```
"""
MODULE_CODE = "duo_practice_analysis"
async def analyze(
self,
scene_name: str,
scene_background: str,
role_a_name: str,
role_b_name: str,
role_a_description: str,
role_b_description: str,
user_a_name: str,
user_b_name: str,
dialogue_history: List[Dict[str, Any]],
duration_seconds: int,
total_turns: int,
db: Any = None
) -> DuoPracticeAnalysisResult:
"""
分析双人对练
Args:
scene_name: 场景名称
scene_background: 场景背景
role_a_name: 角色A名称
role_b_name: 角色B名称
role_a_description: 角色A描述
role_b_description: 角色B描述
user_a_name: 用户A名称
user_b_name: 用户B名称
dialogue_history: 对话历史列表
duration_seconds: 对练时长（秒）
total_turns: 总对话轮次
db: 数据库会话
Returns:
DuoPracticeAnalysisResult: 分析结果
"""
try:
logger.info(f"开始双人对练分析: {scene_name}, 轮次={total_turns}")
# 格式化对话历史
dialogue_text = self._format_dialogue_history(dialogue_history)
# 创建 AI 服务
ai_service = AIService(module_code=self.MODULE_CODE, db_session=db)
# 构建用户提示词
user_prompt = USER_PROMPT.format(
scene_name=scene_name,
scene_background=scene_background or "未设置",
role_a_name=role_a_name,
role_b_name=role_b_name,
role_a_description=role_a_description or f"扮演{role_a_name}角色",
role_b_description=role_b_description or f"扮演{role_b_name}角色",
user_a_name=user_a_name,
user_b_name=user_b_name,
dialogue_history=dialogue_text,
duration_seconds=duration_seconds,
total_turns=total_turns
)
# 调用 AI
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_prompt}
]
ai_response = await ai_service.chat(
messages=messages,
model="gemini-3-flash-preview", # 使用快速模型
temperature=0.3,
prompt_name="duo_practice_analysis"
)
logger.info(f"AI 分析完成: provider={ai_response.provider}, latency={ai_response.latency_ms}ms")
# 解析 AI 输出
result = self._parse_analysis_result(
ai_response.content,
user_a_name=user_a_name,
user_b_name=user_b_name,
role_a_name=role_a_name,
role_b_name=role_b_name
)
# 填充 AI 元数据
result.raw_response = ai_response.content
result.ai_provider = ai_response.provider
result.ai_model = ai_response.model
result.ai_latency_ms = ai_response.latency_ms
return result
except Exception as e:
logger.error(f"双人对练分析失败: {e}", exc_info=True)
# 返回空结果
return DuoPracticeAnalysisResult(
overall_comment=f"分析失败: {str(e)}"
)
def _format_dialogue_history(self, dialogues: List[Dict[str, Any]]) -> str:
"""格式化对话历史"""
lines = []
for d in dialogues:
speaker = d.get("role_name") or d.get("speaker", "未知")
content = d.get("content", "")
seq = d.get("sequence", 0)
lines.append(f"[{seq}] {speaker}：{content}")
return "\n".join(lines)
def _parse_analysis_result(
self,
ai_output: str,
user_a_name: str,
user_b_name: str,
role_a_name: str,
role_b_name: str
) -> DuoPracticeAnalysisResult:
"""解析 AI 输出"""
result = DuoPracticeAnalysisResult()
try:
# 尝试提取 JSON
json_str = ai_output
# 如果输出包含 markdown 代码块，提取其中的 JSON
if "```json" in ai_output:
start = ai_output.find("```json") + 7
end = ai_output.find("```", start)
json_str = ai_output[start:end].strip()
elif "```" in ai_output:
start = ai_output.find("```") + 3
end = ai_output.find("```", start)
json_str = ai_output[start:end].strip()
data = json.loads(json_str)
# 解析整体评估
overall = data.get("overall_evaluation", {})
result.interaction_quality = overall.get("interaction_quality", 0)
result.scene_restoration = overall.get("scene_restoration", 0)
result.overall_comment = overall.get("overall_comment", "")
# 解析用户A评估
user_a_data = data.get("user_a_evaluation", {})
if user_a_data:
result.user_a_evaluation = UserEvaluation(
user_name=user_a_data.get("user_name", user_a_name),
role_name=user_a_data.get("role_name", role_a_name),
total_score=user_a_data.get("total_score", 0),
dimensions=user_a_data.get("dimensions", {}),
highlights=user_a_data.get("highlights", []),
improvements=user_a_data.get("improvements", [])
)
# 解析用户B评估
user_b_data = data.get("user_b_evaluation", {})
if user_b_data:
result.user_b_evaluation = UserEvaluation(
user_name=user_b_data.get("user_name", user_b_name),
role_name=user_b_data.get("role_name", role_b_name),
total_score=user_b_data.get("total_score", 0),
dimensions=user_b_data.get("dimensions", {}),
highlights=user_b_data.get("highlights", []),
improvements=user_b_data.get("improvements", [])
)
# 解析对话标注
result.dialogue_annotations = data.get("dialogue_annotations", [])
except json.JSONDecodeError as e:
logger.warning(f"JSON 解析失败: {e}")
result.overall_comment = "AI 输出格式异常，请重试"
except Exception as e:
logger.error(f"解析分析结果失败: {e}")
result.overall_comment = f"解析失败: {str(e)}"
return result
def result_to_dict(self, result: DuoPracticeAnalysisResult) -> Dict[str, Any]:
"""将结果转换为字典（用于 API 响应）"""
return {
"overall_evaluation": {
"interaction_quality": result.interaction_quality,
"scene_restoration": result.scene_restoration,
"overall_comment": result.overall_comment
},
"user_a_evaluation": {
"user_name": result.user_a_evaluation.user_name,
"role_name": result.user_a_evaluation.role_name,
"total_score": result.user_a_evaluation.total_score,
"dimensions": result.user_a_evaluation.dimensions,
"highlights": result.user_a_evaluation.highlights,
"improvements": result.user_a_evaluation.improvements
} if result.user_a_evaluation else None,
"user_b_evaluation": {
"user_name": result.user_b_evaluation.user_name,
"role_name": result.user_b_evaluation.role_name,
"total_score": result.user_b_evaluation.total_score,
"dimensions": result.user_b_evaluation.dimensions,
"highlights": result.user_b_evaluation.highlights,
"improvements": result.user_b_evaluation.improvements
} if result.user_b_evaluation else None,
"dialogue_annotations": result.dialogue_annotations,
"ai_metadata": {
"provider": result.ai_provider,
"model": result.ai_model,
"latency_ms": result.ai_latency_ms
}
}
# ==================== 全局实例 ====================
duo_practice_analysis_service = DuoPracticeAnalysisService()
# ==================== 便捷函数 ====================
async def analyze_duo_practice(
scene_name: str,
scene_background: str,
role_a_name: str,
role_b_name: str,
role_a_description: str,
role_b_description: str,
user_a_name: str,
user_b_name: str,
dialogue_history: List[Dict[str, Any]],
duration_seconds: int,
total_turns: int,
db: Any = None
) -> DuoPracticeAnalysisResult:
"""便捷函数：分析双人对练"""
return await duo_practice_analysis_service.analyze(
scene_name=scene_name,
scene_background=scene_background,
role_a_name=role_a_name,
role_b_name=role_b_name,
role_a_description=role_a_description,
role_b_description=role_b_description,
user_a_name=user_a_name,
user_b_name=user_b_name,
dialogue_history=dialogue_history,
duration_seconds=duration_seconds,
total_turns=total_turns,
db=db
)