main.py

from typing import Literal
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from model.chatgpt_prompting import generate_sentence
from model.summary_dialogue_prompt import summary_dialogue
from model.recommend_check_prompt import recommend_check
from tts_model.google_tts import synthesize_speech_base64
from prometheus_fastapi_instrumentator import Instrumentator # 모니터링
from typing import Optional

app = FastAPI()

# CORS 설정 추가
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # 프론트엔드 Origin만 허용
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Instrumentator().instrument(app).expose(app) #모니터링

# 캐시 딕셔너리
cache = {}  # {"room_number": ["문장1", "문장2"]}
total_conversation_cache = {}  # {"room_number": ["전체 문장1", "전체 문장2"]}

# 문장 데이터를 위한 Pydantic 모델 정의
class DialogueRequest(BaseModel):
    room_number: str
    sentence: Optional[str] = None  # 기본값 None으로 설정
    
class SummaryRequest(BaseModel):
    room_number: str
    sentence: Optional[str] = None  # 기본값 None으로 설정

# TTS 요청 바디 스키마 정의
class TTSRequest(BaseModel):
    room_number: str
    voice_type: Literal["male1", "male2", "female1", "female2"]
    text: str

@app.get("/health")
async def health_check():
    return {"message": "Hello ParroTalk!"}

# 캐시에 문장 추가 (중복 확인)
def add_sentence_to_cache(room_number: str, sentence: str):
    # 캐시 초기화
    if room_number not in cache:
        cache[room_number] = []
    if room_number not in total_conversation_cache:
        total_conversation_cache[room_number] = []
    
    # 중복된 문장 추가 방지
    if not cache[room_number] or cache[room_number][-1] != sentence:
        cache[room_number].append(sentence)
    
    # 전체 대화 누적 캐시에 추가
    total_conversation_cache[room_number].append(sentence)

@app.post("/recommendations")
async def get_recommendations(request: DialogueRequest):
    sentence = request.sentence.strip() if request.sentence else ""
    room_number = request.room_number

    if not sentence:  # sentence가 None이거나 비어 있을 경우
        return {
            "room_number": room_number,
            "sentence": "",
            "is_recommend": False,
            "recommendations": []
        }

    try:
        # 문장 캐시에 추가
        add_sentence_to_cache(room_number, sentence)

        # 룸 넘버별 캐시에서 문장 조합
        combined_text = " ".join(cache[room_number])
        total_combined_text = " ".join(total_conversation_cache[room_number])  # 전체 누적 대화 텍스트

        # 추천 여부 판단
        is_recommend_combined = recommend_check(combined_text)

        if is_recommend_combined:
            # 추천 가능하면 캐시를 비우고 결과 반환
            cache[room_number].clear()

            # 추천 실행 (전체 누적 문장 및 최신 문장 활용)
            result = generate_sentence(total_combined_text, sentence)
            # 반환값 정규화
            if '응답' in result:
                if isinstance(result['응답'], dict):
                    recommendations = [
                        result['응답'].get('추천 문장 1', ""),
                        result['응답'].get('추천 문장 2', ""),
                        result['응답'].get('추천 문장 3', "")
                    ]
                elif isinstance(result['응답'], list):
                    recommendations = [
                        result['응답'][0].get('추천 문장 1', ""),
                        result['응답'][0].get('추천 문장 2', ""),
                        result['응답'][0].get('추천 문장 3', "")
                    ]
                else:
                    recommendations = []
            else:
                recommendations = [
                    result.get('추천 문장 1', ""),
                    result.get('추천 문장 2', ""),
                    result.get('추천 문장 3', "")
                ]

            return {
                "room_number": room_number,
                "sentence": sentence,
                "is_recommend": True,
                "recommendations": recommendations
            }
        else:
            # 추천 불가능하면 결합된 문장만 반환
            return {
                "room_number": room_number,
                "sentence": combined_text,
                "is_recommend": False,
                "recommendations": []
            }
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error checking recommendation: {str(e)}")

@app.post("/summary")
async def summarize_dialogue(request: SummaryRequest):
    dialogue_content = request.sentence.strip()

    # 입력값이 공백인 경우 처리
    if not dialogue_content:
        return {
            "summary": [],
            "todo": []
        }

    summary_result = summary_dialogue(dialogue_content)

    if summary_result is None:
        return {
            "summary": [],
            "todo": []
        }

    return {
        "summary": summary_result['summary'],
        "todo": summary_result['todo']
    }


# Text-to-Speech API 엔드포인트
@app.post("/tts")
async def synthesize_tts(request: TTSRequest):
    try:
        # Google TTS 호출
        audio_base64 = synthesize_speech_base64(request.voice_type, request.text)
        return {"status": "success", "audio_base64": audio_base64}
    except ValueError as ve:
        raise HTTPException(status_code=400, detail=str(ve))
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")
    
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8080)