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 # [추가] 요청 처리 시간 측정을 위한 모듈 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, } # --------------------------- # ✅ Gemini 클라이언트 (v1) # --------------------------- # 이전 라이브러리 버전과의 호환성을 위해 client_options를 제거합니다. 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 = {} # --------------------------- # 운영시간 체크 (KST) # --------------------------- def is_allowed_now_kst() -> bool: now = datetime.now(ZoneInfo("Asia/Seoul")) wd = now.weekday() # 월=0 ... 일=6 t = now.time() # [수정] 주석과 코드를 일치시켜 토요일(5), 일요일(6)에 24시간 허용하도록 명확히 변경합니다. # 이전 코드는 월요일을 제외한 대부분의 요일에 24시간을 허용하여 의도와 다를 수 있었습니다. if wd in (5, 6): # 토/일 24시간 허용 # if wd in (1,2,3,4,5, 6): # 토/일 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) 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_json(system_instruction: str, user_message: str) -> dict: """ 모델이 JSON 텍스트를 반환한다고 기대. 실패 시 평문을 response에 담아 반환. """ combined_message = f"""[시스템 명령어] {system_instruction} --- [사용자 질문] {user_message}""" def _one_shot(model_name: str): return client.models.generate_content( model=model_name, contents=[ {"role": "user", "parts": [{"text": combined_message}]} ], config={"response_mime_type": "application/json"}, # 아주 오래된 라이브러리 버전에서는 request_options를 지원하지 않아 제거합니다. ) text = "" last_err = None for model_name in ("models/gemini-2.5-flash",): try: resp = _one_shot(model_name) text = (getattr(resp, "text", "") or "").strip() if not text and getattr(resp, "candidates", None): cand = resp.candidates[0] if getattr(cand, "content", None) and getattr(cand.content, "parts", None): part0 = cand.content.parts[0] text = (getattr(part0, "text", "") or str(getattr(part0, "inline_data", "")) or "").strip() if text: break except Exception as e: last_err = e print(f"[오류] Gemini({model_name}) 호출 오류: {e}") if not text: return {"block": False, "response": "(챗봇) 현재 답변 엔진 점검 중입니다. 잠시 후 다시 시도해주세요."} try: return json.loads(text) except Exception: return {"block": False, "response": text[:500]} # --------------------------- # Webhook 엔드포인트 # --------------------------- @app.route('/webhook/', methods=['POST']) def webhook(bot_name): # [추가] 요청 처리 시작 시간 기록 request_start_time = timer.time() print(f"\n[INFO] -------- Request received for '{bot_name}' at {datetime.now()} --------") data = request.json or {} if not is_allowed_now_kst(): # [추가] 무응답 원인을 파악하기 위해 로그를 남깁니다. print("[INFO] 운영시간이 아니므로 응답하지 않습니다.") return ('', 204) if bot_name == 'habby': current_system_prompt = system_prompt_habby elif bot_name == 'granceed': current_system_prompt = system_prompt_granceed elif bot_name == 'granceed_tools': current_system_prompt = system_prompt_granceed_tools elif bot_name == 'yonex-daegu': current_system_prompt = system_prompt_yonex_daegu else: current_system_prompt = system_prompt_habby user_message = (data.get('textContent', {}).get('text', '') or '').strip() user_id = data.get('user', '') or '' # [수정] 'inputType' 체크를 제거하고, 대신 메시지 내용이 실제로 있는지 확인합니다. # 이렇게 하면 어떤 방식(타이핑, 버튼 클릭 등)으로 메시지가 전달되어도 내용을 처리하여 안정성을 높입니다. if not user_message: print("[INFO] 내용이 없는 메시지를 수신하여 무시합니다.") return ('', 204) user_query_count[user_id] = user_query_count.get(user_id, 0) + 1 if user_query_count[user_id] > MAX_USER_QUERIES: return jsonify({ "event": "send", "textContent": { "text": "(챗봇) 질의 횟수를 초과하셨습니다. 영업시간(9시~17시)에 문의 부탁드립니다." } }) if len(user_message) > MAX_MESSAGE_LENGTH: return jsonify({ "event": "send", "textContent": { "text": "(챗봇) 질문이 너무 길어요! 100자 이내로 짧게 다시 질문해주세요~" } }) reply_text = "" try: print(f"[INFO] User message from {user_id}: '{user_message}'") # [추가] AI API 호출 시간 측정 api_call_start_time = timer.time() print("[INFO] Calling Gemini API...") reply_dict = call_gemini_json(current_system_prompt, user_message) api_call_duration = timer.time() - api_call_start_time print(f"[INFO] Gemini API call finished in {api_call_duration:.2f} seconds.") if reply_dict.get('block'): reply_text = "(챗봇) 현재는 배송 및 매장 안내 외 다른 상담은 불가합니다. 다른 상담은 영업시간(오전 9시~오후 5시)에 문의해주세요." else: reply_text = reply_dict.get('response', "(챗봇) 죄송합니다. 답변 생성 오류입니다.") except Exception as e: print(f"[오류] Main webhook process error: {e}") reply_text = "(챗봇) 죄송합니다. 시스템 오류가 발생했습니다. 잠시 후 다시 시도해주세요." # [추가] 전체 요청 처리 시간 로깅 total_duration = timer.time() - request_start_time print(f"[INFO] Sending reply: '{reply_text[:50]}...'") print(f"[INFO] -------- Total request time: {total_duration:.2f} seconds --------") return jsonify({ "event": "send", "textContent": {"text": reply_text} }) # --------------------------- # 앱 시작 시 프롬프트 캐싱 # --------------------------- if __name__ == '__main__': # ------------------------------------------------------------------------------------ # [!!!][매우 중요] 성능 문제의 핵심 원인 및 해결책 # ------------------------------------------------------------------------------------ # 현재 `app.run()` 방식은 테스트 및 개발용 서버입니다. # 이 서버는 한 번에 '하나'의 요청만 처리할 수 있는 '1차선 도로'와 같습니다. # 여러 사용자가 동시에 메시지를 보내면, 도로가 막히듯 나머지 요청들은 하염없이 대기하게 됩니다. # 이것이 바로 챗봇이 느려지고, 심하면 응답이 없는 것처럼 보이는 현상의 '주범'입니다. # # [해결책] Gunicorn (운영용 WSGI 서버) 사용 # Gunicorn은 여러 개의 '차선'을 만들어 동시에 여러 요청을 처리할 수 있게 해줍니다. # 아래 방법으로 챗봇을 실행하면 속도와 안정성이 크게 향상됩니다. # # 1. Gunicorn 설치 (최초 1회만 실행): # pip install gunicorn # # 2. Gunicorn으로 챗봇 실행 (python3 ... 대신 아래 명령어를 사용): # gunicorn --workers 4 --bind 0.0.0.0:5000 habby_granceed_ai_chabot:app # # - `--workers 4`: 4개의 차선(프로세스)으로 요청을 동시에 처리하라는 의미입니다. # - `habby_granceed_ai_chabot:app`: '파이썬파일이름:Flask객체이름' 을 의미합니다. # ------------------------------------------------------------------------------------ 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("\n[WARNING] Flask 개발용 서버로 실행 중입니다. 성능 저하 및 응답 누락이 발생할 수 있습니다.") print("[RECOMMEND] 운영 환경에서는 Gunicorn을 사용해 주세요.\n") app.run(host='0.0.0.0', port=5000)