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 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): 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자 이내로 짧게 다시 질문해주세요~" } }) try: reply_dict = call_gemini_json(current_system_prompt, user_message) if reply_dict.get('block'): reply_text = "(챗봇) 현재는 배송 및 매장 안내 외 다른 상담은 불가합니다. 다른 상담은 영업시간(오전 9시~오후 5시)에 문의해주세요." else: reply_text = reply_dict.get('response', "(챗봇) 죄송합니다. 답변 생성 오류입니다.") except Exception as e: print(f"[오류] Gemini API 호출 오류: {e}") reply_text = "(챗봇) 죄송합니다. 시스템 오류가 발생했습니다. 잠시 후 다시 시도해주세요." return jsonify({ "event": "send", "textContent": {"text": reply_text} }) # --------------------------- # 앱 시작 시 프롬프트 캐싱 # --------------------------- if __name__ == '__main__': # --------------------------- # [성능 개선 제안] # 현재 방식(app.run)은 개발용 서버로, 동시에 여러 요청을 처리하기 어렵습니다. # 사용자가 늘어나면 챗봇이 느려지는 가장 큰 원인이 됩니다. # Gunicorn과 같은 '운영용(Production) WSGI 서버'를 사용하면 성능이 크게 향상됩니다. # # [사용 방법] # 1. Gunicorn 설치: pip install gunicorn # 2. Gunicorn으로 앱 실행 (터미널에서 아래 명령어 입력) # gunicorn --workers 4 --bind 0.0.0.0:5000 habby_granceed_ai_chabot:app # (habby_granceed_ai_chabot은 현재 파이썬 파일 이름입니다) # --------------------------- 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) app.run(host='0.0.0.0', port=5000)