# 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 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": , "reasons": "", "advices": ""}}``` ### 步骤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 }