from fastapi import APIRouter, HTTPException from fastapi.responses import RedirectResponse import openai from openai import OpenAI import boto3 import os import time from bs4 import BeautifulSoup from schemas import SpeechText, UserQuestion from config import logger, openai_api_key from datetime import datetime import chromadb from chromadb.config import Settings import re from datetime import datetime import ast from urllib.parse import urljoin, urlparse from selenium import webdriver from selenium.webdriver.chrome.options import Options from fastapi import BackgroundTasks # 🔧 ChromaDB 설정 chroma_client = chromadb.HttpClient( host="localhost", port=8001, settings=Settings(chroma_api_impl="rest") ) collection_saved_qna = chroma_client.get_collection("segue_saved_qna11") collection_qna = chroma_client.get_collection("segue_qna_by_question3") collection_profile = chroma_client.get_collection("segue_profile_split6") # 🔧 OpenAI 설정 openai.api_key = openai_api_key client = OpenAI(api_key=openai_api_key) # 🔧 S3 설정 s3_client = boto3.client('s3') bucket_name = 'shanri-ai-chatbot-for-text-to-speech' router = APIRouter() # ✅ 음성 합성 async def synthesize_speech(text, user_id): response = client.audio.speech.create( model="tts-1", voice="nova", input=text, ) audio_file = f"tmp/audio-{user_id}-{time.time()}.mp3" with open(audio_file, 'wb') as f: for chunk in response.iter_bytes(): f.write(chunk) s3_key = f"{user_id}-{time.time()}.mp3" s3_client.upload_file(audio_file, bucket_name, s3_key) os.remove(audio_file) return f"https://{bucket_name}.s3.amazonaws.com/{s3_key}" # ✅ page_type 복정 함수 def correct_page_type_by_keyword(question: str, original_page_type: str) -> str: if any(k in question for k in ["キャッシュフロー", "貸借対照表", "財務諸表"]): print("[DEBUG] キーワードにより 'finance' に上書きされました") return "finance" elif any(k in question for k in ["売上高", "営業利益", "経常利益", "純利益"]): print("[DEBUG] キーワードにより 'performance' に上書きされました") return "performance" elif any(k in question for k in ["商品", "製品"]): print("[DEBUG] キーワードにより 'business' に上書きされました") return "business" return original_page_type # ✅ Profile 검색 (메타데이터 기반 전체 반환) async def search_profile_context_by_metadata(question: str) -> str | None: # ✅ Step 1: GPT로 page_type 추론 system_prompt = """ 以下のカテゴリの中から、質問が該当するカテゴリを1つ選び、カテゴリ名（英語）だけを出力してください。 - overview - history - welfare - training - message - performance - executives - sustainability - strategy - stock - business - finance - philosophy 【注意】 - 売上高、営業利益、経常利益、純利益 などの業績に関する質問は「performance」としてください。 - 財務諸表、貸借対照表、キャッシュフロー に関する質問は「finance」としてください。 回答はカテゴリ名だけで、他の言葉は含めないでください。 """ chat = client.chat.completions.create( model="gpt-4o", messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": question} ] ) page_type = chat.choices[0].message.content.strip().lower() page_type = correct_page_type_by_keyword(question, page_type) print(f"[DEBUG] 推論された page_type: {page_type}") # ✅ Step 2: GPT 또는 룰로 year 추출 # def extract_year_range(question: str) -> list[int]: # current_year = datetime.now().year # years = set() # # ① 명시적 연도 (예: 2021年, 2022年) # matches = re.findall(r"(20[0-3][0-9])年", question) # years.update(int(y) for y in matches) # # ② 연도 범위 (예: 2022〜2023年 or 2022-2023年) # range_match = re.search(r"(20[0-3][0-9])\s*[〜~\-]\s*(20[0-3][0-9])年", question) # if range_match: # start, end = int(range_match.group(1)), int(range_match.group(2)) # years.update(range(start, end + 1)) # # ③ 「過去3年」「過去5年」 등의 표현 # past_match = re.search(r"過去(\d)年", question) # if past_match: # n = int(past_match.group(1)) # years.update(current_year - i for i in range(1, n + 1)) # # ④ 「昨年」「一昨年」 # if "昨年" in question: # years.add(current_year - 1) # if "一昨年" in question: # years.add(current_year - 2) # return sorted(years) def extract_year_range(question: str) -> list[int]: prompt = """ 次の質問文に関連する西暦年（例: 2021, 2022 など）をすべて配列で抽出してください。 質問文には過去の年や年数に関する言及が含まれている場合があります。 数値の配列だけを返してください。形式は以下に従ってください。 例）[2021, 2022, 2023] 質問文: """ + question response = client.chat.completions.create( model="gpt-4o", messages=[ {"role": "system", "content": "あなたは日本語テキストから年（西暦）を抽出する専門家です。"}, {"role": "user", "content": prompt} ] ) content = response.choices[0].message.content.strip() try: years = ast.literal_eval(content) # 🎯 유효한 int 리스트인지 확인 if isinstance(years, list) and all(isinstance(y, int) and 2000 <= y <= datetime.now().year + 1 for y in years): return sorted(set(years)) except Exception as e: logger.warning(f"[GPT] 年抽出失敗: {content}, error: {e}") # ❗ 실패 시 안전하게 빈 리스트 반환 return [] year_list = extract_year_range(question) # ✅ Step 3: where 조건 구성 if page_type == "performance" and year_list: where = { "$and": [ {"page_type": page_type}, {"year": {"$in": year_list}} ] } else: where = {"page_type": page_type} # ✅ Step 4: 문서 전체 검색 # results = collection_profile.get( # where=where, # limit=100 # ) response = client.embeddings.create( model="text-embedding-3-small", input=question ) query_embedding = response.data[0].embedding results = collection_profile.query( query_embeddings=[query_embedding], n_results=100, where=where ) docs = results.get("documents", []) if not docs or not docs[0]: return None joined_docs = "\n\n".join(docs[0]) return joined_docs # ✅ 저장된 QNA 검색 async def search_saved_qna_answer(question: str) -> tuple[str, float]: collection = chroma_client.get_collection("segue_saved_qna11") response = client.embeddings.create( model="text-embedding-3-small", input=question ) query_embedding = response.data[0].embedding results = collection.query( query_embeddings=[query_embedding], n_results=1 ) if not results or not results['documents'] or not results['documents'][0]: return "情報がありません。", 1.0 answer = results['metadatas'][0][0].get("answer", "") distance = results['distances'][0][0] return answer, distance # ✅ 재무재표 검색 async def search_qna_answer(question: str) -> tuple[str, float]: response = client.embeddings.create( model="text-embedding-3-small", input=question ) query_embedding = response.data[0].embedding results = collection_qna.query( query_embeddings=[query_embedding], n_results=1 ) if not results or not results['documents'] or not results['documents'][0]: return "情報がありません。", 1.0 answer = results['metadatas'][0][0].get("answer", "") distance = results['distances'][0][0] return answer, distance async def choose_hompage_url(question): """homepage_url을 선택하는 함수""" messages = [ {"role": "system", "content": "質問を分析し、回答を作成する際に必要な情報を持つURLを返す必要があります。質問を見てURLを選択してください。"}, {"role": "system", "content": "最も情報がある可能性が高いURLを配列で返す必要があります。"}, {"role": "system", "content": "関数の戻り値のように、配列のみを返す必要があります。 例) ['https://segue-g.jp/company/boardmember/index.html','https://segue-g.jp/ir/results/settle.html']"}, {"role": "system", "content": "企業理念 : https://segue-g.jp/company/sdgs/index.html"}, {"role": "system", "content": "IR 情報 : https://segue-g.jp/ir/results/index.html"}, {"role": "system", "content": "財務、業績報告、売上 : https://segue-g.jp/ir/results/settle.html"}, {"role": "system", "content": "株式情報 : https://segue-g.jp/ir/stock/index.html"}, {"role": "system", "content": "当社の強み : https://segue-g.jp/ir/investor/strong_point/index.html"}, {"role": "system", "content": "成長戦略 : https://segue-g.jp/ir/investor/strategy/index.html"}, {"role": "system", "content": "会社概要 : https://segue-g.jp/company/basic/index.html"}, {"role": "system", "content": "企業理念 : https://segue-g.jp/company/brand/index.html"}, {"role": "system", "content": "沿革 : https://segue-g.jp/company/history/index.html"}, {"role": "system", "content": "役員一覧 : https://segue-g.jp/company/boardmember/index.html"}, {"role": "system", "content": "事業紹介 : https://segue-g.jp/business/index.html"}, {"role": "user", "content": "質問 : " + question} ] response = client.chat.completions.create( model="gpt-4o", messages=messages ) return response.choices[0].message.content # ✅ 특정 웹사이트(segue-g) 크롤링 함수 def init_driver(): options = Options() options.add_argument("--headless") options.add_argument("--disable-gpu") options.add_argument("--no-sandbox") return webdriver.Chrome(options=options) def is_same_domain(base_url, test_url): return urlparse(base_url).netloc == urlparse(test_url).netloc async def fetch_all_pages_with_selenium(url_list): """url_list에 있는 각 페이지의 sections_group HTML만 크롤링 (내부 링크 X)""" driver = init_driver() all_html = "" try: for url in url_list: try: print(f"\n🟢 크롤링 중: {url}") driver.get(url) soup = BeautifulSoup(driver.page_source, "html.parser") # ✅ sections_group 클래스를 가진 요소만 가져오기 sections = soup.find_all(class_="sections_group") page_html = "" for section in sections: # ✅ graph 클래스가 포함된 div 제거 for graph_div in section.find_all("div", class_="graph"): graph_div.decompose() # 해당 요소를 제거 page_html += section.prettify() all_html += f"\n{page_html}\n\n" except Exception as e: print(f"[에러] {url}: {e}") return all_html finally: driver.quit() # ✅ GPT 대답 생성 로직 (QnA → Profile + GPT → 정보 없음) async def generate_gpt_answer(question: str) -> str: # ✅ Step 1: 재무 QnA 먼저 검색 qna_answer, qna_distance = await search_qna_answer(question) logger.info(f"📊 財務類似質問 - distance: {qna_distance}, answer: {repr(qna_answer)}") if qna_distance < 0.275: logger.info("📊 財務QnA回答を返します。") return qna_answer # ✅ Step 2: 저장된 QnA 검색 answer, distance = await search_saved_qna_answer(question) logger.info(f"🎯 類似質問あり - distance: {distance}, answer: {repr(answer)}") if distance < 0.275: logger.info("🎯 類似質問が見つかり、保存された回答を返します。") return answer # ✅ Step 3: Profile에서 검색 (엑셀 기반 정보) profile_context = await search_profile_context_by_metadata(question) if profile_context: logger.info("📄 プロフィール情報から回答を生成します。") messages = [ {"role": "system", "content": "あなたは「セグエグループ」の会社紹介および就職希望者向けのチャットボットです。"}, {"role": "system", "content": "以下のルールに従って、ユーザーと会話してください。"}, {"role": "system", "content": "1.会話スタイル：丁寧で信頼感のあるビジネス口調で応答してください。"}, {"role": "system", "content": "2.トーン：敬語を用いながらも、丁寧で親しみやすい表現を心がけてください。"}, {"role": "system", "content": "3.情報の制限：返答に使用できるのは、提供されたプロフィールExcelファイルの情報のみとします。その他の情報は使用しないでください。"}, {"role": "system", "content": "4.数値データ形式：数値がある場合は、マークダウン形式の表で表示してください。"}, {"role": "system", "content": f"企業プロフィール情報:\n{profile_context}"}, {"role": "user", "content": question} ] print(datetime.now(), "Profile 기반 GPT 응답 생성 시작") response = client.chat.completions.create( model="gpt-4o", messages=messages ) return response.choices[0].message.content.strip() # ✅ Step 4: 크롤링 + GPT 생성 print(datetime.now(), "url 선택 시작") url_list_str = await choose_hompage_url(question) url_list = ast.literal_eval(url_list_str) print(datetime.now(), "url 선택 끝") print(datetime.now(), "크롤링시작") website_data = await fetch_all_pages_with_selenium(url_list) print(datetime.now(), "크롤링끝") messages = [ {"role": "system", "content": "あなたは「セグエグループ」の会社紹介および就職希望者向けのチャットボットです。"}, {"role": "system", "content": "以下のルールに従って、ユーザーと会話してください。"}, {"role": "system", "content": "1.会話スタイル：丁寧で信頼感のあるビジネス口調で応答してください。"}, {"role": "system", "content": "2.トーン：敬語を用いながらも、丁寧で親しみやすい表現を心がけてください。"}, {"role": "system", "content": "3.情報の制限：返答に使用できるのは、提供されたURLとExcelファイルに含まれる情報のみとします。その他の情報は使用しないでください。"}, {"role": "system", "content": "4.数値データ形式：数値がある場合は、マークダウン形式の表で表示してください。"}, {"role": "system", "content": f"企業サイト情報\n{website_data}"}, {"role": "user", "content": question} ] print(datetime.now(), "GPT 응답 생성 시작") response = client.chat.completions.create( model="gpt-4o", messages=messages ) print(datetime.now(), "GPT 응답 생성 완료") return response.choices[0].message.content.strip() # ✅ QnA 저장 함수 (async 함수 주의!) async def store_answer_to_saved_qna(question: str, answer: str): if any(x in answer for x in ["申し訳", "情報がありません", "わかりかねます"]): return try: # 1. 질문 임베딩 response = client.embeddings.create( model="text-embedding-3-small", input=question ) embedding = response.data[0].embedding # 2. 유사한 질문 있는지 확인 (기존 저장된 컬렉션에서) collection = chroma_client.get_collection("segue_saved_qna11") results = collection.query( query_embeddings=[embedding], n_results=1 ) # 3. 거리 임계값 체크 (0.15 이하면 유사한 질문으로 간주 → 저장 안 함) if results and results["distances"] and results["distances"][0][0] < 0.15: logger.info("類似の質問が既に存在しているため、保存しません。") return # 4. 저장 collection.add( ids=[f"saved_{int(time.time())}"], documents=[question], embeddings=[embedding], metadatas=[{"answer": answer}] ) logger.info("ChromaDBに回答を保存しました。") except Exception as e: logger.error(f"ChromaDB保存失敗: {e}") # ✅ 건강 체크 @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} # ✅ 질문 API @router.post("/apige/ask_question") async def gemini_question(user_question: UserQuestion, background_tasks: BackgroundTasks): question_text = user_question.question.strip() if not question_text: raise HTTPException(status_code=400, detail="Question is required") answer = await generate_gpt_answer(question_text) background_tasks.add_task(store_answer_to_saved_qna, question_text, answer) return {"answer": answer}