ai-meeting-server/app/apis/asr.py

import asyncio
import json
import time
import websockets
import aiofiles
from fastapi import APIRouter, UploadFile, WebSocket
from datetime import datetime

from app.config.config import settings
from app.models.ai_meeting import AIMeetingModel, AIMeetingStatus
from app.models.ai_meeting_minutes import AIMeetingMinutesRecordModel, AIMeetingSpeecher
from app.models.asr import AsrRecordModel, AsrRecordStatus
from app.models.common import Callback, Page
from app.services.ai_meeting import AIMeetingService
from app.services.ai_meeting_minutes import AIMeetingMinutesRecordService
from app.services.asr import AsrRecordService
from app.services.hotword import ReplaceHotwordService, ASRHotwordService
from app.utils.common import uuid_generate, text_replace, match_content_keyword
from app.utils.savefile import save_file, StreamFileSaver, rename_to_md5, FileStatus
# from app.utils.audio import convert_to_mp3
from app.utils.oss import upload_file


router = APIRouter()
WS_URL = f"ws://{settings.meeting_file_base_url}/ws/transcribe"
LANG = "zh"
SV = 1  # 1 for enabled, 0 for disabled
SPEAKER_VERIFICATION = 1

@router.websocket("/realtime")
async def realtime(websocket: WebSocket):
    await websocket.accept()
    times = 0
    pause = 0
    start = ""
    saver: StreamFileSaver = StreamFileSaver()
    is_update = False
    content_list = []
    try:
        async with websockets.connect(WS_URL+ f"?lang={LANG}&sv={int(SPEAKER_VERIFICATION)}") as ws:

            # 接收前端传输的音频流
            data = await websocket.receive_json()
            await saver.aset_ext(data["ext"])
            meeting_id = data["meeting_id"]
            meeting = await AIMeetingService.get_ai_meeting(meeting_id)
            if not meeting or meeting.status != AIMeetingStatus.RECRODING:
                await websocket.send_json(
                    {"type": "error", "data": "会议信息异常！"}
                )
                raise Exception("meeting error")

            is_update = True
            hotword_dict = {i.origin_word:i.hotword for i in await ReplaceHotwordService.list_replace_hotwords(Page(page_size=1000))}
            hotwords = " ".join([i.hotword for i in
                            await ASRHotwordService.list_asr_hotwords(Page(page_size=1000))])
            await ws.send(hotwords)  # 发送文件到实时ASR服务
            async def receive():
                nonlocal pause
                async with saver:
                    while True:

                        audio_data = await websocket.receive_bytes()
                        await ws.send(audio_data)  # 发送文件到实时ASR服务
                        saver.audio_data_list.append(audio_data)

                        try:
                            message = await asyncio.wait_for(websocket.receive_json(), timeout=0.1)
                            if message.get("action") == "pause":
                                pause_time = int(time.time())
                                while True:
                                    # 接收前端传输的音频流
                                    tmp_data = await websocket.receive_json()
                                    if tmp_data.get("action") == "start":
                                        pause += int(time.time()) - pause_time
                                        break
                        except Exception:
                            ...
            async def send():
                start_time = int(time.time())
                nonlocal times
                nonlocal start
                async for message in ws:
                    try:
                        res_json = json.loads(message)
                        if res_json.get('code') == 0:
                            recv_data = res_json.get('data', {})
                            if recv_data:
                                speaker = recv_data.get("speaker")
                                text = recv_data.get("text")
                                if text:
                                    text = await text_replace(text, hotword_dict)
                                    print(f"user:{speaker}---{text}")
                                    start = datetime.now().strftime('%Y/%m/%d %H:%M:%S')
                                    times = int(time.time()) - start_time - pause
                                    send_json = {"type": "stream", "text": text, "times": times, "start": start, "speaker":speaker}
                                    await websocket.send_json(
                                        send_json
                                    )
                                    # await websocket.send_json(
                                    #     {"type": "close", "data": ""}
                                    # )
                                    content_list.append(send_json)
                                    await AIMeetingService.update_ai_meeting_record(
                                        AIMeetingModel(id=meeting_id, content=send_json)
                                    )

                        else:
                            print("Error:", res_json.get('code'), res_json.get('message'))
                    except json.JSONDecodeError as e:
                        print("Failed to parse response data:", e)
                    # 接收实时ASR的结果
            await asyncio.gather(receive(), send())
    # except WebSocketDisconnect:
    #     print("---------------------关闭连接11111-------------------------")

    except Exception as e:
        print("------------------------------------------------")
    finally:
        print("---------------------关闭连接:finally-------------------------")
        try:
            await ws.close()
        except:
            ...
        try:
            await websocket.close()
        except:
            ...
        if is_update:
            print("----------------更新状态------------------------")
            object_name = f"{await uuid_generate()}{saver.ext}"
            if saver.status != FileStatus.CLOSED:
                await saver.__aexit__(None, None, None)

            duration = await saver.get_duration(saver.get_file_path())
            if duration:
                file_url = await upload_file(object_name, saver.get_file_path())

                minutes_record = await AIMeetingMinutesRecordService.generate_meeting_summary(AIMeetingMinutesRecordModel(view_content=content_list))

                print(minutes_record)
                await AIMeetingMinutesRecordService.create_ai_meeting_minutes_record(AIMeetingMinutesRecordModel(meeting_id=meeting_id, content=minutes_record.get("summary", "生成会议记录失败！"), speecher=AIMeetingSpeecher.AI))
                await AIMeetingService.update_ai_meeting(
                    AIMeetingModel(id=meeting_id, file_url=file_url, duration=duration, description= await match_content_keyword(minutes_record.get("summary", "")),
                                   status=AIMeetingStatus.FINISHED)
                )
            else:
                await AIMeetingService.update_ai_meeting(
                    AIMeetingModel(id=meeting_id, file_url="", duration=duration,
                                   status=AIMeetingStatus.FAILED)
                )



        # await saver.remove_file(saver.get_file_path())


@router.post("/callback")
async def callback(call_back: Callback):
    task_id = call_back.task_id
    result = call_back.result
    # print(result)
    hotword_dict = {i.origin_word: i.hotword for i in
                    await ReplaceHotwordService.list_replace_hotwords(Page(page_size=1000))}
    if task_id and isinstance(result, list):
        view_list = [
                {"type": "stream", "text": await text_replace(i["text"], hotword_dict), "times": i["start"] // 1000 if isinstance(i["start"], int) else 0,
                 "start": "", "speaker": i["speaker"]} for i in result
            ]
        print(view_list)
        asr_record = await AsrRecordService.update_asr_record(AsrRecordModel(
            task_id=task_id,
            status=AsrRecordStatus.FINISHED
        ))
        minutes_record = await AIMeetingMinutesRecordService.generate_meeting_summary(
            AIMeetingMinutesRecordModel(view_content=view_list))
        print(minutes_record)
        # if minutes_record:
        await AIMeetingMinutesRecordService.create_ai_meeting_minutes_record(
            AIMeetingMinutesRecordModel(meeting_id=asr_record.meeting_id, content=minutes_record.get("summary", "生成会议记录失败！"),
                                        speecher=AIMeetingSpeecher.AI))
        await AIMeetingService.update_ai_meeting(
            AIMeetingModel(id=asr_record.meeting_id, view_content=json.dumps(view_list), status=AIMeetingStatus.FINISHED,
                           description=await match_content_keyword(minutes_record.get("summary", ""))))

    print(f"Received callback for task {task_id}: {result}")
    return {"status": "success"}