155 lines
5.3 KiB
Python
155 lines
5.3 KiB
Python
"""
|
|
音频上传后台处理服务
|
|
|
|
将上传后的重操作放到后台线程执行，避免请求长时间阻塞：
|
|
1. 音频预处理
|
|
2. 更新音频文件记录
|
|
3. 启动转录
|
|
4. 启动自动总结监控
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from app.core.config import BACKGROUND_TASK_CONFIG, BASE_DIR
|
|
from app.services.async_transcription_service import AsyncTranscriptionService
|
|
from app.services.audio_preprocess_service import audio_preprocess_service
|
|
from app.services.audio_service import handle_audio_upload
|
|
from app.services.background_task_runner import KeyedBackgroundTaskRunner
|
|
|
|
|
|
upload_task_runner = KeyedBackgroundTaskRunner(
|
|
max_workers=max(1, int(BACKGROUND_TASK_CONFIG.get("upload_workers", 2))),
|
|
thread_name_prefix="imeeting-audio-upload",
|
|
)
|
|
|
|
|
|
class AudioUploadTaskService:
|
|
def __init__(self):
|
|
self.transcription_service = AsyncTranscriptionService()
|
|
|
|
def enqueue_upload_processing(
|
|
self,
|
|
*,
|
|
meeting_id: int,
|
|
original_file_path: str,
|
|
current_user: dict,
|
|
auto_summarize: bool,
|
|
prompt_id: Optional[int] = None,
|
|
model_code: Optional[str] = None,
|
|
) -> str:
|
|
task_id = self.transcription_service.create_local_processing_task(
|
|
meeting_id=meeting_id,
|
|
status="processing",
|
|
progress=5,
|
|
)
|
|
upload_task_runner.submit(
|
|
f"audio-upload:{task_id}",
|
|
self._process_uploaded_audio,
|
|
task_id,
|
|
meeting_id,
|
|
original_file_path,
|
|
current_user,
|
|
auto_summarize,
|
|
prompt_id,
|
|
model_code,
|
|
)
|
|
return task_id
|
|
|
|
def _process_uploaded_audio(
|
|
self,
|
|
task_id: str,
|
|
meeting_id: int,
|
|
original_file_path: str,
|
|
current_user: dict,
|
|
auto_summarize: bool,
|
|
prompt_id: Optional[int],
|
|
model_code: Optional[str],
|
|
) -> None:
|
|
source_absolute_path = BASE_DIR / original_file_path.lstrip("/")
|
|
processed_absolute_path: Optional[Path] = None
|
|
handoff_to_audio_service = False
|
|
|
|
try:
|
|
self.transcription_service.update_local_processing_task(task_id, "processing", 15, None)
|
|
|
|
preprocess_result = audio_preprocess_service.preprocess(source_absolute_path)
|
|
processed_absolute_path = preprocess_result.file_path
|
|
audio_duration = preprocess_result.metadata.duration_seconds
|
|
file_path = "/" + str(processed_absolute_path.relative_to(BASE_DIR))
|
|
|
|
print(
|
|
f"[AudioUploadTaskService] 音频预处理完成: source={source_absolute_path.name}, "
|
|
f"target={processed_absolute_path.name}, duration={audio_duration}s, "
|
|
f"applied={preprocess_result.applied}"
|
|
)
|
|
|
|
self.transcription_service.update_local_processing_task(task_id, "processing", 40, None)
|
|
|
|
handoff_to_audio_service = True
|
|
result = handle_audio_upload(
|
|
file_path=file_path,
|
|
file_name=preprocess_result.file_name,
|
|
file_size=preprocess_result.file_size,
|
|
meeting_id=meeting_id,
|
|
current_user=current_user,
|
|
auto_summarize=auto_summarize,
|
|
background_tasks=None,
|
|
prompt_id=prompt_id,
|
|
model_code=model_code,
|
|
duration=audio_duration,
|
|
transcription_task_id=task_id,
|
|
)
|
|
|
|
if not result["success"]:
|
|
raise RuntimeError(self._extract_response_message(result["response"]))
|
|
|
|
if preprocess_result.applied and processed_absolute_path != source_absolute_path and source_absolute_path.exists():
|
|
try:
|
|
os.remove(source_absolute_path)
|
|
except OSError:
|
|
pass
|
|
|
|
except Exception as exc:
|
|
error_message = str(exc)
|
|
print(f"[AudioUploadTaskService] 音频后台处理失败, task_id={task_id}, meeting_id={meeting_id}: {error_message}")
|
|
self.transcription_service.update_local_processing_task(task_id, "failed", 0, error_message)
|
|
|
|
if handoff_to_audio_service:
|
|
return
|
|
|
|
cleanup_targets = []
|
|
if processed_absolute_path:
|
|
cleanup_targets.append(processed_absolute_path)
|
|
if source_absolute_path.exists():
|
|
cleanup_targets.append(source_absolute_path)
|
|
|
|
deduped_targets: list[Path] = []
|
|
for target in cleanup_targets:
|
|
if target not in deduped_targets:
|
|
deduped_targets.append(target)
|
|
|
|
for target in deduped_targets:
|
|
if target.exists():
|
|
try:
|
|
os.remove(target)
|
|
except OSError:
|
|
pass
|
|
|
|
@staticmethod
|
|
def _extract_response_message(response) -> str:
|
|
body = getattr(response, "body", None)
|
|
if not body:
|
|
return "音频处理失败"
|
|
try:
|
|
payload = json.loads(body.decode("utf-8"))
|
|
return payload.get("message") or "音频处理失败"
|
|
except Exception:
|
|
return "音频处理失败"
|
|
|
|
|
|
audio_upload_task_service = AudioUploadTaskService()
|