import gradio as gr import pandas as pd import os import time import threading import tempfile import logging import random import uuid import shutil import glob from datetime import datetime from gradio_client import Client # 환경변수 로드를 위한 dotenv 사용 try: from dotenv import load_dotenv load_dotenv() except ImportError: pass # 로깅 설정 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.StreamHandler(), logging.FileHandler('control_tower_app.log', mode='a') ] ) logger = logging.getLogger(__name__) # API 클라이언트 설정 def get_api_client(): """환경변수에서 API 엔드포인트를 가져와 클라이언트 생성""" endpoint = os.getenv('API_ENDPOINT', '').strip() # 디버깅을 위한 환경변수 체크 (실제 값은 로그에 남기지 않음) if not endpoint: logger.error("API_ENDPOINT 환경변수가 설정되지 않았습니다.") logger.info("사용 가능한 환경변수들:") for key in os.environ.keys(): if 'API' in key.upper(): logger.info(f" {key}: {'설정됨' if os.environ[key] else '비어있음'}") raise ValueError("API_ENDPOINT 환경변수가 설정되지 않았습니다.") if endpoint.startswith('#') or len(endpoint) == 0: logger.error("API_ENDPOINT 환경변수가 올바르지 않은 형식입니다.") raise ValueError("API_ENDPOINT 환경변수가 올바르게 설정되지 않았습니다.") # 엔드포인트 형식 검증 if '/' not in endpoint: logger.error("API_ENDPOINT는 'username/repo-name' 형식이어야 합니다.") raise ValueError("API_ENDPOINT 형식이 올바르지 않습니다.") return Client(endpoint) # 세션별 임시 파일 관리를 위한 딕셔너리 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, "*keyword*.xlsx"))) session_files.extend(glob.glob(os.path.join(temp_dir, "*keyword*.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(), "control_tower_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['CONTROL_TOWER_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('CONTROL_TOWER_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(random.randint(1000, 9999)) # 애플리케이션 전용 임시 디렉토리 사용 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 wrapper_modified(keyword, korean_only, apply_main_keyword_option, exclude_zero_volume, session_id): """키워드 검색 및 처리 래퍼 함수 (API 클라이언트 사용)""" update_session_activity(session_id) try: client = get_api_client() result = client.predict( keyword=keyword, korean_only=korean_only, apply_main_keyword=apply_main_keyword_option, exclude_zero_volume=exclude_zero_volume, api_name="/process_search_results" ) # API 결과 처리 table_html, cat_choices, vol_choices, selected_cat, download_file = result # 로컬 파일로 다운로드 파일 복사 local_file = None if download_file: local_file = create_session_temp_file(session_id, '.xlsx') shutil.copy(download_file, local_file) return ( gr.update(value=table_html), gr.update(choices=cat_choices), gr.update(choices=vol_choices), None, # DataFrame은 클라이언트에서 관리하지 않음 gr.update(choices=cat_choices, value=selected_cat), local_file, gr.update(visible=True), gr.update(visible=True), keyword ) except Exception as e: logger.error(f"API 호출 오류: {e}") return ( gr.update(value="
검색 중 오류가 발생했습니다. 다시 시도해주세요.
"), gr.update(choices=["전체 보기"]), gr.update(choices=["전체"]), None, gr.update(choices=["전체 보기"], value="전체 보기"), None, gr.update(visible=False), gr.update(visible=False), keyword ) def analyze_with_auto_download(analysis_keywords, selected_category, state_df, session_id): """카테고리 일치 분석 실행 및 자동 다운로드 (API 클라이언트 사용)""" update_session_activity(session_id) try: client = get_api_client() result = client.predict( analysis_keywords=analysis_keywords, selected_category=selected_category, api_name="/process_analyze_results" ) analysis_result, download_file = result # 로컬 파일로 다운로드 파일 복사 local_file = None if download_file: local_file = create_session_temp_file(session_id, '.xlsx') shutil.copy(download_file, local_file) return analysis_result, local_file, gr.update(visible=True) except Exception as e: logger.error(f"분석 API 호출 오류: {e}") return "분석 중 오류가 발생했습니다. 다시 시도해주세요.", None, gr.update(visible=False) def filter_and_sort_table(df, selected_cat, keyword_sort, total_volume_sort, usage_count_sort, selected_volume_range, exclude_zero_volume, session_id): """테이블 필터링 및 정렬 함수 (API 클라이언트 사용)""" update_session_activity(session_id) try: client = get_api_client() result = client.predict( selected_cat=selected_cat, keyword_sort=keyword_sort, total_volume_sort=total_volume_sort, usage_count_sort=usage_count_sort, selected_volume_range=selected_volume_range, exclude_zero_volume=exclude_zero_volume, api_name="/filter_and_sort_table" ) return result except Exception as e: logger.error(f"필터링 API 호출 오류: {e}") return "" def update_category_selection(selected_cat, session_id): """카테고리 필터 선택 시 분석할 카테고리도 같은 값으로 업데이트""" update_session_activity(session_id) try: client = get_api_client() result = client.predict( selected_cat=selected_cat, api_name="/update_category_selection" ) return gr.update(value=result) except Exception as e: logger.error(f"카테고리 선택 API 호출 오류: {e}") return gr.update(value=selected_cat) 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] = [] try: client = get_api_client() result = client.predict( api_name="/reset_interface" ) # API 결과를 그대로 반환 return result except Exception as e: logger.error(f"리셋 API 호출 오류: {e}") return ( "", # 검색 키워드 True, # 한글만 추출 False, # 검색량 0 키워드 제외 "메인키워드 적용", # 조합 방식 "", # HTML 테이블 ["전체 보기"], # 카테고리 필터 "전체 보기", # 카테고리 필터 선택 ["전체"], # 검색량 구간 필터 "전체", # 검색량 구간 선택 "정렬 없음", # 총검색량 정렬 "정렬 없음", # 키워드 사용횟수 정렬 ["전체 보기"], # 분석할 카테고리 "전체 보기", # 분석할 카테고리 선택 "", # 키워드 입력 "", # 분석 결과 None # 다운로드 파일 ) # 래퍼 함수들 def search_with_loading(keyword, korean_only, apply_main_keyword, exclude_zero_volume, session_id): update_session_activity(session_id) try: client = get_api_client() result = client.predict( keyword=keyword, korean_only=korean_only, apply_main_keyword=apply_main_keyword, exclude_zero_volume=exclude_zero_volume, api_name="/search_with_loading" ) return ( gr.update(visible=True), gr.update(visible=False) ) except Exception as e: logger.error(f"검색 로딩 API 호출 오류: {e}") return ( gr.update(visible=False), gr.update(visible=True) ) def process_search_results(keyword, korean_only, apply_main_keyword, exclude_zero_volume, session_id): update_session_activity(session_id) result = wrapper_modified(keyword, korean_only, apply_main_keyword, exclude_zero_volume, session_id) table_html, cat_choices, vol_choices, df, selected_cat, excel, keyword_section_vis, cat_section_vis, new_keyword_state = result if excel is not None: empty_placeholder_vis = False keyword_section_visibility = True execution_section_visibility = True else: empty_placeholder_vis = True keyword_section_visibility = False execution_section_visibility = False return ( table_html, cat_choices, vol_choices, df, selected_cat, excel, gr.update(visible=keyword_section_visibility), gr.update(visible=cat_section_vis), gr.update(visible=False), gr.update(visible=empty_placeholder_vis), gr.update(visible=execution_section_visibility), new_keyword_state ) def analyze_with_loading(analysis_keywords, selected_category, state_df, session_id): update_session_activity(session_id) try: client = get_api_client() result = client.predict( analysis_keywords=analysis_keywords, selected_category=selected_category, api_name="/analyze_with_loading" ) return gr.update(visible=True) except Exception as e: logger.error(f"분석 로딩 API 호출 오류: {e}") return gr.update(visible=False) def process_analyze_results(analysis_keywords, selected_category, state_df, session_id): update_session_activity(session_id) results = analyze_with_auto_download(analysis_keywords, selected_category, state_df, 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_html = """ """ # CSS 파일 로드 try: with open('style.css', 'r', encoding='utf-8') as f: custom_css = f.read() except: custom_css = "" 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) # 키워드 상태 관리 keyword_state = gr.State("") # 입력 섹션 with gr.Column(elem_classes="custom-frame fade-in"): gr.HTML('순번 | 조합 키워드 | PC검색량 | 모바일검색량 | 총검색량 | 검색량구간 | 키워드 사용자순위 | 키워드 사용횟수 | 상품 등록 카테고리 |
---|---|---|---|---|---|---|---|---|
검색을 실행하면 여기에 결과가 표시됩니다 |