428 lines
16 KiB
Python
428 lines
16 KiB
Python
# myapp/services/course_service.py
|
||
import re
|
||
from typing import Dict, List
|
||
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 is_chapter:
|
||
for chapter in material['chapters']:
|
||
if chapter['chapter_name'] == chapter_name:
|
||
break
|
||
return False
|
||
for chapter in material['chapters']:
|
||
if chapter['chapter_name'] == new_name:
|
||
return False
|
||
else:
|
||
for chapter in material['chapters']:
|
||
if chapter['chapter_name'] == chapter_name:
|
||
break
|
||
return False
|
||
for lesson in chapter['lessons']:
|
||
if lesson['lesson_name'] == lesson_name:
|
||
break
|
||
for lesson in chapter['lessons']:
|
||
if lesson['lesson_name'] == new_name:
|
||
return False
|
||
|
||
if is_chapter:
|
||
mongo.db.materials.update_one(
|
||
{'_id': ObjectId(material_id)},
|
||
{'$set': {'chapters.$[c].chapter_name': new_name}},
|
||
array_filters=[{'c.chapter_name': chapter_name}]
|
||
)
|
||
else:
|
||
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
|
||
chapter_map = {ch["chapter_name"]: ch for ch in material["chapters"]}
|
||
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]
|
||
|
||
# 建立 lesson_name -> lesson dict
|
||
lesson_map = {ls["lesson_name"]: ls for ls in original_ch["lessons"]}
|
||
|
||
# 按顺序重排 lessons
|
||
new_lessons = []
|
||
for ls_name in chapter_order['lessons']:
|
||
if ls_name in lesson_map:
|
||
new_lessons.append(lesson_map[ls_name])
|
||
|
||
# 如果还有遗漏（前端没传的），可以追加到最后，避免丢数据
|
||
for ls_name, ls in lesson_map.items():
|
||
if ls not in new_lessons:
|
||
new_lessons.append(ls)
|
||
|
||
# 构造新的 chapter
|
||
new_chapter = {**original_ch, "lessons": new_lessons}
|
||
new_chapters.append(new_chapter)
|
||
|
||
# 更新 material
|
||
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}
|
||
> Auto-generated at {ts}
|
||
|
||
## 阶段一：示例阶段
|
||
### 步骤A
|
||
在这里编写该步骤的教学内容（文字/代码/图片链接等）。
|
||
|
||
### 步骤B
|
||
继续补充该阶段的其它步骤内容。
|
||
"""
|
||
|
||
prompt_md = f"""# {lesson_name}
|
||
> Prompt for LLM (auto-generated at {ts})
|
||
|
||
## 阶段一：示例阶段
|
||
### 步骤A
|
||
请基于“步骤A”的教学目标与材料，生成指导性提示词（尽量结构化，给出评估标准和示例）。
|
||
|
||
### 步骤B
|
||
请基于“步骤B”的教学目标与材料，生成指导性提示词。
|
||
"""
|
||
|
||
score_prompt_md = f"""# {lesson_name}
|
||
> Scoring Prompt for LLM (auto-generated at {ts})
|
||
|
||
## 评分准则
|
||
- 正确性、完整性、可读性、效率（按需调整权重）
|
||
|
||
## 阶段一：示例阶段
|
||
### 步骤A
|
||
对学员在“步骤A”的输出进行评分与文字反馈，分数区间 0-100，并输出 JSON：```{{"score": <int>, "reasons": "<string>", "advices": "<string>"}}```
|
||
### 步骤B
|
||
同上，对“步骤B”输出进行评分与反馈。
|
||
"""
|
||
return lesson_md, prompt_md, score_prompt_md
|