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 import trafilatura # 🔧 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/settle.html"}, {"role": "system", "content": "経営成績、売上高、営業利益、経常利益、親会社株主に帰属する当期純利益 : https://segue-g.jp/ir/results/index.html"}, {"role": "system", "content": "財政状況、総資産、純資産、自己資本比率、１株当たり純資産 : https://segue-g.jp/ir/results/finance.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() # ✅ 뉴 크롤링 함수 async def crawl_with_trafilatura(question: str) -> str | None: # url_list_str = await choose_hompage_url(question) # ✅ 테스트용 하드코딩 URL 목록 url_list = [ "https://segue-g.jp/company/basic/index.html", "https://segue-g.jp/company/message/index.html", "https://segue-g.jp/company/brand/index.html", "https://segue-g.jp/company/boardmember/index.html", "https://segue-g.jp/company/group/index.html", "https://segue-g.jp/company/sdgs/index.html", "https://segue-g.jp/business/index.html", "https://segue-g.jp/ir/investor/strategy/index.html", "https://segue-g.jp/ir/investor/strong_point/index.html", "https://segue-g.jp/ir/governance/index.html", "https://revoworks.jp/about/", "https://revoworks.jp/media/case-study" ] all_text = "" for url in url_list: try: print(f"🌐 Trafialtura 크롤링 중: {url}") downloaded = trafilatura.fetch_url(url) if downloaded: result = trafilatura.extract(downloaded, include_comments=False, include_tables=True) if result: all_text += ( f"\n\n========== [URL] {url} ==========\n" f"{result.strip()}\n" f"========== [END] {url} ==========\n" ) except Exception as e: logger.warning(f"Trafialtura 크롤링 실패 - {url}: {e}") return all_text if all_text else None # ✅ GPT 대답 생성 로직 (QnA → Profile + GPT → 정보 없음) async def generate_gpt_answer(question: str) -> str: # # ✅ Step 0: trafilatura 테스트 출력 # trafilatura_content = await crawl_with_trafilatura(question) # if trafilatura_content: # logger.info("🔍 trafilatura 크롤링 결과:") # print(trafilatura_content) # # print(trafilatura_content[:2000]) # 너무 길 경우 일부만 출력 # else: # logger.info("🔍 trafilatura 크롤링 결과 없음 또는 실패") # # ✅ 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 응답 생성 완료") # ✅ 테스트용 Table 기반 프롬프트 prompt = f""" 以下は、会社に関する２つの情報カテゴリ（Table1とTable2）です。 [Table1] 売上高 営業利益 営業利益率 経常利益 親会社株主に帰属する当期純利益 1株当たり当期純利益 自己資本当期純利益率 総資産 純資産 総負債 自己資本比率 １株当たり純資産 営業活動によるキャッシュ・フロー 投資活動によるキャッシュ・フロー 財務活動によるキャッシュ・フロー 現金及び現金同等物の残高 [Table2] 沿革 決算短信 決算説明資料 有価証券報告書 株主総会関連資料 株主通信 その他IR情報 ユーザーの質問に対して、次の3つの情報を特定してください： 1. 該当するテーブル名（Table1 または Table2。どちらにも該当しない場合は「なし」） 2. 該当するキーワード（例：「売上高」「営業利益率」「有価証券報告書」など。該当しない場合は「該当項目なし」） 3. 質問文に明示された西暦年、または以下の表現から自然に推定できる西暦年の配列： ※ 年が明示されている場合（例：「2023年」「2024年度」など）は、必ずその年を優先して使用してください。 ※ 明示された年と、「過去3年」「昨年」「前年同月比」「改善」「推移」などが両方含まれる場合は、明示された年を中心に考え、その他の年は補足的に扱って構いません。 - 「過去3年」「過去5年」など → 現在を含む直近 {datetime.now().year - 3}〜{datetime.now().year - 1} 年を推定 - 「昨年」「前年同月比」 → {datetime.now().year - 1} 年 - 「一昨年」 → {datetime.now().year - 2} 年 - 質問文に「改善」「推移」「変化」「動向」「変遷」「増減」などの言葉が含まれており、かつ年が明示されていない場合は、直近3年間（{datetime.now().year - 2}, {datetime.now().year - 1}, {datetime.now().year}）を推定して構いません。 --- さらに、Table1にもTable2にも該当しない場合（"なし"と判定した場合）は、以下の候補リンクの中から「質問内容に関係がありそうなページURL」を配列形式で推薦してください。複数選んでも構いません。 [候補リンク一覧] 「会社の設立年・所在地・資本金・従業員数・事業内容など、基本的な企業情報を知りたい場合」に選択すべきURLです。 - https://segue-g.jp/company/basic/index.html 「経営トップからのメッセージや企業のビジョン・経営方針を知りたい場合」に選択すべきURLです。 - https://segue-g.jp/company/message/index.html 「企業理念・ブランドスローガン・存在意義（パーパス）を知りたい場合」に選択すべきURLです。 - https://segue-g.jp/company/brand/index.html 「代表取締役や取締役など、経営陣の氏名・役職を知りたい場合」に選択すべきURLです。 - https://segue-g.jp/company/boardmember/index.html 「グループ会社の名称・所在地・事業内容など、グループ全体の構成を知りたい場合」に選択すべきURLです。 - https://segue-g.jp/company/group/index.html 「当社のSDGs（持続可能な開発目標）への取り組み内容や社会貢献活動を知りたい場合」に選択すべきURLです。 - https://segue-g.jp/company/sdgs/index.html 「当社の主要事業領域や提供サービス、自社開発製品（RevoWorks等）の概要を知りたい場合」に選択すべきURLです。 - https://segue-g.jp/business/index.html 「セグエグループの中期経営計画や成長戦略、将来ビジョンに関する質問」の場合、このURLを選択すべきです。 - https://segue-g.jp/ir/investor/strategy/index.html 「当社の競争優位性や強み（技術力、顧客基盤、製品特徴など）を知りたい場合」に選択すべきURLです。 - https://segue-g.jp/ir/investor/strong_point/index.html 「取締役会の構成・監査体制・コンプライアンス方針など、企業統治（コーポレート・ガバナンス）に関する情報を知りたい場合」に選択すべきURLです。 - https://segue-g.jp/ir/governance/index.html 「社債の発行状況や格付機関による信用格付情報を知りたい場合」に選択すべきURLです。 - https://segue-g.jp/ir/stock/rating.html 「「当社の主要事業領域や提供サービス、自社開発製品（RevoWorks等）の概要」に選択すべきURLです。 - https://revoworks.jp/about/ 「RevoWorks製品の具体的な導入事例や活用実績、顧客の声を知りたい場合」に選択すべきURLです。 - https://revoworks.jp/media/case-study --- 【出力形式】 1. Tableに該当する場合: Table名 - キーワード - [西暦年リスト] 2. 該当しない場合（なし）: なし - 該当項目なし - [] - 推奨リンク: [リンク1, リンク2, ...] --- 【出力例】 Table1 - 売上高 - [2022, 2023, 2024] Table1 - 売上高, 営業利益率 - [2023] Table2 - 有価証券報告書 - [2023] なし - 該当項目なし - [] - 推奨リンク: [https://segue-g.jp/company/basic/index.html, https://segue-g.jp/business/index.html] --- 質問文: {question} """ response = client.chat.completions.create( model="gpt-4o", messages=[ {"role": "system", "content": "あなたは質問文を分類するAIアシスタントです。質問がどのテーブルに該当するか判断してください。"}, {"role": "user", "content": prompt} ] ) 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}