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 import requests import json from dotenv import load_dotenv # 환경변수 로드 load_dotenv() # 로깅 설정 (API 정보 완전 차단) logging.basicConfig(level=logging.WARNING) logger = logging.getLogger(__name__) # 세션별 임시 파일 관리를 위한 딕셔너리 session_temp_files = {} session_data = {} def get_api_client(): """환경변수에서 API 엔드포인트를 가져와 요청 함수 생성""" endpoint = os.getenv('API_ENDPOINT') if not endpoint: raise ValueError("API_ENDPOINT 환경변수가 필요합니다.") def make_request(api_name, **kwargs): try: # gradio_client와 동일한 방식으로 API 호출 if not endpoint.startswith('http'): base_url = f"https://{endpoint}.hf.space" else: base_url = endpoint # Gradio API 엔드포인트 형식 맞추기 url = f"{base_url}/call{api_name}" # 매개변수를 순서대로 배열로 변환 if api_name == "/process_search_results": data = [kwargs.get('keyword', ''), kwargs.get('korean_only', True), kwargs.get('apply_main_keyword', '메인키워드 적용'), kwargs.get('exclude_zero_volume', False)] elif api_name == "/search_with_loading": data = [kwargs.get('keyword', ''), kwargs.get('korean_only', True), kwargs.get('apply_main_keyword', '메인키워드 적용'), kwargs.get('exclude_zero_volume', False)] elif api_name == "/filter_and_sort_table": data = [kwargs.get('selected_cat', '전체 보기'), kwargs.get('keyword_sort', '정렬 없음'), kwargs.get('total_volume_sort', '정렬 없음'), kwargs.get('usage_count_sort', '정렬 없음'), kwargs.get('selected_volume_range', '전체'), kwargs.get('exclude_zero_volume', False)] elif api_name == "/update_category_selection": data = [kwargs.get('selected_cat', '전체 보기')] elif api_name == "/process_analyze_results": data = [kwargs.get('analysis_keywords', ''), kwargs.get('selected_category', '전체 보기')] elif api_name == "/analyze_with_loading": data = [kwargs.get('analysis_keywords', ''), kwargs.get('selected_category', '전체 보기')] elif api_name == "/reset_interface": data = [] elif api_name == "/get_session_id": data = [] else: data = [] response = requests.post(url, json={"data": data}, timeout=60) if response.status_code == 200: result = response.json() return result.get('data', []) else: raise Exception(f"API 호출 실패: {response.status_code}") except Exception as e: raise Exception(f"API 연결 오류: {str(e)}") return type('APIClient', (), {'predict': lambda self, **kwargs: make_request(kwargs.pop('api_name'), **kwargs)})() def cleanup_huggingface_temp_folders(): """허깅페이스 임시 폴더 초기 정리""" try: temp_dirs = [tempfile.gettempdir(), "/tmp", "/var/tmp"] 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"))) for file_path in session_files: try: if os.path.getmtime(file_path) < time.time() - 3600: os.remove(file_path) cleanup_count += 1 except Exception: pass except Exception: pass logger.info(f"✅ 임시 폴더 정리 완료 - {cleanup_count}개 파일 삭제") except Exception as e: logger.error(f"임시 폴더 정리 중 오류: {e}") def setup_clean_temp_environment(): """깨끗한 임시 환경 설정""" try: cleanup_huggingface_temp_folders() 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) 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 생성""" try: client = get_api_client() result = client.predict(api_name="/get_session_id") return result[0] if result else str(uuid.uuid4()) except Exception: 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: 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 search_with_loading(keyword, korean_only, apply_main_keyword, exclude_zero_volume): """원본 API: /search_with_loading""" 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 result[0] if result else "" except Exception as e: logger.error(f"search_with_loading API 호출 오류: {e}") return "" def process_search_results(keyword, korean_only, apply_main_keyword, exclude_zero_volume): """원본 API: /process_search_results""" 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="/process_search_results" ) # 결과 안전하게 처리 if len(result) >= 5: table_html, cat_choices, vol_choices, selected_cat, download_file = result[:5] # 다운로드 파일이 있는 경우 로컬로 복사 local_download_file = None if download_file: session_id = get_session_id() local_download_file = create_session_temp_file(session_id, '.xlsx') try: shutil.copy2(download_file, local_download_file) except Exception as e: logger.error(f"파일 복사 오류: {e}") local_download_file = None return table_html, cat_choices, vol_choices, selected_cat, local_download_file else: return ( "

검색 결과가 없습니다.

", ["전체 보기"], ["전체"], "전체 보기", None ) except Exception as e: logger.error(f"process_search_results API 호출 오류: {e}") return ( "

서비스 연결에 문제가 발생했습니다. 잠시 후 다시 시도해주세요.

", ["전체 보기"], ["전체"], "전체 보기", None ) def filter_and_sort_table(selected_cat, keyword_sort, total_volume_sort, usage_count_sort, selected_volume_range, exclude_zero_volume): """원본 API: /filter_and_sort_table""" 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[0] if result else "" except Exception as e: logger.error(f"filter_and_sort_table API 호출 오류: {e}") return "

필터링 서비스 연결에 문제가 발생했습니다.

" def update_category_selection(selected_cat): """원본 API: /update_category_selection""" try: client = get_api_client() result = client.predict( selected_cat=selected_cat, api_name="/update_category_selection" ) return gr.update(value=result[0] if result else selected_cat) except Exception as e: logger.error(f"update_category_selection API 호출 오류: {e}") return gr.update(value=selected_cat) def analyze_with_loading(analysis_keywords, selected_category): """원본 API: /analyze_with_loading""" try: client = get_api_client() result = client.predict( analysis_keywords=analysis_keywords, selected_category=selected_category, api_name="/analyze_with_loading" ) return result[0] if result else "" except Exception as e: logger.error(f"analyze_with_loading API 호출 오류: {e}") return "" def process_analyze_results(analysis_keywords, selected_category): """원본 API: /process_analyze_results""" try: client = get_api_client() result = client.predict( analysis_keywords=analysis_keywords, selected_category=selected_category, api_name="/process_analyze_results" ) if len(result) >= 2: analysis_result, download_file = result[:2] # 다운로드 파일이 있는 경우 로컬로 복사 local_download_file = None if download_file: session_id = get_session_id() local_download_file = create_session_temp_file(session_id, '.xlsx') try: shutil.copy2(download_file, local_download_file) except Exception as e: logger.error(f"분석 결과 파일 복사 오류: {e}") local_download_file = None return analysis_result, local_download_file else: return "분석 결과가 없습니다.", None except Exception as e: logger.error(f"process_analyze_results API 호출 오류: {e}") return "분석 서비스 연결에 문제가 발생했습니다. 잠시 후 다시 시도해주세요.", None def reset_interface(): """원본 API: /reset_interface""" try: client = get_api_client() result = client.predict(api_name="/reset_interface") return result if result else get_default_reset_values() except Exception as e: logger.error(f"reset_interface API 호출 오류: {e}") return get_default_reset_values() def get_default_reset_values(): """기본 리셋 값 반환""" return ( "", True, False, "메인키워드 적용", "", ["전체 보기"], "전체 보기", ["전체"], "전체", "정렬 없음", "정렬 없음", ["전체 보기"], "전체 보기", "", "", None ) # UI 처리 래퍼 함수들 def wrapper_search_with_loading(keyword, korean_only, apply_main_keyword, exclude_zero_volume): """검색 로딩 UI 처리""" search_with_loading(keyword, korean_only, apply_main_keyword, exclude_zero_volume) return ( gr.update(visible=True), # progress_section gr.update(visible=False) # empty_table_html ) def wrapper_process_search_results(keyword, korean_only, apply_main_keyword, exclude_zero_volume): """검색 결과 처리 UI""" result = process_search_results(keyword, korean_only, apply_main_keyword, exclude_zero_volume) table_html, cat_choices, vol_choices, selected_cat, download_file = result # UI 표시 여부 결정 if table_html and "검색 결과가 없습니다" not in table_html and "문제가 발생했습니다" not in table_html: keyword_section_visibility = True category_section_visibility = True empty_placeholder_vis = False execution_section_visibility = True else: keyword_section_visibility = False category_section_visibility = False empty_placeholder_vis = True execution_section_visibility = False # 가상의 state_df state_df = pd.DataFrame() return ( table_html, # table_output cat_choices, # category_filter choices vol_choices, # search_volume_filter choices state_df, # state_df selected_cat, # selected_category value download_file, # download_output gr.update(visible=keyword_section_visibility), # keyword_analysis_section gr.update(visible=category_section_visibility), # category_analysis_section gr.update(visible=False), # progress_section gr.update(visible=empty_placeholder_vis), # empty_table_html gr.update(visible=execution_section_visibility), # execution_section keyword # keyword_state ) def wrapper_analyze_with_loading(analysis_keywords, selected_category, state_df): """분석 로딩 UI 처리""" analyze_with_loading(analysis_keywords, selected_category) return gr.update(visible=True) # progress_section def wrapper_process_analyze_results(analysis_keywords, selected_category, state_df): """분석 결과 처리 UI""" analysis_result, download_file = process_analyze_results(analysis_keywords, selected_category) return ( analysis_result, # analysis_result download_file, # download_output gr.update(visible=True), # analysis_output_section gr.update(visible=False) # progress_section ) # 세션 정리 스케줄러 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("🧹 컨트롤 타워 애플리케이션 시작 - 초기 정리 작업 시작...") cleanup_huggingface_temp_folders() app_temp_dir = setup_clean_temp_environment() 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 = """ :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; } """ 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) # 키워드 상태 관리 keyword_state = gr.State("") # 입력 섹션 with gr.Column(elem_classes="custom-frame fade-in"): gr.HTML('
검색 입력
') with gr.Row(): with gr.Column(scale=1): keyword = gr.Textbox( label="메인 키워드", placeholder="예: 오징어" ) with gr.Column(scale=1): search_btn = gr.Button( "메인키워드 분석", elem_classes="custom-button" ) with gr.Accordion("옵션 설정", open=False): with gr.Row(): with gr.Column(scale=1): korean_only = gr.Checkbox( label="한글만 추출", value=True ) with gr.Column(scale=1): exclude_zero_volume = gr.Checkbox( label="검색량 0 키워드 제외", value=False ) with gr.Row(): with gr.Column(scale=1): apply_main_keyword = gr.Radio( ["메인키워드 적용", "메인키워드 미적용"], label="조합 방식", value="메인키워드 적용" ) with gr.Column(scale=1): gr.HTML("") # 진행 상태 표시 섹션 with gr.Column(elem_classes="custom-frame fade-in", visible=False) as progress_section: gr.HTML('
분석 진행 상태
') progress_html = gr.HTML("""
키워드 데이터를 분석중입니다. 잠시만 기다려주세요...
""") # 메인키워드 분석 결과 섹션 with gr.Column(elem_classes="custom-frame fade-in") as main_keyword_section: gr.HTML('
메인키워드 분석 결과
') empty_table_html = gr.HTML("""
순번 조합 키워드 PC검색량 모바일검색량 총검색량 검색량구간 키워드 사용자순위 키워드 사용횟수 상품 등록 카테고리
검색을 실행하면 여기에 결과가 표시됩니다
""") with gr.Column(visible=False) as keyword_analysis_section: with gr.Row(): with gr.Column(scale=1): category_filter = gr.Dropdown( choices=["전체 보기"], label="카테고리 필터", value="전체 보기", interactive=True ) with gr.Column(scale=1): total_volume_sort = gr.Dropdown( choices=["정렬 없음", "오름차순", "내림차순"], label="총검색량 정렬", value="정렬 없음", interactive=True ) with gr.Row(): with gr.Column(scale=1): search_volume_filter = gr.Dropdown( choices=["전체"], label="검색량 구간 필터", value="전체", interactive=True ) with gr.Column(scale=1): usage_count_sort = gr.Dropdown( choices=["정렬 없음", "오름차순", "내림차순"], label="키워드 사용횟수 정렬", value="정렬 없음", interactive=True ) gr.HTML("
") table_output = gr.HTML(elem_classes="fade-in") # 카테고리 분석 섹션 with gr.Column(elem_classes="custom-frame fade-in", visible=False) as category_analysis_section: gr.HTML('
키워드 분석
') with gr.Row(): with gr.Column(scale=1): analysis_keywords = gr.Textbox( label="키워드 입력 (최대 20개, 쉼표 또는 엔터로 구분)", placeholder="예: 오징어볶음, 오징어 손질, 오징어 요리...", lines=5 ) with gr.Column(scale=1): selected_category = gr.Dropdown( label="분석할 카테고리(분석 전 반드시 선택해주세요)", choices=["전체 보기"], value="전체 보기", interactive=True ) # 실행 섹션 with gr.Column(elem_classes="execution-section", visible=False) as execution_section: gr.HTML('
실행
') with gr.Row(): with gr.Column(scale=1): analyze_btn = gr.Button( "카테고리 일치 분석", elem_classes=["execution-button", "primary-button"] ) with gr.Column(scale=1): reset_btn = gr.Button( "모든 입력 초기화", elem_classes=["execution-button", "secondary-button"] ) # 분석 결과 출력 섹션 with gr.Column(elem_classes="custom-frame fade-in", visible=False) as analysis_output_section: gr.HTML('
분석 결과 요약
') analysis_result = gr.HTML(elem_classes="fade-in") with gr.Row(): download_output = gr.File( label="키워드 목록 다운로드", visible=True ) # 상태 저장용 변수 state_df = gr.State() # 이벤트 연결 search_btn.click( fn=wrapper_search_with_loading, inputs=[keyword, korean_only, apply_main_keyword, exclude_zero_volume], outputs=[progress_section, empty_table_html] ).then( fn=wrapper_process_search_results, inputs=[keyword, korean_only, apply_main_keyword, exclude_zero_volume], outputs=[ table_output, category_filter, search_volume_filter, state_df, selected_category, download_output, keyword_analysis_section, category_analysis_section, progress_section, empty_table_html, execution_section, keyword_state ] ) # 필터 및 정렬 변경 이벤트 연결 category_filter.change( fn=filter_and_sort_table, inputs=[ category_filter, gr.Textbox(value="정렬 없음", visible=False), total_volume_sort, usage_count_sort, search_volume_filter, exclude_zero_volume ], outputs=[table_output] ) category_filter.change( fn=update_category_selection, inputs=[category_filter], outputs=[selected_category] ) total_volume_sort.change( fn=filter_and_sort_table, inputs=[ category_filter, gr.Textbox(value="정렬 없음", visible=False), total_volume_sort, usage_count_sort, search_volume_filter, exclude_zero_volume ], outputs=[table_output] ) usage_count_sort.change( fn=filter_and_sort_table, inputs=[ category_filter, gr.Textbox(value="정렬 없음", visible=False), total_volume_sort, usage_count_sort, search_volume_filter, exclude_zero_volume ], outputs=[table_output] ) search_volume_filter.change( fn=filter_and_sort_table, inputs=[ category_filter, gr.Textbox(value="정렬 없음", visible=False), total_volume_sort, usage_count_sort, search_volume_filter, exclude_zero_volume ], outputs=[table_output] ) exclude_zero_volume.change( fn=filter_and_sort_table, inputs=[ category_filter, gr.Textbox(value="정렬 없음", visible=False), total_volume_sort, usage_count_sort, search_volume_filter, exclude_zero_volume ], outputs=[table_output] ) # 카테고리 분석 버튼 이벤트 analyze_btn.click( fn=wrapper_analyze_with_loading, inputs=[analysis_keywords, selected_category, state_df], outputs=[progress_section] ).then( fn=wrapper_process_analyze_results, inputs=[analysis_keywords, selected_category, state_df], outputs=[analysis_result, download_output, analysis_output_section, progress_section] ) # 리셋 버튼 이벤트 연결 reset_btn.click( fn=reset_interface, inputs=[], outputs=[ keyword, korean_only, exclude_zero_volume, apply_main_keyword, table_output, category_filter, category_filter, search_volume_filter, search_volume_filter, total_volume_sort, usage_count_sort, selected_category, selected_category, analysis_keywords, analysis_result, download_output ] ) return demo if __name__ == "__main__": # ========== 시작 시 전체 초기화 ========== print("===== Application Startup at %s =====" % time.strftime("%Y-%m-%d %H:%M:%S")) logger.info("🚀 컨트롤 타워 애플리케이션 시작...") # 1. 첫 번째: 허깅페이스 임시 폴더 정리 및 환경 설정 app_temp_dir = cleanup_on_startup() # 2. 세션 정리 스케줄러 시작 start_session_cleanup_scheduler() # 3. API 연결 테스트 try: test_client = get_api_client() logger.info("✅ API 연결 테스트 성공") except Exception as e: logger.error("❌ API 연결 실패 - 환경변수 API_ENDPOINT를 확인하세요") print("❌ API_ENDPOINT 환경변수가 설정되지 않았습니다.") print("💡 .env 파일에 다음과 같이 설정하세요:") print("API_ENDPOINT=your-endpoint-here") raise SystemExit(1) logger.info("===== 컨트롤 타워 애플리케이션 시작 완료 at %s =====", time.strftime("%Y-%m-%d %H:%M:%S")) logger.info(f"📁 임시 파일 저장 위치: {app_temp_dir}") # ========== 앱 실행 ========== try: app = create_app() print("🚀 Gradio 애플리케이션이 시작됩니다...") app.launch( share=False, # 보안을 위해 share 비활성화 server_name="0.0.0.0", # 모든 IP에서 접근 허용 server_port=7860, # 포트 지정 max_threads=40, # 멀티유저를 위한 스레드 수 증가 auth=None, # 필요시 인증 추가 가능 show_error=True, # 에러 표시 quiet=False, # 로그 표시 favicon_path=None, # 파비콘 설정 ssl_verify=False, # SSL 검증 비활성화 (개발용) inbrowser=False, # 자동 브라우저 열기 비활성화 prevent_thread_lock=False # 스레드 잠금 방지 비활성화 ) except Exception as e: logger.error(f"애플리케이션 실행 실패: {e}") print(f"❌ 애플리케이션 실행 실패: {e}") raise SystemExit(1) finally: # 애플리케이션 종료 시 정리 logger.info("🧹 컨트롤 타워 애플리케이션 종료 - 최종 정리 작업...") try: cleanup_huggingface_temp_folders() if os.path.exists(app_temp_dir): shutil.rmtree(app_temp_dir, ignore_errors=True) logger.info("✅ 최종 정리 완료") except Exception as e: logger.error(f"최종 정리 중 오류: {e}")