import os import time import asyncio from typing import Dict, List, Tuple import uuid from playwright.async_api import async_playwright from openai import AsyncOpenAI import boto3 # FastAPI及び関連ライブラリ from fastapi import APIRouter, HTTPException # ⚙️ ローカルモジュールのインポート from config import logger, openai_api_key # [修正] 必要なスキーマをすべてインポートします from schemas import SpeechText, ChatRequest import db_module # =================================================== # 🔧 初期設定 # =================================================== client = AsyncOpenAI(api_key=openai_api_key) s3_client = boto3.client('s3') bucket_name = 'shanri-ai-chatbot-for-text-to-speech' router = APIRouter() # =================================================== # ⚙️ サービスロジック # =================================================== # --- スクレイピング関連のサービス --- class KeywordGenerator: # // 書籍のタイトルからキーワードを生成します async def generate_keywords(self, title: str) -> List[str]: prompt = f""" 以下の書籍のタイトルから、検索キーワードを生成してください。 以下の条件を満たすようにしてください： - 本の内容を推測できるキーワードを抽出 - 一般的な検索キーワードとして使えそうなものを選択 - 単語や短いフレーズを抽出 - 重複を避ける - 数字や括弧などの記号は含めない タイトル: {title} キーワードはカンマ区切りで出力してください。 """ try: response = await client.chat.completions.create( model="gpt-3.5-turbo", messages=[ {"role": "system", "content": "あなたは書籍の内容を分析し、適切な検索キーワードを生成する専門家です。"}, {"role": "user", "content": prompt} ], temperature=0.7, max_tokens=100 ) keywords_text = response.choices[0].message.content return [kw.strip() for kw in keywords_text.split(",") if kw.strip()] except Exception as e: logger.error(f"Error generating keywords for '{title}': {e}") return [] class RakutenScraper: # // 楽天ブックスのランキングをスクレイピングします def __init__(self): self.base_url = "https://books.rakuten.co.jp" self.keyword_generator = KeywordGenerator() async def scrape(self) -> List[Dict]: logger.info("Rakuten Books scraping process started.") try: await db_module.clear_ranking_data() async with async_playwright() as p: browser = await p.chromium.launch() page = await browser.new_page() try: page.set_default_timeout(60000) url = f"{self.base_url}/ranking/hourly/001/" await page.goto(url, wait_until="networkidle") selector = ".item" items = await page.query_selector_all(selector) if not items: return [] scraped_results = [] for item in items: title_element = await item.query_selector(".title") if not title_element: continue title = (await title_element.text_content()).strip() author_element = await item.query_selector(".author") author = (await author_element.text_content()).strip() if author_element else "不明" scraped_results.append({"title": title, "author": author, "source": "rakuten"}) await db_module.save_book_rankings(scraped_results) for book in scraped_results: keywords = await self.keyword_generator.generate_keywords(book["title"]) book["keywords"] = keywords if keywords: # // スクレイピングで取得したキーワードをkeywordsテーブルにも保存 await db_module.save_keywords_list(keywords) await db_module.link_keywords_to_book(book["title"], keywords) logger.info(f"Successfully processed {len(scraped_results)} items.") return scraped_results finally: await browser.close() except Exception as e: logger.error(f"Scraping failed: {e}") raise e # --- チャット関連のサービス --- SYSTEM_PROMPT_CHAT = """あなたは正確な情報提供を重視するAIアシスタントです。利用者が探している本について親切に質問し、どんな本が読みたいか幅広く聞き出してください。 重要な指示: 1. 推測や曖昧な情報は提供しない 2. 本のジャンル、雰囲気、感情、時代背景など、検索につながる特徴を聞き出してください。 3. 200文字程度に収まるように会話してください 重要: JSONは絶対に返さないでください。常に自然な日本語の文章のみで応答してください。""" async def extract_keywords_from_conversation(conversation_history: List[str]) -> List[str]: if not conversation_history: return [] try: keyword_prompt = "ユーザーが言及した単語をそのまま抽出してください。カンマ区切りで最大3つまで。" user_messages = [msg for i, msg in enumerate(conversation_history) if i % 2 == 0] all_user_text = " | ".join(user_messages) response = await client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": keyword_prompt}, {"role": "user", "content": f"会話内容: {all_user_text}"} ], max_tokens=100, temperature=0.3 ) keywords_text = response.choices[0].message.content.strip() return [k.strip() for k in keywords_text.split(',') if k.strip()][:3] except Exception as e: logger.error(f"AI keyword extraction from conversation failed: {e}") return [] async def generate_chat_response_and_keywords(message: str, conversation_history: List[str]) -> Tuple[str, List[str]]: # // 1. AIに応答を生成させる messages_for_response = [{"role": "system", "content": SYSTEM_PROMPT_CHAT}] for i, msg in enumerate(conversation_history): role = "user" if i % 2 == 0 else "assistant" messages_for_response.append({"role": role, "content": msg}) messages_for_response.append({"role": "user", "content": message}) try: response_completion = await client.chat.completions.create( model="gpt-4-turbo", messages=messages_for_response, max_tokens=500, temperature=0.7 ) assistant_response = response_completion.choices[0].message.content.strip() except Exception as e: logger.error(f"Error generating chat response: {e}") assistant_response = "申し訳ございません。応答の生成中にエラーが発生しました。" # // 2. AIにキーワードを抽出させる current_full_history = conversation_history + [message, assistant_response] keywords = await extract_keywords_from_conversation(current_full_history) return assistant_response, keywords # =================================================== # 🔊 音声合成関連ロジック # =================================================== async def synthesize_speech(text: str, user_id: str) -> str: # // (音声合成ロジックは変更なし) try: response = await client.audio.speech.create( model="tts-1", voice="nova", input=text, ) if not os.path.exists("tmp"): os.makedirs("tmp") unique_id = uuid.uuid4() audio_file_path = f"tmp/audio-{user_id}-{unique_id}.mp3" response.stream_to_file(audio_file_path) s3_key = f"speech-audio/{user_id}-{unique_id}.mp3" s3_client.upload_file(audio_file_path, bucket_name, s3_key) os.remove(audio_file_path) logger.info(f"음성 합성 완료 및 S3 업로드: {s3_key}") return f"https://{bucket_name}.s3.amazonaws.com/{s3_key}" except Exception as e: logger.error(f"음성 합성 실패: {e}") raise HTTPException(status_code=500, detail="음성 합성에 실패했습니다.") # =================================================== # 📡 APIエンドポイント # =================================================== @router.get("/health", tags=["System"]) async def health_check(): return {"status": "healthy"} @router.post("/yokohama/api/speech", tags=["Speech"]) async def speech(speech_data: SpeechText): # // (既存のコード) if not speech_data.text: raise HTTPException(status_code=400, detail="Text is required") audio_url = await synthesize_speech(speech_data.text, speech_data.chat_token) return {"audio_url": audio_url} @router.post("/yokohama/api/scrape-rankings", tags=["Scraping"]) async def scrape_rakuten_rankings(): # // (既存のコード) try: scraper = RakutenScraper() scraped_data = await scraper.scrape() return { "status": "success", "message": "Scraping completed.", "scraped_items_count": len(scraped_data) } except Exception as e: logger.error(f"Scraping endpoint failed: {e}") raise HTTPException(status_code=500, detail=str(e)) @router.get("/yokohama/api/random-keywords", tags=["Keywords"]) async def get_random_keywords_endpoint(limit: int = 4): # // (既存のコード) try: keywords = await db_module.get_random_keywords(limit=limit) if not keywords: return {"keywords": ["村上春樹", "ミステリー", "最新刊", "料理レシピ"]} return {"keywords": keywords} except Exception as e: logger.error(f"Failed to get random keywords: {e}") raise HTTPException(status_code=500, detail="Could not fetch keywords.") @router.get("/yokohama/api/chat/history/{session_id}", tags=["Chat"]) async def get_chat_history_by_session(session_id: str): """ // 指定されたsession_idの全会話履歴をDBから取得して返します。 """ history = await db_module.get_chat_history(session_id) return {"history": history} @router.post("/yokohama/api/chat/{session_id}", tags=["Chat"]) async def process_chat_message(session_id: str, request: ChatRequest): """ // ユーザーからの新しいメッセージを受け取り、AIの応答とキーワードを生成し、 // 会話履歴をDBに保存します。 """ user_message = request.message # // 1. DBから過去の会話履歴を取得 history_records = await db_module.get_chat_history(session_id) conversation_history = [] for record in history_records: if record["user_message"]: conversation_history.append(record["user_message"]) if record["assistant_response"]: conversation_history.append(record["assistant_response"]) # // 2. 応答生成とキーワード抽出を同時に行うロジックを呼び出し assistant_response, keywords = await generate_chat_response_and_keywords(user_message, conversation_history) # // 3. 新しい会話ターンをDBに保存 await db_module.save_chat_turn(session_id, user_message, assistant_response) # // 4. フロントエンドに応答とキーワードを返す return {"response": assistant_response, "keywords": keywords}