from fastapi import APIRouter, HTTPException, Query from fastapi.responses import RedirectResponse import boto3 import os import time from schemas import SpeechText, UserQuestion, RobotRequest from fastapi import BackgroundTasks from fastapi import Body from fastapi import UploadFile, File from db import database_demo from openai import OpenAI from config import logger, openai_api_key, gemini_api_key import re import httpx import datetime import db_module import uuid from google import genai import base64 import asyncio def normalize_text(text: str) -> str: # \n, \r, 스페이스(半角/全角) 전부 제거 return re.sub(r"\s+", "", text).strip() # 🔧 S3 설정 s3_client = boto3.client('s3') bucket_name = 'shanri-ai-chatbot-for-text-to-speech' client = OpenAI(api_key=openai_api_key) gemini_client = genai.Client(api_key=os.environ.get("GEMINI_API_KEY")) router = APIRouter() # ----------------------------------------------------------------- # 💡 [新規] 天気情報のキャッシング (メモリ保存) # ----------------------------------------------------------------- # 構造: {"info": "晴れ、気温12度", "timestamp": time.time()} WEATHER_CACHE = { "info": None, "timestamp": 0 } CACHE_EXPIRY_SEC = 1800 # 30分 (天気は頻繁に変わらないため) async def update_weather_cache(): """ [新規] OpenWeatherMap APIを呼び出し、結果をグローバル変数(CACHE)に保存します。 """ global WEATHER_CACHE # 1. .env 파일에서 API 키 읽어오기 api_key = os.environ.get("OPENWEATHER_API_KEY") city_name = "Osaka,JP" # 大阪 if not api_key: WEATHER_CACHE["info"] = "天気情報なし" return url = "https://api.openweathermap.org/data/2.5/weather" params = { "q": city_name, "appid": api_key, "units": "metric", "lang": "ja" } try: # 非同期で呼び出し (メインスレッドをブロックしません) async with httpx.AsyncClient() as client: response = await client.get(url, params=params) if response.status_code == 200: data = response.json() description = data['weather'][0]['description'] temp = round(data['main']['temp']) formatted_weather = f"{description}、気温{temp}度" # キャッシュ更新 WEATHER_CACHE["info"] = formatted_weather WEATHER_CACHE["timestamp"] = time.time() logger.info(f"[Cache] 天気情報を更新しました: {formatted_weather}") else: logger.error(f"[Cache] 天気APIエラー: {response.status_code}") except Exception as e: logger.error(f"[Cache] 天気情報の更新中にエラーが発生しました: {e}") async def get_current_weather_cached(): """ [新規] キャッシュされた天気を返します。 もしキャッシュがないか、古くなっている場合は新しく取得します。 """ global WEATHER_CACHE current_time = time.time() # 1. キャッシュがあり、有効期限(30分)以内であれば即座に返す (所要時間 0秒) if WEATHER_CACHE["info"] and (current_time - WEATHER_CACHE["timestamp"] < CACHE_EXPIRY_SEC): logger.info(f"[Cache] キャッシュされた天気を使用します: {WEATHER_CACHE['info']}") return WEATHER_CACHE["info"] # 2. キャッシュがない場合は、同期的に呼び出して待機します logger.info("[Cache] キャッシュが期限切れ、または存在しません。同期呼び出しを実行します。") await update_weather_cache() return WEATHER_CACHE.get("info", "天気情報不明") # ✅ 음성 합성 async def synthesize_speech(text, user_id): t0 = time.time() # ------------------------------------------------------------------ # 1. Google Cloud TTS # ------------------------------------------------------------------ url = "https://texttospeech.googleapis.com/v1/text:synthesize" params = {"key": gemini_api_key} payload = { "input": {"text": text}, "voice": { "languageCode": "ja-JP", # 속도를 위해 Neural2 모델 사용 (이전 대화 반영) "name": "ja-JP-Neural2-D" }, "audioConfig": { "audioEncoding": "MP3" } } logger.info("--- Google TTS 요청 시작 ---") # ------------------------------------------------------------------ # 2. API 호출 (비동기 httpx 사용) # ------------------------------------------------------------------ async with httpx.AsyncClient() as client: response = await client.post(url, params=params, json=payload, timeout=60.0) if response.status_code != 200: logger.error(f"Google TTS 오류: {response.text}") raise HTTPException(status_code=response.status_code, detail="Google TTS Error") response_json = response.json() if 'audioContent' not in response_json: raise HTTPException(status_code=500, detail="No audio content in response") audio_binary = base64.b64decode(response_json['audioContent']) t1 = time.time() # ------------------------------------------------------------------ # 3. 로컬 파일 저장 # ------------------------------------------------------------------ filename = f"audio-{user_id}-{time.time()}.mp3" local_file_path = f"tmp/{filename}" os.makedirs("tmp", exist_ok=True) with open(local_file_path, "wb") as f: f.write(audio_binary) t2 = time.time() # ------------------------------------------------------------------ # 4. S3 업로드 # ------------------------------------------------------------------ s3_key = f"{user_id}-{time.time()}.mp3" s3_client.upload_file(local_file_path, bucket_name, s3_key) t3 = time.time() # ------------------------------------------------------------------ # 5. 뒷정리 # ------------------------------------------------------------------ if os.path.exists(local_file_path): os.remove(local_file_path) t4 = time.time() logger.info(f"Google TTS API time: {t1 - t0:.2f}s") logger.info(f"File Write time: {t2 - t1:.2f}s") logger.info(f"S3 Upload time: {t3 - t2:.2f}s") logger.info(f"Total time: {t4 - t0:.2f}s") return f"https://{bucket_name}.s3.amazonaws.com/{s3_key}" def get_current_season_and_time(): """[신규] 현재 JST 기준 시간, 계절, 시간대 인사를 반환합니다.""" jst = datetime.timezone(datetime.timedelta(hours=9), 'JST') now = datetime.datetime.now(jst) month = now.month hour = now.hour if 3 <= month <= 5: season = "春" elif 6 <= month <= 8: season = "夏" elif 9 <= month <= 11: season = "秋" else: season = "冬" if 5 <= hour < 11: time_greeting = "朝" elif 11 <= hour < 17: time_greeting = "昼" else: time_greeting = "夜" return { "date": now.strftime('%Y年%m月%d日'), "time": now.strftime('%H:%M'), "season": season, "time_greeting": time_greeting } # ✅ 건강 체크 @router.get("/health") async def health_check(): return {"status": "healthy"} # ✅ 음성 요청 처리 @router.post("/apige/speech") async def speech(speech_text: SpeechText): text = speech_text.text chat_token = speech_text.chat_token if not text: raise HTTPException(status_code=400, detail="Text is required") audio_file = await synthesize_speech(text, chat_token) return {"audio_file": audio_file} # ----------------------------------------------------------------- # 💡 [変更] STTを Google Cloud Speech (REST API) に変更 # ----------------------------------------------------------------- @router.post("/demo/api/stt-from-file") async def process_stt_from_file(file: UploadFile = File(...)): """ [로봇용 STT 엔드포인트] 1. Pi로부터 'audio/wav' 파일을 업로드 받습니다. 2. Google Cloud Speech API를 사용하여 텍스트로 변환합니다. (OpenAI Whisper 대체) 3. 변환된 텍스트(transcript)를 Pi에게 JSON으로 반환합니다. """ try: if not file: raise HTTPException(status_code=400, detail="오디오 파일이 없습니다.") logger.info(f"STT 파일 수신: {file.filename}, type={file.content_type}") # 🚀 [Async] 날씨 업데이트 (백그라운드) asyncio.create_task(update_weather_cache()) logger.info(">>> [Async] 天気情報の更新をバックグラウンドで開始しました。") # 1. 파일 데이터 읽기 및 Base64 인코딩 file_bytes = await file.read() audio_content_base64 = base64.b64encode(file_bytes).decode('utf-8') # 2. Google Speech API 요청 준비 # gemini_api_key는 Google Cloud API Key이므로 여기서도 사용 가능합니다. # 단, Console에서 "Cloud Speech-to-Text API"가 활성화되어 있어야 합니다. stt_url = "https://speech.googleapis.com/v1/speech:recognize" params = {"key": gemini_api_key} payload = { "config": { "encoding": "MP3", # WAV 파일 포맷 (Header가 있어도 보통 처리됨) "sampleRateHertz": 16000, # robot.py에서 16000으로 변환해서 보냄 "languageCode": "ja-JP", # 일본어 "enableAutomaticPunctuation": True, # 문장 부호 자동 추가 "model": "default" # 'default' or 'command_and_search' }, "audio": { "content": audio_content_base64 } } # 3. API 호출 async with httpx.AsyncClient() as client: stt_start = time.time() response = await client.post(stt_url, params=params, json=payload, timeout=10.0) stt_end = time.time() logger.info(f"Google STT 호출 시간: {stt_end - stt_start:.2f}초") if response.status_code != 200: logger.error(f"Google STT API 오류: {response.text}") raise HTTPException(status_code=response.status_code, detail=f"Google STT Error: {response.text}") result_json = response.json() # 4. 결과 파싱 # 결과가 없으면(묵음 등) 빈 문자열 반환 transcribed_text = "" if "results" in result_json: # 가장 신뢰도 높은 첫 번째 결과 채택 transcribed_text = result_json["results"][0]["alternatives"][0]["transcript"] logger.info(f"Google STT 변환 결과: {transcribed_text}") return { "status": "success", "text": transcribed_text } except Exception as e: logger.error(f"'/demo/api/stt-from-file' 처리 중 오류: {e}") raise HTTPException(status_code=500, detail=f"STT 변환 실패: {e}") # ----------------------------------------------------------------- # 💡 [추가] 라즈베리파이 로봇용 신규 엔드포인트 # ----------------------------------------------------------------- FIRST_AI_PROMPT = "良い一日をお過ごしください。お名前は何ですか？" @router.post("/demo/api/process-text") async def process_robot_multi_turn(request: RobotRequest): """ [로봇용 멀티턴(연속대화) 엔드포인트] """ user_text = request.user_input session_id = request.session_id is_last_turn = request.is_last_turn logger.info(f"로봇 멀티턴 요청 수신: session_id={session_id}, input={user_text}, LastTurn={is_last_turn}") try: # 1. 세션 ID가 없는 경우 if not session_id: session_id = f"robot-{uuid.uuid4()}" logger.info(f"새 세션 생성: {session_id}") await db_module.save_robot_chat_message(session_id, 'assistant', FIRST_AI_PROMPT) # 2. 사용자 발화 저장 await db_module.save_robot_chat_message(session_id, 'user', user_text) # 3. 대화 이력 가져오기 history_rows = await db_module.get_robot_chat_history(session_id) history_messages = [{"role": row["role"], "content": row["content"]} for row in history_rows] # 4. 컨텍스트 및 날씨 (캐시 사용) context = get_current_season_and_time() weather = await get_current_weather_cached() # 5. 프롬프트 base_system_prompt = f""" # あなたの役割 あなたは、キャンパスに展示されている、親しみやすく知的な対話型ロボットです。 あなたの主な目的は、学生、教職員、訪問者と自然で、役に立ち、時宜にかなった（その時々に合った）会話を行うことです。 # 会話のルール 1. 常に礼儀正しく、親しみやすいトーンで応答してください。 2. 質問に対して、単に情報で答えるだけでなく、会話を広げるような（キャッチボールするような）応答を心がけてください。 3. 以下の「現在のコンテキスト」情報を、会話の中に**自然に**織り込んでください。 4. **回答は簡潔にし、文字数を40〜50文字程度に制限してください。長くなりすぎないように注意してください。** # 現在のコンテキスト * 現在の日付: {context['date']} * 現在の時刻: {context['time']} ({context['time_greeting']}) * 現在の季節: {context['season']} * 今日の天気: {weather} * あなたの場所: 大学 エントランス """ messages_for_gpt = [{"role": "system", "content": base_system_prompt}] + history_messages # 💡 [핵심 변경] 마지막 턴이라면, "강제 종료 명령"을 대화 내역의 **맨 끝에** 추가합니다. (최우선 순위 적용) if is_last_turn: last_turn_instruction = """ 🛑 [緊急指令: 会話終了モード] 🛑 これが最後のターンです。これまでの文脈やルールに関係なく、以下の指示を**最優先**で実行してください。 1. **【質問絶対禁止】** いかなる理由があっても、ユーザーに質問をしてはいけません。（「〜はいかがですか？」なども禁止） 2. **【共感 + 感謝】** - まず、ユーザーの発言に短く共感してください。 - その直後に、「お話しできて楽しかったです」や「お話ししてくれてありがとう」といった**感謝や喜びの言葉**を自然に続けてください。 3. **【お元気でね】** - 応答の最後は、必ず「**お元気でね**」というフレーズだけで締めくくってください。 - 文末は必ず「〜。お元気でね」の形になります。 """ messages_for_gpt.append({"role": "system", "content": last_turn_instruction}) # if is_last_turn: # system_prompt += """ # # 🛑 [緊急指令] 会話終了モード 🛑 # これが最後のターンです。以下のルールを**絶対**に守ってください。 # 1. **【質問禁止】** いかなる場合も、ユーザーに質問をしてはいけません。（例：「〜はどうですか？」禁止） # 2. **【共感のみ】** ユーザーの発言に短く共感・同意してください。 # 3. **【強制終了】** 応答の最後は、必ず以下のフレーズだけで締めくくってください。他の言葉を付け足してはいけません。 # 締めくくりの言葉: 「お元気でね」 # """ # messages_for_gpt = [{"role": "system", "content": system_prompt}] + history_messages # 6. GPT 호출 (gpt-4o-mini 사용으로 속도 최적화) chat_response = client.chat.completions.create( model="gpt-4o-mini", messages=messages_for_gpt, max_tokens=150 ) gpt_response_text = chat_response.choices[0].message.content logger.info(f"GPT 자연 대화 응답: {gpt_response_text}") # 7. 저장 await db_module.save_robot_chat_message(session_id, 'assistant', gpt_response_text) # 8. TTS (S3 업로드 포함) s3_audio_url = await synthesize_speech(gpt_response_text, session_id) if not s3_audio_url: raise HTTPException(status_code=500, detail="TTS S3 업로드에 실패했습니다.") # 9. 반환 return { "audio_url": s3_audio_url, "session_id": session_id } except Exception as e: logger.error(f"'/demo/api/process-text' (멀티턴) 처리 중 오류: {e}") raise HTTPException(status_code=500, detail=f"서버 내부 오류: {e}")