728 lines
33 KiB
Python
728 lines
33 KiB
Python
from sqlalchemy.orm import Session
|
|
from app.core.db import SessionLocal
|
|
from typing import Callable, Awaitable, Optional, List
|
|
from app.models import Meeting, SummarizeTask, PromptTemplate, AIModel, TranscriptSegment, TranscriptTask, MeetingAudio, Hotword
|
|
from app.services.llm_service import LLMService
|
|
import uuid
|
|
import json
|
|
import logging
|
|
import math
|
|
import httpx
|
|
import asyncio
|
|
from datetime import datetime
|
|
from app.core.redis import redis_client
|
|
from app.core.config import get_settings
|
|
from pathlib import Path
|
|
|
|
logger = logging.getLogger(__name__)
|
|
settings = get_settings()
|
|
|
|
class MeetingService:
|
|
@staticmethod
|
|
async def create_transcript_task(
|
|
db: Session,
|
|
meeting_id: int,
|
|
model_id: int,
|
|
language: str = "auto"
|
|
):
|
|
# 1. 验证会议和音频
|
|
meeting = db.query(Meeting).filter(Meeting.meeting_id == meeting_id).first()
|
|
if not meeting:
|
|
raise Exception("Meeting not found")
|
|
|
|
# 获取最新的音频文件
|
|
audio = db.query(MeetingAudio).filter(
|
|
MeetingAudio.meeting_id == meeting_id
|
|
).order_by(MeetingAudio.upload_time.desc()).first()
|
|
|
|
if not audio:
|
|
raise Exception("No audio file found for this meeting")
|
|
|
|
# 2. 创建转译任务
|
|
task_id = str(uuid.uuid4())
|
|
new_task = TranscriptTask(
|
|
task_id=task_id,
|
|
meeting_id=meeting_id,
|
|
model_id=model_id,
|
|
language=language,
|
|
status="pending",
|
|
progress=0,
|
|
created_at=datetime.utcnow()
|
|
)
|
|
db.add(new_task)
|
|
|
|
# 更新会议状态
|
|
meeting.status = "transcribing"
|
|
db.commit()
|
|
|
|
# 3. 进入 Redis 队列 (transcribe queue)
|
|
# Note: In a real worker system, we would push to a queue.
|
|
# For now, we return the task so the API can trigger background processing.
|
|
# await redis_client.lpush("meeting:transcribe:queue", task_id)
|
|
|
|
return new_task
|
|
|
|
@staticmethod
|
|
async def process_transcript_task(db: Session, task_id: str):
|
|
print(f"[DEBUG] Processing transcript task {task_id}")
|
|
task = db.query(TranscriptTask).filter(TranscriptTask.task_id == task_id).first()
|
|
if not task:
|
|
print(f"[ERROR] Task {task_id} not found in DB")
|
|
return
|
|
|
|
try:
|
|
task.status = "processing"
|
|
task.progress = 10
|
|
await asyncio.to_thread(db.commit)
|
|
|
|
# 1. 获取模型配置
|
|
model_config = db.query(AIModel).filter(AIModel.model_id == task.model_id).first()
|
|
if not model_config:
|
|
# 如果没有指定模型，尝试找一个默认的 ASR 模型
|
|
model_config = db.query(AIModel).filter(AIModel.model_type == "asr").first()
|
|
|
|
if not model_config:
|
|
raise Exception("No ASR model configuration found")
|
|
|
|
# 2. 获取音频文件
|
|
audio = db.query(MeetingAudio).filter(
|
|
MeetingAudio.meeting_id == task.meeting_id
|
|
).order_by(MeetingAudio.upload_time.desc()).first()
|
|
|
|
if not audio:
|
|
raise Exception("Audio file missing")
|
|
|
|
task.progress = 20
|
|
await asyncio.to_thread(db.commit)
|
|
|
|
# 3. 调用 ASR 服务 (Local Model Only)
|
|
# Use model config base_url if available, otherwise fallback to settings
|
|
asr_base_url = model_config.base_url if model_config.base_url else settings.asr_api_base_url
|
|
|
|
meeting = db.query(Meeting).filter(Meeting.meeting_id == task.meeting_id).first()
|
|
if not meeting:
|
|
raise Exception("Meeting not found")
|
|
|
|
hotword_filters = [Hotword.scope.in_(["public", "global"])]
|
|
if meeting.user_id:
|
|
hotword_filters.append(
|
|
(Hotword.scope == "personal") & (Hotword.user_id == meeting.user_id)
|
|
)
|
|
hotwords = db.query(Hotword).filter(
|
|
(hotword_filters[0]) if len(hotword_filters) == 1 else (hotword_filters[0] | hotword_filters[1])
|
|
).all()
|
|
hotword_entries = []
|
|
hotword_string_parts = []
|
|
for hw in hotwords:
|
|
hotword_entries.append({"word": hw.word, "weight": hw.weight})
|
|
if hw.weight and hw.weight != 1:
|
|
hotword_string_parts.append(f"{hw.word}:{hw.weight}")
|
|
else:
|
|
hotword_string_parts.append(hw.word)
|
|
hotword_string = " ".join([p for p in hotword_string_parts if p])
|
|
|
|
logger.info(f"Task {task_id}: Starting transcription with ASR Model at {asr_base_url}")
|
|
print(f"[DEBUG] Calling ASR at {asr_base_url} for file: {audio.file_path}")
|
|
|
|
# Define progress callback
|
|
async def update_progress(p: int, msg: str = None):
|
|
# Map ASR progress (0-100) to Task progress (20-80)
|
|
# Ensure we don't go backwards or exceed bounds
|
|
new_progress = 20 + int(p * 0.6)
|
|
print(f"[DEBUG] update_progress called: p={p}, new_progress={new_progress}, current={task.progress}")
|
|
|
|
if new_progress > task.progress:
|
|
task.progress = new_progress
|
|
# Note: We must be careful with db.commit() in async loop if db session is shared
|
|
try:
|
|
# Offload blocking DB commit to thread to avoid freezing the event loop
|
|
await asyncio.to_thread(db.commit)
|
|
print(f"[DEBUG] DB Updated: progress={new_progress}")
|
|
except Exception as dbe:
|
|
logger.error(f"DB Commit failed: {dbe}")
|
|
db.rollback()
|
|
|
|
# Update status message in Redis
|
|
if msg:
|
|
try:
|
|
await redis_client.setex(f"task:status:{task_id}", 3600, msg)
|
|
print(f"[DEBUG] Redis Updated: key=task:status:{task_id}, msg={msg}")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to update task status in Redis: {e}")
|
|
|
|
# 使用本地模型 API 进行转译
|
|
segments_data = await MeetingService._call_local_asr_api(
|
|
audio_path=audio.file_path,
|
|
base_url=asr_base_url,
|
|
language=task.language,
|
|
hotwords=hotword_entries if hotword_entries else None,
|
|
hotword_string=hotword_string,
|
|
progress_callback=update_progress
|
|
)
|
|
print(f"[DEBUG] Received {len(segments_data)} segments from ASR")
|
|
|
|
task.progress = 80
|
|
await asyncio.to_thread(db.commit)
|
|
|
|
# 4. 保存转译结果
|
|
last_saved_end_ms = 0
|
|
|
|
def parse_number(v):
|
|
if isinstance(v, (int, float)):
|
|
return float(v)
|
|
if isinstance(v, str):
|
|
try:
|
|
return float(v.strip())
|
|
except Exception:
|
|
return None
|
|
return None
|
|
|
|
for seg in segments_data:
|
|
# Local API returns: { "text": "...", "timestamp": [[start, end]], "speaker": "..." }
|
|
# We need to adapt it to our schema
|
|
|
|
text = (seg.get("text") or "").strip()
|
|
if not text:
|
|
continue
|
|
|
|
start_ms = None
|
|
end_ms = None
|
|
ts = seg.get("timestamp")
|
|
if ts:
|
|
if isinstance(ts, list) and len(ts) > 0:
|
|
try:
|
|
if isinstance(ts[0], list):
|
|
starts = []
|
|
ends = []
|
|
for pair in ts:
|
|
if not isinstance(pair, (list, tuple)) or len(pair) < 2:
|
|
continue
|
|
s = parse_number(pair[0])
|
|
e = parse_number(pair[1])
|
|
if s is not None and e is not None:
|
|
starts.append(s)
|
|
ends.append(e)
|
|
if starts and ends:
|
|
raw_start = min(starts)
|
|
raw_end = max(ends)
|
|
if raw_end < raw_start:
|
|
raise ValueError("timestamp end < start")
|
|
if raw_end < 1000:
|
|
raw_start *= 1000.0
|
|
raw_end *= 1000.0
|
|
start_ms = int(raw_start)
|
|
end_ms = int(raw_end)
|
|
elif len(ts) >= 2:
|
|
raw_start = parse_number(ts[0])
|
|
raw_end = parse_number(ts[1])
|
|
if raw_start is None or raw_end is None:
|
|
raise ValueError("timestamp not numeric")
|
|
if raw_end < raw_start:
|
|
raise ValueError("timestamp end < start")
|
|
if raw_end < 1000:
|
|
raw_start *= 1000.0
|
|
raw_end *= 1000.0
|
|
start_ms = int(raw_start)
|
|
end_ms = int(raw_end)
|
|
except Exception:
|
|
start_ms = None
|
|
end_ms = None
|
|
|
|
if start_ms is None or end_ms is None:
|
|
bt = parse_number(seg.get("begin_time") or seg.get("start_time"))
|
|
et = parse_number(seg.get("end_time"))
|
|
if bt is not None and et is not None:
|
|
if bt < 1000 and et < 1000:
|
|
bt *= 1000.0
|
|
et *= 1000.0
|
|
start_ms = int(bt)
|
|
end_ms = int(et)
|
|
|
|
if start_ms is None or end_ms is None or end_ms < start_ms:
|
|
start_ms = last_saved_end_ms + 1
|
|
end_ms = start_ms + 1000
|
|
|
|
last_saved_end_ms = max(last_saved_end_ms, end_ms)
|
|
|
|
transcript_segment = TranscriptSegment(
|
|
meeting_id=task.meeting_id,
|
|
audio_id=audio.audio_id,
|
|
speaker_id=0,
|
|
speaker_tag=seg.get("speaker", "Unknown"),
|
|
start_time_ms=start_ms,
|
|
end_time_ms=end_ms,
|
|
text_content=text
|
|
)
|
|
db.add(transcript_segment)
|
|
|
|
# 5. 完成任务
|
|
task.status = "completed"
|
|
task.progress = 100
|
|
task.completed_at = datetime.utcnow()
|
|
|
|
# 更新音频状态
|
|
audio.processing_status = "completed"
|
|
|
|
# 更新会议状态 (如果需要自动开始总结，可以在这里触发，或者由用户触发)
|
|
meeting = db.query(Meeting).filter(Meeting.meeting_id == task.meeting_id).first()
|
|
if meeting:
|
|
meeting.status = "transcribed" # distinct status before summarizing
|
|
|
|
# Auto-trigger summarization for uploaded meetings
|
|
if meeting.type == 'upload':
|
|
try:
|
|
# Find default LLM model
|
|
llm_model = None
|
|
if meeting.summary_model_id:
|
|
llm_model = db.query(AIModel).filter(
|
|
AIModel.model_id == meeting.summary_model_id,
|
|
AIModel.status == 1
|
|
).first()
|
|
if not llm_model:
|
|
llm_model = db.query(AIModel).filter(
|
|
AIModel.model_type == 'llm',
|
|
AIModel.is_default == 1,
|
|
AIModel.status == 1
|
|
).first()
|
|
|
|
prompt_tmpl = None
|
|
if meeting.summary_prompt_id:
|
|
prompt_tmpl = db.query(PromptTemplate).filter(
|
|
PromptTemplate.id == meeting.summary_prompt_id
|
|
).first()
|
|
if not prompt_tmpl:
|
|
prompt_tmpl = db.query(PromptTemplate).filter(
|
|
PromptTemplate.status == 1
|
|
).order_by(PromptTemplate.is_system.desc(), PromptTemplate.id.asc()).first()
|
|
|
|
if llm_model and prompt_tmpl:
|
|
logger.info(f"Auto-triggering summary for meeting {meeting.meeting_id}")
|
|
# Create summarize task
|
|
# We need to call create_summarize_task but it's an async static method
|
|
# and we are inside an async static method.
|
|
# However, create_summarize_task commits to DB, so we should be careful with session.
|
|
# Since we are in the same session `db`, we can just call it.
|
|
# But create_summarize_task takes `db` session.
|
|
|
|
# Note: create_summarize_task commits. We should commit our changes first.
|
|
await asyncio.to_thread(db.commit)
|
|
|
|
new_sum_task = await MeetingService.create_summarize_task(
|
|
db,
|
|
meeting_id=meeting.meeting_id,
|
|
prompt_id=prompt_tmpl.id,
|
|
model_id=llm_model.model_id
|
|
)
|
|
# Trigger background worker for summary
|
|
# Since we are in a worker, we can just call process_summarize_task directly or via background task
|
|
# But better to use the same mechanism (add to background tasks if possible, or just call it)
|
|
# Here we don't have access to FastAPI BackgroundTasks object.
|
|
# We can use asyncio.create_task to run it in background
|
|
# import asyncio # Already imported at top level
|
|
# Use a separate worker method that creates its own session
|
|
# to avoid "Session is closed" error since the current session
|
|
# will be closed when this function returns.
|
|
asyncio.create_task(MeetingService.run_summarize_worker(new_sum_task.task_id))
|
|
|
|
# Update meeting status to summarizing
|
|
meeting.status = "summarizing"
|
|
else:
|
|
logger.warning(f"Skipping auto-summary: No default LLM or Prompt found (LLM: {llm_model}, Prompt: {prompt_tmpl})")
|
|
except Exception as sum_e:
|
|
logger.error(f"Failed to auto-trigger summary: {sum_e}")
|
|
|
|
await asyncio.to_thread(db.commit)
|
|
logger.info(f"Task {task_id} transcription completed")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Task {task_id} failed: {str(e)}")
|
|
task.status = "failed"
|
|
task.error_message = str(e)
|
|
|
|
# Update audio status
|
|
audio = db.query(MeetingAudio).filter(
|
|
MeetingAudio.meeting_id == task.meeting_id
|
|
).order_by(MeetingAudio.upload_time.desc()).first()
|
|
if audio:
|
|
audio.processing_status = "failed"
|
|
audio.error_message = str(e)
|
|
|
|
# Update meeting status to failed so frontend knows to stop polling
|
|
meeting = db.query(Meeting).filter(Meeting.meeting_id == task.meeting_id).first()
|
|
if meeting:
|
|
meeting.status = "failed"
|
|
|
|
await asyncio.to_thread(db.commit)
|
|
|
|
@staticmethod
|
|
async def _call_local_asr_api(
|
|
audio_path: str,
|
|
base_url: str = "http://localhost:3050",
|
|
language: str = "auto",
|
|
hotwords: Optional[List[dict]] = None,
|
|
hotword_string: Optional[str] = None,
|
|
progress_callback: Optional[Callable[[int, Optional[str]], Awaitable[None]]] = None
|
|
) -> list:
|
|
"""
|
|
Call local ASR API for transcription.
|
|
Flow: Create Task -> Poll Status -> Get Result
|
|
"""
|
|
# import asyncio # Already imported at top level
|
|
|
|
create_url = f"{base_url.rstrip('/')}/api/tasks/recognition"
|
|
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
# 1. Create Task
|
|
try:
|
|
normalized_hotwords = {}
|
|
if hotwords:
|
|
for hw in hotwords:
|
|
if isinstance(hw, dict):
|
|
word = hw.get("word")
|
|
weight = hw.get("weight")
|
|
if word:
|
|
normalized_hotwords[word] = int(weight) if weight is not None else 20
|
|
elif isinstance(hw, str):
|
|
normalized_hotwords[hw] = 20
|
|
|
|
payload = {
|
|
"file_path": audio_path,
|
|
"language": language,
|
|
"use_spk_id": True
|
|
}
|
|
if normalized_hotwords:
|
|
payload["hotwords"] = normalized_hotwords
|
|
elif hotword_string:
|
|
payload["hotword"] = hotword_string
|
|
response = await client.post(create_url, json=payload)
|
|
if response.status_code >= 500:
|
|
base_payload = {
|
|
"file_path": audio_path,
|
|
"language": language,
|
|
"use_spk_id": True
|
|
}
|
|
response = await client.post(create_url, json=base_payload)
|
|
if response.status_code == 422:
|
|
alt_payload = {
|
|
"audio_path": audio_path,
|
|
"language": language,
|
|
"use_spk_id": True
|
|
}
|
|
if normalized_hotwords:
|
|
alt_payload["hotwords"] = normalized_hotwords
|
|
elif hotword_string:
|
|
alt_payload["hotword"] = hotword_string
|
|
response = await client.post(create_url, json=alt_payload)
|
|
if response.status_code >= 500:
|
|
alt_base_payload = {
|
|
"audio_path": audio_path,
|
|
"language": language,
|
|
"use_spk_id": True
|
|
}
|
|
response = await client.post(create_url, json=alt_base_payload)
|
|
if response.status_code == 422 or response.status_code >= 500:
|
|
with open(audio_path, "rb") as f:
|
|
files = {
|
|
"file": (Path(audio_path).name, f, "application/octet-stream")
|
|
}
|
|
data_fields = {
|
|
"language": language,
|
|
"use_spk_id": "true"
|
|
}
|
|
if normalized_hotwords:
|
|
data_fields["hotwords"] = json.dumps(normalized_hotwords, ensure_ascii=False)
|
|
elif hotword_string:
|
|
data_fields["hotword"] = hotword_string
|
|
response = await client.post(create_url, data=data_fields, files=files)
|
|
if response.status_code >= 400:
|
|
try:
|
|
logger.error(f"ASR create response: {response.status_code} {response.text}")
|
|
except Exception:
|
|
pass
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
# Handle nested data structure {code: 200, data: {task_id: ...}}
|
|
if "data" in data and isinstance(data["data"], dict) and "task_id" in data["data"]:
|
|
task_id = data["data"]["task_id"]
|
|
else:
|
|
task_id = data.get("task_id")
|
|
|
|
if not task_id:
|
|
raise Exception("Failed to get task_id from ASR service")
|
|
except Exception as e:
|
|
logger.error(f"Failed to create ASR task: {e}")
|
|
# Fallback for testing/mock if service is not running
|
|
if "Connection refused" in str(e) or "ConnectError" in str(e):
|
|
logger.warning("ASR Service not reachable, using mock data for testing")
|
|
await asyncio.sleep(2)
|
|
return [
|
|
{"timestamp": [[0, 2500]], "text": "This is a mock transcription (Local ASR unreachable).", "speaker": "System"},
|
|
{"timestamp": [[2500, 5000]], "text": "Please ensure the ASR service is running at localhost:3050.", "speaker": "System"}
|
|
]
|
|
raise
|
|
|
|
# 2. Poll Status
|
|
status_url = f"{base_url.rstrip('/')}/api/tasks/{task_id}"
|
|
# Increase timeout for polling
|
|
poll_client = httpx.AsyncClient(timeout=10.0)
|
|
|
|
max_retries = 600 # 20 minutes max wait (assuming 2s sleep)
|
|
|
|
try:
|
|
for _ in range(max_retries):
|
|
try:
|
|
res = await poll_client.get(status_url)
|
|
res.raise_for_status()
|
|
status_data = res.json()
|
|
|
|
progress = 0
|
|
|
|
# Handle nested data structure
|
|
if "data" in status_data and isinstance(status_data["data"], dict):
|
|
inner_data = status_data["data"]
|
|
status = inner_data.get("status")
|
|
result = inner_data.get("result", {})
|
|
error_msg = inner_data.get("msg") or status_data.get("msg")
|
|
# Try to get progress
|
|
if "progress" in inner_data:
|
|
raw_progress = inner_data["progress"]
|
|
elif "percent" in inner_data:
|
|
raw_progress = inner_data["percent"]
|
|
elif "percentage" in inner_data:
|
|
raw_progress = inner_data["percentage"]
|
|
else:
|
|
raw_progress = 0
|
|
|
|
# Update error_msg with message if available
|
|
if "message" in inner_data and inner_data["message"]:
|
|
error_msg = inner_data["message"]
|
|
else:
|
|
status = status_data.get("status")
|
|
result = status_data.get("result", {})
|
|
error_msg = status_data.get("msg")
|
|
# Try to get progress
|
|
if "progress" in status_data:
|
|
raw_progress = status_data["progress"]
|
|
elif "percent" in status_data:
|
|
raw_progress = status_data["percent"]
|
|
elif "percentage" in status_data:
|
|
raw_progress = status_data["percentage"]
|
|
else:
|
|
raw_progress = 0
|
|
|
|
# Update error_msg with message if available
|
|
if "message" in status_data and status_data["message"]:
|
|
error_msg = status_data["message"]
|
|
|
|
# Handle float progress (0.0 - 1.0) or int (0-100)
|
|
progress = 0
|
|
try:
|
|
if isinstance(raw_progress, (int, float)):
|
|
if 0.0 <= raw_progress <= 1.0 and isinstance(raw_progress, float):
|
|
progress = int(raw_progress * 100)
|
|
else:
|
|
progress = int(raw_progress)
|
|
elif isinstance(raw_progress, str):
|
|
# Handle string like "45%" or "0.45"
|
|
if raw_progress.endswith("%"):
|
|
progress = int(float(raw_progress.strip("%")))
|
|
else:
|
|
val = float(raw_progress)
|
|
if 0.0 <= val <= 1.0:
|
|
progress = int(val * 100)
|
|
else:
|
|
progress = int(val)
|
|
except (ValueError, TypeError):
|
|
progress = 0
|
|
|
|
# Log raw status for debugging
|
|
print(f"[DEBUG] ASR Polling: status={status}, raw_progress={raw_progress}, progress={progress}, msg={error_msg}")
|
|
|
|
# Update progress if callback provided
|
|
if progress_callback:
|
|
try:
|
|
status_msg = error_msg if error_msg else f"正在转写中... {progress}%"
|
|
print(f"[DEBUG] Invoking callback with progress={progress}, msg={status_msg}")
|
|
await progress_callback(progress, status_msg)
|
|
except Exception as e:
|
|
logger.warning(f"Progress callback error: {e}")
|
|
|
|
if status == "completed" or status == "success":
|
|
# Return the segments list.
|
|
if isinstance(result, dict) and "segments" in result:
|
|
return result["segments"]
|
|
elif isinstance(result, list):
|
|
return result
|
|
else:
|
|
# Fallback or empty
|
|
return []
|
|
elif status == "failed":
|
|
raise Exception(f"ASR Task failed: {error_msg}")
|
|
|
|
# Still processing
|
|
await asyncio.sleep(2)
|
|
except httpx.RequestError as e:
|
|
logger.warning(f"Error polling ASR task {task_id}: {e}")
|
|
await asyncio.sleep(2)
|
|
finally:
|
|
await poll_client.aclose()
|
|
|
|
raise Exception("ASR Task timed out")
|
|
|
|
@staticmethod
|
|
async def create_summarize_task(
|
|
db: Session,
|
|
meeting_id: int,
|
|
prompt_id: int,
|
|
model_id: int,
|
|
extra_prompt: str = ""
|
|
):
|
|
# 1. 基础数据校验
|
|
meeting = db.query(Meeting).filter(Meeting.meeting_id == meeting_id).first()
|
|
if not meeting:
|
|
raise Exception("Meeting not found")
|
|
|
|
# 2. 格式化会议转译内容 (作为 user_prompt 素材)
|
|
segments = db.query(TranscriptSegment).filter(
|
|
TranscriptSegment.meeting_id == meeting_id
|
|
).order_by(TranscriptSegment.start_time_ms).all()
|
|
|
|
formatted_content = []
|
|
for s in segments:
|
|
secs = int(s.start_time_ms // 1000)
|
|
m, sc = divmod(secs, 60)
|
|
timestamp = f"[{m:02d}:{sc:02d}]"
|
|
speaker = s.speaker_tag or f"发言人{s.speaker_id or '?'}"
|
|
formatted_content.append(f"{timestamp} {speaker}: {s.text_content}")
|
|
|
|
meeting_text = "\n".join(formatted_content)
|
|
|
|
# 组合最终 user_prompt (素材 + 用户的附加要求)
|
|
user_prompt_content = f"### 会议转译内容 ###\n{meeting_text}"
|
|
if extra_prompt:
|
|
user_prompt_content += f"\n\n### 用户的额外指令 ###\n{extra_prompt}"
|
|
|
|
# 3. 创建任务记录 (按照数据库实际字段)
|
|
task_id = str(uuid.uuid4())
|
|
new_task = SummarizeTask(
|
|
task_id=task_id,
|
|
meeting_id=meeting_id,
|
|
prompt_id=prompt_id,
|
|
model_id=model_id,
|
|
user_prompt=user_prompt_content,
|
|
status="pending",
|
|
progress=0,
|
|
created_at=datetime.utcnow()
|
|
)
|
|
db.add(new_task)
|
|
|
|
# 更新会议状态为“总结中”
|
|
meeting.status = "summarizing"
|
|
await asyncio.to_thread(db.commit)
|
|
|
|
# 4. 进入 Redis 队列
|
|
await redis_client.lpush("meeting:summarize:queue", task_id)
|
|
|
|
return new_task
|
|
|
|
@staticmethod
|
|
async def run_summarize_worker(task_id: str):
|
|
"""
|
|
Worker entry point for summarize task that manages its own DB session.
|
|
This is safe to call via asyncio.create_task from other contexts.
|
|
"""
|
|
db = SessionLocal()
|
|
try:
|
|
await MeetingService.process_summarize_task(db, task_id)
|
|
except Exception as e:
|
|
logger.error(f"Summarize worker (task {task_id}) failed: {e}")
|
|
finally:
|
|
db.close()
|
|
|
|
@staticmethod
|
|
async def process_summarize_task(db: Session, task_id: str):
|
|
"""
|
|
后台 Worker 真实执行逻辑
|
|
"""
|
|
task = db.query(SummarizeTask).filter(SummarizeTask.task_id == task_id).first()
|
|
if not task:
|
|
return
|
|
|
|
try:
|
|
task.status = "processing"
|
|
task.progress = 15
|
|
await asyncio.to_thread(db.commit)
|
|
|
|
# 1. 获取模型配置
|
|
model_config = db.query(AIModel).filter(AIModel.model_id == task.model_id).first()
|
|
if not model_config:
|
|
raise Exception("AI 模型配置不存在")
|
|
|
|
# 2. 实时获取提示词模板内容
|
|
prompt_tmpl = db.query(PromptTemplate).filter(PromptTemplate.id == task.prompt_id).first()
|
|
system_prompt = prompt_tmpl.content if prompt_tmpl else "请根据提供的会议转译内容生成准确的总结。"
|
|
|
|
task.progress = 30
|
|
await asyncio.to_thread(db.commit)
|
|
|
|
# 3. 构建消息结构
|
|
messages = [
|
|
{"role": "system", "content": system_prompt},
|
|
{"role": "user", "content": task.user_prompt}
|
|
]
|
|
|
|
# 解析模型参数
|
|
config_params = model_config.config or {}
|
|
temperature = config_params.get("temperature", 0.7)
|
|
top_p = config_params.get("top_p", 0.9)
|
|
|
|
logger.info(f"Task {task_id}: Launching LLM request to {model_config.model_name}")
|
|
task.progress = 50
|
|
await asyncio.to_thread(db.commit)
|
|
|
|
# 4. 调用大模型服务
|
|
summary_result = await LLMService.chat_completion(
|
|
api_key=model_config.api_key,
|
|
base_url=model_config.base_url or "https://api.openai.com/v1",
|
|
model_name=model_config.model_name,
|
|
messages=messages,
|
|
api_path=model_config.api_path or "/chat/completions",
|
|
temperature=float(temperature),
|
|
top_p=float(top_p)
|
|
)
|
|
|
|
# 5. 任务完成，回写结果
|
|
task.result = summary_result
|
|
task.status = "completed"
|
|
task.progress = 100
|
|
task.completed_at = datetime.utcnow()
|
|
|
|
# 同步更新会议主表摘要和状态
|
|
meeting = db.query(Meeting).filter(Meeting.meeting_id == task.meeting_id).first()
|
|
if meeting:
|
|
meeting.summary = summary_result
|
|
meeting.status = "completed"
|
|
|
|
await asyncio.to_thread(db.commit)
|
|
logger.info(f"Task {task_id} completed successfully")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Task {task_id} execution error: {str(e)}")
|
|
task.status = "failed"
|
|
task.error_message = str(e)
|
|
|
|
# Restore meeting status to transcribed (so user can retry summary) instead of draft
|
|
meeting = db.query(Meeting).filter(Meeting.meeting_id == task.meeting_id).first()
|
|
if meeting:
|
|
meeting.status = "transcribed"
|
|
|
|
await asyncio.to_thread(db.commit)
|
|
|
|
@staticmethod
|
|
def get_task_status(db: Session, task_id: str):
|
|
task = db.query(SummarizeTask).filter(SummarizeTask.task_id == task_id).first()
|
|
if task:
|
|
return task
|
|
return db.query(TranscriptTask).filter(TranscriptTask.task_id == task_id).first()
|