""" API 관련 유틸리티 함수 모음 (환경변수 버전) - API 키 관리 (환경변수에서 로드) - 시그니처 생성 - API 헤더 생성 - Gemini API 키 랜덤 로테이션 추가 """ import os import time import hmac import hashlib import base64 import requests import threading import random import google.generativeai as genai import logging logger = logging.getLogger(__name__) # 환경변수에서 API 설정 로드 def get_api_configs(): # 환경변수 'API_CONFIGS'에서 전체 설정을 가져옴 api_configs_str = os.getenv('API_CONFIGS', '') if not api_configs_str: logger.error("API_CONFIGS 환경변수가 설정되지 않았습니다.") return [], [], [], [] try: # 환경변수 값을 exec로 실행하여 설정 로드 local_vars = {} exec(api_configs_str, {}, local_vars) return ( local_vars.get('NAVER_API_CONFIGS', []), local_vars.get('NAVER_SHOPPING_CONFIGS', []), local_vars.get('NAVER_DATALAB_CONFIGS', []), local_vars.get('GEMINI_API_CONFIGS', []) ) except Exception as e: logger.error(f"환경변수 파싱 오류: {e}") return [], [], [], [] # API 설정 로드 NAVER_API_CONFIGS, NAVER_SHOPPING_CONFIGS, NAVER_DATALAB_CONFIGS, GEMINI_API_CONFIGS = get_api_configs() # 순차 사용을 위한 인덱스와 락 current_api_index = 0 current_shopping_api_index = 0 current_datalab_api_index = 0 current_gemini_api_index = 0 api_lock = threading.Lock() shopping_lock = threading.Lock() datalab_lock = threading.Lock() gemini_lock = threading.Lock() # Gemini 모델 캐시 _gemini_models = {} # API 설정 초기화 함수 추가 def initialize_api_configs(): """API 설정을 초기화하고 랜덤하게 정렬""" global NAVER_API_CONFIGS, NAVER_SHOPPING_CONFIGS, NAVER_DATALAB_CONFIGS, GEMINI_API_CONFIGS # API 설정을 다시 로드 NAVER_API_CONFIGS, NAVER_SHOPPING_CONFIGS, NAVER_DATALAB_CONFIGS, GEMINI_API_CONFIGS = get_api_configs() # API 설정을 랜덤하게 섞기 if NAVER_API_CONFIGS: random.shuffle(NAVER_API_CONFIGS) if NAVER_SHOPPING_CONFIGS: random.shuffle(NAVER_SHOPPING_CONFIGS) if NAVER_DATALAB_CONFIGS: random.shuffle(NAVER_DATALAB_CONFIGS) if GEMINI_API_CONFIGS: random.shuffle(GEMINI_API_CONFIGS) print(f"API 설정 초기화 완료:") print(f" - 검색광고 API: {len(NAVER_API_CONFIGS)}개") print(f" - 쇼핑 API: {len(NAVER_SHOPPING_CONFIGS)}개") print(f" - 데이터랩 API: {len(NAVER_DATALAB_CONFIGS)}개") print(f" - Gemini API: {len(GEMINI_API_CONFIGS)}개") def generate_signature(timestamp, method, uri, secret_key): """시그니처 생성 함수""" message = f"{timestamp}.{method}.{uri}" digest = hmac.new(secret_key.encode("utf-8"), message.encode("utf-8"), hashlib.sha256).digest() return base64.b64encode(digest).decode() def get_header(method, uri, api_key, secret_key, customer_id): """API 헤더 생성 함수""" timestamp = str(round(time.time() * 1000)) signature = generate_signature(timestamp, method, uri, secret_key) return { "Content-Type": "application/json; charset=UTF-8", "X-Timestamp": timestamp, "X-API-KEY": api_key, "X-Customer": str(customer_id), "X-Signature": signature } def get_next_api_config(): """순차적으로 다음 API 설정을 반환 (스레드 안전)""" global current_api_index if not NAVER_API_CONFIGS: logger.error("네이버 검색광고 API 설정이 없습니다.") return None with api_lock: config = NAVER_API_CONFIGS[current_api_index] current_api_index = (current_api_index + 1) % len(NAVER_API_CONFIGS) return config def get_next_shopping_api_config(): """순차적으로 다음 쇼핑 API 설정을 반환 (오류 키 건너뛰기 추가)""" global current_shopping_api_index if not NAVER_SHOPPING_CONFIGS: logger.error("네이버 쇼핑 API 설정이 없습니다.") return None with shopping_lock: # 최대 전체 키 수만큼 시도 (무한 루프 방지) for _ in range(len(NAVER_SHOPPING_CONFIGS)): config = NAVER_SHOPPING_CONFIGS[current_shopping_api_index] current_shopping_api_index = (current_shopping_api_index + 1) % len(NAVER_SHOPPING_CONFIGS) # 기본값 체크 if config["CLIENT_ID"] and not config["CLIENT_ID"].startswith("YOUR_"): return config # 모든 키가 기본값인 경우 첫 번째 키 반환 return NAVER_SHOPPING_CONFIGS[0] if NAVER_SHOPPING_CONFIGS else None def get_next_datalab_api_config(): """순차적으로 다음 데이터랩 API 설정을 반환 (스레드 안전)""" global current_datalab_api_index if not NAVER_DATALAB_CONFIGS: logger.error("네이버 데이터랩 API 설정이 없습니다.") return None with datalab_lock: # API 키가 설정되지 않았으면 None 반환 if not NAVER_DATALAB_CONFIGS[0]["CLIENT_ID"] or NAVER_DATALAB_CONFIGS[0]["CLIENT_ID"].startswith("YOUR_"): return None config = NAVER_DATALAB_CONFIGS[current_datalab_api_index] current_datalab_api_index = (current_datalab_api_index + 1) % len(NAVER_DATALAB_CONFIGS) return config def get_next_gemini_api_key(): """순차적으로 다음 Gemini API 키를 반환 (스레드 안전)""" global current_gemini_api_index if not GEMINI_API_CONFIGS: logger.warning("사용 가능한 Gemini API 키가 없습니다.") return None with gemini_lock: # 최대 전체 키 수만큼 시도 (무한 루프 방지) for _ in range(len(GEMINI_API_CONFIGS)): api_key = GEMINI_API_CONFIGS[current_gemini_api_index] current_gemini_api_index = (current_gemini_api_index + 1) % len(GEMINI_API_CONFIGS) # 기본값이 아닌 키만 반환 if api_key and not api_key.startswith("YOUR_") and api_key.strip(): return api_key # 모든 키가 기본값인 경우 None 반환 logger.warning("사용 가능한 Gemini API 키가 없습니다.") return None def get_gemini_model(): """캐시된 Gemini 모델을 반환하거나 새로 생성""" api_key = get_next_gemini_api_key() if not api_key: logger.error("Gemini API 키를 가져올 수 없습니다.") return None # 캐시에서 모델 확인 if api_key in _gemini_models: return _gemini_models[api_key] try: # 새 모델 생성 genai.configure(api_key=api_key) model = genai.GenerativeModel("gemini-2.0-flash-exp") # 캐시에 저장 _gemini_models[api_key] = model logger.info(f"Gemini 모델 생성 성공: {api_key[:8]}***{api_key[-4:]}") return model except Exception as e: logger.error(f"Gemini 모델 생성 실패 ({api_key[:8]}***): {e}") return None def validate_api_config(api_config): """API 설정 유효성 검사""" if not api_config: return False, "API 설정이 없습니다." API_KEY = api_config.get("API_KEY", "") SECRET_KEY = api_config.get("SECRET_KEY", "") CUSTOMER_ID_STR = api_config.get("CUSTOMER_ID", "") if not all([API_KEY, SECRET_KEY, CUSTOMER_ID_STR]): return False, "API 키가 설정되지 않았습니다." if CUSTOMER_ID_STR.startswith("YOUR_") or API_KEY.startswith("YOUR_"): return False, "API 키가 플레이스홀더입니다." try: CUSTOMER_ID = int(CUSTOMER_ID_STR) except ValueError: return False, f"CUSTOMER_ID 변환 오류: '{CUSTOMER_ID_STR}'는 유효한 숫자가 아닙니다." return True, "유효한 API 설정입니다." def validate_datalab_config(datalab_config): """데이터랩 API 설정 유효성 검사""" if not datalab_config: return False, "데이터랩 API 설정이 없습니다." CLIENT_ID = datalab_config.get("CLIENT_ID", "") CLIENT_SECRET = datalab_config.get("CLIENT_SECRET", "") if not all([CLIENT_ID, CLIENT_SECRET]): return False, "데이터랩 API 키가 설정되지 않았습니다." if CLIENT_ID.startswith("YOUR_") or CLIENT_SECRET.startswith("YOUR_"): return False, "데이터랩 API 키가 플레이스홀더입니다." return True, "유효한 데이터랩 API 설정입니다." def validate_gemini_config(): """Gemini API 설정 유효성 검사""" valid_keys = 0 for api_key in GEMINI_API_CONFIGS: if api_key and not api_key.startswith("YOUR_") and api_key.strip(): valid_keys += 1 if valid_keys == 0: return False, "사용 가능한 Gemini API 키가 없습니다." return True, f"{valid_keys}개의 유효한 Gemini API 키가 설정되어 있습니다."