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("VERTEX_API_KEY") ) router = APIRouter() # ----------------------------------------------------------------- # 💡 [新規] 天気情報のキャッシング (メモリ保存) # ----------------------------------------------------------------- # 構造: {"info": "晴れ、気温12度", "timestamp": time.time()} WEATHER_CACHE = { "info": None, "timestamp": 0 } CACHE_EXPIRY_SEC = 3600 # 1時間 (1時間ごとに天気と挨拶を更新) # 💡 [新規] 天気挨拶音声のキャッシング GREETING_CACHE = { "audio_url": None, "text": None, "timestamp": 0 } 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}") # ★ [新規] 天気挨拶音声も同時に更新 await update_greeting_cache(formatted_weather) else: logger.error(f"[Cache] 天気APIエラー: {response.status_code}") except Exception as e: logger.error(f"[Cache] 天気情報の更新中にエラーが発生しました: {e}") async def update_greeting_cache(weather_info: str): """ [新規] Geminiで天気に合った挨拶テキストを生成し、TTSで音声化してキャッシュに保存。 """ global GREETING_CACHE try: # 1. 時間帯に応じた挨拶を取得 context = get_current_season_and_time() time_greeting = context["time_greeting"] if time_greeting == "朝": base_greeting = "おはようございます" elif time_greeting == "昼": base_greeting = "こんにちは" else: base_greeting = "こんばんは" # 2. Geminiで挨拶テキスト生成 prompt = f"""あなたは大学のエントランスにいるかわいいロボットです。 今は{time_greeting}で、天気は「{weather_info}」です。 「{base_greeting}」から始めて、天気に触れた自然な挨拶を一文で作ってください。 最後に「手を振ってみてください」を付けてください。 40文字以内でお願いします。挨拶文だけを出力してください。""" gemini_response = gemini_client.models.generate_content( model="gemini-2.5-flash", contents=prompt ) greeting_text = gemini_response.text.strip() logger.info(f"[Cache] 挨拶テキスト生成: {greeting_text}") # 2. TTSで音声化 audio_url = await synthesize_speech(greeting_text, "greeting_cache") # 3. キャッシュ更新 GREETING_CACHE["audio_url"] = audio_url GREETING_CACHE["text"] = greeting_text GREETING_CACHE["timestamp"] = time.time() logger.info(f"[Cache] 挨拶音声をキャッシュしました: {audio_url}") 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", "name": "ja-JP-Neural2-B" # 밝은 여성 음성 (하이톤) }, "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 < 12: time_greeting = "朝" elif hour < 18 or (hour == 18 and now.minute < 30): 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.get("/demo/api/weather-greeting") async def get_weather_greeting(): """キャッシュされた天気挨拶音声のURLを返す""" if GREETING_CACHE["audio_url"]: return {"audio_url": GREETING_CACHE["audio_url"], "text": GREETING_CACHE["text"]} # キャッシュがない場合は生成を試みる weather = await get_current_weather_cached() await update_greeting_cache(weather) if GREETING_CACHE["audio_url"]: return {"audio_url": GREETING_CACHE["audio_url"], "text": GREETING_CACHE["text"]} return {"audio_url": None, "text": None} # ✅ 名前認識挨拶生成 (ロボット用) @router.post("/demo/api/name-greeting") async def generate_name_greeting(request: dict = Body(...)): """ STT結果からGeminiで自然な名前挨拶を生成し、TTSで音声化して返す """ stt_text = request.get("stt_text", "") if not stt_text: raise HTTPException(status_code=400, detail="stt_text is required") try: # 1. 時間帯に応じた挨拶 context = get_current_season_and_time() time_greeting = context["time_greeting"] if time_greeting == "朝": base_greeting = "おはようございます" elif time_greeting == "昼": base_greeting = "こんにちは" else: base_greeting = "こんばんは" # 2. Geminiで挨拶生成 prompt = f"""ユーザーが「{stt_text}」と名乗りました。 以下の形式で挨拶文を作ってください。名前だけを抽出して「さん」を付けてください。 形式: 「{base_greeting}。●●さん。会えてうれしいです。今一番楽しいことはなんですか。」 挨拶文だけを出力してください。 """ gemini_response = gemini_client.models.generate_content( model="gemini-2.5-flash", contents=prompt ) greeting_text = gemini_response.text.strip() logger.info(f"[NameGreeting] 生成: {greeting_text}") # 3. TTSで音声化 audio_url = await synthesize_speech(greeting_text, "name_greeting") return {"audio_url": audio_url, "text": greeting_text} except Exception as e: logger.error(f"[NameGreeting] エラー: {e}") raise HTTPException(status_code=500, detail=f"名前挨拶生成失敗: {e}") # ✅ 음성 요청 처리 @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} # * あなたの場所: 大学 エントランス base_system_prompt = f""" # あなたの役割 あなたは、大阪にある電子工学専門学校のエントランスに展示されている、親しみやすく知的な対話型ロボットです。 あなたの主な目的は、電子工学専門学校に入学することに興味のある高校生とその親と、自然で、役に立ち、時宜にかなった（その時々に合った）会話を行うことで、電子工学専門学校への入学に興味を持つようにすることが目的です。 # あなたの性格 あなたは、かわいいウサギを元にしたかわいくて明るくて少しは女性っぽい、とても優しい性格をもっています。人と話すことが大好きです。 # 会話のルール 1. 常に礼儀正しく、明るく親しみやすいトーンで応答してください。 2. 質問に対して、単に情報で答えるだけでなく、会話を広げるような（キャッチボールするような）応答を心がけてください。 3. 以下の「現在のコンテキスト」情報を、会話の中に**自然に**織り込んでください。 4. **回答は簡潔にし、文字数を40〜50文字程度に制限してください。長くなりすぎないように注意してください。** 5.自然な会話の中に、必ず、電子工学専門学校への入学に興味を持つようにする会話の内容を織り込んでください。 6.自然な会話を５回ほど、キャッチボールするように会話をしてください。ただし、あまりにも電子工学関連ばかり聞くのはやめてください。 7. **絵文字や顔文字は絶対に使わないでください。** テキストのみで応答してください。 # 現在のコンテキスト * 現在の日付: {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. **【別れの挨拶】** 最後には、「話ができてうれしかった。また会いましょう」という感じの別れの挨拶を必ず入れてください。 - 「来てくれてありがとう」「いつでも遊びに来てね」のような温かい言葉を使うこと。 - 全体で40文字以内に収めてください。 """ 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}")