249 lines
14 KiB
Python
249 lines
14 KiB
Python
import os
|
||
import sys
|
||
import whisper
|
||
import torch
|
||
import shutil
|
||
import subprocess
|
||
import time
|
||
import re
|
||
import json
|
||
import requests
|
||
from celery import Celery, Task
|
||
from opencc import OpenCC
|
||
from moviepy import VideoFileClip
|
||
from dotenv import load_dotenv
|
||
|
||
# Load environment variables from .env file
|
||
load_dotenv()
|
||
|
||
# ========== Dify API Configuration ==========
|
||
DIFY_API_KEY = os.environ.get("DIFY_API_KEY")
|
||
DIFY_API_BASE_URL = os.environ.get("DIFY_API_BASE_URL")
|
||
|
||
# ========== Dify API Client Function ==========
|
||
def ask_dify(prompt: str, user_id: str = "default-tk-user-resume", inputs: dict = None, response_mode: str = "streaming", conversation_id: str = None, timeout_seconds: int = 1200) -> dict:
|
||
if not DIFY_API_KEY or not DIFY_API_BASE_URL:
|
||
return {"answer": "❌ 錯誤：DIFY_API_KEY 或 DIFY_API_BASE_URL 未在環境變數中設定。", "conversation_id": conversation_id}
|
||
|
||
if inputs is None:
|
||
inputs = {}
|
||
url = f"{DIFY_API_BASE_URL}/chat-messages"
|
||
headers = {"Authorization": f"Bearer {DIFY_API_KEY}", "Content-Type": "application/json"}
|
||
payload = {"inputs": inputs, "query": prompt, "user": user_id, "response_mode": response_mode}
|
||
if conversation_id:
|
||
payload["conversation_id"] = conversation_id
|
||
returned_conversation_id = conversation_id
|
||
error_from_stream_message = None
|
||
print(f"\n--- [ASK_DIFY] Sending request to Dify ---")
|
||
try:
|
||
is_streaming_request = (response_mode == "streaming")
|
||
response = requests.post(url, headers=headers, json=payload, timeout=timeout_seconds, stream=is_streaming_request)
|
||
response.raise_for_status()
|
||
if is_streaming_request:
|
||
full_answer_chunks = []
|
||
for line in response.iter_lines():
|
||
if line:
|
||
decoded_line = line.decode('utf-8')
|
||
if decoded_line.startswith("data:"):
|
||
try:
|
||
data_json_str = decoded_line[len("data:"):]
|
||
data_obj = json.loads(data_json_str)
|
||
event_type = data_obj.get("event")
|
||
if event_type == "agent_message" or event_type == "message":
|
||
if "answer" in data_obj and data_obj["answer"] is not None:
|
||
full_answer_chunks.append(data_obj["answer"])
|
||
elif event_type == "message_end":
|
||
if "conversation_id" in data_obj:
|
||
returned_conversation_id = data_obj["conversation_id"]
|
||
break
|
||
elif event_type == "error":
|
||
error_from_stream_message = data_obj.get('message', 'Dify API 返回未知的流式錯誤')
|
||
returned_conversation_id = data_obj.get("conversation_id", returned_conversation_id)
|
||
break
|
||
except json.JSONDecodeError:
|
||
pass
|
||
if error_from_stream_message:
|
||
clean_error_answer = re.sub(r"<think>.*?</think>\s*", "", error_from_stream_message, flags=re.DOTALL).strip()
|
||
return {"answer": f"❌ Dify API 流處理錯誤: {clean_error_answer}", "conversation_id": returned_conversation_id}
|
||
raw_answer = "".join(full_answer_chunks) if full_answer_chunks else "⚠️ 流式響應未包含有效回答或內容為空"
|
||
else:
|
||
response_data = response.json()
|
||
raw_answer = response_data.get("answer", "⚠️ 回應中未找到 'answer' 欄位或內容為空")
|
||
returned_conversation_id = response_data.get("conversation_id", returned_conversation_id)
|
||
clean_answer = re.sub(r"<think>.*?</think>\s*", "", raw_answer, flags=re.DOTALL).strip()
|
||
return {"answer": clean_answer, "conversation_id": returned_conversation_id}
|
||
except requests.exceptions.Timeout:
|
||
return {"answer": f"⚠️ 請求 Dify API 逾時 (超過 {timeout_seconds} 秒)", "conversation_id": conversation_id}
|
||
except requests.exceptions.HTTPError as http_err:
|
||
error_message_detail = f" - 原始響應: {http_err.response.text}"
|
||
final_error_message = f"❌ Dify API HTTP 錯誤: {http_err.response.status_code}{error_message_detail}"
|
||
return {"answer": final_error_message, "conversation_id": conversation_id}
|
||
except Exception as e:
|
||
return {"answer": f"❌ 處理 Dify API 請求或響應時發生未預期錯誤: {type(e).__name__}: {str(e)}", "conversation_id": conversation_id}
|
||
# ========== Dify API Client Function END ==========
|
||
|
||
|
||
# ========== Celery 設定 ==========
|
||
celery = Celery(
|
||
'tasks',
|
||
broker=os.environ.get('CELERY_BROKER_URL', 'redis://localhost:6379/0'),
|
||
backend=os.environ.get('CELERY_RESULT_BACKEND', 'redis://localhost:6379/0')
|
||
)
|
||
|
||
class ProgressTask(Task):
|
||
def update_progress(self, current, total, status, extra_info=None):
|
||
meta = {'current': current, 'total': total, 'status': status}
|
||
if extra_info:
|
||
meta.update(extra_info)
|
||
self.update_state(state='PROGRESS', meta=meta)
|
||
|
||
# ========== Demucs 輔助函式 ==========
|
||
def separate_vocals_with_demucs(self, audio_path, project_root):
|
||
self.update_progress(10, 100, "🎛️ 使用 Demucs 分離人聲...")
|
||
output_dir = os.path.join(project_root, 'demucs_separated')
|
||
cmd = [
|
||
sys.executable, '-m', 'demucs.separate', # <-- 將 'python' 改為 sys.executable
|
||
'-n', 'htdemucs_ft',
|
||
'--two-stems=vocals',
|
||
'-o', output_dir,
|
||
audio_path
|
||
]
|
||
try:
|
||
result = subprocess.run(cmd, check=True, capture_output=True)
|
||
print("Demucs stdout:", result.stdout)
|
||
except subprocess.CalledProcessError as e:
|
||
stderr_str = "No stderr output"
|
||
if e.stderr:
|
||
try:
|
||
stderr_str = e.stderr.decode('utf-8')
|
||
except UnicodeDecodeError:
|
||
stderr_str = e.stderr.decode(sys.getdefaultencoding(), errors='replace')
|
||
print("Demucs stderr:", stderr_str)
|
||
raise RuntimeError(f"Demucs 人聲分離失敗: {stderr_str}")
|
||
original_filename_base = os.path.splitext(os.path.basename(audio_path))[0]
|
||
vocals_path = os.path.join(output_dir, 'htdemucs_ft', original_filename_base, 'vocals.wav')
|
||
if not os.path.exists(vocals_path):
|
||
raise FileNotFoundError(f"找不到 Demucs 分離出的人聲音訊檔案: {vocals_path}")
|
||
return vocals_path
|
||
|
||
# ========== Task 1: 影片轉音訊 ==========
|
||
@celery.task(base=ProgressTask, bind=True)
|
||
def extract_audio_task(self, input_video_path, output_audio_path):
|
||
try:
|
||
self.update_progress(0, 100, "準備轉換影片...")
|
||
video_clip = VideoFileClip(input_video_path)
|
||
self.update_progress(50, 100, "正在提取音訊...")
|
||
video_clip.audio.write_audiofile(output_audio_path, codec="pcm_s16le")
|
||
video_clip.close()
|
||
self.update_progress(100, 100, "音訊提取完成！")
|
||
return {'status': '完成', 'result_path': output_audio_path}
|
||
except Exception as e:
|
||
self.update_state(state='FAILURE', meta={'exc_type': type(e).__name__, 'exc_message': str(e)})
|
||
return {'status': '錯誤', 'error': str(e)}
|
||
|
||
# ========== Task 2: 音訊轉文字 (Whisper) ==========
|
||
@celery.task(base=ProgressTask, bind=True)
|
||
def transcribe_audio_task(self, input_audio_path, output_txt_path, language, use_demucs=False):
|
||
try:
|
||
self.update_progress(0, 100, "準備開始轉錄...")
|
||
current_audio_path = input_audio_path
|
||
if use_demucs:
|
||
project_root = os.path.dirname(os.path.dirname(output_txt_path)) # uploads folder is inside project_root
|
||
current_audio_path = separate_vocals_with_demucs(self, input_audio_path, project_root)
|
||
self.update_progress(25, 100, "✅ 人聲分離完成，準備載入 Whisper...")
|
||
|
||
device = "cuda" if torch.cuda.is_available() else "cpu"
|
||
progress_after_load = 40 if use_demucs else 20
|
||
self.update_progress(progress_after_load - 10, 100, f"載入 Whisper 'medium' 模型 (使用 {device})...")
|
||
model = whisper.load_model("medium", device=device)
|
||
self.update_progress(progress_after_load, 100, "模型載入完畢，開始轉錄音訊...")
|
||
transcription_result = model.transcribe(
|
||
audio=current_audio_path,
|
||
language=language if language != 'auto' else None,
|
||
fp16=(device == "cuda"),
|
||
verbose=False
|
||
)
|
||
self.update_progress(85, 100, "轉錄完成，進行繁體轉換與格式化...")
|
||
cc = OpenCC('s2twp')
|
||
with open(output_txt_path, "w", encoding="utf-8") as f_out:
|
||
for segment in transcription_result["segments"]:
|
||
start_time_s = int(segment["start"])
|
||
end_time_s = int(segment["end"])
|
||
text_content = cc.convert(segment["text"].strip())
|
||
formatted_line = f"[{start_time_s:04d}s - {end_time_s:04d}s] {text_content}\n"
|
||
f_out.write(formatted_line)
|
||
self.update_progress(100, 100, "逐字稿完成！")
|
||
return {'status': '完成', 'result_path': output_txt_path}
|
||
except Exception as e:
|
||
import traceback
|
||
traceback.print_exc()
|
||
self.update_state(state='FAILURE', meta={'exc_type': type(e).__name__, 'exc_message': str(e)})
|
||
return {'status': '錯誤', 'error': str(e)}
|
||
|
||
# ========== Task 3: 逐段翻譯 (Dify) ==========
|
||
glossary = {"LLM": "大型語言模型", "prompt": "提示詞", "API": "應用程式介面"}
|
||
def generate_translation_prompt_for_dify(english_text_segment, target_language):
|
||
glossary_entries = "\n".join([f"- {k}: {v}" for k, v in glossary.items()])
|
||
return (
|
||
f"動作: 翻譯\n目標語言: {target_language}\n參考字典:\n{glossary_entries}\n\n"
|
||
f"需翻譯內容or需總結內容:\n---\n{english_text_segment}\n---\n你的結果:\n"
|
||
)
|
||
def split_segments(text_content):
|
||
pattern = re.compile(r"(\[\d{4}s\s*-\s*\d{4}s\])(.*?)(?=\n\[\d{4}s|$)", re.DOTALL)
|
||
return [f"{match.group(1).strip()} {match.group(2).strip()}" for match in pattern.finditer(text_content)]
|
||
def parse_segment(segment_text):
|
||
match = re.match(r"(\[\d{4}s\s*-\s*\d{4}s\])\s*(.*)", segment_text.strip(), re.DOTALL)
|
||
return (match.group(1), match.group(2)) if match else (None, segment_text.strip())
|
||
|
||
@celery.task(base=ProgressTask, bind=True)
|
||
def translate_segments_task(self, input_txt_path, output_txt_path, target_language):
|
||
try:
|
||
with open(input_txt_path, "r", encoding="utf-8") as f:
|
||
segments = split_segments(f.read())
|
||
total_segments = len(segments)
|
||
if total_segments == 0:
|
||
self.update_progress(100, 100, "完成，但輸入檔無內容。")
|
||
return {'status': '完成', 'result_path': None, 'content': '(輸入檔案為空)'}
|
||
conversation_id = None
|
||
full_translated_content = ""
|
||
for idx, segment_text in enumerate(segments):
|
||
timestamp, original_text = parse_segment(segment_text)
|
||
status_msg = f"正在翻譯第 {idx + 1}/{total_segments} 段..."
|
||
self.update_progress(idx, total_segments, status_msg, {'preview': full_translated_content})
|
||
if not original_text.strip():
|
||
translated_text = "(原始內容為空)"
|
||
else:
|
||
prompt = generate_translation_prompt_for_dify(original_text, target_language)
|
||
response = ask_dify(prompt, conversation_id=conversation_id)
|
||
translated_text = response.get("answer", "翻譯失敗")
|
||
conversation_id = response.get("conversation_id")
|
||
line_break = "\n\n" if full_translated_content else ""
|
||
segment_result = f"{timestamp}\n{original_text}\n👉{translated_text}"
|
||
full_translated_content += f"{line_break}{segment_result}"
|
||
with open(output_txt_path, "w", encoding="utf-8") as f:
|
||
f.write(full_translated_content)
|
||
self.update_progress(total_segments, total_segments, "全部翻譯完成！")
|
||
return {'status': '完成', 'result_path': output_txt_path, 'content': full_translated_content}
|
||
except Exception as e:
|
||
self.update_state(state='FAILURE', meta={'exc_type': type(e).__name__, 'exc_message': str(e)})
|
||
return {'status': '錯誤', 'error': str(e)}
|
||
|
||
# ========== Task 4: 會議結論整理 (Dify) ==========
|
||
@celery.task(base=ProgressTask, bind=True)
|
||
def summarize_text_task(self, text_content, target_language, conversation_id=None, revision_instruction=None):
|
||
try:
|
||
self.update_progress(1, 100, "準備提示詞...")
|
||
if revision_instruction:
|
||
prompt = f"Existing Summary:\n---\n{text_content}\n---\n\nUser's Revision Instructions:\n---\n{revision_instruction}\n---\nPlease provide the revised meeting summary in {target_language}:"
|
||
else:
|
||
prompt = f"Please act as a professional meeting analyst and summarize the following meeting transcript into key points, action items, and conclusions. The summary should be in {target_language}.\n\nTranscript:\n---\n{text_content}\n---"
|
||
self.update_progress(20, 100, "正在請求 Dify API...")
|
||
response = ask_dify(prompt, conversation_id=conversation_id)
|
||
summary = response.get("answer", "總結失敗")
|
||
new_conv_id = response.get("conversation_id")
|
||
self.update_progress(100, 100, "總結已生成！")
|
||
return {'status': '完成', 'summary': summary, 'conversation_id': new_conv_id}
|
||
except Exception as e:
|
||
self.update_state(state='FAILURE', meta={'exc_type': type(e).__name__, 'exc_message': str(e)})
|
||
return {'status': '錯誤', 'error': str(e)}
|