当前位置：首页 > news >正文

Audio Pixel Studio实战教程：用Streamlit Session State管理多任务音频队列

news 2026/7/24 6:43:04

Audio Pixel Studio实战教程：用Streamlit Session State管理多任务音频队列

1. 引言

想象一下这个场景：你正在用Audio Pixel Studio处理一个音频项目，需要先合成一段语音旁白，然后从一首歌里分离出人声，最后再合成另一段说明。如果每次只能做一件事，做完一个任务再手动开始下一个，整个过程就会变得断断续续，效率低下。

这就是我们今天要解决的问题。Audio Pixel Studio作为一个功能强大的音频处理工具，在实际使用中，用户往往需要连续处理多个任务。传统的同步处理方式会让用户等待当前任务完成才能进行下一步操作，体验不够流畅。

本文将带你深入实战，学习如何利用Streamlit的核心功能——Session State，为Audio Pixel Studio构建一个智能的多任务音频队列管理系统。学完这篇教程，你将掌握：

Streamlit Session State的基本原理和核心用法
如何设计一个可靠的任务队列数据结构
实现任务的状态管理、执行和回调机制
构建用户友好的队列监控界面
处理任务执行过程中的各种边界情况

无论你是Streamlit的初学者，还是有一定经验的开发者，这篇教程都将用具体的代码示例和实际场景，带你一步步实现这个实用的功能升级。

2. 理解Streamlit Session State

2.1 为什么需要Session State

在深入代码之前，我们先要明白一个关键问题：为什么普通的Python变量在Streamlit中行不通？

Streamlit的工作方式很特别——每次用户与界面交互（比如点击按钮、输入文本），整个脚本都会从头到尾重新执行一遍。这意味着如果你用普通的Python变量来存储状态：

# 这样是行不通的！ task_queue = [] # 每次页面刷新都会被清空 def add_task(): task_queue.append("新任务") # 添加的任务在下一次交互时就消失了

你会发现，task_queue列表在每次页面刷新时都会被重新初始化为空列表，之前添加的任务全部丢失。这就是Streamlit的无状态特性带来的挑战。

Session State就是为解决这个问题而生的。它为每个用户会话提供了一个持久化的存储空间，数据在页面刷新和交互过程中得以保留。

2.2 Session State的基本用法

让我们从最简单的例子开始，看看Session State怎么用：

import streamlit as st # 初始化Session State if 'task_counter' not in st.session_state: st.session_state.task_counter = 0 # 使用Session State st.write(f"当前任务数: {st.session_state.task_counter}") # 修改Session State if st.button("添加任务"): st.session_state.task_counter += 1 st.rerun() # 触发页面刷新，显示更新后的值

这段代码展示了Session State的几个关键点：

初始化检查：使用if 'key' not in st.session_state:来确保变量只初始化一次
数据存取：通过st.session_state.key来读写数据
状态持久：即使页面刷新，task_counter的值也会保持不变

2.3 Session State的高级特性

除了基本的键值存储，Session State还有一些高级特性特别适合我们的任务队列场景：

回调函数支持

def on_task_add(): # 这个函数会在任务添加时自动执行 st.session_state.last_added = datetime.now() # 将回调函数与按钮关联 st.button("添加任务", on_click=on_task_add)

状态变更监听

# 监听特定状态的变化 if st.session_state.get('queue_changed'): # 队列发生变化时执行的操作 update_queue_display() st.session_state.queue_changed = False

复杂数据结构存储

# Session State可以存储任何Python对象 st.session_state.task_queue = { 'pending': [], # 等待中的任务 'running': None, # 正在执行的任务 'completed': [], # 已完成的任务 'failed': [] # 失败的任务 }

理解了这些基础，我们就可以开始设计音频任务队列的具体实现了。

3. 设计音频任务队列系统

3.1 定义任务数据结构

一个好的任务队列系统始于清晰的数据结构设计。对于音频处理任务，我们需要记录足够的信息来唯一标识和管理每个任务。

from dataclasses import dataclass from datetime import datetime from enum import Enum from typing import Optional, Any class TaskType(Enum): """任务类型枚举""" TTS = "语音合成" UVR = "人声分离" MERGE = "音频合并" class TaskStatus(Enum): """任务状态枚举""" PENDING = "等待中" RUNNING = "执行中" COMPLETED = "已完成" FAILED = "失败" CANCELLED = "已取消" @dataclass class AudioTask: """音频任务数据类""" task_id: str # 唯一标识符 task_type: TaskType # 任务类型 created_at: datetime # 创建时间 status: TaskStatus # 当前状态 progress: float = 0.0 # 进度(0.0-1.0) # 任务参数（根据任务类型不同而不同） params: Optional[dict] = None # 任务结果 result: Optional[Any] = None error_message: Optional[str] = None completed_at: Optional[datetime] = None def to_dict(self): """转换为字典，便于存储和显示""" return { 'id': self.task_id, 'type': self.task_type.value, 'created': self.created_at.strftime("%H:%M:%S"), 'status': self.status.value, 'progress': f"{self.progress*100:.1f}%", 'has_result': self.result is not None }

这个AudioTask类定义了每个任务的核心属性：

task_id：使用UUID或时间戳确保唯一性
task_type：区分是语音合成还是人声分离
status：跟踪任务的生命周期状态
progress：实时显示执行进度
params和result：存储任务输入和输出

3.2 设计队列管理器

有了任务对象，接下来需要设计一个管理器来协调多个任务的执行：

class AudioTaskQueue: """音频任务队列管理器""" def __init__(self): """初始化队列""" if 'task_queue' not in st.session_state: st.session_state.task_queue = { 'tasks': {}, # 所有任务，按ID索引 'pending_ids': [], # 等待执行的任务ID列表 'running_id': None, # 当前执行的任务ID 'max_concurrent': 1, # 最大并发数（音频处理通常串行） 'history_size': 50 # 保留的历史任务数量 } self.queue = st.session_state.task_queue def add_task(self, task_type: TaskType, params: dict) -> str: """添加新任务到队列""" import uuid from datetime import datetime # 生成唯一任务ID task_id = f"{task_type.name}_{uuid.uuid4().hex[:8]}" # 创建任务对象 task = AudioTask( task_id=task_id, task_type=task_type, created_at=datetime.now(), status=TaskStatus.PENDING, params=params ) # 存储任务 self.queue['tasks'][task_id] = task self.queue['pending_ids'].append(task_id) # 触发状态更新 st.session_state.queue_updated = True return task_id def get_next_task(self) -> Optional[AudioTask]: """获取下一个待执行的任务""" if not self.queue['pending_ids']: return None if self.queue['running_id'] is not None: # 已经有任务在执行，串行处理 return None # 取出下一个任务 task_id = self.queue['pending_ids'].pop(0) task = self.queue['tasks'][task_id] # 更新状态 task.status = TaskStatus.RUNNING self.queue['running_id'] = task_id return task def complete_task(self, task_id: str, result: Any = None): """标记任务完成""" if task_id in self.queue['tasks']: task = self.queue['tasks'][task_id] task.status = TaskStatus.COMPLETED task.progress = 1.0 task.result = result task.completed_at = datetime.now() # 清理运行状态 if self.queue['running_id'] == task_id: self.queue['running_id'] = None # 限制历史记录大小 self._cleanup_history() def fail_task(self, task_id: str, error_msg: str): """标记任务失败""" if task_id in self.queue['tasks']: task = self.queue['tasks'][task_id] task.status = TaskStatus.FAILED task.error_message = error_msg if self.queue['running_id'] == task_id: self.queue['running_id'] = None def cancel_task(self, task_id: str): """取消任务""" if task_id in self.queue['tasks']: task = self.queue['tasks'][task_id] if task.status == TaskStatus.PENDING: # 从等待队列中移除 if task_id in self.queue['pending_ids']: self.queue['pending_ids'].remove(task_id) task.status = TaskStatus.CANCELLED elif task.status == TaskStatus.RUNNING: # 运行中的任务需要特殊处理 task.status = TaskStatus.CANCELLED self.queue['running_id'] = None def _cleanup_history(self): """清理过期的历史任务""" all_tasks = list(self.queue['tasks'].items()) if len(all_tasks) > self.queue['history_size']: # 按完成时间排序，移除最早的任务 completed_tasks = [ (task_id, task) for task_id, task in all_tasks if task.status in [TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.CANCELLED] ] # 按完成时间排序 completed_tasks.sort(key=lambda x: x[1].completed_at or x[1].created_at) # 移除最早的任务，直到数量符合要求 while len(all_tasks) > self.queue['history_size'] and completed_tasks: task_id, _ = completed_tasks.pop(0) del self.queue['tasks'][task_id] all_tasks = list(self.queue['tasks'].items()) def get_queue_status(self) -> dict: """获取队列状态统计""" tasks = list(self.queue['tasks'].values()) return { 'total': len(tasks), 'pending': len([t for t in tasks if t.status == TaskStatus.PENDING]), 'running': 1 if self.queue['running_id'] else 0, 'completed': len([t for t in tasks if t.status == TaskStatus.COMPLETED]), 'failed': len([t for t in tasks if t.status == TaskStatus.FAILED]), 'cancelled': len([t for t in tasks if t.status == TaskStatus.CANCELLED]) }

这个队列管理器提供了完整的任务生命周期管理：

任务添加：支持不同类型的音频任务
任务调度：自动从等待队列中取出任务执行
状态管理：跟踪任务的执行、完成、失败等状态
历史清理：自动管理历史记录，避免内存溢出
状态统计：提供实时的队列状态信息

3.3 集成到Audio Pixel Studio

现在我们需要把这个队列系统集成到现有的Audio Pixel Studio应用中。关键是要保持原有功能的完整性，同时增加队列管理能力。

import streamlit as st import asyncio from concurrent.futures import ThreadPoolExecutor import edge_tts import librosa import soundfile as sf import numpy as np from pathlib import Path import tempfile class AudioPixelStudioWithQueue: """增强版的Audio Pixel Studio，支持任务队列""" def __init__(self): """初始化应用""" st.set_page_config( page_title="Audio Pixel Studio Pro", page_icon="🎵", layout="wide" ) # 初始化任务队列 self.task_queue = AudioTaskQueue() # 创建线程池用于异步任务执行 if 'thread_pool' not in st.session_state: st.session_state.thread_pool = ThreadPoolExecutor(max_workers=1) # 初始化任务执行器状态 if 'task_runner_active' not in st.session_state: st.session_state.task_runner_active = False # 启动任务执行器 self._start_task_runner() def _start_task_runner(self): """启动后台任务执行器""" if not st.session_state.task_runner_active: st.session_state.task_runner_active = True # 使用Streamlit的异步支持 import threading def run_task_loop(): """任务执行循环""" while st.session_state.task_runner_active: # 获取下一个任务 task = self.task_queue.get_next_task() if task: try: # 执行任务 self._execute_task(task) except Exception as e: # 任务执行失败 self.task_queue.fail_task(task.task_id, str(e)) # 短暂休眠，避免CPU占用过高 import time time.sleep(0.1) # 启动后台线程 thread = threading.Thread(target=run_task_loop, daemon=True) thread.start() def _execute_task(self, task: AudioTask): """执行具体的音频任务""" try: # 更新进度 task.progress = 0.1 self._update_task_display() if task.task_type == TaskType.TTS: result = self._execute_tts_task(task) elif task.task_type == TaskType.UVR: result = self._execute_uvr_task(task) else: raise ValueError(f"未知的任务类型: {task.task_type}") # 标记任务完成 task.progress = 1.0 self.task_queue.complete_task(task.task_id, result) except Exception as e: # 任务执行失败 self.task_queue.fail_task(task.task_id, str(e)) def _execute_tts_task(self, task: AudioTask) -> str: """执行TTS任务""" params = task.params # 提取参数 text = params.get('text', '') voice = params.get('voice', 'zh-CN-XiaoxiaoNeural') rate = params.get('rate', '+0%') # 更新进度 task.progress = 0.3 self._update_task_display() # 执行TTS communicate = edge_tts.Communicate(text, voice, rate=rate) # 创建临时文件保存结果 with tempfile.NamedTemporaryFile(suffix='.mp3', delete=False) as tmp_file: output_path = tmp_file.name # 更新进度 task.progress = 0.6 self._update_task_display() # 保存音频文件 async def save_audio(): await communicate.save(output_path) # 运行异步函数 import asyncio asyncio.run(save_audio()) # 更新进度 task.progress = 0.9 self._update_task_display() return output_path def _execute_uvr_task(self, task: AudioTask) -> dict: """执行人声分离任务""" params = task.params input_path = params.get('audio_path') if not input_path or not Path(input_path).exists(): raise FileNotFoundError(f"音频文件不存在: {input_path}") # 更新进度 task.progress = 0.2 self._update_task_display() # 加载音频文件 y, sr = librosa.load(input_path, sr=None, mono=False) # 如果是立体声，转换为单声道 if len(y.shape) > 1: y = librosa.to_mono(y) # 更新进度 task.progress = 0.4 self._update_task_display() # 执行简单的人声分离（这里使用简化的频谱方法） # 在实际应用中，你可以替换为更复杂的UVR5算法 # 计算短时傅里叶变换 D = librosa.stft(y) # 更新进度 task.progress = 0.6 self._update_task_display() # 简单的频谱掩码分离（示例方法） magnitude, phase = librosa.magphase(D) # 创建掩码（这里使用简化的方法） # 实际应用中应该使用训练好的模型 mask = np.ones_like(magnitude) * 0.5 # 分离人声和伴奏 vocals_mag = magnitude * mask accompaniment_mag = magnitude * (1 - mask) # 更新进度 task.progress = 0.8 self._update_task_display() # 重建音频 vocals_stft = vocals_mag * phase accompaniment_stft = accompaniment_mag * phase vocals = librosa.istft(vocals_stft) accompaniment = librosa.istft(accompaniment_stft) # 保存结果 output_dir = Path("logs") / "uvr_output" output_dir.mkdir(exist_ok=True) base_name = Path(input_path).stem vocals_path = output_dir / f"{base_name}_vocals.wav" accompaniment_path = output_dir / f"{base_name}_accompaniment.wav" sf.write(str(vocals_path), vocals, sr) sf.write(str(accompaniment_path), accompaniment, sr) return { 'vocals_path': str(vocals_path), 'accompaniment_path': str(accompaniment_path), 'sample_rate': sr } def _update_task_display(self): """更新任务显示状态""" # 设置一个标志，让Streamlit知道需要更新显示 st.session_state.queue_updated = True def render_ui(self): """渲染用户界面""" # 应用标题 st.title("🎵 Audio Pixel Studio Pro") st.markdown("### 高质量语音合成 · 智能人声分离 · 多任务队列管理") # 创建标签页 tab1, tab2, tab3 = st.tabs(["🎙️ 语音合成", "🎤 人声分离", "📊 任务队列"]) with tab1: self._render_tts_tab() with tab2: self._render_uvr_tab() with tab3: self._render_queue_tab() # 如果有任务更新，触发重新运行 if st.session_state.get('queue_updated', False): st.session_state.queue_updated = False st.rerun() def _render_tts_tab(self): """渲染语音合成标签页""" st.header("语音合成设置") # 文本输入 text = st.text_area( "输入要合成的文本", height=150, placeholder="在这里输入你想要转换为语音的文本内容..." ) col1, col2 = st.columns(2) with col1: # 音色选择 voice_options = { "晓晓 (女声)": "zh-CN-XiaoxiaoNeural", "云希 (女声)": "zh-CN-YunxiNeural", "云扬 (男声)": "zh-CN-YunyangNeural", "晓辰 (女声)": "zh-CN-XiaochenNeural" } voice = st.selectbox("选择音色", list(voice_options.keys())) voice_code = voice_options[voice] with col2: # 语速调整 rate = st.slider("语速调整", -50, 50, 0, format="%d%%") rate_str = f"{'+' if rate >= 0 else ''}{rate}%" # 操作按钮 col1, col2, col3 = st.columns(3) with col1: if st.button("🎯 立即合成", type="primary", use_container_width=True): if text.strip(): # 创建TTS任务 task_id = self.task_queue.add_task( TaskType.TTS, { 'text': text, 'voice': voice_code, 'rate': rate_str } ) st.success(f"已添加TTS任务: {task_id}") else: st.warning("请输入要合成的文本") with col2: if st.button("📝 添加到队列", use_container_width=True): if text.strip(): task_id = self.task_queue.add_task( TaskType.TTS, { 'text': text, 'voice': voice_code, 'rate': rate_str } ) st.success(f"已添加到队列: {task_id}") else: st.warning("请输入要合成的文本") with col3: if st.button("🔄 清空文本", use_container_width=True): st.rerun() def _render_uvr_tab(self): """渲染人声分离标签页""" st.header("人声分离设置") # 文件上传 audio_file = st.file_uploader( "上传音频文件", type=['mp3', 'wav', 'ogg', 'flac'], help="支持 MP3, WAV, OGG, FLAC 格式" ) if audio_file is not None: # 保存上传的文件 upload_dir = Path("logs") / "uploads" upload_dir.mkdir(exist_ok=True) file_path = upload_dir / audio_file.name with open(file_path, "wb") as f: f.write(audio_file.getbuffer()) st.success(f"已上传: {audio_file.name}") # 显示音频信息 col1, col2 = st.columns(2) with col1: st.audio(audio_file.getvalue()) with col2: st.info(f""" **文件信息** - 文件名: {audio_file.name} - 文件大小: {audio_file.size / 1024:.1f} KB - 类型: {audio_file.type} """) # 分离选项 st.subheader("分离设置") col1, col2 = st.columns(2) with col1: separation_strength = st.slider( "分离强度", min_value=1, max_value=10, value=5, help="数值越大，人声和伴奏分离越彻底" ) with col2: output_format = st.selectbox( "输出格式", ["WAV (高质量)", "MP3 (较小体积)"] ) # 操作按钮 col1, col2 = st.columns(2) with col1: if st.button("🎛️ 立即分离", type="primary", use_container_width=True): task_id = self.task_queue.add_task( TaskType.UVR, { 'audio_path': str(file_path), 'separation_strength': separation_strength, 'output_format': output_format } ) st.success(f"已添加UVR任务: {task_id}") with col2: if st.button("📥 添加到队列", use_container_width=True): task_id = self.task_queue.add_task( TaskType.UVR, { 'audio_path': str(file_path), 'separation_strength': separation_strength, 'output_format': output_format } ) st.success(f"已添加到队列: {task_id}") def _render_queue_tab(self): """渲染任务队列标签页""" st.header("📊 任务队列管理") # 队列状态统计 status = self.task_queue.get_queue_status() # 显示统计卡片 cols = st.columns(5) metrics = [ ("总任务数", status['total'], "📋"), ("等待中", status['pending'], "⏳"), ("执行中", status['running'], "⚡"), ("已完成", status['completed'], "✅"), ("失败", status['failed'], "❌") ] for col, (label, value, icon) in zip(cols, metrics): with col: st.metric(label, value) # 任务列表 st.subheader("任务列表") # 获取所有任务 tasks = list(self.task_queue.queue['tasks'].values()) if not tasks: st.info("当前没有任务") return # 按状态分组显示 status_groups = { TaskStatus.RUNNING: "执行中的任务", TaskStatus.PENDING: "等待中的任务", TaskStatus.COMPLETED: "已完成的任务", TaskStatus.FAILED: "失败的任务", TaskStatus.CANCELLED: "已取消的任务" } for status_type, group_title in status_groups.items(): group_tasks = [t for t in tasks if t.status == status_type] if group_tasks: st.markdown(f"### {group_title}") for task in group_tasks: self._render_task_card(task) def _render_task_card(self, task: AudioTask): """渲染单个任务卡片""" # 根据任务类型选择图标 type_icons = { TaskType.TTS: "🎙️", TaskType.UVR: "🎤", TaskType.MERGE: "🔀" } # 根据状态选择颜色 status_colors = { TaskStatus.PENDING: "gray", TaskStatus.RUNNING: "blue", TaskStatus.COMPLETED: "green", TaskStatus.FAILED: "red", TaskStatus.CANCELLED: "orange" } icon = type_icons.get(task.task_type, "📄") color = status_colors.get(task.status, "gray") # 创建任务卡片 with st.container(): col1, col2, col3, col4 = st.columns([1, 2, 1, 1]) with col1: st.markdown(f"### {icon}") st.caption(task.task_type.value) with col2: st.markdown(f"**ID:** `{task.task_id}`") st.caption(f"创建时间: {task.created_at.strftime('%H:%M:%S')}") if task.status == TaskStatus.RUNNING: # 显示进度条 progress = task.progress st.progress(progress) st.caption(f"进度: {progress*100:.1f}%") with col3: # 状态标签 status_text = task.status.value st.markdown( f'<span style="color: {color}; font-weight: bold;">{status_text}</span>', unsafe_allow_html=True ) if task.completed_at: st.caption(f"完成: {task.completed_at.strftime('%H:%M:%S')}") with col4: # 操作按钮 if task.status == TaskStatus.PENDING: if st.button("❌", key=f"cancel_{task.task_id}", help="取消任务"): self.task_queue.cancel_task(task.task_id) st.rerun() elif task.status == TaskStatus.COMPLETED and task.result: # 显示结果 if task.task_type == TaskType.TTS: audio_path = task.result if Path(audio_path).exists(): st.audio(audio_path) elif task.task_type == TaskType.UVR: result = task.result if st.button("📥", key=f"download_{task.task_id}", help="下载结果"): # 这里可以添加下载逻辑 st.success("下载开始") elif task.status == TaskStatus.FAILED: if task.error_message: st.error("失败") with st.expander("查看错误详情"): st.code(task.error_message) st.divider() # 运行应用 if __name__ == "__main__": app = AudioPixelStudioWithQueue() app.render_ui()

这个完整的实现展示了如何将队列系统集成到Audio Pixel Studio中。关键改进包括：

后台任务执行器：自动从队列中取出任务执行
进度实时更新：任务执行过程中实时更新进度条
状态持久化：所有任务状态在页面刷新后保持不变
用户友好界面：清晰的任务状态显示和操作按钮
错误处理：完善的任务失败处理和错误信息展示

4. 高级功能与优化建议

4.1 任务优先级管理

在实际使用中，用户可能希望某些任务能够优先执行。我们可以为任务队列添加优先级支持：

class Priority(Enum): """任务优先级""" HIGH = 3 NORMAL = 2 LOW = 1 @dataclass class AudioTask: """增强版音频任务，支持优先级""" # ... 原有字段 ... priority: Priority = Priority.NORMAL # 添加优先级字段 class AudioTaskQueue: """支持优先级的任务队列""" def add_task(self, task_type: TaskType, params: dict, priority: Priority = Priority.NORMAL) -> str: """添加任务，支持优先级""" # ... 创建任务 ... task.priority = priority # 根据优先级插入到合适位置 self._insert_by_priority(task_id, priority) def _insert_by_priority(self, task_id: str, priority: Priority): """按优先级插入任务""" priority_value = priority.value # 找到插入位置 insert_index = 0 for i, pending_id in enumerate(self.queue['pending_ids']): pending_task = self.queue['tasks'][pending_id] if pending_task.priority.value < priority_value: insert_index = i break else: insert_index = i + 1 # 插入任务 self.queue['pending_ids'].insert(insert_index, task_id)

4.2 批量任务处理

对于需要处理大量相似任务的场景，可以添加批量处理功能：

def add_batch_tts_tasks(self, texts: List[str], voice: str, rate: str = "+0%"): """批量添加TTS任务""" task_ids = [] for i, text in enumerate(texts): task_id = self.add_task( TaskType.TTS, { 'text': text, 'voice': voice, 'rate': rate, 'batch_index': i, 'batch_total': len(texts) } ) task_ids.append(task_id) return task_ids def get_batch_progress(self, task_ids: List[str]) -> dict: """获取批量任务进度""" total = len(task_ids) completed = 0 failed = 0 running = 0 for task_id in task_ids: if task_id in self.queue['tasks']: task = self.queue['tasks'][task_id] if task.status == TaskStatus.COMPLETED: completed += 1 elif task.status == TaskStatus.FAILED: failed += 1 elif task.status == TaskStatus.RUNNING: running += 1 return { 'total': total, 'completed': completed, 'failed': failed, 'running': running, 'progress': (completed + failed) / total if total > 0 else 0 }

4.3 任务依赖关系

有些任务可能需要等待其他任务完成才能开始，比如先进行人声分离，然后对分离出的人声进行语音增强：

@dataclass class AudioTask: """支持依赖关系的任务""" # ... 原有字段 ... dependencies: List[str] = None # 依赖的任务ID列表 dependent_tasks: List[str] = None # 依赖于此任务的任务列表 class AudioTaskQueue: """支持任务依赖的队列""" def add_task_with_dependencies(self, task_type: TaskType, params: dict, dependencies: List[str] = None) -> str: """添加有依赖关系的任务""" task_id = self.add_task(task_type, params) if dependencies: task = self.queue['tasks'][task_id] task.dependencies = dependencies # 标记为等待依赖 task.status = TaskStatus.PENDING # 更新依赖任务的关系 for dep_id in dependencies: if dep_id in self.queue['tasks']: dep_task = self.queue['tasks'][dep_id] if dep_task.dependent_tasks is None: dep_task.dependent_tasks = [] dep_task.dependent_tasks.append(task_id) return task_id def _check_dependencies(self, task_id: str) -> bool: """检查任务依赖是否满足""" task = self.queue['tasks'][task_id] if not task.dependencies: return True for dep_id in task.dependencies: if dep_id in self.queue['tasks']: dep_task = self.queue['tasks'][dep_id] if dep_task.status != TaskStatus.COMPLETED: return False return True def get_next_task(self) -> Optional[AudioTask]: """获取下一个可执行的任务（考虑依赖关系）""" for task_id in self.queue['pending_ids']: task = self.queue['tasks'][task_id] # 检查依赖是否满足 if self._check_dependencies(task_id): # 从等待队列中移除 self.queue['pending_ids'].remove(task_id) # 更新状态 task.status = TaskStatus.RUNNING self.queue['running_id'] = task_id return task return None