import gradio as gr import pandas as pd import os import time import threading import tempfile import logging import uuid import shutil import glob from datetime import datetime import sys import types # 로깅 설정 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.StreamHandler(), logging.FileHandler('category_analysis_app.log', mode='a') ] ) logger = logging.getLogger(__name__) # 환경변수에서 모듈 코드 로드 및 동적 생성 def load_module_from_env(module_name, env_var_name): """환경변수에서 모듈 코드를 로드하여 동적으로 모듈 생성""" try: module_code = os.getenv(env_var_name) if not module_code: raise ValueError(f"환경변수 {env_var_name}가 설정되지 않았습니다.") # 새 모듈 생성 module = types.ModuleType(module_name) # 모듈에 필요한 기본 임포트들 추가 module.__dict__.update({ 'os': __import__('os'), 'time': __import__('time'), 'logging': __import__('logging'), 'pandas': __import__('pandas'), 'requests': __import__('requests'), 'tempfile': __import__('tempfile'), 'threading': __import__('threading'), 're': __import__('re'), 'random': __import__('random'), 'uuid': __import__('uuid'), 'shutil': __import__('shutil'), 'glob': __import__('glob'), 'datetime': __import__('datetime'), 'types': __import__('types'), 'collections': __import__('collections'), 'Counter': __import__('collections').Counter, 'defaultdict': __import__('collections').defaultdict, 'hmac': __import__('hmac'), 'hashlib': __import__('hashlib'), 'base64': __import__('base64'), }) # 코드 실행 exec(module_code, module.__dict__) # 시스템 모듈에 등록 sys.modules[module_name] = module logger.info(f"✅ 모듈 {module_name} 로드 완료") return module except Exception as e: logger.error(f"❌ 모듈 {module_name} 로드 실패: {e}") raise # 필요한 모듈들을 환경변수에서 로드 logger.info("🔄 모듈 로드 시작...") try: # 1. api_utils 모듈 로드 api_utils = load_module_from_env('api_utils', 'API_UTILS_CODE') # 2. text_utils 모듈 로드 (다른 모듈들이 의존하므로 먼저 로드) text_utils = load_module_from_env('text_utils', 'TEXT_UTILS_CODE') # 3. keyword_search 모듈 로드 keyword_search = load_module_from_env('keyword_search', 'KEYWORD_SEARCH_CODE') # 4. product_search 모듈 로드 (text_utils, keyword_search 의존) product_search_module = load_module_from_env('product_search', 'PRODUCT_SEARCH_CODE') # product_search 모듈에 의존성 주입 product_search_module.api_utils = api_utils product_search_module.text_utils = text_utils product_search = product_search_module # 5. keyword_processor 모듈 로드 keyword_processor_module = load_module_from_env('keyword_processor', 'KEYWORD_PROCESSOR_CODE') # keyword_processor 모듈에 의존성 주입 keyword_processor_module.text_utils = text_utils keyword_processor_module.keyword_search = keyword_search keyword_processor_module.product_search = product_search keyword_processor = keyword_processor_module # 6. export_utils 모듈 로드 export_utils = load_module_from_env('export_utils', 'EXPORT_UTILS_CODE') # 7. category_analysis 모듈 로드 (모든 모듈 의존) category_analysis_module = load_module_from_env('category_analysis', 'CATEGORY_ANALYSIS_CODE') # category_analysis 모듈에 의존성 주입 category_analysis_module.text_utils = text_utils category_analysis_module.product_search = product_search category_analysis_module.keyword_search = keyword_search category_analysis = category_analysis_module logger.info("✅ 모든 모듈 로드 완료") except Exception as e: logger.error(f"❌ 모듈 로드 중 치명적 오류: {e}") logger.error("필요한 환경변수들이 설정되었는지 확인하세요:") logger.error("- API_UTILS_CODE") logger.error("- TEXT_UTILS_CODE") logger.error("- KEYWORD_SEARCH_CODE") logger.error("- PRODUCT_SEARCH_CODE") logger.error("- KEYWORD_PROCESSOR_CODE") logger.error("- EXPORT_UTILS_CODE") logger.error("- CATEGORY_ANALYSIS_CODE") raise # 세션별 임시 파일 관리를 위한 딕셔너리 session_temp_files = {} session_data = {} def cleanup_huggingface_temp_folders(): """허깅페이스 임시 폴더 초기 정리""" try: # 일반적인 임시 디렉토리들 temp_dirs = [ tempfile.gettempdir(), "/tmp", "/var/tmp", os.path.join(os.getcwd(), "temp"), os.path.join(os.getcwd(), "tmp"), "/gradio_cached_examples", "/flagged" ] cleanup_count = 0 for temp_dir in temp_dirs: if os.path.exists(temp_dir): try: # 기존 세션 파일들 정리 session_files = glob.glob(os.path.join(temp_dir, "session_*.xlsx")) session_files.extend(glob.glob(os.path.join(temp_dir, "session_*.csv"))) session_files.extend(glob.glob(os.path.join(temp_dir, "*category*.xlsx"))) session_files.extend(glob.glob(os.path.join(temp_dir, "*category*.csv"))) session_files.extend(glob.glob(os.path.join(temp_dir, "*analysis*.xlsx"))) session_files.extend(glob.glob(os.path.join(temp_dir, "*analysis*.csv"))) session_files.extend(glob.glob(os.path.join(temp_dir, "tmp*.xlsx"))) session_files.extend(glob.glob(os.path.join(temp_dir, "tmp*.csv"))) for file_path in session_files: try: # 파일이 1시간 이상 오래된 경우만 삭제 if os.path.getmtime(file_path) < time.time() - 3600: os.remove(file_path) cleanup_count += 1 logger.info(f"초기 정리: 오래된 임시 파일 삭제 - {file_path}") except Exception as e: logger.warning(f"파일 삭제 실패 (무시됨): {file_path} - {e}") except Exception as e: logger.warning(f"임시 디렉토리 정리 실패 (무시됨): {temp_dir} - {e}") logger.info(f"✅ 허깅페이스 임시 폴더 초기 정리 완료 - {cleanup_count}개 파일 삭제") # Gradio 캐시 폴더도 정리 try: gradio_temp_dir = os.path.join(os.getcwd(), "gradio_cached_examples") if os.path.exists(gradio_temp_dir): shutil.rmtree(gradio_temp_dir, ignore_errors=True) logger.info("Gradio 캐시 폴더 정리 완료") except Exception as e: logger.warning(f"Gradio 캐시 폴더 정리 실패 (무시됨): {e}") except Exception as e: logger.error(f"초기 임시 폴더 정리 중 오류 (계속 진행): {e}") def setup_clean_temp_environment(): """깨끗한 임시 환경 설정""" try: # 1. 기존 임시 파일들 정리 cleanup_huggingface_temp_folders() # 2. 애플리케이션 전용 임시 디렉토리 생성 app_temp_dir = os.path.join(tempfile.gettempdir(), "category_analysis_app") if os.path.exists(app_temp_dir): shutil.rmtree(app_temp_dir, ignore_errors=True) os.makedirs(app_temp_dir, exist_ok=True) # 3. 환경 변수 설정 (임시 디렉토리 지정) os.environ['CATEGORY_APP_TEMP'] = app_temp_dir logger.info(f"✅ 애플리케이션 전용 임시 디렉토리 설정: {app_temp_dir}") return app_temp_dir except Exception as e: logger.error(f"임시 환경 설정 실패: {e}") return tempfile.gettempdir() def get_app_temp_dir(): """애플리케이션 전용 임시 디렉토리 반환""" return os.environ.get('CATEGORY_APP_TEMP', tempfile.gettempdir()) def get_session_id(): """세션 ID 생성""" return str(uuid.uuid4()) def cleanup_session_files(session_id, delay=300): """세션별 임시 파일 정리 함수""" def cleanup(): time.sleep(delay) if session_id in session_temp_files: files_to_remove = session_temp_files[session_id].copy() del session_temp_files[session_id] for file_path in files_to_remove: try: if os.path.exists(file_path): os.remove(file_path) logger.info(f"세션 {session_id[:8]}... 임시 파일 삭제: {file_path}") except Exception as e: logger.error(f"세션 {session_id[:8]}... 파일 삭제 오류: {e}") threading.Thread(target=cleanup, daemon=True).start() def register_session_file(session_id, file_path): """세션별 파일 등록""" if session_id not in session_temp_files: session_temp_files[session_id] = [] session_temp_files[session_id].append(file_path) def cleanup_old_sessions(): """오래된 세션 데이터 정리""" current_time = time.time() sessions_to_remove = [] for session_id, data in session_data.items(): if current_time - data.get('last_activity', 0) > 3600: # 1시간 초과 sessions_to_remove.append(session_id) for session_id in sessions_to_remove: # 파일 정리 if session_id in session_temp_files: for file_path in session_temp_files[session_id]: try: if os.path.exists(file_path): os.remove(file_path) logger.info(f"오래된 세션 {session_id[:8]}... 파일 삭제: {file_path}") except Exception as e: logger.error(f"오래된 세션 파일 삭제 오류: {e}") del session_temp_files[session_id] # 세션 데이터 정리 if session_id in session_data: del session_data[session_id] logger.info(f"오래된 세션 데이터 삭제: {session_id[:8]}...") def update_session_activity(session_id): """세션 활동 시간 업데이트""" if session_id not in session_data: session_data[session_id] = {} session_data[session_id]['last_activity'] = time.time() def create_session_temp_file(session_id, suffix='.xlsx'): """세션별 임시 파일 생성 (전용 디렉토리 사용)""" timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") random_suffix = str(time.time_ns())[-4:] # 애플리케이션 전용 임시 디렉토리 사용 temp_dir = get_app_temp_dir() filename = f"session_{session_id[:8]}_{timestamp}_{random_suffix}{suffix}" temp_file_path = os.path.join(temp_dir, filename) # 빈 파일 생성 with open(temp_file_path, 'w') as f: pass register_session_file(session_id, temp_file_path) return temp_file_path def analyze_product_terms_wrapper(product_name, main_keyword, current_state, session_id): """상품명 키워드 분석 래퍼 함수 - 세션 ID 추가""" update_session_activity(session_id) if not product_name: return "상품명을 입력해주세요.", current_state, None, gr.update(visible=False) # 분석 수행 - HTML 결과와 키워드 분석 결과 함께 받기 result_html, keyword_results = category_analysis.analyze_product_terms(product_name, main_keyword) # 새로운 상태 생성 if current_state is None or not isinstance(current_state, dict): current_state = {} # 분석 결과를 상태에 추가 current_state["keyword_analysis_results"] = keyword_results current_state["product_name"] = product_name current_state["main_keyword"] = main_keyword # 세션별 엑셀 파일 다운로드 - 자동 다운로드 excel_path = download_analysis(current_state, session_id) # 출력 섹션 표시 return result_html, current_state, excel_path, gr.update(visible=True) def download_analysis(result, session_id): """카테고리 분석 결과 다운로드 (세션별)""" update_session_activity(session_id) if not result or not isinstance(result, dict): logger.warning(f"세션 {session_id[:8]}... 분석 결과가 없습니다.") return None try: # 상품명 분석 결과가 있는지 확인 if "keyword_analysis_results" in result: logger.info(f"세션 {session_id[:8]}... 키워드 분석 결과 포함하여 다운로드: {len(result['keyword_analysis_results'])}개 키워드") # 세션별 임시 파일 생성 temp_filename = create_session_temp_file(session_id, '.xlsx') # 데이터프레임 생성 keywords = [] pc_volumes = [] mobile_volumes = [] total_volumes = [] ranges = [] category_items = [] for kw_result in result["keyword_analysis_results"]: keywords.append(kw_result.get("키워드", "")) pc_volumes.append(kw_result.get("PC검색량", 0)) mobile_volumes.append(kw_result.get("모바일검색량", 0)) total_volumes.append(kw_result.get("총검색량", 0)) ranges.append(kw_result.get("검색량구간", "")) category_items.append(kw_result.get("카테고리항목", "")) # 데이터프레임으로 변환 df = pd.DataFrame({ "키워드": keywords, "PC검색량": pc_volumes, "모바일검색량": mobile_volumes, "총검색량": total_volumes, "검색량구간": ranges, "카테고리항목": category_items }) with pd.ExcelWriter(temp_filename, engine="xlsxwriter") as writer: df.to_excel(writer, sheet_name="상품명 검증 결과", index=False) ws = writer.sheets["상품명 검증 결과"] # 줄바꿈 + 위쪽 정렬 서식 wrap_fmt = writer.book.add_format({ "text_wrap": True, "valign": "top" }) # F열('카테고리항목') 전체에 서식 적용 + 열 너비 ws.set_column("F:F", 40, wrap_fmt) # 열 너비 설정 worksheet = writer.sheets['상품명 검증 결과'] worksheet.set_column('A:A', 20) # 키워드 worksheet.set_column('B:B', 12) # PC검색량 worksheet.set_column('C:C', 12) # 모바일검색량 worksheet.set_column('D:D', 12) # 총검색량 worksheet.set_column('E:E', 12) # 검색량구간 worksheet.set_column('F:F', 40) # 카테고리항목 # 헤더 서식 지정 header_format = writer.book.add_format({ 'bold': True, 'bg_color': '#FB7F0D', 'color': 'white', 'border': 1 }) # 헤더에 서식 적용 for col_num, value in enumerate(df.columns.values): worksheet.write(0, col_num, value, header_format) logger.info(f"세션 {session_id[:8]}... 엑셀 파일 저장 완료: {temp_filename}") return temp_filename else: logger.warning(f"세션 {session_id[:8]}... 키워드 분석 결과가 없습니다.") return None except Exception as e: logger.error(f"세션 {session_id[:8]}... 다운로드 중 오류 발생: {e}") import traceback logger.error(traceback.format_exc()) return None def reset_interface(session_id): """인터페이스 리셋 함수 - 세션별 데이터 초기화""" update_session_activity(session_id) # 세션별 임시 파일 정리 if session_id in session_temp_files: for file_path in session_temp_files[session_id]: try: if os.path.exists(file_path): os.remove(file_path) logger.info(f"세션 {session_id[:8]}... 리셋 시 파일 삭제: {file_path}") except Exception as e: logger.error(f"세션 {session_id[:8]}... 리셋 시 파일 삭제 오류: {e}") session_temp_files[session_id] = [] return ( "", # 메인 키워드 입력 "", # 상품명 입력 "", # 분석 결과 출력 None, # 다운로드 파일 None, # 상태 변수 gr.update(visible=False) # 분석 결과 섹션 ) def product_analyze_with_loading(product_name, main_keyword, current_state, session_id): """로딩 표시 함수""" update_session_activity(session_id) return gr.update(visible=True) def process_product_analyze(product_name, main_keyword, current_state, session_id): """실제 분석 수행""" update_session_activity(session_id) results = analyze_product_terms_wrapper(product_name, main_keyword, current_state, session_id) # 로딩 인디케이터 숨기기 return results + (gr.update(visible=False),) # 세션 정리 스케줄러 def start_session_cleanup_scheduler(): """세션 정리 스케줄러 시작""" def cleanup_scheduler(): while True: time.sleep(600) # 10분마다 실행 cleanup_old_sessions() # 추가로 허깅페이스 임시 폴더도 주기적 정리 cleanup_huggingface_temp_folders() threading.Thread(target=cleanup_scheduler, daemon=True).start() def cleanup_on_startup(): """애플리케이션 시작 시 전체 정리""" logger.info("🧹 카테고리 분석 애플리케이션 시작 - 초기 정리 작업 시작...") # 1. 허깅페이스 임시 폴더 정리 cleanup_huggingface_temp_folders() # 2. 깨끗한 임시 환경 설정 app_temp_dir = setup_clean_temp_environment() # 3. 전역 변수 초기화 global session_temp_files, session_data session_temp_files.clear() session_data.clear() logger.info(f"✅ 초기 정리 작업 완료 - 앱 전용 디렉토리: {app_temp_dir}") return app_temp_dir # Gradio 인터페이스 생성 def create_app(): # FontAwesome 아이콘 포함 fontawesome_html = """ """ # CSS 파일 로드 try: with open('style.css', 'r', encoding='utf-8') as f: custom_css = f.read() except: # CSS 파일이 없는 경우 기본 스타일 사용 custom_css = """ :root { --primary-color: #FB7F0D; --secondary-color: #ff9a8b; } .custom-button { background: linear-gradient(135deg, var(--primary-color), var(--secondary-color)) !important; color: white !important; border-radius: 30px !important; height: 45px !important; font-size: 16px !important; font-weight: bold !important; width: 100% !important; text-align: center !important; display: flex !important; align-items: center !important; justify-content: center !important; } .reset-button { background: linear-gradient(135deg, #6c757d, #495057) !important; color: white !important; border-radius: 30px !important; height: 45px !important; font-size: 16px !important; font-weight: bold !important; width: 100% !important; text-align: center !important; display: flex !important; align-items: center !important; justify-content: center !important; } .section-title { border-bottom: 2px solid #FB7F0D; font-weight: bold; padding-bottom: 5px; margin-bottom: 15px; } .loading-indicator { display: flex; align-items: center; justify-content: center; padding: 15px; background-color: #f8f9fa; border-radius: 5px; margin: 10px 0; border: 1px solid #ddd; } .loading-spinner { border: 4px solid rgba(0, 0, 0, 0.1); width: 24px; height: 24px; border-radius: 50%; border-left-color: #FB7F0D; animation: spin 1s linear infinite; margin-right: 10px; } @keyframes spin { 0% { transform: rotate(0deg); } 100% { transform: rotate(360deg); } } .progress-bar { height: 10px; background-color: #FB7F0D; border-radius: 5px; width: 0%; animation: progressAnim 2s ease-in-out infinite; } @keyframes progressAnim { 0% { width: 10%; } 50% { width: 70%; } 100% { width: 10%; } } .execution-section { margin-top: 20px; background-color: #f9f9f9; border-radius: 8px; padding: 15px; border: 1px solid #e5e5e5; } .session-info { background-color: #e8f4f8; padding: 8px 12px; border-radius: 4px; font-size: 12px; color: #0c5460; margin-bottom: 10px; text-align: center; } """ with gr.Blocks(css=custom_css, theme=gr.themes.Default( primary_hue="orange", secondary_hue="orange", font=[gr.themes.GoogleFont("Noto Sans KR"), "ui-sans-serif", "system-ui"] )) as demo: gr.HTML(fontawesome_html) # 세션 ID 상태 (각 사용자별로 고유) session_id = gr.State(get_session_id) # 입력 섹션 with gr.Column(elem_classes="custom-frame fade-in"): gr.HTML('