'commit'

AI/T1_Mp4ToWav.py

@@ -47,6 +47,3 @@ def convert_mp4_to_wav(
     except FileNotFoundError:
         print(f"FFmpeg未找到，请确认路径是否正确：{ffmpeg_path}")
 
-
-# 使用示例（替换为实际MP4文件名）
-convert_mp4_to_wav(r"D:\backup\七年级第三单元复习课.mp4", output_path=r"D:\backup\七年级第三单元复习课.wav")

AI/T1_Start.py

@@ -0,0 +1,19 @@
+from T1_Mp4ToWav import *
+from T2_WavToText import *
+from T3_TextSummarize import *
+if __name__ == '__main__':
+    # 1、根据云校的视频课程下载视频文件
+
+    # 2、调用ffmpeg将视频文件转成wav文件
+    # convert_mp4_to_wav(r"D:\backup\七年级第三单元复习课.mp4",
+    #                   output_path=r"D:\backup\七年级第三单元复习课.wav")
+    # 3、将转换完成的WAV上传到阿里云
+
+    # 4、调用阿里云的语音识别API，将wav文件转成文字
+    #audio_url = "https://ylt.oss-cn-hangzhou.aliyuncs.com/HuangHai/123.wav"
+    # 5、 调用dashscope的API，将文字转成markdown文件
+    #ShiBie(audio_url, "识别结果.txt")
+    # 6、生成总结
+    input_file = Path(r"D:\dsWork\QingLong\AI\音频文本.txt")
+    output_file = Path(r"D:\dsWork\QingLong\AI\分析结果.txt")
+    analyzerAction(input_file, output_file)

AI/T2_WavToText.py

@@ -1,143 +1,189 @@
-# -*- coding: utf8 -*-
+# -*- coding: utf-8 -*-
 import json
 import time
+from pathlib import Path
+from typing import Dict, Tuple
+
 from aliyunsdkcore.acs_exception.exceptions import ClientException, ServerException
 from aliyunsdkcore.client import AcsClient
 from aliyunsdkcore.request import CommonRequest
-# 录音文件识别
-# https://common-buy.aliyun.com/?spm=5176.11801677.0.0.44e77a33oMMyT6&commodityCode=nlsService&orderType=UPGRADE&instanceId=1546399445482588
-# 1. 开通商用即为付费使用服务，根据后付费梯度计费，标准单价为2.5元/小时；
-# 2. 商用服务可在最长3小时内完成识别并返回识别文本；
-# 3. 注意：已经“商用”后再切换“试用”，等同于立即关停线上服务，请谨慎操作；
-
-def fileTrans(akId, akSecret, appKey, fileLink):
-    # 服务配置
-    REGION_ID = "cn-shanghai"
-    PRODUCT = "nls-filetrans"
-    DOMAIN = "filetrans.cn-shanghai.aliyuncs.com"
-    API_VERSION = "2018-08-17"
-
-    # 请求参数
-    KEY_APP_KEY = "appkey"
-    KEY_FILE_LINK = "file_link"
-    KEY_VERSION = "version"
-    KEY_ENABLE_WORDS = "enable_words"
-    KEY_TASK = "Task"
-    KEY_TASK_ID = "TaskId"
-    KEY_STATUS_TEXT = "StatusText"
-    KEY_RESULT = "Result"
-
-    # 状态码
-    STATUS_SUCCESS = "SUCCESS"
-    STATUS_RUNNING = "RUNNING"
-    STATUS_QUEUEING = "QUEUEING"
-
-    # 初始化客户端
-    client = AcsClient(akId, akSecret, REGION_ID)
-
-    # 构建任务参数
+
+# 服务常量配置
+CONFIG = {
+    "REGION_ID": "cn-shanghai",
+    "PRODUCT": "nls-filetrans",
+    "DOMAIN": "filetrans.cn-shanghai.aliyuncs.com",
+    "API_VERSION": "2018-08-17",
+    "MAX_RETRIES": 20,
+    "POLL_INTERVAL": 10
+}
+
+
+class TranscriptionError(Exception):
+    """自定义语音识别异常基类"""
+    pass
+
+
+class APIConnectionError(TranscriptionError):
+    """API连接异常"""
+    pass
+
+
+class TaskSubmissionError(TranscriptionError):
+    """任务提交异常"""
+    pass
+
+
+class TaskTimeoutError(TranscriptionError):
+    """任务超时异常"""
+    pass
+
+
+def submit_transcription_task(client: AcsClient, app_key: str, file_link: str) -> str:
+    """提交语音识别任务
+
+    :param client: 阿里云客户端实例
+    :param app_key: 应用密钥
+    :param file_link: 音频文件URL
+    :return: 任务ID
+    :raises TaskSubmissionError: 任务提交失败时抛出
+    """
     task_config = {
-        KEY_APP_KEY: appKey,
-        KEY_FILE_LINK: fileLink,
-        KEY_VERSION: "4.0",
-        KEY_ENABLE_WORDS: False
+        "appkey": app_key,
+        "file_link": file_link,
+        "version": "4.0",
+        "enable_words": False
     }
-    task_json = json.dumps(task_config)
-    print("任务配置:", task_json)
-
-    # 提交识别请求
-    postRequest = CommonRequest()
-    postRequest.set_domain(DOMAIN)
-    postRequest.set_version(API_VERSION)
-    postRequest.set_product(PRODUCT)
-    postRequest.set_action_name("SubmitTask")
-    postRequest.set_method('POST')
-    postRequest.add_body_params(KEY_TASK, task_json)
+
+    request = CommonRequest()
+    request.set_domain(CONFIG["DOMAIN"])
+    request.set_version(CONFIG["API_VERSION"])
+    request.set_product(CONFIG["PRODUCT"])
+    request.set_action_name("SubmitTask")
+    request.set_method('POST')
+    request.add_body_params("Task", json.dumps(task_config))
 
     try:
-        postResponse = client.do_action_with_exception(postRequest)
-        postData = json.loads(postResponse)
-        print("提交响应:", postData)
+        response = client.do_action_with_exception(request)
+        response_data = json.loads(response)
 
-        if postData.get(KEY_STATUS_TEXT) != STATUS_SUCCESS:
-            print(f"任务提交失败: {postData.get(KEY_STATUS_TEXT, '未知错误')}")
-            return None
+        if response_data.get("StatusText") != "SUCCESS":
+            raise TaskSubmissionError(f"任务提交失败: {response_data.get('StatusText', '未知错误')}")
 
-        taskId = postData[KEY_TASK_ID]
-        print(f"任务ID: {taskId}")
+        return response_data["TaskId"]
 
     except (ServerException, ClientException) as e:
-        print(f"API请求异常: {e}")
-        return None
-    except KeyError:
-        print("响应缺少关键字段")
-        return None
-
-    # 查询任务结果
-    getRequest = CommonRequest()
-    getRequest.set_domain(DOMAIN)
-    getRequest.set_version(API_VERSION)
-    getRequest.set_product(PRODUCT)
-    getRequest.set_action_name("GetTaskResult")
-    getRequest.set_method('GET')
-    getRequest.add_query_param(KEY_TASK_ID, taskId)
-
-    retry_count = 0
-    max_retries = 20  # 最多等待200秒 (20次*10秒)
-
-    while retry_count < max_retries:
+        raise APIConnectionError(f"API连接异常: {str(e)}") from e
+    except KeyError as e:
+        raise TaskSubmissionError("响应缺少TaskId字段") from e
+
+
+def poll_transcription_result(client: AcsClient, task_id: str) -> Dict:
+    """轮询识别任务结果
+
+    :param client: 阿里云客户端实例
+    :param task_id: 任务ID
+    :return: 识别结果字典
+    :raises TaskTimeoutError: 超时未完成时抛出
+    """
+    request = CommonRequest()
+    request.set_domain(CONFIG["DOMAIN"])
+    request.set_version(CONFIG["API_VERSION"])
+    request.set_product(CONFIG["PRODUCT"])
+    request.set_action_name("GetTaskResult")
+    request.set_method('GET')
+    request.add_query_param("TaskId", task_id)
+
+    retries = 0
+    while retries < CONFIG["MAX_RETRIES"]:
         try:
-            getResponse = client.do_action_with_exception(getRequest)
-            resultData = json.loads(getResponse)
-            print(f"轮询结果({retry_count + 1}/{max_retries}):", resultData)
-
-            status = resultData.get(KEY_STATUS_TEXT, "")
-            if status == STATUS_SUCCESS:
-                return resultData.get(KEY_RESULT, {})
-            elif status in (STATUS_RUNNING, STATUS_QUEUEING):
-                time.sleep(10)
-                retry_count += 1
+            response = client.do_action_with_exception(request)
+            result = json.loads(response)
+            status = result.get("StatusText", "")
+
+            if status == "SUCCESS":
+                return result.get("Result", {})
+            if status in ("RUNNING", "QUEUEING"):
+                time.sleep(CONFIG["POLL_INTERVAL"])
+                retries += 1
             else:
-                print(f"识别失败，最终状态: {status}")
-                return None
+                raise TaskSubmissionError(f"识别失败，状态: {status}")
 
         except (ServerException, ClientException) as e:
-            print(f"查询异常: {e}")
-            return None
-
-    print("超过最大重试次数，任务未完成")
-    return None
-
-
-# 配置信息（请替换为实际值）
-accessKeyId = "LTAI5tE4tgpGcKWhbZg6C4bh"
-accessKeySecret = "oizcTOZ8izbGUouboC00RcmGE8vBQ1"
-appKey = "OIpiw501l4o6MYEe"
-fileLink = "https://ylt.oss-cn-hangzhou.aliyuncs.com/HuangHai/123.wav"
-
-# 执行识别
-result = fileTrans(accessKeyId, accessKeySecret, appKey, fileLink)
-
-# 处理结果
-if not result:
-    print("未获取到有效结果")
-    exit(1)
-
-text_content = []
-if 'Sentences' in result:
-    for sentence in result['Sentences']:
-        text_content.append(sentence.get("Text", ""))
-    full_text = ''.join(text_content)
-else:
-    print("响应中未包含有效文本")
-    exit(1)
-
-# 保存结果
-output_file = "识别结果.txt"
-try:
-    with open(output_file, 'w', encoding='utf-8') as f:
-        f.write(full_text)
-    print(f"结果已保存至 {output_file}")
-except Exception as e:
-    print(f"文件保存失败: {str(e)}")
-    exit(1)
+            raise APIConnectionError(f"查询异常: {str(e)}") from e
+
+    raise TaskTimeoutError(f"超过最大重试次数({CONFIG['MAX_RETRIES']})，任务未完成")
+
+
+def save_transcription_result(result: Dict, output_path: Path) -> Path:
+    """保存识别结果到文件
+
+    :param result: 识别结果字典
+    :param output_path: 输出文件路径
+    :return: 实际保存路径
+    :raises IOError: 文件保存失败时抛出
+    """
+    if not result.get('Sentences'):
+        raise ValueError("识别结果为空")
+
+    try:
+        full_text = ''.join(s.get("Text", "") for s in result['Sentences'])
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        output_path.write_text(full_text, encoding='utf-8')
+        return output_path.resolve()
+    except (IOError, PermissionError) as e:
+        raise IOError(f"文件保存失败: {str(e)}") from e
+
+
+def transcribe_audio_file(
+        access_key_id: str,
+        access_key_secret: str,
+        app_key: str,
+        audio_url: str,
+        output_path: Path = Path("识别结果.txt")
+) -> Tuple[bool, Path, str]:
+    """语音识别主流程
+
+    :param access_key_id: 阿里云访问密钥ID
+    :param access_key_secret: 阿里云访问密钥
+    :param app_key: 应用密钥
+    :param audio_url: 音频文件URL
+    :param output_path: 输出文件路径
+    :return: (成功标志, 输出路径, 错误信息)
+    """
+    client = AcsClient(access_key_id, access_key_secret, CONFIG["REGION_ID"])
+
+    try:
+        task_id = submit_transcription_task(client, app_key, audio_url)
+        print(f"任务已提交，ID: {task_id}")
+
+        result = poll_transcription_result(client, task_id)
+        saved_path = save_transcription_result(result, output_path)
+
+        return True, saved_path, ""
+
+    except TranscriptionError as e:
+        return False, Path(), str(e)
+    except Exception as e:
+        return False, Path(), f"未处理的异常: {str(e)}"
+
+def ShiBie(audio_url, output_path):
+    # 配置参数（应通过环境变量或配置文件获取）
+    CREDENTIALS = {
+        "access_key_id": "LTAI5tE4tgpGcKWhbZg6C4bh",
+        "access_key_secret": "oizcTOZ8izbGUouboC00RcmGE8vBQ1",
+        "app_key": "OIpiw501l4o6MYEe",
+        "audio_url":audio_url,
+        "output_path": output_path
+    }
+
+    # 执行识别
+    success, path, error = transcribe_audio_file(**CREDENTIALS)
+
+    if success:
+        print(f"✅ 识别成功，文件已保存至: {path}")
+    else:
+        print(f"❌ 识别失败: {error}")
+        exit(1)
+
+

AI/T3_TextSummarize.py

@@ -1,28 +1,177 @@
-from openai import OpenAI
+# -*- coding: utf-8 -*-
+import time
 from pathlib import Path
-# https://help.aliyun.com/zh/model-studio/developer-reference/deepseek?spm=a2c4g.11186623.0.0.274b1d1c4GY0Zd
-
-API_KEY = "sk-01d13a39e09844038322108ecdbd1bbc"
-client = OpenAI(
-    api_key=API_KEY,
-    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
-)
-
-# 读取文本文件内容
-file_path = Path(r"D:\dsWork\QingLong\音频文本.txt")
-
-if file_path.exists():
-    # 自动处理文件编码（默认utf-8），读取内容
-    content = file_path.read_text(encoding='utf-8')
-else:
-    print(f"文件 {file_path} 不存在")
-    exit(0)
-
-completion = client.chat.completions.create(
-    model="deepseek-v3",  # 此处以 deepseek-r1 为例，可按需更换模型名称。
-    messages=[
-        {'role': 'user', 'content': "帮我梳理：这节课分了几个部分，每部分的名称和开始的时间是多少："+content}
-    ]
-)
-# 通过content字段打印最终答案
-print(completion.choices[0].message.content)
+from typing import Optional, Tuple
+
+from openai import OpenAI, APIError
+
+
+class ContentAnalyzer:
+    """课程内容分析器（带Emoji进度展示）"""
+
+    def __init__(
+            self,
+            api_key: str = "sk-01d13a39e09844038322108ecdbd1bbc",
+            base_url: str = "https://dashscope.aliyuncs.com/compatible-mode/v1",
+            model: str = "deepseek-r1"
+    ):
+        """
+        初始化分析器
+        :param api_key: 阿里云API密钥
+        :param base_url: API基础地址
+        :param model: 使用的模型名称
+        """
+        self.client = OpenAI(api_key=api_key, base_url=base_url)
+        self.model = model
+
+    def _show_progress(self, emoji: str, message: str, level: int = 1, delay: float = 0.3):
+        """进度展示方法
+        :param emoji: 表情符号
+        :param message: 进度信息
+        :param level: 缩进层级（0-无缩进，1-一级缩进）
+        :param delay: 展示后的延迟时间
+        """
+        indent = "  " * level
+        progress_line = f"{indent}{emoji} {message}"
+        print(progress_line)
+        time.sleep(delay)  # 增加进度展示的节奏感
+
+    def analyze_content(
+            self,
+            content: str,
+            prompt_template: str = "帮我梳理：这节课分了几个部分，每部分的名称和开始的时间是多少：{}"
+    ) -> Tuple[bool, str]:
+        """
+        分析课程内容结构
+        :param content: 需要分析的文本内容
+        :param prompt_template: 提示词模板
+        :return: (成功标志, 分析结果或错误信息)
+        """
+        try:
+            # 构建提示词
+            self._show_progress("📝", "正在构建提示词...", level=1)
+            full_prompt = prompt_template.format(content)
+
+            # 发送API请求
+            self._show_progress("📡", "正在发送API请求...", level=1)
+            start_time = time.time()
+            completion = self.client.chat.completions.create(
+                model=self.model,
+                messages=[{'role': 'user', 'content': full_prompt}]
+            )
+            cost_time = time.time() - start_time
+
+            # 处理响应
+            self._show_progress("✅", f"收到API响应（耗时{cost_time:.1f}s）", level=1)
+            if not completion.choices:
+                self._show_progress("❌", "响应内容为空", level=2)
+                return False, "API响应中未包含有效结果"
+
+            # 解析结果
+            self._show_progress("🔍", "正在解析结果内容...", level=1)
+            return True, completion.choices[0].message.content
+
+        except APIError as e:
+            self._show_progress("🚨", f"API异常: {str(e)}", level=2)
+            return False, f"API调用失败: {str(e)}"
+        except Exception as e:
+            self._show_progress("💥", f"未处理的异常: {str(e)}", level=2)
+            return False, f"未处理的异常: {str(e)}"
+
+    def analyze_file(
+            self,
+            file_path: Path,
+            output_path: Optional[Path] = None,
+            encoding: str = 'utf-8'
+    ) -> Tuple[bool, str]:
+        """
+        从文件读取内容并分析
+        :param file_path: 输入文件路径
+        :param output_path: 输出文件路径（可选）
+        :param encoding: 文件编码格式
+        :return: (成功标志, 最终结果或错误信息)
+        """
+        try:
+            self._show_progress("📂", f"开始处理文件: {file_path}", level=0)
+
+            # 检查文件存在性
+            self._show_progress("🔎", "检查文件是否存在...", level=1)
+            if not file_path.exists():
+                self._show_progress("❌", "文件不存在", level=2)
+                return False, f"文件不存在: {file_path}"
+
+            # 读取文件内容
+            self._show_progress("📖", "正在读取文件内容...", level=1)
+            try:
+                content = file_path.read_text(encoding=encoding)
+            except UnicodeDecodeError:
+                self._show_progress("🔠", f"解码失败（当前编码: {encoding}）", level=2)
+                return False, f"文件解码失败，请尝试使用正确的编码格式"
+
+            # 分析内容
+            self._show_progress("🧠", "开始分析内容...", level=1)
+            success, result = self.analyze_content(content)
+
+            if not success:
+                self._show_progress("⚠️", "内容分析失败", level=2)
+                return False, result
+
+            # 保存结果
+            if output_path:
+                self._show_progress("💾", f"正在保存结果到: {output_path}", level=1)
+                try:
+                    output_path.parent.mkdir(parents=True, exist_ok=True)
+                    output_path.write_text(result, encoding=encoding)
+                    self._show_progress("✅", "文件保存成功", level=2)
+                except IOError as e:
+                    self._show_progress("❌", f"保存失败: {str(e)}", level=2)
+                    return False, f"结果保存失败: {str(e)}"
+
+            self._show_progress("🎉", "文件处理完成", level=0)
+            return True, result
+
+        except Exception as e:
+            self._show_progress("💣", f"意外错误: {str(e)}", level=2)
+            return False, f"文件处理异常: {str(e)}"
+
+
+def show_banner():
+    """显示启动横幅"""
+    print("\n" + "=" * 50)
+    print("🌟  课程内容智能分析系统  🌟".center(50))
+    print("=" * 50 + "\n")
+
+
+def analyzer_action(input_file: str, output_file: Optional[str] = None):
+    """执行分析流程
+    :param input_file: 输入文件路径
+    :param output_file: 输出文件路径（可选）
+    """
+    show_banner()
+
+    analyzer = ContentAnalyzer()
+    input_path = Path(input_file)
+    output_path = Path(output_file) if output_file else None
+
+    success, result = analyzer.analyze_file(input_path, output_path)
+
+    print("\n" + "=" * 50)
+    if success:
+        print("✅ 最终分析结果：".center(50))
+        print("-" * 50)
+        print(result)
+        print("-" * 50)
+        if output_path:
+            print(f"结果已保存至：{output_path.resolve()}")
+    else:
+        print(f"❌ 分析失败：{result}")
+    print("=" * 50 + "\n")
+
+
+if __name__ == "__main__":
+    # 示例配置
+    input_file = r"D:\dsWork\QingLong\AI\音频文本.txt"
+    output_file = r"D:\dsWork\QingLong\AI\分析结果.txt"
+
+    # 执行分析
+    analyzer_action(input_file, output_file)