import sys sys.path.insert(0, '/usr/local/lib/python3.10/dist-packages') from flask import Flask, request, jsonify import google.generativeai as genai # pip install -U google-generativeai import os import json import pymysql from datetime import datetime, time from zoneinfo import ZoneInfo app = Flask(__name__) # --------------------------- # ⚠️ 중요 보안 경고 ⚠️ # 요청에 따라 API 키와 DB 정보를 코드에 그대로 두었지만, # 이 방식은 매우 위험하며 실제 서비스에서는 절대 사용하면 안 됩니다. # 키가 노출되면 해킹 및 금전적 피해가 발생할 수 있습니다. # 반드시 os.getenv("...") 부분의 기본값을 제거하고, # 서버 환경변수를 사용하여 키를 관리하세요. # --------------------------- 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 클라이언트 설정 (최신 방식) # --------------------------- # 이 코드는 라이브러리 업데이트(pip install -U google-generativeai)가 필요합니다. try: genai.configure(api_key=API_KEY) except Exception as e: print(f"[오류] Gemini 설정 실패: API 키를 확인하세요. {e}") # --------------------------- # 글로벌 변수 # --------------------------- 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시간 허용 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: conn = None 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: if conn: conn.close() # 안전 폴백 return '{"block": true, "response": "(챗봇) 내부 오류로 응답할 수 없습니다."}' # --------------------------- # Gemini 호출 헬퍼 (개선된 버전) # --------------------------- def call_gemini_json(system_instruction: str, user_message: str) -> dict: """ 모델이 JSON 텍스트를 반환한다고 기대. 실패 시 평문을 response에 담아 반환. """ try: # 1. 사용할 모델 객체 생성 (json 출력을 위해 generation_config 설정) model = genai.GenerativeModel( 'gemini-1.5-flash-latest', system_instruction=system_instruction, generation_config={"response_mime_type": "application/json"} ) # 2. 콘텐츠 생성 response = model.generate_content(user_message) # 3. 결과 파싱 (response.text는 JSON 형식의 문자열) return json.loads(response.text) except Exception as e: print(f"[오류] Gemini JSON 호출 오류: {e}") # 오류 발생 시, 일반 텍스트로 재시도 (폴백) try: model_text = genai.GenerativeModel( 'gemini-1.5-flash-latest', system_instruction=system_instruction ) response_text = model_text.generate_content(user_message) # API가 JSON을 생성하지 못했을 때 받은 평문을 그대로 반환 return {"block": False, "response": response_text.text[:500]} except Exception as final_e: print(f"[오류] Gemini 텍스트 재시도 실패: {final_e}") return {"block": False, "response": "(챗봇) 현재 답변 엔진 점검 중입니다. 잠시 후 다시 시도해주세요."} # --------------------------- # 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"[오류] Webhook 처리 중 오류 발생: {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)