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 = 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() # 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): # 'system_instruction' 인수를 제거하고, # 합쳐진 메시지를 'user' 역할로 전달합니다. return client.models.generate_content( model=model_name, contents=[ {"role": "user", "parts": [{"text": combined_message}]} ], # 아주 오래된 라이브러리 버전을 위해 'generation_config'를 'config'로 변경합니다. config={"response_mime_type": "application/json"} ) 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(): 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 '' user_id = data.get('user', '') or '' input_type = data.get('textContent', {}).get('inputType', '') or '' if input_type != "typing": 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__': 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)