CakeCN/hsa
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

调整course_service 文件长度

Browse Source
This commit is contained in:
CakeCN
2025-12-09 13:25:05 +08:00
parent c3c76788fe
commit a509b2efb2
3 changed files with 674 additions and 644 deletions
Download Patch File Download Diff File Expand all files Collapse all files

648
Html/apps/services/course_service.py
View File

@@ -8,650 +8,10 @@ from urllib.parse import urlparse
from datetime import datetime from datetime import datetime
from bson import ObjectId from bson import ObjectId
from ..services.cos_service import upload_file, delete_file from services.cos_service import upload_file, delete_file
from ..models.material import Material from models.material import Material
from .course_services.material_process import *
def get_new_edit_materials(num: int)->List[Material]: from .course_services.learning_progess_process import *
mongo = current_app.extensions["mongo"]
materials = mongo.db.materials.find().sort("updated_at", -1).limit(num)
returnCode = []
for material in materials:
returnCode.append(Material(**material))
returnCode[-1]._id = str(material['_id'])
return returnCode
def user_selected_course(user_obj):
briefs = []
for cid in getattr(user_obj, "select_course", []):
course_obj = load_material(cid)
brief={}
brief["_id"] = cid
brief["name"] = course_obj.name
brief["description"] = course_obj.description
brief["image_url"] = course_obj.image_url
brief["created_at"] = course_obj.created_at.strftime("%Y-%m-%d %H:%M:%S")
brief["updated_at"] = course_obj.updated_at.strftime("%Y-%m-%d %H:%M:%S")
brief["chapters"] = [chapter.model_dump() for chapter in course_obj.chapters]
briefs.append(brief)
return briefs
def user_selected_course_briefs(user_obj):
"""根据用户对象的选课列表，拼装课程简要信息数组"""
briefs = []
for cid in getattr(user_obj, "select_course", []):
course_obj = load_material(cid)
brief={}
brief["_id"] = cid
brief["name"] = course_obj.name
brief["description"] = course_obj.description
brief["image_url"] = course_obj.image_url
brief["created_at"] = course_obj.created_at.strftime("%Y-%m-%d %H:%M:%S")
brief["updated_at"] = course_obj.updated_at.strftime("%Y-%m-%d %H:%M:%S")
briefs.append(brief)
return briefs
def create_material(teacher_id, material_name, description, chapters, image_url):
material = Material(
name=material_name,
description=description,
teacher_id=teacher_id,
created_at=datetime.now(),
updated_at=datetime.now(),
chapters=chapters,
image_url=image_url
)
mongo = current_app.extensions["mongo"]
material_id = mongo.db.materials.insert_one(material.model_dump()).inserted_id
return str(material_id)
def load_material(material_id):
mongo = current_app.extensions["mongo"]
material = mongo.db.materials.find_one({"_id": ObjectId(material_id)})
material['_id'] = material_id
return Material(**material)
def update_material(material_id, chapters):
mongo = current_app.extensions["mongo"]
material = mongo.db.materials.find_one({'_id': ObjectId(material_id)})
if material:
mongo.db.materials.update_one(
{'_id': ObjectId(material_id)},
{'$set': {'chapters': chapters, 'updated_at': datetime.now()}}
)
return True
return False
def add_material_chapter(material_id, chapter_name, teacher_id):
mongo = current_app.extensions["mongo"]
material = mongo.db.materials.find_one({'_id': ObjectId(material_id)})
if material['teacher_id'] != teacher_id:
return False
if material:
mongo.db.materials.update_one(
{'_id': ObjectId(material_id)},
{'$push': {'chapters': {'chapter_name': chapter_name, 'lessons': []}}})
return True
def rename_material_structure(material_id, is_chapter, chapter_name, lesson_name, new_name):
mongo = current_app.extensions["mongo"]
material = mongo.db.materials.find_one({'_id': ObjectId(material_id)})
if not material:
return False # 材料不存在
# 检查名称是否存在于列表中的辅助函数
def name_exists_in_list(item_list, name_key, target_name):
return any(item[name_key] == target_name for item in item_list)
# 查找包含指定章节名的章节
def find_chapter_by_name(chapters, target_name):
for chapter in chapters:
if chapter['chapter_name'] == target_name:
return chapter
return None
# 检查章节是否存在
target_chapter = find_chapter_by_name(material['chapters'], chapter_name)
if not target_chapter:
return False
if is_chapter:
# 检查新章节名是否已存在
if name_exists_in_list(material['chapters'], 'chapter_name', new_name):
return False
# 更新章节名
mongo.db.materials.update_one(
{'_id': ObjectId(material_id)},
{'$set': {'chapters.$[c].chapter_name': new_name}},
array_filters=[{'c.chapter_name': chapter_name}]
)
else:
# 检查课时是否存在
if not name_exists_in_list(target_chapter['lessons'], 'lesson_name', lesson_name):
return False
# 检查新课时名是否已存在
if name_exists_in_list(target_chapter['lessons'], 'lesson_name', new_name):
return False
# 更新课时名
mongo.db.materials.update_one(
{'_id': ObjectId(material_id)},
{'$set': {'chapters.$[c].lessons.$[l].lesson_name': new_name}},
array_filters=[{'c.chapter_name': chapter_name}, {'l.lesson_name': lesson_name}]
)
return True
def delete_material_chapter_lesson(material_id, chapter_name, lesson_name, teacher_id):
"""
删除指定课程的课时。
删除后，更新 MongoDB 中对应的章节中的 lesson 列表。
"""
mongo = current_app.extensions["mongo"]
# 查询 material 是否存在，且确保该 teacher_id 对应的 material
material = mongo.db.materials.find_one({'_id': ObjectId(material_id), 'teacher_id': teacher_id})
for chapter in material['chapters']:
if chapter['chapter_name'] == chapter_name:
for lesson in chapter['lessons']:
if lesson['lesson_name'] == lesson_name:
if lesson['markdown_lesson_file_link']!='':
delete_file(lesson['markdown_lesson_file_link'])
if lesson['markdown_prompt_file_name']!='':
delete_file(lesson['markdown_prompt_file_name'])
if lesson['markdown_score_prompt_file_link']!='':
delete_file(lesson['markdown_score_prompt_file_link'])
if not material:
return False # 找不到对应的 material 或 teacher_id 不匹配
# 删除 lesson 的条件
res = mongo.db.materials.update_one(
{
"_id": ObjectId(material_id),
"teacher_id": teacher_id # 确保是该教师的课程
},
{
"$pull": {
"chapters.$[c].lessons": {
"lesson_name": lesson_name # 根据 lesson_name 来删除
}
},
"$set": {"updated_at": datetime.now()} # 更新修改时间
},
array_filters=[{"c.chapter_name": chapter_name}] # 找到匹配的章节
)
# 如果删除操作成功，res.modified_count 应为 1
return res.modified_count == 1
def reorder_material_structure(material_id, order):
mongo = current_app.extensions["mongo"]
material = mongo.db.materials.find_one({'_id': ObjectId(material_id)})
if not material:
return False
# 1. 构建原章节映射（保留章节元信息）和原章节的课时映射
chapter_map = {ch["chapter_name"]: ch for ch in material["chapters"]}
# 每个章节的课时映射：{章节名: {课时名: 课时对象}}
original_chapter_lessons = {
ch["chapter_name"]: {ls["lesson_name"]: ls for ls in ch["lessons"]}
for ch in material["chapters"]
}
# 2. 收集所有章节“少掉的课时”到全局池（仅包含原章节有但新排序没包含的课时）
# 用列表存储，避免同名课时被覆盖（处理可能的重名场景）
global_extra_lessons = []
for ch_name in chapter_map:
original_lessons = original_chapter_lessons[ch_name] # 原章节的所有课时
# 找到当前章节在order中的配置（如果没有，默认该章节所有课时都是“少掉的”）
chapter_order = next((item for item in order if item["chapter_name"] == ch_name), None)
if not chapter_order:
# 章节未在order中出现，所有原课时都是“少掉的”
global_extra_lessons.extend(original_lessons.values())
continue
# 章节在order中出现，找出原课时中不在新排序里的（即“少掉的”）
new_lesson_names = chapter_order["lessons"]
for ls_name, ls in original_lessons.items():
if ls_name not in new_lesson_names:
global_extra_lessons.append(ls)
print("global_extra_lessons:", global_extra_lessons)
new_chapters = []
for chapter_order in order:
ch_name = chapter_order["chapter_name"]
if ch_name not in chapter_map:
continue # 跳过不存在的章节
original_ch = chapter_map[ch_name]
original_lessons = original_chapter_lessons[ch_name] # 本章节原课时
new_lessons = []
# 3. 构建当前章节的新课时列表
for ls_name in chapter_order["lessons"]:
# 优先使用本章节原有的课时
if ls_name in original_lessons:
new_lessons.append(original_lessons[ls_name])
continue
# 本章节没有，从全局池（其他章节少掉的课时）中找
# 处理可能的重名：找到第一个匹配的课时
for i, extra_ls in enumerate(global_extra_lessons):
if extra_ls["lesson_name"] == ls_name:
new_lessons.append(extra_ls)
del global_extra_lessons[i] # 从全局池移除，避免重复
break
# 4. 构造新章节（保留原章节元信息，替换课时列表）
new_chapter = {**original_ch, "lessons": new_lessons}
new_chapters.append(new_chapter)
# 5. 处理全局池中剩余的课时（未被任何章节引用的，追加到最后一个章节）
if global_extra_lessons and new_chapters:
new_chapters[-1]["lessons"].extend(global_extra_lessons)
print("new_chapters:", new_chapters)
# 更新数据库
mongo.db.materials.update_one(
{"_id": ObjectId(material_id)},
{
"$set": {
"chapters": new_chapters,
"updated_at": datetime.now()
}
}
)
return True
def save_materials_markdown_cos(
material_id: str,
chapter_name: str,
lesson_name: str,
docs: Dict[str, Dict[str, str]],
version: str | None = None, # 目前不做并发控制，保留参数
actor=None # 可记录操作者到审计日志
) -> Dict[str, str]:
"""
docs 结构：
{
"lesson": {"cdn_link": "...lesson.md", "content": "..."},
"prompt": {"cdn_link": "...prompt.md", "content": "..."},
"score": {"cdn_link": "...score_prompt.md", "content": "..."}
}
约定：cdn_link 与对象存储的 key 一一映射（同路径/同文件名），
我们通过解析 URL 的 path 作为 key 进行覆盖写入（不改文件名）。
"""
def _cdn_to_key(cdn_link: str) -> str:
"""
把 CDN URL 映射成对象存储的 key：
- 若 CDN 与存储是“同 key 直出”，则直接取 path（去掉前导 /）
- 若你有自定义映射规则，在此处改写
"""
p = urlparse(cdn_link)
key = p.path.lstrip("/")
if not key:
raise ValueError(f"非法 cdn_link：{cdn_link}")
return key
# 1) 覆盖写入三份文件
results: Dict[str, Dict[str, str]] = {}
for kind in ("lesson", "prompt", "score"):
item = docs.get(kind)
if not item or "cdn_link" not in item or "content" not in item:
raise ValueError(f"缺少 {kind} 的 cdn_link 或 content")
key = _cdn_to_key(item["cdn_link"])
content = item["content"]
# 覆盖写入（注意：upload_file 内部需实现覆盖同 key）
try:
# 约定：upload_file 返回可访问的 URL（覆盖后可与原相同）
url = upload_file(content.encode("utf-8"), key)
except Exception as e:
current_app.logger.exception("覆盖写入失败 kind=%s key=%s", kind, key)
raise
# 简单返回：保留原 cdn_link（一般与 url 相同），以及 md5 作为伪版本
import hashlib
etag = hashlib.md5(content.encode("utf-8")).hexdigest()
results[kind] = {"cdn_link": url or item["cdn_link"], "version": etag}
# 2) 写库：更新该 lesson 的更新时间（不改文件名/链接）
_touch_material_lesson_updated_at(material_id, chapter_name, lesson_name, actor=actor)
results["updated_at"] = datetime.now().isoformat() + "Z"
return results
def _touch_material_lesson_updated_at(material_id: str, chapter_name: str, lesson_name: str, actor=None) -> None:
"""仅刷新 materials 文档中该课时的更新时间戳（以及顶层 updated_at）"""
mongo = current_app.extensions["mongo"]
now = datetime.now()
# 更新顶层 updated_at
mongo.db.materials.update_one(
{"_id": ObjectId(material_id)},
{"$set": {"updated_at": now}}
)
# 更新嵌套 lesson 的更新时间（如果你在 lesson 内有该字段）
# 需要使用 arrayFilters 精确定位到对应章节/课时
mongo.db.materials.update_one(
{"_id": ObjectId(material_id)},
{
"$set": {
"chapters.$[c].lessons.$[l].updated_at": now,
# 可选：记录最近编辑人
"chapters.$[c].lessons.$[l].last_editor": getattr(actor, "id", None) if actor else None
}
},
array_filters=[
{"c.chapter_name": chapter_name},
{"l.lesson_name": lesson_name}
]
)
def add_material_chapter_lesson(material_id, chapter_name, lesson_name, teacher_id):
"""
1) 生成占位 Markdown（lesson_prompt_score_prompt）
2) 上传到对象存储（返回 URL）
3) 将 lesson 文档（含三份 URL）push 到指定章节
"""
mongo = current_app.extensions["mongo"]
# 组织对象存储的文件路径
chapter_slug = _slugify(chapter_name or "")
lesson_slug = _slugify(lesson_name or "")
timestamp = datetime.now().strftime('%Y%m%dT%H%M%SZ')
base_prefix = f"materials_{material_id}_{chapter_slug}_{lesson_slug}_{timestamp}"
# 占位文本
lesson_md, prompt_md, score_prompt_md = _default_markdowns(lesson_name or "未命名课时")
# 生成文件名（带层级路径）
lesson_key = f"{base_prefix}_lesson.md"
prompt_key = f"{base_prefix}_prompt.md"
score_key = f"{base_prefix}_score_prompt.md"
# 上传（任意一个失败则终止，不写库）
try:
lesson_url = upload_file(lesson_md.encode("utf-8"), lesson_key)
prompt_url = upload_file(prompt_md.encode("utf-8"), prompt_key)
score_url = upload_file(score_prompt_md.encode("utf-8"), score_key)
except Exception as e:
current_app.logger.exception("Upload default markdown failed: %s", e)
return False
# 组装 lesson_doc
lesson_doc = {
"lesson_name": lesson_name or "",
"markdown_lesson_file_link": lesson_url, # lesson 主体
"markdown_prompt_file_name": prompt_url, # 给大模型的 prompt
"markdown_score_prompt_file_link": score_url, # 评分 prompt
}
# 将 lesson push 到指定章节
res = mongo.db.materials.update_one(
{
"_id": ObjectId(material_id),
"teacher_id": teacher_id, # 注意：若存的是 ObjectId，请确保传入的是同类型
},
{
"$push": {"chapters.$[c].lessons": lesson_doc},
"$set": {"updated_at": datetime.now()},
},
array_filters=[{"c.chapter_name": chapter_name}],
)
# 可选：如果 res.modified_count != 1，说明 push 失败（可能章节名不匹配）
# 这里可以考虑是否回滚 COS 上刚上传的 3 个文件（需实现删除逻辑）。
return res.modified_count == 1
def get_materials_by_teacher(teacher_id)->List[Material]:
mongo = current_app.extensions["mongo"]
materials = mongo.db.materials.find({'teacher_id': teacher_id})
return [Material(**material) for material in materials]
def get_materials_by_teacher_dict(teacher_id)->List[Dict]:
mongo = current_app.extensions["mongo"]
materials = mongo.db.materials.find({'teacher_id': teacher_id})
result = []
for material in materials:
material['_id'] = str(material['_id'])
result.append(material)
return result
def _slugify(name: str) -> str:
"""将章节/课时名转为 URL 友好的 slug。"""
s = name.strip().lower()
s = re.sub(r'\s+', '-', s) # 空白 -> -
s = re.sub(r'[^a-z0-9\-_]+', '', s) # 仅保留安全字符
return s or 'untitled'
def _default_markdowns(lesson_name: str):
"""三份占位 Markdown 文本。可按需调整模板内容。"""
ts = datetime.now().strftime('%Y-%m-%d %H:%M')
lesson_md = f"""# {lesson_name}
## 阶段一：示例阶段
### 步骤A
在这里编写该步骤的教学内容（文字/代码/图片链接等）。
### 步骤B
继续补充该阶段的其它步骤内容。
"""
prompt_md = f"""# {lesson_name}
## 阶段一：示例阶段
### 步骤A
请基于"步骤A"的教学目标与材料，生成指导性提示词（尽量结构化，给出评估标准和示例）。
### 步骤B
请基于"步骤B"的教学目标与材料，生成指导性提示词。
"""
score_prompt_md = f"""# {lesson_name}
## 评分准则
- 正确性、完整性、可读性、效率（按需调整权重）
## 阶段一：示例阶段
### 步骤A
对学员在"步骤A"的输出进行评分与文字反馈，分数区间 0-100，并输出 JSON：```{{"score": <int>, "reasons": "<string>", "advices": "<string>"}}```
### 步骤B
同上，对"步骤B"输出进行评分与反馈。
"""
return lesson_md, prompt_md, score_prompt_md
def save_learning_progress(user_uuid: str, chatmanager, user_id: str) -> bool:
"""
保存用户学习进度到MongoDB
参数：
- user_uuid: 用户UUID
- chatmanager: ChatManager实例，包含学习进度数据
- user_id: 用户ID
返回：
- bool: 保存是否成功
"""
try:
# 获取MongoDB连接
mongo = current_app.extensions["mongo"]
# 获取MultiAgents框架中的对话列表
dialog_list = get_multiagents_dialog_list(user_id)
his_data = dialog_list['data']
# 构建学习进度数据
progress_data = {
"user_uuid": user_uuid,
"user_id": user_id,
"material_id": chatmanager.material_id,
"chapter_name": chatmanager.chapter_name,
"lesson_name": chatmanager.lesson_name,
"chapter_chain_now": chatmanager.chapter_chain_now,
"chat_historys": chatmanager.chat_historys,
"scores": chatmanager.scores,
"multiagents_dialogs": his_data,
"updated_at": datetime.now()
}
# 保存到MongoDB，使用upsert操作（如果存在则更新，不存在则插入）
mongo.db.learning_progress.update_one(
{
"user_uuid": user_uuid,
"material_id": chatmanager.material_id,
"chapter_name": chatmanager.chapter_name,
"lesson_name": chatmanager.lesson_name
},
{"$set": progress_data},
upsert=True
)
current_app.logger.info(f"学习进度保存成功: user_uuid={user_uuid}, lesson={chatmanager.lesson_name}")
return True
except Exception as e:
current_app.logger.error(f"保存学习进度失败: {str(e)}")
return False
def get_multiagents_dialog_list(user_id: str) -> List[Dict]:
"""
从MultiAgents框架获取用户对话列表
参数：
- user_id: 用户ID
返回：
- Dict[List]: 对话列表(key 为 History database)
"""
try:
# 从配置中获取ASE引擎URL和token
config = current_app.config
url_token = config.get("ASE_ENGINE_URL_TOKEN", "")
# 构建请求URL
base_url = config.get("ASE_ENGINE_URL", "")
api_url = f"{base_url}/api/sync/download"
# 构建请求参数
payload = {
"namespace_url": url_token, # 使用配置中的token作为namespace_url
"user_id": user_id
}
# 发送POST请求
response = requests.post(
api_url,
json=payload,
headers={"Content-Type": "application/json"},
timeout=10
)
# 检查响应
if response.status_code == 200:
return response.json()
else:
current_app.logger.error(f"获取MultiAgents对话列表失败: HTTP {response.status_code}")
return []
except Exception as e:
current_app.logger.error(f"获取MultiAgents对话列表异常: {str(e)}")
return []
def load_learning_progress(user_uuid: str, material_id: str, chapter_name: str, lesson_name: str) -> Optional[Dict]:
"""
加载用户学习进度
参数：
- user_uuid: 用户UUID
- material_id: 课程ID
- chapter_name: 章节名
- lesson_name: 课时名
返回：
- Optional[Dict]: 学习进度数据，如果不存在则返回None
"""
try:
mongo = current_app.extensions["mongo"]
progress = mongo.db.learning_progress.find_one({
"user_uuid": user_uuid,
"material_id": material_id,
"chapter_name": chapter_name,
"lesson_name": lesson_name
})
if progress:
# 转换ObjectId为字符串
if "_id" in progress:
progress["_id"] = str(progress["_id"])
return progress
return None
except Exception as e:
current_app.logger.error(f"加载学习进度失败: {str(e)}")
return None
def upload_learning_progress_to_cloud(progress: Dict) -> Dict:
"""
将学习进度上传到云服务
参数：
- progress: 学习进度数据字典{'databases':{}, 'global_databases':{}, 'session_info_out':{}}
返回：
- Dict: 上传结果，包含success和message字段
"""
try:
# 从配置中获取ASE引擎URL和token
config = current_app.config
url_token = config.get("ASE_ENGINE_URL_TOKEN", "")
# 构建请求URL
base_url = config.get("ASE_ENGINE_URL", "")
api_url = f"{base_url}/api/sync/upload"
# 构建请求参数
payload = {
"namespace_url": url_token, # 使用配置中的token作为namespace_url
"user_id": progress.get("user_id", ""),
"data": progress
}
# 发送POST请求
response = requests.post(
api_url,
json=payload,
headers={"Content-Type": "application/json"},
timeout=10
)
# 处理响应
if response.status_code == 200:
result = response.json()
current_app.logger.info(f"学习进度上传成功: user_id={progress.get('user_id')}")
return {
"success": True,
"message": "学习进度上传成功",
"data": result
}
else:
error_msg = f"上传失败，HTTP状态码: {response.status_code}"
current_app.logger.error(f"学习进度上传失败: {error_msg}")
return {
"success": False,
"message": error_msg
}
except Exception as e:
error_msg = f"上传异常: {str(e)}"
current_app.logger.error(f"学习进度上传异常: {error_msg}")
return {
"success": False,
"message": error_msg
}

197
Html/apps/services/course_services/learning_progess_process.py Normal file
View File

@@ -0,0 +1,197 @@
# myapp/services/course_service.py
import re
import requests
import json
from typing import Dict, List, Optional
from flask import current_app
from urllib.parse import urlparse
from datetime import datetime
from bson import ObjectId
from services.cos_service import upload_file, delete_file
from models.material import Material
def save_learning_progress(user_uuid: str, chatmanager, user_id: str) -> bool:
"""
保存用户学习进度到MongoDB
参数：
- user_uuid: 用户UUID
- chatmanager: ChatManager实例，包含学习进度数据
- user_id: 用户ID
返回：
- bool: 保存是否成功
"""
try:
# 获取MongoDB连接
mongo = current_app.extensions["mongo"]
# 获取MultiAgents框架中的对话列表
dialog_list = get_multiagents_dialog_list(user_id)
his_data = dialog_list['data']
# 构建学习进度数据
progress_data = {
"user_uuid": user_uuid,
"user_id": user_id,
"material_id": chatmanager.material_id,
"chapter_name": chatmanager.chapter_name,
"lesson_name": chatmanager.lesson_name,
"chapter_chain_now": chatmanager.chapter_chain_now,
"chat_historys": chatmanager.chat_historys,
"scores": chatmanager.scores,
"multiagents_dialogs": his_data,
"updated_at": datetime.now()
}
# 保存到MongoDB，使用upsert操作（如果存在则更新，不存在则插入）
mongo.db.learning_progress.update_one(
{
"user_uuid": user_uuid,
"material_id": chatmanager.material_id,
"chapter_name": chatmanager.chapter_name,
"lesson_name": chatmanager.lesson_name
},
{"$set": progress_data},
upsert=True
)
current_app.logger.info(f"学习进度保存成功: user_uuid={user_uuid}, lesson={chatmanager.lesson_name}")
return True
except Exception as e:
current_app.logger.error(f"保存学习进度失败: {str(e)}")
return False
def get_multiagents_dialog_list(user_id: str) -> List[Dict]:
"""
从MultiAgents框架获取用户对话列表
参数：
- user_id: 用户ID
返回：
- Dict[List]: 对话列表(key 为 History database)
"""
try:
# 从配置中获取ASE引擎URL和token
config = current_app.config
url_token = config.get("ASE_ENGINE_URL_TOKEN", "")
# 构建请求URL
base_url = config.get("ASE_ENGINE_URL", "")
api_url = f"{base_url}/api/sync/download"
# 构建请求参数
payload = {
"namespace_url": url_token, # 使用配置中的token作为namespace_url
"user_id": user_id
}
# 发送POST请求
response = requests.post(
api_url,
json=payload,
headers={"Content-Type": "application/json"},
timeout=10
)
# 检查响应
if response.status_code == 200:
return response.json()
else:
current_app.logger.error(f"获取MultiAgents对话列表失败: HTTP {response.status_code}")
return []
except Exception as e:
current_app.logger.error(f"获取MultiAgents对话列表异常: {str(e)}")
return []
def load_learning_progress(user_uuid: str, material_id: str, chapter_name: str, lesson_name: str) -> Optional[Dict]:
"""
加载用户学习进度
参数：
- user_uuid: 用户UUID
- material_id: 课程ID
- chapter_name: 章节名
- lesson_name: 课时名
返回：
- Optional[Dict]: 学习进度数据，如果不存在则返回None
"""
try:
mongo = current_app.extensions["mongo"]
progress = mongo.db.learning_progress.find_one({
"user_uuid": user_uuid,
"material_id": material_id,
"chapter_name": chapter_name,
"lesson_name": lesson_name
})
if progress:
# 转换ObjectId为字符串
if "_id" in progress:
progress["_id"] = str(progress["_id"])
return progress
return None
except Exception as e:
current_app.logger.error(f"加载学习进度失败: {str(e)}")
return None
def upload_learning_progress_to_cloud(progress: Dict) -> Dict:
"""
将学习进度上传到云服务
参数：
- progress: 学习进度数据字典{'databases':{}, 'global_databases':{}, 'session_info_out':{}}
返回：
- Dict: 上传结果，包含success和message字段
"""
try:
# 从配置中获取ASE引擎URL和token
config = current_app.config
url_token = config.get("ASE_ENGINE_URL_TOKEN", "")
# 构建请求URL
base_url = config.get("ASE_ENGINE_URL", "")
api_url = f"{base_url}/api/sync/upload"
# 构建请求参数
payload = {
"namespace_url": url_token, # 使用配置中的token作为namespace_url
"user_id": progress.get("user_id", ""),
"data": progress
}
# 发送POST请求
response = requests.post(
api_url,
json=payload,
headers={"Content-Type": "application/json"},
timeout=10
)
# 处理响应
if response.status_code == 200:
result = response.json()
current_app.logger.info(f"学习进度上传成功: user_id={progress.get('user_id')}")
return {
"success": True,
"message": "学习进度上传成功",
"data": result
}
else:
error_msg = f"上传失败，HTTP状态码: {response.status_code}"
current_app.logger.error(f"学习进度上传失败: {error_msg}")
return {
"success": False,
"message": error_msg
}
except Exception as e:
error_msg = f"上传异常: {str(e)}"
current_app.logger.error(f"学习进度上传异常: {error_msg}")
return {
"success": False,
"message": error_msg
}

473
Html/apps/services/course_services/material_process.py Normal file
View File

@@ -0,0 +1,473 @@
import re
import requests
import json
from typing import Dict, List, Optional
from flask import current_app
from urllib.parse import urlparse
from datetime import datetime
from bson import ObjectId
from services.cos_service import upload_file, delete_file
from models.material import Material
def get_new_edit_materials(num: int)->List[Material]:
mongo = current_app.extensions["mongo"]
materials = mongo.db.materials.find().sort("updated_at", -1).limit(num)
returnCode = []
for material in materials:
returnCode.append(Material(**material))
returnCode[-1]._id = str(material['_id'])
return returnCode
def user_selected_course(user_obj):
briefs = []
for cid in getattr(user_obj, "select_course", []):
course_obj = load_material(cid)
brief={}
brief["_id"] = cid
brief["name"] = course_obj.name
brief["description"] = course_obj.description
brief["image_url"] = course_obj.image_url
brief["created_at"] = course_obj.created_at.strftime("%Y-%m-%d %H:%M:%S")
brief["updated_at"] = course_obj.updated_at.strftime("%Y-%m-%d %H:%M:%S")
brief["chapters"] = [chapter.model_dump() for chapter in course_obj.chapters]
briefs.append(brief)
return briefs
def user_selected_course_briefs(user_obj):
"""根据用户对象的选课列表，拼装课程简要信息数组"""
briefs = []
for cid in getattr(user_obj, "select_course", []):
course_obj = load_material(cid)
brief={}
brief["_id"] = cid
brief["name"] = course_obj.name
brief["description"] = course_obj.description
brief["image_url"] = course_obj.image_url
brief["created_at"] = course_obj.created_at.strftime("%Y-%m-%d %H:%M:%S")
brief["updated_at"] = course_obj.updated_at.strftime("%Y-%m-%d %H:%M:%S")
briefs.append(brief)
return briefs
def create_material(teacher_id, material_name, description, chapters, image_url):
material = Material(
name=material_name,
description=description,
teacher_id=teacher_id,
created_at=datetime.now(),
updated_at=datetime.now(),
chapters=chapters,
image_url=image_url
)
mongo = current_app.extensions["mongo"]
material_id = mongo.db.materials.insert_one(material.model_dump()).inserted_id
return str(material_id)
def load_material(material_id):
mongo = current_app.extensions["mongo"]
material = mongo.db.materials.find_one({"_id": ObjectId(material_id)})
material['_id'] = material_id
return Material(**material)
def update_material(material_id, chapters):
mongo = current_app.extensions["mongo"]
material = mongo.db.materials.find_one({'_id': ObjectId(material_id)})
if material:
mongo.db.materials.update_one(
{'_id': ObjectId(material_id)},
{'$set': {'chapters': chapters, 'updated_at': datetime.now()}}
)
return True
return False
def add_material_chapter(material_id, chapter_name, teacher_id):
mongo = current_app.extensions["mongo"]
material = mongo.db.materials.find_one({'_id': ObjectId(material_id)})
if material['teacher_id'] != teacher_id:
return False
if material:
mongo.db.materials.update_one(
{'_id': ObjectId(material_id)},
{'$push': {'chapters': {'chapter_name': chapter_name, 'lessons': []}}})
return True
def rename_material_structure(material_id, is_chapter, chapter_name, lesson_name, new_name):
mongo = current_app.extensions["mongo"]
material = mongo.db.materials.find_one({'_id': ObjectId(material_id)})
if not material:
return False # 材料不存在
# 检查名称是否存在于列表中的辅助函数
def name_exists_in_list(item_list, name_key, target_name):
return any(item[name_key] == target_name for item in item_list)
# 查找包含指定章节名的章节
def find_chapter_by_name(chapters, target_name):
for chapter in chapters:
if chapter['chapter_name'] == target_name:
return chapter
return None
# 检查章节是否存在
target_chapter = find_chapter_by_name(material['chapters'], chapter_name)
if not target_chapter:
return False
if is_chapter:
# 检查新章节名是否已存在
if name_exists_in_list(material['chapters'], 'chapter_name', new_name):
return False
# 更新章节名
mongo.db.materials.update_one(
{'_id': ObjectId(material_id)},
{'$set': {'chapters.$[c].chapter_name': new_name}},
array_filters=[{'c.chapter_name': chapter_name}]
)
else:
# 检查课时是否存在
if not name_exists_in_list(target_chapter['lessons'], 'lesson_name', lesson_name):
return False
# 检查新课时名是否已存在
if name_exists_in_list(target_chapter['lessons'], 'lesson_name', new_name):
return False
# 更新课时名
mongo.db.materials.update_one(
{'_id': ObjectId(material_id)},
{'$set': {'chapters.$[c].lessons.$[l].lesson_name': new_name}},
array_filters=[{'c.chapter_name': chapter_name}, {'l.lesson_name': lesson_name}]
)
return True
def delete_material_chapter_lesson(material_id, chapter_name, lesson_name, teacher_id):
"""
删除指定课程的课时。
删除后，更新 MongoDB 中对应的章节中的 lesson 列表。
"""
mongo = current_app.extensions["mongo"]
# 查询 material 是否存在，且确保该 teacher_id 对应的 material
material = mongo.db.materials.find_one({'_id': ObjectId(material_id), 'teacher_id': teacher_id})
for chapter in material['chapters']:
if chapter['chapter_name'] == chapter_name:
for lesson in chapter['lessons']:
if lesson['lesson_name'] == lesson_name:
if lesson['markdown_lesson_file_link']!='':
delete_file(lesson['markdown_lesson_file_link'])
if lesson['markdown_prompt_file_name']!='':
delete_file(lesson['markdown_prompt_file_name'])
if lesson['markdown_score_prompt_file_link']!='':
delete_file(lesson['markdown_score_prompt_file_link'])
if not material:
return False # 找不到对应的 material 或 teacher_id 不匹配
# 删除 lesson 的条件
res = mongo.db.materials.update_one(
{
"_id": ObjectId(material_id),
"teacher_id": teacher_id # 确保是该教师的课程
},
{
"$pull": {
"chapters.$[c].lessons": {
"lesson_name": lesson_name # 根据 lesson_name 来删除
}
},
"$set": {"updated_at": datetime.now()} # 更新修改时间
},
array_filters=[{"c.chapter_name": chapter_name}] # 找到匹配的章节
)
# 如果删除操作成功，res.modified_count 应为 1
return res.modified_count == 1
def reorder_material_structure(material_id, order):
mongo = current_app.extensions["mongo"]
material = mongo.db.materials.find_one({'_id': ObjectId(material_id)})
if not material:
return False
# 1. 构建原章节映射（保留章节元信息）和原章节的课时映射
chapter_map = {ch["chapter_name"]: ch for ch in material["chapters"]}
# 每个章节的课时映射：{章节名: {课时名: 课时对象}}
original_chapter_lessons = {
ch["chapter_name"]: {ls["lesson_name"]: ls for ls in ch["lessons"]}
for ch in material["chapters"]
}
# 2. 收集所有章节“少掉的课时”到全局池（仅包含原章节有但新排序没包含的课时）
# 用列表存储，避免同名课时被覆盖（处理可能的重名场景）
global_extra_lessons = []
for ch_name in chapter_map:
original_lessons = original_chapter_lessons[ch_name] # 原章节的所有课时
# 找到当前章节在order中的配置（如果没有，默认该章节所有课时都是“少掉的”）
chapter_order = next((item for item in order if item["chapter_name"] == ch_name), None)
if not chapter_order:
# 章节未在order中出现，所有原课时都是“少掉的”
global_extra_lessons.extend(original_lessons.values())
continue
# 章节在order中出现，找出原课时中不在新排序里的（即“少掉的”）
new_lesson_names = chapter_order["lessons"]
for ls_name, ls in original_lessons.items():
if ls_name not in new_lesson_names:
global_extra_lessons.append(ls)
print("global_extra_lessons:", global_extra_lessons)
new_chapters = []
for chapter_order in order:
ch_name = chapter_order["chapter_name"]
if ch_name not in chapter_map:
continue # 跳过不存在的章节
original_ch = chapter_map[ch_name]
original_lessons = original_chapter_lessons[ch_name] # 本章节原课时
new_lessons = []
# 3. 构建当前章节的新课时列表
for ls_name in chapter_order["lessons"]:
# 优先使用本章节原有的课时
if ls_name in original_lessons:
new_lessons.append(original_lessons[ls_name])
continue
# 本章节没有，从全局池（其他章节少掉的课时）中找
# 处理可能的重名：找到第一个匹配的课时
for i, extra_ls in enumerate(global_extra_lessons):
if extra_ls["lesson_name"] == ls_name:
new_lessons.append(extra_ls)
del global_extra_lessons[i] # 从全局池移除，避免重复
break
# 4. 构造新章节（保留原章节元信息，替换课时列表）
new_chapter = {**original_ch, "lessons": new_lessons}
new_chapters.append(new_chapter)
# 5. 处理全局池中剩余的课时（未被任何章节引用的，追加到最后一个章节）
if global_extra_lessons and new_chapters:
new_chapters[-1]["lessons"].extend(global_extra_lessons)
print("new_chapters:", new_chapters)
# 更新数据库
mongo.db.materials.update_one(
{"_id": ObjectId(material_id)},
{
"$set": {
"chapters": new_chapters,
"updated_at": datetime.now()
}
}
)
return True
def save_materials_markdown_cos(
material_id: str,
chapter_name: str,
lesson_name: str,
docs: Dict[str, Dict[str, str]],
version: str | None = None, # 目前不做并发控制，保留参数
actor=None # 可记录操作者到审计日志
) -> Dict[str, str]:
"""
docs 结构：
{
"lesson": {"cdn_link": "...lesson.md", "content": "..."},
"prompt": {"cdn_link": "...prompt.md", "content": "..."},
"score": {"cdn_link": "...score_prompt.md", "content": "..."}
}
约定：cdn_link 与对象存储的 key 一一映射（同路径/同文件名），
我们通过解析 URL 的 path 作为 key 进行覆盖写入（不改文件名）。
"""
def _cdn_to_key(cdn_link: str) -> str:
"""
把 CDN URL 映射成对象存储的 key：
- 若 CDN 与存储是“同 key 直出”，则直接取 path（去掉前导 /）
- 若你有自定义映射规则，在此处改写
"""
p = urlparse(cdn_link)
key = p.path.lstrip("/")
if not key:
raise ValueError(f"非法 cdn_link：{cdn_link}")
return key
# 1) 覆盖写入三份文件
results: Dict[str, Dict[str, str]] = {}
for kind in ("lesson", "prompt", "score"):
item = docs.get(kind)
if not item or "cdn_link" not in item or "content" not in item:
raise ValueError(f"缺少 {kind} 的 cdn_link 或 content")
key = _cdn_to_key(item["cdn_link"])
content = item["content"]
# 覆盖写入（注意：upload_file 内部需实现覆盖同 key）
try:
# 约定：upload_file 返回可访问的 URL（覆盖后可与原相同）
url = upload_file(content.encode("utf-8"), key)
except Exception as e:
current_app.logger.exception("覆盖写入失败 kind=%s key=%s", kind, key)
raise
# 简单返回：保留原 cdn_link（一般与 url 相同），以及 md5 作为伪版本
import hashlib
etag = hashlib.md5(content.encode("utf-8")).hexdigest()
results[kind] = {"cdn_link": url or item["cdn_link"], "version": etag}
# 2) 写库：更新该 lesson 的更新时间（不改文件名/链接）
_touch_material_lesson_updated_at(material_id, chapter_name, lesson_name, actor=actor)
results["updated_at"] = datetime.now().isoformat() + "Z"
return results
def _touch_material_lesson_updated_at(material_id: str, chapter_name: str, lesson_name: str, actor=None) -> None:
"""仅刷新 materials 文档中该课时的更新时间戳（以及顶层 updated_at）"""
mongo = current_app.extensions["mongo"]
now = datetime.now()
# 更新顶层 updated_at
mongo.db.materials.update_one(
{"_id": ObjectId(material_id)},
{"$set": {"updated_at": now}}
)
# 更新嵌套 lesson 的更新时间（如果你在 lesson 内有该字段）
# 需要使用 arrayFilters 精确定位到对应章节/课时
mongo.db.materials.update_one(
{"_id": ObjectId(material_id)},
{
"$set": {
"chapters.$[c].lessons.$[l].updated_at": now,
# 可选：记录最近编辑人
"chapters.$[c].lessons.$[l].last_editor": getattr(actor, "id", None) if actor else None
}
},
array_filters=[
{"c.chapter_name": chapter_name},
{"l.lesson_name": lesson_name}
]
)
def add_material_chapter_lesson(material_id, chapter_name, lesson_name, teacher_id):
"""
1) 生成占位 Markdown（lesson_prompt_score_prompt）
2) 上传到对象存储（返回 URL）
3) 将 lesson 文档（含三份 URL）push 到指定章节
"""
mongo = current_app.extensions["mongo"]
# 组织对象存储的文件路径
chapter_slug = _slugify(chapter_name or "")
lesson_slug = _slugify(lesson_name or "")
timestamp = datetime.now().strftime('%Y%m%dT%H%M%SZ')
base_prefix = f"materials_{material_id}_{chapter_slug}_{lesson_slug}_{timestamp}"
# 占位文本
lesson_md, prompt_md, score_prompt_md = _default_markdowns(lesson_name or "未命名课时")
# 生成文件名（带层级路径）
lesson_key = f"{base_prefix}_lesson.md"
prompt_key = f"{base_prefix}_prompt.md"
score_key = f"{base_prefix}_score_prompt.md"
# 上传（任意一个失败则终止，不写库）
try:
lesson_url = upload_file(lesson_md.encode("utf-8"), lesson_key)
prompt_url = upload_file(prompt_md.encode("utf-8"), prompt_key)
score_url = upload_file(score_prompt_md.encode("utf-8"), score_key)
except Exception as e:
current_app.logger.exception("Upload default markdown failed: %s", e)
return False
# 组装 lesson_doc
lesson_doc = {
"lesson_name": lesson_name or "",
"markdown_lesson_file_link": lesson_url, # lesson 主体
"markdown_prompt_file_name": prompt_url, # 给大模型的 prompt
"markdown_score_prompt_file_link": score_url, # 评分 prompt
}
# 将 lesson push 到指定章节
res = mongo.db.materials.update_one(
{
"_id": ObjectId(material_id),
"teacher_id": teacher_id, # 注意：若存的是 ObjectId，请确保传入的是同类型
},
{
"$push": {"chapters.$[c].lessons": lesson_doc},
"$set": {"updated_at": datetime.now()},
},
array_filters=[{"c.chapter_name": chapter_name}],
)
# 可选：如果 res.modified_count != 1，说明 push 失败（可能章节名不匹配）
# 这里可以考虑是否回滚 COS 上刚上传的 3 个文件（需实现删除逻辑）。
return res.modified_count == 1
def get_materials_by_teacher(teacher_id)->List[Material]:
mongo = current_app.extensions["mongo"]
materials = mongo.db.materials.find({'teacher_id': teacher_id})
return [Material(**material) for material in materials]
def get_materials_by_teacher_dict(teacher_id)->List[Dict]:
mongo = current_app.extensions["mongo"]
materials = mongo.db.materials.find({'teacher_id': teacher_id})
result = []
for material in materials:
material['_id'] = str(material['_id'])
result.append(material)
return result
def _slugify(name: str) -> str:
"""将章节/课时名转为 URL 友好的 slug。"""
s = name.strip().lower()
s = re.sub(r'\s+', '-', s) # 空白 -> -
s = re.sub(r'[^a-z0-9\-_]+', '', s) # 仅保留安全字符
return s or 'untitled'
def _default_markdowns(lesson_name: str):
"""三份占位 Markdown 文本。可按需调整模板内容。"""
ts = datetime.now().strftime('%Y-%m-%d %H:%M')
lesson_md = f"""# {lesson_name}
## 阶段一：示例阶段
### 步骤A
在这里编写该步骤的教学内容（文字/代码/图片链接等）。
### 步骤B
继续补充该阶段的其它步骤内容。
"""
prompt_md = f"""# {lesson_name}
## 阶段一：示例阶段
### 步骤A
请基于"步骤A"的教学目标与材料，生成指导性提示词（尽量结构化，给出评估标准和示例）。
### 步骤B
请基于"步骤B"的教学目标与材料，生成指导性提示词。
"""
score_prompt_md = f"""# {lesson_name}
## 评分准则
- 正确性、完整性、可读性、效率（按需调整权重）
## 阶段一：示例阶段
### 步骤A
对学员在"步骤A"的输出进行评分与文字反馈，分数区间 0-100，并输出 JSON：```{{"score": <int>, "reasons": "<string>", "advices": "<string>"}}```
### 步骤B
同上，对"步骤B"输出进行评分与反馈。
"""
return lesson_md, prompt_md, score_prompt_md