PCNUSMSE/transcript_service
0
1o
2 E��ha@ � @ s2 d Z ddlZddlZddlZddlmZmZ ddlmZmZ ddl m3Z4 ddl m Z ddl
mZmZmZmZmZ ddlmZ d d5lmZ d d lmZ d d lmZ d d
lm Z d dl m Z G dd� de6�Z G dd� de7�Z eG dd� d��Z!eG dd� d��Z"eG dd� d��Z#G dd� d�Z$da%de$fd d �Z&dS ) uT 任务管理模块8 9提供任务状态跟踪、进度管理和任务队列功能。10� N)� dataclass�field)�datetime� timedelta)�Enum)�Path)�Dict�List�Optional�Callable�Any)�ThreadPoolExecutor� )�11get_config)�get_task_logger)�get_file_validator)�get_oss_service)�get_paraformer_servicec @ s, e Zd ZdZdZdZdZdZdZdZ dZ12d S )13�14TaskStatusu 任务状态�pending�15validating� uploading� transcribing� completed�failed� cancelledN) �__name__�16__module__� __qualname__�__doc__�PENDING�17VALIDATING� UPLOADING� TRANSCRIBING� COMPLETED�FAILED� CANCELLED� r' r' �F/Users/chengpeng/MyProject/transcript_service/src/core/task_manager.pyr s r c @ s e Zd ZdZdZdZdZdZdS )� TaskPriorityu 任务优先级� r � � N)r r r r ZLOW�NORMALZHIGHZURGENTr' r' r' r( r) " s r) c @ sf e Zd ZU dZdZeed< dZeed< dZ eed< dZ18eed< d Z e ed19< ddededefd d
�Z
d S )� TaskProgressu 任务进度信息� �stager �current�d �total�message� �20percentageNc C sT |dur|| _ |dur|| _|dur|| _| jdkr(td| j | j d �| _dS dS )u 更新进度信息Nr g Y@r2 )r1 r3 r4 �minr6 )�selfr1 r3 r4 r' r' r( �update3 s 21 �zTaskProgress.update)NNN)r r r r r0 �str�__annotations__r1 �intr3 r4 r6 �floatr9 r' r' r' r( r. * s 22 r. c @ s� e Zd ZU dZdZeed< dZee ed< dZ23ee ed< e e
d�Zee ed< e e
d�Zee ed < dZee ed24< d Zeed < d
e fdd�ZdS )�25TaskResultu 任务结果F�successN�data�
error_message��default_factory�processed_files� failed_files�transcription_resultsr5 �duration�returnc C s" | j | j| j| j| j| j| jd�S )� 转换为字典格式�r? r@ rA rD rE rF rG rJ �r8 r' r' r( �to_dictK s �zTaskResult.to_dict)r r r r r? �boolr; r@ r26 r rA r: r �listrD r rE rF rG r= rL r' r' r' r( r> @ s 27 r> c @ s� e Zd ZU dZedd� d�Zeed< ej Z28eed< e j Z
e ed< eed�Zee ed< eed�Zeed < eed�Zeed29< eejd�Zeed < d Zee ed
< d Zee ed< d Z ee ed< ee d�Z e ee!f ed< de fdd�Z"d S )�Tasku 任务信息c C s t t�� �d d� S )N� )r: �uuid�uuid4r' r' r' r( �<lambda>[ � z
Task.<lambda>rB �id�status�priority�30file_paths�progress�result�31created_atN�32started_at� completed_at�callback�metadatarH c C s� | j | jj| jjt| j�dd� | jD �| jj| jj| jj | jj33| jj d�| j �
� | jr0| j�� nd| jr9| j�� nd| jrB| j�� nd| jd� S )rI c S s g | ]}|j �qS r' )�name)�.0�fpr' r' r( �34<listcomp>n s z Task.to_dict.<locals>.<listcomp>)r0 r1 r3 r6 r4 N) rU rV rW Z35file_countZ36file_namesrY rZ r[ r\ r] r_ )rU rV �valuerW �lenrX rY r0 r1 r3 r6 r4 rZ rL r[ � isoformatr\ r] r_ rK r' r' r( rL g s" ��z Task.to_dict)#r r r r r rU r: r; r r rV r) r- rW rN rX r r r. rY r> rZ r �nowr[ r\ r37 r] r^ r �dictr_ r r rL r' r' r' r( rO X s 38 rO c @ s4 e Zd ZdZdd� Zdefdd�Zdefdd �Ze j39d40fd e e d e d
e
fdd�Zde
d
ee fdd�Zd
e e fdd�Zded
e e fdd�Zde
d
efdd�Zdd� Zd d � Zd d � Zdefd d!�Zded
efd"d#�Zded e e d
e e fd$d%�Z ded&e e
d
efd'd(�Z d/d*e fd+d,�Z d
e fd-d.�Z!d41S )0� TaskManageru 任务管理器c C sp t � | _tdd�| _i | _tj| jjjd�| _ t42� | _ t � | _
t� | _t| jjjd�| _g | _d| _| �� dS )u 初始化任务管理器ztranscript_service.task)� logger_name)�maxsize)Z max_workersFN)r �configr �logger�tasks�asyncio�Queue�task�43queue_size�44task_queuer �file_validatorr � oss_servicer �paraformer_servicer
�app�concurrent_tasks�executor�status_callbacks�_processor_started�_start_task_processorrK r' r' r( �__init__� s zTaskManager.__init__r^ c C s | j �|� dS )ua 添加状态变化回调函数45 46 Args:47 callback: 回调函数48 N)rz �append)r8 r^ r' r' r( �add_status_callback� s z TaskManager.add_status_callbackrq c C sV | j D ]%}z||� W q ty( } z| j�dt|�� �� W Y d}~qd}~ww dS )u 通知状态变化u 回调函数执行失败: N)rz � Exceptionrm �errorr: )r8 rq r^ �er' r' r( �_notify_status_change� s 49 "���z!TaskManager._notify_status_changeNrX rW rH c � sf �| j s| �� t|||pi d�}|| j|j< | j�|j�I dH | j�d|j� dt |�� �� |jS )u� 创建新任务50 51 Args:52 file_paths: 文件路径列表53 priority: 任务优先级54 metadata: 任务元数据55 56 Returns:57 任务ID58 )rX rW r_ Nu 创建任务: u , 文件数量: )59r{ �_ensure_processor_startedrO rn rU rs �putrm �infore )r8 rX rW r_ rq r' r' r( � create_task� s � � zTaskManager.create_task�task_idc C s | j �|�S )u� 获取任务信息60 61 Args:62 task_id: 任务ID63 64 Returns:65 任务对象66 )rn �get)r8 r� r' r' r( �get_task� s zTaskManager.get_taskc C s t | j�� �S )u 获取所有任务)rN rn �valuesrK r' r' r( �
get_all_tasks� s zTaskManager.get_all_tasksrV c s � fdd�| j �� D �S )u 根据状态获取任务c s g | ] }|j � kr|�qS r' �rV )ra rq r� r' r( rc � s z3TaskManager.get_tasks_by_status.<locals>.<listcomp>)rn r� )r8 rV r' r� r( �get_tasks_by_status� s z TaskManager.get_tasks_by_statusc � sh �| � |�}|s67dS |jtjtjtjfv rdS tj|_t�� |_d|j _68| � |� | j �
d|� �� dS )u� 取消任务69 70 Args:71 task_id: 任务ID72 73 Returns:74 是否成功取消75 Fu 任务已取消u 任务已取消: T)r� rV r r$ r% r& r rg r] rY r4 r� rm r� )r8 r� rq r' r' r( � cancel_task� s �76 7778zTaskManager.cancel_taskc C sL z
t �� }t �| �� � W n ty | j�d� d| _Y dS w d| _dS )u 启动任务处理器uB 没有运行的事件循环，任务处理器将在需要时启动FTN)ro �get_running_loopr� �_process_tasks� RuntimeErrorrm �debugr{ �r8 �loopr' r' r( r| � s �79z!TaskManager._start_task_processorc C sN | j s%zt�� }t�| �� � d| _ W dS ty$ | j�d� Y dS w dS )u 确保任务处理器已启动Tu9 无法启动任务处理器：没有运行的事件循环N)r{ ro r� r� r� r� rm �warningr� r' r' r( r� s ��z%TaskManager._ensure_processor_startedc 80 � s� � z+| j �� I dH }| �|�}|r|jtjkr | j �� W q| �|�I dH | j �� W n% tyR } z| j �81dt |�� �� t �
d�I dH W Y d}~nd}~ww q)u 处理任务队列TNu# 处理任务队列时发生错误: r* )rs r� r� rV r r& � task_done�
_execute_taskr� rm � exceptionr: ro �sleep)r8 r� rq r� r' r' r( r� 82 s �8384 ���zTaskManager._process_tasksc 85 � s� ��z@z�| j �|j� tj|_t�� |_d|j _86|j � ddd� | � |� | �
|�I dH \}}|sUtj|_d|j_dd� |D �|j_t�� |_| � |� W W | j �� dS tj|_d |j _87|j � dt|�d88� | � |� | �||�I dH }d d� |D �}|s�tj|_d |j_t�� |_| � |� W W | j �� dS tj|_d
|j _89|j � ddd� | � |� dd� |D �}| �||�I dH \}}} t�� |_|j|j �� |j_|r�tj|_d|j_ ||j_ dd� |D �|j_ |j � ddd� ntj|_| |j_| � |� W n4 t �y2 }90 z'tj|_dt |91�� �|j_t�� |_| j �!d|j� �� | � |� W Y d}92~93n
d}94~95ww W | j �� dS W | j �� dS | j �� w )uK 执行任务96 97 Args:98 task: 任务对象99 u 文件验证r r2 u 开始验证文件Nu 没有有效的文件c S s g | ]}t |d ��qS �r )r: )ra �fr' r' r( rc 3 � z-TaskManager._execute_task.<locals>.<listcomp>u 文件上传u 开始上传文件到OSSc S � g | ]}|d r|�qS �r* r' �ra �rr' r' r( rc ? r� u 文件上传失败u 语音转录u 开始语音转录c S � g | ]}|d �qS )r r' r� r' r' r( rc N rT Tc S r� r� r' r� r' r' r( rc Y rT � 转录完成u 任务执行失败: u 执行任务时发生错误: )"rm Z set_task_idrU r r! rV r rg r\ rY r0 r9 r� �_validate_filesr% rZ rA rE r] Z
clear_task_idr" re �
_upload_filesr# �_transcribe_audio�
total_secondsrG r$ r? rF rD r� r: r� ) r8 rq � valid_files�
invalid_filesZupload_resultsZsuccessful_uploads� file_urlsr? Ztranscription_resultr� r� r' r' r( r� sv �1001011021032�104105106"�107108109��� �zTaskManager._execute_taskc � st �| j �dt|j�� d�� | j�|j�\}}|j�dddt|�� d�� | j �dt|�� dt|�� d�� ||fS ) u 验证文件u
开始验证 �110 个文件r2 u 验证完成: u 个有效文件u 文件验证完成: u 个有效文件, u 个无效文件)rm r� re rX rt Zvalidate_multiple_filesrY r9 )r8 rq r� r� r' r' r( r� j s � "zTaskManager._validate_filesc 111 � s� �| j �dt|�� d�� g }t|�D ]@\}}|jtjkr n5| j�||j �I dH \}}}|�112|j |||f� |j �
|d t|�d|d � dt|�� d�� | �|� q| j �dtdd � |D ��� d113�� |S ) u 上传文件u
开始上传 r� Nr* u114 已上传 �/u 文件上传完成: c S r� r� r' r� r' r' r( rc � r� z-TaskManager._upload_files.<locals>.<listcomp>u115 个成功)rm r� re � enumeraterV r r& ru Z upload_filerU r~ r` rY r9 r� ) r8 rq rX �results�i� file_pathr? Z url_or_error�116object_keyr' r' r( r� u s � . "zTaskManager._upload_filesr� c � s� �| j �dt|�� d�� d}d|jv r"|jd }| j �d|� �� | j�||j|�I dH \}}}|rH|j�ddd� | j �dt|�� d �� n | j � d117|� �� |||fS ) u 转录音频u
开始转录 u 个音频文件N�paraformer_paramsu! 使用自定义Paraformer参数: r2 r� u 转录完成: r� u 转录失败: )118rm r� re r_ rv Zbatch_process_with_retryrU rY r9 r� )r8 rq r� r� r? r� r� r' r' r( r� � s �119120�121z TaskManager._transcribe_audio� �hoursc C s� t �� t|d� }g }| j�� D ] \}}|jtjtjtj fv r,|j122r,|j123|k r,|� |� q|D ]}| j|= q/| j �
dt|�� d�� dS )ud 清理已完成的任务124 125 Args:126 hours: 保留时间（小时）127 )r� u128 清理了 u 个过期任务N)r rg r rn �itemsrV r r$ r% r& r] r~ rm r� re )r8 r� Z cutoff_time� to_remover� rq r' r' r( �cleanup_completed_tasks� s �129130�131 z#TaskManager.cleanup_completed_tasksc C s| t | j�t | �tj��t | �tj��t | �tj��t | �tj��t | �tj��t | �tj ��t | �tj132��| j � � d� }|S )u 获取任务统计信息) Z total_tasksr r r r r r r rr )
re rn r� r r r! r" r# r$ r% r& rs �qsize)r8 �statsr' r' r( �get_statistics� s � zTaskManager.get_statistics)r� )"r r r r r} r r rO r� r) r- r r r: r� r133 r� r� r r� rM r� r| r� r� r� �tupler� r� r� r< r� r r� r' r' r' r( ri ~ s& "
134L ri rH c C s t du rt� a t S )uP 获取任务管理器实例135 136 Returns:137 任务管理器实例138 N)� task_managerri r' r' r' r( �get_task_manager� s r� )'r ro �timerQ � dataclassesr r r r �enumr �pathlibr �typingr r r139 r r �concurrent.futuresr
Z core.configr Z utils.loggerr Zservices.file_validatorr Zservices.oss_servicer Zservices.paraformer_servicer r r) r. r> rO ri r� r� r' r' r' r( �<module> s8 % F