meeting_recorder/main.py

"""
FastAPI 后端 - 会议记录系统 API
"""

import os
import uuid
import json
import asyncio
import wave
import io
from datetime import datetime
from typing import List, Optional, Dict
from contextlib import asynccontextmanager
from threading import Thread, Lock
from queue import Queue

from fastapi import FastAPI, UploadFile, File, HTTPException, Form, Query, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
import openai

from database import init_db, Database
from processor import MeetingProcessor, get_device, RealtimeTranscriber


# 全局变量
db: Database = None
processor: MeetingProcessor = None
task_queue: Queue = Queue()
is_running = True

# 实时转写会话管理
realtime_sessions: Dict[str, Dict] = {}
sessions_lock = Lock()


def background_worker():
    """后台转录工作线程"""
    global is_running, processor, db
    print("🔄 后台转录线程已启动")
    while is_running:
        try:
            meeting_id, audio_path, title = task_queue.get(timeout=1)
            print(f"📝 开始转录会议: {title} (ID: {meeting_id})")
            try:
                result = processor.process_audio(audio_path, title=title)
                db.update_meeting(
                    meeting_id=meeting_id,
                    segments=[s.__dict__ for s in result.segments],
                    status="completed",
                    speaker_count=result.speaker_count,
                    duration=result.duration
                )
                print(f"✅ 转录完成: {title}")
            except Exception as e:
                print(f"❌ 转录失败: {str(e)}")
                db.update_meeting(meeting_id=meeting_id, status="failed")
        except:
            pass
    print("🔄 后台转录线程已停止")


@asynccontextmanager
async def lifespan(app: FastAPI):
    """应用生命周期"""
    global db, processor, is_running
    db = init_db()
    processor = MeetingProcessor(device=get_device())
    
    # 启动后台线程
    is_running = True
    worker_thread = Thread(target=background_worker, daemon=True)
    worker_thread.start()
    print("🚀 后台转录线程已启动")
    
    yield
    
    # 清理
    is_running = False
    print("👋 应用关闭，后台线程将停止")


# 创建 FastAPI 应用
app = FastAPI(
    title="会议记录系统 API",
    description="智能会议记录系统后端接口",
    version="1.0.0",
    lifespan=lifespan
)

# CORS 配置
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


# ============ API 路由 ============

@app.get("/api/health")
async def health_check():
    """健康检查"""
    return {"status": "ok", "timestamp": datetime.now().isoformat()}


@app.get("/api/meetings")
async def get_meetings():
    """获取所有会议"""
    meetings = db.get_all_meetings(limit=1000)
    return {"meetings": meetings, "total": len(meetings)}


@app.get("/api/meetings/{meeting_id}")
async def get_meeting(meeting_id: str):
    """获取单个会议详情"""
    meeting = db.get_meeting(meeting_id)
    if not meeting:
        raise HTTPException(status_code=404, detail="会议不存在")
    return meeting


@app.get("/api/meetings/search")
async def search_meetings(keyword: str = Query(...)):
    """搜索会议"""
    meetings = db.search_meetings(keyword)
    return {"meetings": meetings, "total": len(meetings)}


@app.post("/api/meetings/upload")
async def upload_meeting(
    file: UploadFile = File(...),
    title: str = Form(...)
):
    """上传音频文件，开始异步转录"""
    # 验证文件类型
    allowed_types = ['audio/wav', 'audio/mpeg', 'audio/mp3', 'audio/mp4',
                     'audio/x-m4a', 'audio/flac', 'audio/ogg', 'audio/x-wav']
    content_type = file.content_type or ''

    if not any(ct in content_type for ct in ['audio/', 'video/']) and \
       not file.filename.endswith(('.wav', '.mp3', '.m4a', '.flac', '.ogg')):
        raise HTTPException(status_code=400, detail="不支持的文件类型")

    # 保存文件
    project_root = os.path.dirname(os.path.abspath(__file__))
    data_dir = os.path.join(project_root, "data", "audio")
    os.makedirs(data_dir, exist_ok=True)

    ext = os.path.splitext(file.filename)[1] or ".mp3"
    filename = f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:8]}{ext}"
    filepath = os.path.join(data_dir, filename)

    with open(filepath, "wb") as f:
        content = await file.read()
        f.write(content)

    # 创建会议记录（状态: processing）
    meeting_id = datetime.now().strftime("%Y%m%d_%H%M%S") + "_" + uuid.uuid4().hex[:6]
    
    db.create_meeting(
        meeting_id=meeting_id,
        title=title,
        audio_path=filepath,
        status="processing"
    )
    
    # 将转录任务加入后台队列
    task_queue.put((meeting_id, filepath, title))
    print(f"📋 任务已加入队列: {title} (ID: {meeting_id})")

    # 立即返回会议信息（不等待转录完成）
    meeting = db.get_meeting(meeting_id)
    return meeting


@app.delete("/api/meetings/{meeting_id}")
async def delete_meeting(meeting_id: str):
    """删除会议"""
    meeting = db.get_meeting(meeting_id)
    if not meeting:
        raise HTTPException(status_code=404, detail="会议不存在")

    db.delete_meeting(meeting_id)

    # 删除音频文件
    audio_path = meeting.get('audio_path')
    if audio_path and os.path.exists(audio_path):
        try:
            os.remove(audio_path)
        except:
            pass

    return {"message": "删除成功"}


@app.get("/api/audio/{meeting_id}")
async def get_audio(meeting_id: str):
    """获取音频文件"""
    meeting = db.get_meeting(meeting_id)
    if not meeting:
        raise HTTPException(status_code=404, detail="会议不存在")

    audio_path = meeting.get('audio_path')
    if not audio_path or not os.path.exists(audio_path):
        raise HTTPException(status_code=404, detail="音频文件不存在")

    return FileResponse(
        audio_path,
        media_type="audio/mpeg",
        filename=os.path.basename(audio_path)
    )


@app.get("/api/stats")
async def get_stats():
    """获取统计数据"""
    meetings = db.get_all_meetings(limit=1000)
    return {
        "total": len(meetings),
        "completed": sum(1 for m in meetings if m['status'] == 'completed'),
        "processing": sum(1 for m in meetings if m['status'] == 'processing'),
        "failed": sum(1 for m in meetings if m['status'] == 'failed'),
        "total_duration": sum(m.get('duration', 0) for m in meetings),
        "queue_size": task_queue.qsize()
    }


@app.patch("/api/meetings/{meeting_id}")
async def update_meeting(meeting_id: str, title: str = Form(...)):
    """更新会议标题"""
    meeting = db.get_meeting(meeting_id)
    if not meeting:
        raise HTTPException(status_code=404, detail="会议不存在")
    
    db.update_meeting(meeting_id, title=title)
    return db.get_meeting(meeting_id)


@app.get("/api/meetings/{meeting_id}/status")
async def get_meeting_status(meeting_id: str):
    """获取会议处理状态（用于轮询）"""
    meeting = db.get_meeting(meeting_id)
    if not meeting:
        raise HTTPException(status_code=404, detail="会议不存在")
    return {
        "meeting_id": meeting_id,
        "status": meeting['status'],
        "title": meeting['title'],
        "queue_position": None,
        "segments_count": len(meeting.get('segments', []))
    }


@app.post("/api/meetings/{meeting_id}/summarize")
async def generate_summary(meeting_id: str):
    """生成会议摘要（使用 AI）"""
    import openai
    
    meeting = db.get_meeting(meeting_id)
    if not meeting:
        raise HTTPException(status_code=404, detail="会议不存在")
    
    if meeting['status'] != 'completed':
        raise HTTPException(status_code=400, detail="会议尚未完成转录")
    
    segments = meeting.get('segments', [])
    if not segments:
        raise HTTPException(status_code=400, detail="无转写内容")
    
    # 合并转写文本
    transcript = ""
    for seg in segments:
        speaker = f"说话人{seg.get('speaker', 0)}"
        text = seg.get('text', '')
        transcript += f"{speaker}: {text}\n"
    
    # 调用 DeepSeek API
    api_key = os.environ.get('DEEPSEEK_API_KEY') or 'sk-a8bcfef7ad67444dbdb20e52c854a5c1'
    
    client = openai.OpenAI(
        api_key=api_key,
        base_url="https://api.deepseek.com"
    )
    
    brief_prompt = f"""请为以下会议录音生成简洁的摘要（150字以内），概括会议的主要内容和结论：

{transcript[:3000]}

请用中文回复，格式如下：
【摘要】
...
"""

    detailed_prompt = f"""请为以下会议录音生成详细的分段纪要，按话题整理要点：

{transcript[:5000]}

请用中文回复，格式如下：
【详细纪要】
1. [话题标题]
   - 要点1
   - 要点2
2. [话题标题]
   ...
"""

    try:
        # 生成简要摘要
        brief_response = client.chat.completions.create(
            model="deepseek-chat",
            messages=[
                {"role": "system", "content": "你是一个专业的会议记录助手，擅长总结会议要点。"},
                {"role": "user", "content": brief_prompt}
            ],
            max_tokens=500,
            temperature=0.3
        )
        brief_summary = brief_response.choices[0].message.content
        
        # 生成详细纪要
        detailed_response = client.chat.completions.create(
            model="deepseek-chat",
            messages=[
                {"role": "system", "content": "你是一个专业的会议记录助手，擅长整理会议纪要。"},
                {"role": "user", "content": detailed_prompt}
            ],
            max_tokens=1500,
            temperature=0.3
        )
        detailed_summary = detailed_response.choices[0].message.content
        
        # 清理格式标记
        brief_summary = brief_summary.replace('【摘要】', '').strip()
        detailed_summary = detailed_summary.replace('【详细纪要】', '').strip()
        
        # 更新数据库
        db.update_meeting(
            meeting_id=meeting_id,
            brief_summary=brief_summary,
            detailed_summary=detailed_summary
        )
        
        # 更新本地变量并返回
        meeting['brief_summary'] = brief_summary
        meeting['detailed_summary'] = detailed_summary
        
        return meeting
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"生成摘要失败: {str(e)}")


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8501)


# ============ 实时转写 WebSocket ============

@app.websocket("/ws/realtime/{session_id}")
async def realtime_transcribe(websocket: WebSocket, session_id: str):
    """实时转写 WebSocket"""
    await websocket.accept()

    transcriber = None
    title = "实时录音"
    audio_chunks = []

    try:
        # 初始化转写器
        transcriber = processor.create_realtime_transcriber()

        # 保存会话
        with sessions_lock:
            realtime_sessions[session_id] = {
                "transcriber": transcriber,
                "title": title,
                "segments": [],
                "start_time": datetime.now()
            }

        # 发送就绪消息
        await websocket.send_json({
            "type": "ready",
            "session_id": session_id
        })

        # 处理音频数据
        while True:
            try:
                data = await websocket.receive_bytes()

                # 检查消息类型
                if data[:4] == b'TYPE':
                    # 解析消息类型
                    msg_type = data[4:8].decode('utf-8').strip()
                    if msg_type == 'TITLE':
                        title = data[8:].decode('utf-8')
                        with sessions_lock:
                            if session_id in realtime_sessions:
                                realtime_sessions[session_id]['title'] = title
                    elif msg_type == 'END__':
                        # 录音结束，保存会议
                        break
                    continue

                # 音频数据处理
                audio_chunks.append(data)

                # 转写音频块
                result = transcriber.process_audio_data(data)

                if result:
                    # 发送转写结果
                    await websocket.send_json({
                        "type": "segment",
                        "start": result.start,
                        "end": result.end,
                        "speaker": result.speaker,
                        "text": result.text
                    })

                    # 更新会话
                    with sessions_lock:
                        if session_id in realtime_sessions:
                            realtime_sessions[session_id]['segments'].append({
                                "start": result.start,
                                "end": result.end,
                                "speaker": result.speaker,
                                "text": result.text
                            })

            except WebSocketDisconnect:
                break

        # 录音结束，保存会议
        if transcriber and audio_chunks:
            # 保存录音文件
            project_root = os.path.dirname(os.path.abspath(__file__))
            data_dir = os.path.join(project_root, "data", "audio")
            os.makedirs(data_dir, exist_ok=True)

            audio_filename = f"{datetime.now().strftime('%Y%m%d_%H%M%')}{session_id}.wav"
            audio_path = os.path.join(data_dir, audio_filename)

            # 合并音频块并保存
            with wave.open(audio_path, 'wb') as wav_file:
                wav_file.setnchannels(1)
                wav_file.setsampwidth(2)
                wav_file.setframerate(16000)
                for chunk in audio_chunks:
                    wav_file.writeframes(chunk)

            # 创建会议记录
            meeting_id = datetime.now().strftime("%Y%m%d_%H%M%S") + "_" + session_id[:6]

            with sessions_lock:
                session_data = realtime_sessions.get(session_id, {})
                segments = session_data.get('segments', [])
                speakers = set(s['speaker'] for s in segments)
                duration = sum((s['end'] - s['start']) for s in segments) if segments else 0

            db.create_meeting(
                meeting_id=meeting_id,
                title=title,
                audio_path=audio_path,
                status="completed"
            )

            db.update_meeting(
                meeting_id=meeting_id,
                segments=segments,
                status="completed",
                speaker_count=len(speakers),
                duration=duration
            )

            # 发送保存完成消息
            await websocket.send_json({
                "type": "saved",
                "meeting_id": meeting_id
            })

    except Exception as e:
        print(f"实时转写错误: {e}")
        await websocket.send_json({
            "type": "error",
            "message": str(e)
        })
    finally:
        # 清理会话
        with sessions_lock:
            if session_id in realtime_sessions:
                del realtime_sessions[session_id]


@app.get("/api/realtime/{session_id}/status")
async def get_realtime_status(session_id: str):
    """获取实时转写状态"""
    with sessions_lock:
        if session_id in realtime_sessions:
            session = realtime_sessions[session_id]
            return {
                "session_id": session_id,
                "title": session['title'],
                "segments_count": len(session.get('segments', [])),
                "speaker_count": len(set(s['speaker'] for s in session.get('segments', [])))
            }
    return {"error": "会话不存在"}