from flask import Flask, request, jsonify from google import genai # pip install -U google-genai import os import json import pymysql from datetime import datetime, time from zoneinfo import ZoneInfo import time as timer # 요청 처리 시간 측정을 위한 모듈 import holidays # 대한민국 공휴일 체크를 위한 라이브러리 import threading # 백그라운드 작업을 위한 라이브러리 import requests # 네이버 Send API 호출을 위한 라이브러리 app = Flask(__name__) # --------------------------- # ✅ 환경변수 (운영 권장) — 데모 기본값 포함 # --------------------------- API_KEY = os.getenv("GEMINI_API_KEY", "AIzaSyBn1Xami_b7fgIyRm_rpAuMzwA-xiRlf7Q") DB_CONFIG = { "host": os.getenv("DB_HOST", "119.205.233.11"), "user": os.getenv("DB_USER", "changgi"), "password": os.getenv("DB_PASSWORD", "vkdnjgus13"), "database": os.getenv("DB_NAME", "versatile"), "charset": "utf8mb4", "cursorclass": pymysql.cursors.DictCursor, } # [최종 수정] 봇 이름(key)을 각 Webhook URL과 정확히 일치시킵니다. NAVER_ACCESS_TOKENS = { "habby": os.getenv("NAVER_ACCESS_TOKEN_HABBY", "0kOxJovvQaqwH2zJpfaQ"), "granceed": os.getenv("NAVER_ACCESS_TOKEN_GRANCEED", "BpJzdvb3RzeO3o5rJaMk"), "granceed_tools": os.getenv("NAVER_ACCESS_TOKEN_GRANCEED_TOOLS", "r3jxwzxFSQOzWzeg61Ej"), # granceed-tools -> granceed_tools "yonex-daegu": os.getenv("NAVER_ACCESS_TOKEN_YONEX_DAEGU", "lfZAFRz8SvmZmxynNjJs"), } # --------------------------- # ✅ Gemini 클라이언트 (v1) # --------------------------- client = genai.Client(api_key=API_KEY) # --------------------------- # 글로벌 변수 # --------------------------- system_prompt_habby = "" # id=4 system_prompt_granceed = "" # id=5 system_prompt_granceed_tools = "" # id=6 system_prompt_yonex_daegu = "" # id=7 MAX_MESSAGE_LENGTH = 100 MAX_USER_QUERIES = 30 user_query_count = {} last_message_cache = {} # --------------------------- # 운영시간 체크 (KST) # --------------------------- def is_allowed_now_kst() -> bool: now = datetime.now(ZoneInfo("Asia/Seoul")) today = now.date() wd = now.weekday() t = now.time() kr_holidays = holidays.KR() if wd in (5, 6) or today in kr_holidays: if today in kr_holidays: print(f"[INFO] 오늘은 공휴일({kr_holidays.get(today)})이므로 24시간 운영합니다.") return True return ((time(0, 0, 0) <= t < time(6, 0, 0)) or (time(18, 0, 0) <= t <= time(23, 59, 59))) # --------------------------- # DB에서 프롬프트 1회 로드 # --------------------------- def load_prompt_from_db(prompt_id: int) -> str: try: conn = pymysql.connect(**DB_CONFIG, connect_timeout=5) with conn.cursor() as cur: cur.execute("SELECT prompt_text FROM ai_prompt_table WHERE id=%s", (prompt_id,)) row = cur.fetchone() if row and row.get("prompt_text"): print(f"[INFO] prompt id={prompt_id} 로드 성공") return row["prompt_text"] except Exception as e: print(f"[WARN] 프롬프트 로드 오류 (id={prompt_id}): {e}") finally: try: conn.close() except: pass return '{"block": true, "response": "(챗봇) 내부 오류로 응답할 수 없습니다."}' # --------------------------- # Gemini 호출 헬퍼 (시간제한 없이 기다리는 버전) # --------------------------- def call_gemini_with_long_wait(system_instruction: str, user_message: str) -> dict: """시간제한 없이 Gemini API의 응답을 기다리는 함수""" combined_message = f"""[시스템 명령어] {system_instruction} --- [사용자 질문] {user_message}""" try: resp = client.models.generate_content( model="models/gemini-2.5-flash", contents=[{"role": "user", "parts": [{"text": combined_message}]}], config={"response_mime_type": "application/json"}, ) text = (getattr(resp, "text", "") or "").strip() if not text: return {"block": False, "response": "(챗봇) AI가 답변을 생성하지 못했습니다."} return json.loads(text) except Exception as e: print(f"[오류] Gemini API 호출 중 오류 발생: {e}") return {"block": False, "response": "(챗봇) AI 엔진에 오류가 발생했습니다."} # --------------------------- # 네이버 Send API를 통해 메시지를 보내는 함수 # --------------------------- def send_message_to_naver(user_id: str, message_text: str, access_token: str): """네이버 Send API를 호출하여 사용자에게 메시지를 전송합니다.""" # [최종 수정] ChatGPT의 제안에 따라 API 엔드포인트(URL)를 변경합니다. naver_api_endpoint = "https://gw.talk.naver.com/chatbot/v1/event" # [최종 수정] ChatGPT의 제안에 따라 'bearer' 접두사를 제거합니다. headers = { "Authorization": access_token, "Content-Type": "application/json;charset=UTF-8", } payload = { "event": "send", "user": user_id, "textContent": { "text": message_text } } try: response = requests.post(naver_api_endpoint, headers=headers, json=payload, timeout=10) response.raise_for_status() print(f"[INFO] 네이버 Send API를 통해 성공적으로 메시지를 보냈습니다. (사용자: {user_id})") except requests.exceptions.RequestException as e: print(f"[FATAL] 네이버 Send API 호출 실패: {e}") if e.response is not None: print(f" ㄴ 응답 코드: {e.response.status_code}, 응답 내용: {e.response.text}") else: print(f" ㄴ 응답 없음") # --------------------------- # 백그라운드에서 실행될 작업 함수 # --------------------------- def process_message_in_background(system_prompt: str, user_id: str, user_message: str, access_token: str): """백그라운드 스레드에서 AI 호출 및 네이버로 답변 전송을 처리합니다.""" print(f"[INFO] 백그라운드 작업 시작 (사용자: {user_id}, 메시지: '{user_message}')") reply_dict = call_gemini_with_long_wait(system_prompt, user_message) if reply_dict and reply_dict.get('response'): reply_text = reply_dict['response'] print(f"[INFO] AI 답변 생성 완료. 네이버로 전송합니다: '{reply_text[:50]}...'") send_message_to_naver(user_id, reply_text, access_token) else: print("[WARN] AI가 유효한 답변을 생성하지 못하여 메시지를 보내지 않습니다.") # --------------------------- # Webhook 엔드포인트 # --------------------------- @app.route('/webhook/', methods=['POST']) def webhook(bot_name): print(f"\n[INFO] -------- Request received for '{bot_name}' at {datetime.now()} --------") data = request.json or {} if not is_allowed_now_kst(): return ('', 204) user_message = (data.get('textContent', {}).get('text', '') or "").strip() user_id = data.get('user', '') or '' if not user_message or not user_id: return ('', 204) now = timer.time() if user_id in last_message_cache: last_time, last_msg = last_message_cache[user_id] if (now - last_time < 10) and (last_msg == user_message): print(f"[WARN] 중복 요청(메아리) 감지됨. 무시합니다: {user_message}") return ('', 204) last_message_cache[user_id] = (now, user_message) # [최종 수정] 봇 이름(key)을 각 Webhook URL과 정확히 일치시킵니다. prompts = { 'habby': system_prompt_habby, 'granceed': system_prompt_granceed, 'granceed_tools': system_prompt_granceed_tools, # granceed-tools -> granceed_tools 'yonex-daegu': system_prompt_yonex_daegu } current_system_prompt = prompts.get(bot_name, system_prompt_habby) current_access_token = NAVER_ACCESS_TOKENS.get(bot_name) # [추가] 어떤 토큰이 사용되는지 확인하기 위한 진단용 로그 print(f"[DEBUG] 봇 '{bot_name}'에 대해 로드된 Access Token: '{current_access_token}'") if not current_access_token: print(f"[FATAL] '{bot_name}'에 해당하는 Access Token이 설정되지 않았습니다.") return ('', 204) background_task = threading.Thread( target=process_message_in_background, args=(current_system_prompt, user_id, user_message, current_access_token) ) background_task.daemon = True background_task.start() print(f"[INFO] 메시지 수신 확인. 백그라운드에서 처리를 시작합니다.") return ('', 204) # --------------------------- # 앱 시작 시 프롬프트 캐싱 # --------------------------- try: print("[INFO] 앱 시작... 데이터베이스에서 프롬프트를 로드합니다.") system_prompt_habby = load_prompt_from_db(4) system_prompt_granceed = load_prompt_from_db(5) system_prompt_granceed_tools = load_prompt_from_db(6) system_prompt_yonex_daegu = load_prompt_from_db(7) print("[INFO] 프롬프트 로드 완료. 챗봇 서버를 시작합니다.") except Exception as e: print(f"[FATAL] 앱 시작 중 심각한 오류 발생: {e}") # --------------------------- # 개발용 서버 실행 (python3 ... 로 실행 시) # --------------------------- if __name__ == '__main__': print("\n[WARNING] Flask 개발용 서버로 실행 중입니다. 성능 저하 및 응답 누락이 발생할 수 있습니다.") print("[RECOMMEND] 운영 환경에서는 Gunicorn을 사용해 주세요.\n") app.run(host='0.0.0.0', port=5000)