import env # 加载 .env 到环境变量 import os import dashscope import pydub from pydub import AudioSegment from pydub.silence import split_on_silence from dashscope.audio.asr import Recognition from dashscope import Generation from http import HTTPStatus from mysqlHandle import MySQLDB import logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # 设置环境变量 # os.environ["DASHSCOPE_API_KEY"] = "sk-your-dashscope-key" def convert_mp3_to_wav(mp3_path, output_wav_path): """ 将MP3文件转换为16kHz单声道WAV格式，这是Qwen3-ASR-Flash模型的推荐格式 参数: mp3_path (str): MP3文件路径 output_wav_path (str): 输出WAV文件路径 返回值: str: 转换后的WAV文件路径 """ logger.info(f"开始转换MP3到WAV: {mp3_path}") # 加载MP3文件 audio = AudioSegment.from_file(mp3_path, format="mp3") # 转换为16kHz采样率、单声道、16位深度 audio = audio.set_frame_rate(16000).set_channels(1) # 导出为WAV格式 audio.export(output_wav_path, format="wav") logger.info(f"✓ MP3转换完成: {output_wav_path}") #print(f"✓ MP3转换完成: {output_wav_path}") return output_wav_path def split_audio_by_fixed_duration(audio_path, chunk_duration, output_folder): """ 将音频文件按固定时长分割成多个片段 参数: audio_path (str): 音频文件路径 chunk_duration (int): 分片时长(毫秒) output_folder (str): 输出文件夹路径 返回值: list: 分片文件路径列表 """ # 加载音频文件 audio = AudioSegment.from_file(audio_path) # 计算总时长(毫秒) total_duration = len(audio) # 分片数 num_chunks = total_duration // chunk_duration + 1 # 存储分片文件路径 chunks = [] # 创建输出文件夹 os.makedirs(output_folder, exist_ok=True) logger.info(f"开始音频分割，总时长: {total_duration/1000:.1f}秒，将分割为{num_chunks}个片段") for i in range(num_chunks): # 计算当前分片的起始和结束时间 start_time = i * chunk_duration end_time = (i + 1) * chunk_duration # 提取分片音频 chunk = audio[start_time:end_time] # 生成文件名 chunk_name = f"chunk_{i}.wav" chunk_path = os.path.join(output_folder, chunk_name) # 导出分片音频 chunk.export(chunk_path, format="wav") chunks.append(chunk_path) # 打印处理进度 progress = (i + 1) / num_chunks * 100 logger.info(f"✓ 已完成分片 {i+1}/{num_chunks} ({progress:.1f}%)") logger.info(f"✓ 音频分割完成，共生成{len(chunks)}个分片文件") return chunks def split_audio_by_smart_silence(audio_path, min_silence_len, silence_thresh, output_folder): """ 将音频文件按智能静音检测方式分割成多个片段，每段不超过3分钟 参数: audio_path (str): 音频文件路径 min_silence_len (int): 最小静音长度(毫秒) silence_thresh (int): 静音阈值(dBFS) output_folder (str): 输出文件夹路径 返回值: list: 分片文件路径列表 """ # 加载音频文件 audio = AudioSegment.from_file(audio_path, format="wav") # 按静音分割 segments = split_on_silence( audio, # 静音超过700毫秒则分割 min_silence_len=min_silence_len, # 静音阈值为-40dBFS silence_thresh=silence_thresh, # 保留静音部分 keep_silence=400 ) logger.info(f"✓ 静音分割完成，共{len(segments)}个初始片段") # 合并过短的片段 merged_segments = [] current_segment = None for segment in segments: if current_segment is None: current_segment = segment else: # 合并当前片段和新片段 temp_segment = current_segment + segment # 如果合并后的片段超过3分钟，则单独保存当前片段 if len(temp_segment) > 180000: # 3分钟=180,000毫秒 merged_segments.append(current_segment) current_segment = segment else: current_segment = temp_segment # 添加最后一个片段 if current_segment is not None: merged_segments.append(current_segment) logger.info(f"✓ 片段合并完成，共{len(merged_segments)}个最终片段") # 存储分片文件路径 chunks = [] # 创建输出文件夹 os.makedirs(output_folder, exist_ok=True) logger.info(f"开始导出音频片段到: {output_folder}") for i, segment in enumerate(merged_segments): # 生成文件名 chunk_name = f"chunk_{i}.wav" chunk_path = os.path.join(output_folder, chunk_name) # 导出分片音频 segment.export(chunk_path, format="wav") chunks.append(chunk_path) # 打印处理进度 progress = (i + 1) / len(merged_segments) * 100 logger.info(f"✓ 已完成分片 {i+1}/{len(merged_segments)} ({progress:.1f}%)") logger.info(f"✓ 智能静音分割完成，共生成{len(chunks)}个分片文件") return chunks def transcribe_audio(audio_path): """ 使用Paraformer实时语音识别模型（通过本地文件）转录音频文件 参数: audio_path (str): 音频文件路径（必须是16kHz单声道WAV） 返回值: str: 识别文本，如果失败返回空字符串 """ try: # 确保音频文件存在 if not os.path.exists(audio_path): logger.error(f"音频文件不存在: {audio_path}") return ['', ''] dashscope.api_key = os.getenv('DASHSCOPE_API_KEY', '') # 创建识别对象 recognition = Recognition( model=os.getenv('DASHSCOPE_ASR_MODEL', 'paraformer-realtime-v2'), format='wav', sample_rate=16000, language_hints=['zh','en'], # 中文和英文 callback=None ) # 调用识别 logger.info(f"开始识别音频: {audio_path}") result = recognition.call(audio_path) text=[] if result.status_code == HTTPStatus.OK: # 提取识别结果 logger.info(f"✓ {audio_path} 识别成功") sentence = result.get_sentence() text.append(merge_transcripts(sentence)) logger.info(f"识别文本长度: {len(text[0])}") text.append(analyze_and_correct_text(text[0])) return text else: logger.error(f"❌ 任务失败: {result.message}") return ['', ''] except Exception as e: logger.error(f"识别过程中发生异常: {e}") return ['', ''] def merge_transcripts(transcripts): """ 将多段识别文本合并成完整句子（保留原始段落逻辑，用空格连接） 参数: transcripts (list): 识别结果列表，每个元素为字典{'text': '识别文本'} 返回: str: 合并后的完整文本 """ # 输入参数检查 if not transcripts: return "" # 确保transcripts是可迭代对象 if not hasattr(transcripts, '__iter__'): return "" try: # 提取所有有效的text字段 texts = [] for t in transcripts: try: # 检查是否为字典类型且包含text字段 if isinstance(t, dict) and 'text' in t and t['text']: text = t['text'] # 确保text是字符串类型 if isinstance(text, str) and text.strip(): texts.append(text.strip()) except (KeyError, TypeError, AttributeError): # 忽略单个元素的处理错误，继续处理其他元素 continue # 用空格连接所有段落（根据实际需求可调整连接符） return " ".join(texts) if texts else "" except Exception as e: logger.error(f"合并转录文本时发生错误: {e}") return "" def text_correction(text): """ 使用通义千问模型修正文本中的错误和标点符号 参数: text (str): 需要修正的文本 返回值: str: 修正后的文本 """ logger.info("开始文本修正...") # 构建修正提示词 correction_prompt = """请仔细检查以下文本，修正其中的错误： 1. 错别字和语法错误 2. 标点符号使用错误 3. 语句不通顺的地方 4. 逻辑不清晰的部分 请直接返回修正后的完整文本，不要添加任何解释说明。""" # 构建消息列表 messages = [ {"role": "system", "content": "你是一个专业的文本校对助手，擅长修正文本中的各种错误。"}, {"role": "user", "content": correction_prompt}, {"role": "user", "content": text} ] logger.info("调用通义千问模型进行文本修正...") # 调用DashScope文本生成接口 response = Generation.call( model=os.getenv('DASHSCOPE_LLM_MODEL', 'qwen-plus'), messages=messages, max_tokens=30000, temperature=0.1, # 使用较低的温度以提高确定性 top_p=0.5 ) # 检查API调用是否成功 if response.status_code != 200: logger.warning(f"❌ 文本修正API调用失败: {response.message}") raise Exception(f"文本修正API调用失败: {response.message}") logger.info("✓ 文本修正完成") # 返回修正后的文本 return response.output.text def analyze_and_correct_text(text): """ 分析文本并自动修正错误 参数: text (str): 待分析和修正的文本 prompt (str): 分析提示词 返回值: tuple: (修正后的文本, 分析结果) """ logger.info("开始文本分析和修正流程...") # 首先修正文本错误 corrected_text = text_correction(text) if corrected_text is None: logger.warning("文本修正返回 None，使用原始文本") corrected_text = text logger.info(f"原始文本长度: {len(text)}") logger.info(f"修正后文本长度: {len(corrected_text)}") # 使用修正后的文本进行分析 # analysis_result = analyze_text(corrected_text, prompt) return corrected_text def analyze_text(text, prompt): """ 使用通义千问模型分析文本 参数: text (str): 待分析文本 prompt (str): 分析提示词 返回值: str: 分析结果 """ logger.info("开始文本分析...") # 设置系统提示 system_prompt = "你是一个专业的文本分析助手，擅长根据提示词对长文本进行深入分析。" # 构建消息列表 messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": prompt}, {"role": "user", "content": text} ] logger.info("调用通义千问模型进行文本分析...") # 调用DashScope文本生成接口 response = Generation.call( model=os.getenv('DASHSCOPE_LLM_MODEL', 'qwen-plus'), messages=messages, max_tokens=8190, # 控制生成文本的最大长度 temperature=0.3, # 控制生成文本的确定性 top_p=0.7 # 控制生成文本的多样性 ) # 检查API调用是否成功 if response.status_code != 200: logger.error(f"❌ API调用失败: {response.message}") raise Exception(f"API调用失败: {response.message}") logger.info("✓ 文本分析完成") # 返回分析结果 return response.output.text def process_long_audio(mp3_path, output_folder, date_str): """ 处理长音频文件，分割、识别并分析 参数: mp3_path (str): MP3文件路径 prompt (str): 分析提示词 output_folder (str): 输出文件夹路径 返回值: str: 分析结果 """ logger.info("开始处理长音频...") # 转换MP3为WAV格式 logger.info("步骤1/4: 转换MP3为WAV格式") wav_path = convert_mp3_to_wav( mp3_path, os.path.join(output_folder, "input.wav") ) # 分割音频 # 可以选择固定分片或智能静音分割 # chunks = split_audio_by_fixed_duration(wav_path, 180000, output_folder) logger.info("步骤2/4: 智能静音分割音频") chunks = split_audio_by_smart_silence( wav_path, 700, -40, output_folder ) # 存储所有识别文本 transcribed_text = "" # 识别每个分片 logger.info(f"步骤3/4: 开始识别音频分片，共{len(chunks)}个分片") for i, chunk_path in enumerate(chunks): try: logger.info(f"识别进度: {i+1}/{len(chunks)} ({((i+1)/len(chunks)*100):.1f}%)") # 调用音频识别API text = transcribe_audio(chunk_path) """ if not text[1].startswith('今天的新闻联播节目播送完毕'): prompt='请分析所给文本的新闻内容，返回一个简短标题' text.append(analyze_text(text[1],prompt)) else: text.append('') """ # 新闻标题留空 text.append('') # 添加到总文本 #transcribed_text += text + "\n" # 删除临时文件 os.remove(chunk_path) # 初始化数据库连接 db = MySQLDB() # 使用默认参数连接数据库 try: # 插入数据示例 user_data = { "news_days": date_str, "daily_sub_id": i, "news_raw": text[0], "news_improve": text[1], "news_title": text[2] } user_id = db.insert_data("xwlb_daily", user_data) finally: # 关闭连接 db.close() except Exception as e: logger.error(f"识别失败: {chunk_path}, 错误: {e}") # 可以在这里添加重试逻辑 # 分析识别文本 """ print("步骤4/4: 分析识别文本") print(f"识别文本长度: {len(transcribed_text)}") print(f"识别文本内容: {transcribed_text}") if len(transcribed_text) < 1: print("识别文本为空，跳过分析处理") return "识别文本为空，无法进行分析" analysis_result = analyze_text(transcribed_text, prompt) """ logger.info("✓ 长音频处理完成") # 返回分析结果 return "" # 使用示例 if __name__ == "__main__": # MP3文件路径 mp3_path = "20251002.mp3" # 分析提示词 prompt = "请总结这段由中国中央电视台新闻联播音频转为文字的文本，理解其主要内容并提取其中的关键信息。" # 输出文件夹 output_folder = "audio_processing" # 处理长音频 try: result = process_long_audio( mp3_path, prompt, output_folder ) # 打印分析结果 print("分析结果:\n") print(result) except Exception as e: print(f"处理失败: {e}")