""" 텍스트 처리 관련 유틸리티 함수 모음 - 텍스트 분리 및 정제 - 키워드 추출 - Gemini API 키 통합 관리 적용 """ import re import google.generativeai as genai import os import logging import api_utils # API 키 통합 관리를 위한 임포트 # 로깅 설정 logger = logging.getLogger(__name__) logger.setLevel(logging.INFO) formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') handler = logging.StreamHandler() handler.setFormatter(formatter) logger.addHandler(handler) # ===== Gemini 모델 관리 함수들 ===== def get_gemini_model(): """api_utils에서 Gemini 모델 가져오기 (통합 관리)""" try: model = api_utils.get_gemini_model() if model: logger.info("Gemini 모델 로드 성공 (api_utils 통합 관리)") return model else: logger.warning("사용 가능한 Gemini API 키가 없습니다.") return None except Exception as e: logger.error(f"Gemini 모델 로드 실패: {e}") return None # 텍스트 분리 및 정제 함수 def clean_and_split(text, only_korean=False): """텍스트를 분리하고 정제하는 함수""" text = re.sub(r"[()\[\]-]", " ", text) text = text.replace("/", " ") if only_korean: # 한글만 추출 옵션이 켜진 경우 # 공백이나 쉼표로 구분한 뒤 한글만 추출 words = re.split(r"[ ,]", text) cleaned = [] for word in words: word = word.strip() # 한글만 남기고 다른 문자는 제거 word = re.sub(r"[^가-힣]", "", word) if word and len(word) >= 1: # 빈 문자열이 아니고 1글자 이상인 경우만 추가 cleaned.append(word) else: # 한글만 추출 옵션이 꺼진 경우 - 단어 통째로 처리 # 공백과 쉼표로 구분하여 단어 전체를 유지 words = re.split(r"[,\s]+", text) cleaned = [] for word in words: word = word.strip() if word and len(word) >= 1: # 빈 문자열이 아니고 1글자 이상인 경우만 추가 cleaned.append(word) return cleaned def filter_keywords_with_gemini(pairs, gemini_model=None): """Gemini AI를 사용하여 키워드 조합 필터링 (개선버전) - API 키 통합 관리""" if gemini_model is None: # api_utils에서 Gemini 모델 가져오기 gemini_model = get_gemini_model() if gemini_model is None: logger.error("Gemini 모델을 가져올 수 없습니다. 모든 키워드를 유지합니다.") # 안전하게 처리: 모든 키워드를 유지 all_keywords = set() for pair in pairs: for keyword in pair: all_keywords.add(keyword) return list(all_keywords) # 모든 키워드를 목록으로 추출 (제거된 키워드 확인용) all_keywords = set() for pair in pairs: for keyword in pair: all_keywords.add(keyword) # 너무 많은 쌍이 있으면 제한 max_pairs = 50 # 최대 50개 쌍만 처리 pairs_to_process = list(pairs)[:max_pairs] if len(pairs) > max_pairs else pairs logger.info(f"필터링할 키워드 쌍: 총 {len(pairs)}개 중 {len(pairs_to_process)}개 처리") # 보수적인 프롬프트 사용 - 키워드 제거 최소화 prompt = ( "다음은 소비자가 검색할 가능성이 있는 키워드 쌍 목록입니다.\n" "각 쌍은 같은 단어 조합이지만 순서만 다른 경우입니다 (예: 손질오징어 vs 오징어손질).\n\n" "아래의 기준에 따라 각 쌍에서 더 자연스러운 키워드를 선택해주세요:\n" "1. 소비자가 일상적으로 사용하는 자연스러운 표현을 우선 선택하세요.\n" "2. 두 키워드가 모두 자연스럽거나 의미가 약간 다르다면, 반드시 둘 다 유지하세요.\n" "3. 확실히 비자연스럽거나 어색한 경우에만 제거하세요.\n" "4. 불확실한 경우에는 반드시 키워드를 유지하세요.\n" "5. 숫자나 영어가 포함된 키워드는 한글 메인 키워드가 앞쪽에 오는 형태를 선택하세요. (예: '10kg 오징어' 보다 '오징어 10kg' 선택)\n" "6. 검색량이 0인 키워드라도 일상적인 표현이라면 가능한 유지하세요. 명백하게 비정상적인 표현만 제거하세요.\n\n" "주의: 기본적으로 대부분의 키워드를 유지하고, 매우 명확하게 비자연스러운 것만 제거하세요.\n\n" "결과는 다음 형식으로 제공해주세요:\n" "- 선택된 키워드 (이유: 자연스러운 표현이기 때문)\n" "- 선택된 키워드1, 선택된 키워드2 (이유: 둘 다 자연스럽고 의미가 조금 다름)\n\n" ) # 키워드 쌍 목록 formatted = "\n".join([f"- {a}, {b}" for a, b in pairs_to_process]) full_prompt = prompt + formatted try: # 타임아웃 추가 logger.info(f"Gemini API 호출 시작 - {len(pairs_to_process)}개 키워드 쌍 처리 중...") # 응답 받기 (타임아웃 기능이 있으면 추가) response = gemini_model.generate_content(full_prompt) logger.info("Gemini API 응답 성공") lines = response.text.strip().split("\n") # 선택된 키워드 추출 (쉼표로 구분된 경우 모두 포함) final_keywords = [] for line in lines: if line.startswith("-"): # 이유 부분 제거 keywords_part = line.strip("- ").split("(이유:")[0].strip() # 쉼표로 구분된 키워드 모두 추가 for kw in keywords_part.split(","): kw = kw.strip() if kw: final_keywords.append(kw) # 처리되지 않은 쌍의 첫 번째 키워드도 추가 (LLM이 처리하지 않은 키워드) if len(pairs) > max_pairs: logger.info(f"추가 키워드 처리: 남은 {len(pairs) - max_pairs}개 쌍의 첫 번째 키워드 추가") for pair in list(pairs)[max_pairs:]: # 각 쌍의 첫 번째 키워드만 사용 final_keywords.append(pair[0]) # 선택된 키워드가 없으면 기존 키워드 모두 반환 if not final_keywords: logger.warning("경고: 선택된 키워드가 없어 모든 키워드를 유지합니다.") final_keywords = list(all_keywords) # 순서 강제 수정 corrected_keywords = [] # 단위와 숫자 관련 정규식 패턴 unit_pattern = re.compile(r'(?i)(kg|g|mm|cm|ml|l|리터|개|팩|박스|세트|2l|l2)') number_pattern = re.compile(r'\d+') for kw in final_keywords: # 공백으로 분리 if ' ' in kw: parts = kw.split() first_part = parts[0] # 첫 부분이 단위나 숫자를 포함하는지 확인 if (unit_pattern.search(first_part) or number_pattern.search(first_part)) and len(parts) > 1: # 순서 바꾸기: 단위/숫자 부분을 뒤로 이동 corrected_kw = " ".join(parts[1:] + [first_part]) logger.info(f"키워드 순서 강제 수정: '{kw}' -> '{corrected_kw}'") corrected_keywords.append(corrected_kw) else: corrected_keywords.append(kw) else: corrected_keywords.append(kw) # 특별 처리: "L 오징어", "2L 오징어" 같은 경우를 명시적으로 확인하고 수정 specific_fixes = [] for kw in corrected_keywords: # 특정 패턴 체크 l_pattern = re.compile(r'^([0-9]*L) (.+)$', re.IGNORECASE) match = l_pattern.match(kw) if match: # L 단위를 뒤로 이동 l_part = match.group(1) main_part = match.group(2) fixed_kw = f"{main_part} {l_part}" logger.info(f"특수 패턴 수정: '{kw}' -> '{fixed_kw}'") specific_fixes.append(fixed_kw) else: specific_fixes.append(kw) # 제거된 키워드 목록 확인 selected_set = set(specific_fixes) removed_keywords = all_keywords - selected_set # 제거된 키워드 출력 logger.info("\n=== LLM에 의해 제거된 키워드 목록 ===") for kw in removed_keywords: logger.info(f" - {kw}") logger.info(f"총 {len(all_keywords)}개 중 {len(removed_keywords)}개 제거됨 ({len(selected_set)}개 유지)\n") return specific_fixes except Exception as e: logger.error(f"Gemini 오류: {e}") logger.error("오류 발생으로 인해 모든 키워드를 유지합니다.") logger.error(f"오류 유형: {type(e).__name__}") import traceback traceback.print_exc() # 안전하게 처리: 모든 키워드를 유지 logger.info(f"안전 모드: {len(all_keywords)}개 키워드 모두 유지") return list(all_keywords) def get_search_volume_range(total_volume): """총 검색량을 기반으로 검색량 구간을 반환""" if total_volume == 0: return "100미만" elif total_volume <= 100: return "100미만" elif total_volume <= 1000: return "1000미만" elif total_volume <= 2000: return "2000미만" elif total_volume <= 5000: return "5000미만" elif total_volume <= 10000: return "10000미만" else: return "10000이상"