import gradio as gr import pandas as pd import os import logging from datetime import datetime import pytz import time import tempfile import zipfile import re import json # 로깅 설정 logging.basicConfig(level=logging.WARNING, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # 외부 라이브러리 로그 비활성화 logging.getLogger('gradio').setLevel(logging.WARNING) logging.getLogger('gradio_client').setLevel(logging.WARNING) logging.getLogger('httpx').setLevel(logging.WARNING) logging.getLogger('urllib3').setLevel(logging.WARNING) # ===== API 클라이언트 설정 ===== def get_api_client(): """환경변수에서 API 엔드포인트를 가져와 클라이언트 생성""" try: from gradio_client import Client # 환경변수에서 API 엔드포인트 가져오기 api_endpoint = os.getenv('API_ENDPOINT') if not api_endpoint: logger.error("API_ENDPOINT 환경변수가 설정되지 않았습니다.") raise ValueError("API_ENDPOINT 환경변수가 설정되지 않았습니다.") client = Client(api_endpoint) logger.info("원격 API 클라이언트 초기화 성공") return client except Exception as e: logger.error(f"API 클라이언트 초기화 실패: {e}") return None # ===== 한국시간 관련 함수 ===== def get_korean_time(): """한국시간 반환""" korea_tz = pytz.timezone('Asia/Seoul') return datetime.now(korea_tz) def format_korean_datetime(dt=None, format_type="filename"): """한국시간 포맷팅""" if dt is None: dt = get_korean_time() if format_type == "filename": return dt.strftime("%y%m%d_%H%M") elif format_type == "display": return dt.strftime('%Y년 %m월 %d일 %H시 %M분') elif format_type == "full": return dt.strftime('%Y-%m-%d %H:%M:%S') else: return dt.strftime("%y%m%d_%H%M") # ===== 데이터 처리 및 검증 함수들 ===== def create_export_data_from_html(analysis_keyword, main_keyword, analysis_html, step1_data=None): """분석 HTML과 1단계 데이터를 기반으로 export용 데이터 구조 생성 (더미 데이터 제거)""" logger.info("=== 📊 Export 데이터 구조 생성 시작 (더미 데이터 제거 버전) ===") # 기본 export 데이터 구조 export_data = { "main_keyword": main_keyword or analysis_keyword, "analysis_keyword": analysis_keyword, "analysis_html": analysis_html, "main_keywords_df": None, "related_keywords_df": None, "analysis_completed": True, "created_at": get_korean_time().isoformat() } # 1단계 데이터에서 main_keywords_df 추출 (실제 데이터만) if step1_data and isinstance(step1_data, dict): if "keywords_df" in step1_data: keywords_df = step1_data["keywords_df"] if isinstance(keywords_df, dict): try: export_data["main_keywords_df"] = pd.DataFrame(keywords_df) logger.info(f"✅ 1단계 키워드 데이터를 DataFrame으로 변환: {export_data['main_keywords_df'].shape}") except Exception as e: logger.warning(f"⚠️ 1단계 데이터 변환 실패: {e}") export_data["main_keywords_df"] = None elif hasattr(keywords_df, 'shape'): export_data["main_keywords_df"] = keywords_df logger.info(f"✅ 1단계 키워드 DataFrame 사용: {keywords_df.shape}") else: logger.info("📋 1단계 키워드 데이터가 유효하지 않음 - None으로 유지") export_data["main_keywords_df"] = None # 분석 HTML에서 연관검색어 정보 추출 시도 (실제 데이터만) if analysis_html and "연관검색어 분석" in analysis_html: logger.info("🔍 분석 HTML에서 연관검색어 정보 발견 - 실제 파싱 필요") # 실제 HTML 파싱 로직이 필요한 부분 # 현재는 더미 데이터 대신 None으로 유지 export_data["related_keywords_df"] = None logger.info("💡 실제 HTML 파싱 로직 구현 필요 - 연관검색어 데이터는 None으로 유지") logger.info(f"📊 Export 데이터 구조 생성 완료 (더미 데이터 없음):") logger.info(f" - analysis_keyword: {export_data['analysis_keyword']}") logger.info(f" - main_keywords_df: {export_data['main_keywords_df'].shape if export_data['main_keywords_df'] is not None else 'None'}") logger.info(f" - related_keywords_df: {export_data['related_keywords_df'].shape if export_data['related_keywords_df'] is not None else 'None'}") logger.info(f" - analysis_html: {len(str(export_data['analysis_html']))} 문자") return export_data def validate_and_repair_export_data(export_data): """Export 데이터 유효성 검사 및 복구 (더미 데이터 제거)""" logger.info("🔧 Export 데이터 유효성 검사 및 복구 시작 (더미 데이터 제거 버전)") if not export_data or not isinstance(export_data, dict): logger.warning("⚠️ Export 데이터가 없거나 딕셔너리가 아님 - 기본 구조 생성") return { "main_keyword": "기본키워드", "analysis_keyword": "기본분석키워드", "analysis_html": "
기본 분석 결과
", "main_keywords_df": None, # 더미 데이터 대신 None "related_keywords_df": None, # 더미 데이터 대신 None "analysis_completed": True } # 필수 키들 확인 및 복구 required_keys = { "analysis_keyword": "분석키워드", "main_keyword": "메인키워드", "analysis_html": "
분석 완료
", "analysis_completed": True } for key, default_value in required_keys.items(): if key not in export_data or not export_data[key]: export_data[key] = default_value logger.info(f"🔧 {key} 키 복구: {default_value}") # DataFrame 데이터 검증 및 변환 (더미 데이터 생성 안함) for df_key in ["main_keywords_df", "related_keywords_df"]: if df_key in export_data and export_data[df_key] is not None: df_data = export_data[df_key] # 딕셔너리를 DataFrame으로 변환 if isinstance(df_data, dict): try: # 빈 딕셔너리는 None으로 처리 if not df_data: export_data[df_key] = None logger.info(f"📋 {df_key} 빈 딕셔너리 - None으로 설정") else: export_data[df_key] = pd.DataFrame(df_data) logger.info(f"✅ {df_key} 딕셔너리를 DataFrame으로 변환 성공") except Exception as e: logger.warning(f"⚠️ {df_key} 변환 실패: {e}") export_data[df_key] = None elif not hasattr(df_data, 'shape'): logger.warning(f"⚠️ {df_key}가 DataFrame이 아님 - None으로 설정") export_data[df_key] = None logger.info("✅ Export 데이터 유효성 검사 및 복구 완료 (더미 데이터 없음)") return export_data # ===== 파일 출력 함수들 ===== def create_timestamp_filename(analysis_keyword): """타임스탬프가 포함된 파일명 생성 - 한국시간 적용""" timestamp = format_korean_datetime(format_type="filename") safe_keyword = re.sub(r'[^\w\s-]', '', analysis_keyword).strip() safe_keyword = re.sub(r'[-\s]+', '_', safe_keyword) return f"{safe_keyword}_{timestamp}_분석결과" def export_to_excel(main_keyword, main_keywords_df, analysis_keyword, related_keywords_df, filename_base): """엑셀 파일로 출력 (실제 데이터만)""" try: # 실제 데이터가 있는지 확인 has_main_data = main_keywords_df is not None and not main_keywords_df.empty has_related_data = related_keywords_df is not None and not related_keywords_df.empty if not has_main_data and not has_related_data: logger.info("📋 생성할 데이터가 없어 엑셀 파일 생성 건너뜀") return None excel_filename = f"{filename_base}.xlsx" excel_path = os.path.join(tempfile.gettempdir(), excel_filename) with pd.ExcelWriter(excel_path, engine='xlsxwriter') as writer: # 워크북과 워크시트 스타일 설정 workbook = writer.book # 헤더 스타일 header_format = workbook.add_format({ 'bold': True, 'text_wrap': True, 'valign': 'top', 'fg_color': '#D7E4BC', 'border': 1 }) # 데이터 스타일 data_format = workbook.add_format({ 'text_wrap': True, 'valign': 'top', 'border': 1 }) # 숫자 포맷 number_format = workbook.add_format({ 'num_format': '#,##0', 'text_wrap': True, 'valign': 'top', 'border': 1 }) # 첫 번째 시트: 메인키워드 조합키워드 (실제 데이터만) if has_main_data: main_keywords_df.to_excel(writer, sheet_name=f'{main_keyword}_조합키워드', index=False) worksheet1 = writer.sheets[f'{main_keyword}_조합키워드'] # 헤더 스타일 적용 for col_num, value in enumerate(main_keywords_df.columns.values): worksheet1.write(0, col_num, value, header_format) # 데이터 스타일 적용 for row_num in range(1, len(main_keywords_df) + 1): for col_num, value in enumerate(main_keywords_df.iloc[row_num-1]): if isinstance(value, (int, float)) and col_num in [1, 2, 3]: # 검색량 컬럼 worksheet1.write(row_num, col_num, value, number_format) else: worksheet1.write(row_num, col_num, value, data_format) # 열 너비 자동 조정 for i, col in enumerate(main_keywords_df.columns): max_len = max( main_keywords_df[col].astype(str).map(len).max(), len(str(col)) ) worksheet1.set_column(i, i, min(max_len + 2, 50)) logger.info(f"✅ 메인키워드 시트 생성: {main_keywords_df.shape}") # 두 번째 시트: 분석키워드 연관검색어 (실제 데이터만) if has_related_data: related_keywords_df.to_excel(writer, sheet_name=f'{analysis_keyword}_연관검색어', index=False) worksheet2 = writer.sheets[f'{analysis_keyword}_연관검색어'] # 헤더 스타일 적용 for col_num, value in enumerate(related_keywords_df.columns.values): worksheet2.write(0, col_num, value, header_format) # 데이터 스타일 적용 for row_num in range(1, len(related_keywords_df) + 1): for col_num, value in enumerate(related_keywords_df.iloc[row_num-1]): if isinstance(value, (int, float)) and col_num in [1, 2, 3]: # 검색량 컬럼 worksheet2.write(row_num, col_num, value, number_format) else: worksheet2.write(row_num, col_num, value, data_format) # 열 너비 자동 조정 for i, col in enumerate(related_keywords_df.columns): max_len = max( related_keywords_df[col].astype(str).map(len).max(), len(str(col)) ) worksheet2.set_column(i, i, min(max_len + 2, 50)) logger.info(f"✅ 연관검색어 시트 생성: {related_keywords_df.shape}") logger.info(f"엑셀 파일 생성 완료: {excel_path}") return excel_path except Exception as e: logger.error(f"엑셀 파일 생성 오류: {e}") return None def export_to_html(analysis_html, filename_base): """HTML 파일로 출력 - 한국시간 적용""" try: html_filename = f"{filename_base}.html" html_path = os.path.join(tempfile.gettempdir(), html_filename) # 한국시간으로 생성 시간 표시 korean_time = format_korean_datetime(format_type="display") # 완전한 HTML 문서 생성 full_html = f""" 키워드 심충분석 결과

키워드 심충분석 결과

AI 상품 소싱 분석 시스템 v3.2 (더미 데이터 제거 버전)

{analysis_html}
생성 시간: {korean_time} (한국시간)
""" with open(html_path, 'w', encoding='utf-8') as f: f.write(full_html) logger.info(f"HTML 파일 생성 완료: {html_path}") return html_path except Exception as e: logger.error(f"HTML 파일 생성 오류: {e}") return None def create_zip_file(excel_path, html_path, filename_base): """압축 파일 생성""" try: zip_filename = f"{filename_base}.zip" zip_path = os.path.join(tempfile.gettempdir(), zip_filename) files_added = 0 with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf: if excel_path and os.path.exists(excel_path): zipf.write(excel_path, f"{filename_base}.xlsx") logger.info(f"엑셀 파일 압축 추가: {filename_base}.xlsx") files_added += 1 if html_path and os.path.exists(html_path): zipf.write(html_path, f"{filename_base}.html") logger.info(f"HTML 파일 압축 추가: {filename_base}.html") files_added += 1 if files_added == 0: logger.warning("압축할 파일이 없음") return None logger.info(f"압축 파일 생성 완료: {zip_path} ({files_added}개 파일)") return zip_path except Exception as e: logger.error(f"압축 파일 생성 오류: {e}") return None def export_analysis_results_enhanced(export_data): """강화된 분석 결과 출력 메인 함수 (더미 데이터 제거)""" try: logger.info("=== 📊 강화된 출력 함수 시작 (더미 데이터 제거 버전) ===") # 데이터 유효성 검사 및 복구 export_data = validate_and_repair_export_data(export_data) analysis_keyword = export_data.get("analysis_keyword", "기본키워드") analysis_html = export_data.get("analysis_html", "
분석 완료
") main_keyword = export_data.get("main_keyword", analysis_keyword) main_keywords_df = export_data.get("main_keywords_df") related_keywords_df = export_data.get("related_keywords_df") logger.info(f"🔍 처리할 데이터:") logger.info(f" - analysis_keyword: '{analysis_keyword}'") logger.info(f" - main_keyword: '{main_keyword}'") logger.info(f" - analysis_html: {len(str(analysis_html))} 문자") logger.info(f" - main_keywords_df: {main_keywords_df.shape if main_keywords_df is not None else 'None'}") logger.info(f" - related_keywords_df: {related_keywords_df.shape if related_keywords_df is not None else 'None'}") # 파일명 생성 (한국시간 적용) filename_base = create_timestamp_filename(analysis_keyword) logger.info(f"📁 출력 파일명: {filename_base}") # HTML 파일은 분석 결과가 있으면 생성 html_path = None if analysis_html and len(str(analysis_html).strip()) > 20: # 의미있는 HTML인지 확인 logger.info("🌐 HTML 파일 생성 시작...") html_path = export_to_html(analysis_html, filename_base) if html_path: logger.info(f"✅ HTML 파일 생성 성공: {html_path}") else: logger.error("❌ HTML 파일 생성 실패") else: logger.info("📄 분석 HTML이 없어 HTML 파일 생성 건너뜀") # 엑셀 파일 생성 (실제 DataFrame이 있는 경우만) excel_path = None if (main_keywords_df is not None and not main_keywords_df.empty) or \ (related_keywords_df is not None and not related_keywords_df.empty): logger.info("📊 엑셀 파일 생성 시작...") excel_path = export_to_excel( main_keyword, main_keywords_df, analysis_keyword, related_keywords_df, filename_base ) if excel_path: logger.info(f"✅ 엑셀 파일 생성 성공: {excel_path}") else: logger.warning("⚠️ 엑셀 파일 생성 실패") else: logger.info("📊 실제 DataFrame 데이터가 없어 엑셀 파일 생성 생략") # 생성된 파일이 있는지 확인 if not html_path and not excel_path: logger.warning("⚠️ 생성된 파일이 없음") return None, "⚠️ 생성할 수 있는 데이터가 없습니다. 분석을 먼저 완료해주세요." # 압축 파일 생성 logger.info("📦 압축 파일 생성 시작...") zip_path = create_zip_file(excel_path, html_path, filename_base) if zip_path: file_types = [] if html_path: file_types.append("HTML") if excel_path: file_types.append("엑셀") file_list = " + ".join(file_types) logger.info(f"✅ 압축 파일 생성 성공: {zip_path} ({file_list})") return zip_path, f"✅ 분석 결과가 성공적으로 출력되었습니다!\n파일명: {filename_base}.zip\n포함 파일: {file_list}\n\n💡 더미 데이터 제거 버전 - 실제 분석 데이터만 포함됩니다." else: logger.error("❌ 압축 파일 생성 실패") return None, "압축 파일 생성에 실패했습니다." except Exception as e: logger.error(f"❌ 강화된 출력 함수 전체 오류: {e}") import traceback logger.error(f"스택 트레이스:\n{traceback.format_exc()}") return None, f"출력 중 오류가 발생했습니다: {str(e)}" # ===== 로딩 애니메이션 ===== def create_loading_animation(): """로딩 애니메이션 HTML""" return """

분석 중입니다...

원격 서버에서 데이터를 수집하고 AI가 분석하고 있습니다.
잠시만 기다려주세요.

""" # ===== 에러 처리 함수 ===== def generate_error_response(error_message): """에러 응답 생성""" return f'''

❌ 연결 오류

{error_message}

해결 방법:

''' # ===== 원격 API 호출 함수들 ===== def call_collect_data_api(keyword): """1단계: 상품 데이터 수집 API 호출""" try: client = get_api_client() if not client: return generate_error_response("API 클라이언트를 초기화할 수 없습니다."), {} logger.info("원격 API 호출: 상품 데이터 수집") result = client.predict( keyword=keyword, api_name="/on_collect_data" ) logger.info(f"데이터 수집 API 결과 타입: {type(result)}") # 결과가 튜플인 경우 첫 번째 요소는 HTML, 두 번째는 세션 데이터 if isinstance(result, tuple) and len(result) == 2: html_result, session_data = result # 세션 데이터가 제대로 있는지 확인 if isinstance(session_data, dict): logger.info(f"데이터 수집 세션 데이터 수신: {list(session_data.keys()) if session_data else '빈 딕셔너리'}") return html_result, session_data else: logger.warning("세션 데이터가 딕셔너리가 아닙니다.") return html_result, {} else: logger.warning("예상과 다른 데이터 수집 결과 형태") return str(result), {"keywords_collected": True} except Exception as e: logger.error(f"상품 데이터 수집 API 호출 오류: {e}") return generate_error_response(f"원격 서버 연결 실패: {str(e)}"), {} def call_analyze_keyword_api_enhanced(analysis_keyword, base_keyword, keywords_data): """3단계: 강화된 키워드 심충분석 API 호출 (더미 데이터 제거)""" try: client = get_api_client() if not client: return generate_error_response("API 클라이언트를 초기화할 수 없습니다."), {} logger.info("=== 🚀 강화된 키워드 심충분석 API 호출 (더미 데이터 제거) ===") logger.info(f"파라미터 - analysis_keyword: '{analysis_keyword}'") logger.info(f"파라미터 - base_keyword: '{base_keyword}'") logger.info(f"파라미터 - keywords_data 타입: {type(keywords_data)}") # 원격 API 호출 result = client.predict( analysis_keyword, base_keyword, keywords_data, api_name="/on_analyze_keyword" ) logger.info(f"📡 원격 API 응답 수신:") logger.info(f" - 응답 타입: {type(result)}") logger.info(f" - 응답 길이: {len(result) if hasattr(result, '__len__') else 'N/A'}") # 응답 처리 및 Export 데이터 구조 생성 if isinstance(result, tuple) and len(result) == 2: html_result, remote_export_data = result logger.info(f"📊 원격 export 데이터:") logger.info(f" - 타입: {type(remote_export_data)}") logger.info(f" - 키들: {list(remote_export_data.keys()) if isinstance(remote_export_data, dict) else 'None'}") # HTML 결과가 있으면 Export 데이터 구조 생성 (더미 데이터 없이) if html_result: logger.info("🔧 Export 데이터 구조 생성 시작 (더미 데이터 제거)") enhanced_export_data = create_export_data_from_html( analysis_keyword=analysis_keyword, main_keyword=base_keyword, analysis_html=html_result, step1_data=keywords_data ) # 원격에서 온 실제 데이터가 있으면 병합 if isinstance(remote_export_data, dict) and remote_export_data: logger.info("🔗 원격 실제 데이터와 로컬 데이터 병합") for key, value in remote_export_data.items(): if value is not None and key in ["main_keywords_df", "related_keywords_df"]: # DataFrame 데이터만 검증하여 병합 if isinstance(value, dict) and value: # 빈 딕셔너리가 아닌 경우만 enhanced_export_data[key] = value logger.info(f" - {key} 원격 실제 데이터로 업데이트") elif hasattr(value, 'shape') and not value.empty: # DataFrame이고 비어있지 않은 경우 enhanced_export_data[key] = value logger.info(f" - {key} 원격 DataFrame 데이터로 업데이트") elif value is not None and key not in ["main_keywords_df", "related_keywords_df"]: enhanced_export_data[key] = value logger.info(f" - {key} 원격 데이터로 업데이트") logger.info(f"✅ 최종 Export 데이터 구조 (더미 데이터 없음):") logger.info(f" - 키 개수: {len(enhanced_export_data)}") logger.info(f" - 키 목록: {list(enhanced_export_data.keys())}") return html_result, enhanced_export_data else: logger.warning("⚠️ HTML 결과가 비어있음") return str(result), {} else: logger.warning("⚠️ 예상과 다른 API 응답 형태") # HTML만 반환된 경우도 처리 if isinstance(result, str) and len(result) > 100: # HTML일 가능성이 높음 logger.info("📄 HTML 문자열로 추정되는 응답 - Export 데이터 생성 (더미 데이터 없이)") enhanced_export_data = create_export_data_from_html( analysis_keyword=analysis_keyword, main_keyword=base_keyword, analysis_html=result, step1_data=keywords_data ) return result, enhanced_export_data else: return str(result), {} except Exception as e: logger.error(f"❌ 키워드 심충분석 API 호출 오류: {e}") import traceback logger.error(f"스택 트레이스:\n{traceback.format_exc()}") return generate_error_response(f"원격 서버 연결 실패: {str(e)}"), {} # ===== 그라디오 인터페이스 ===== def create_interface(): # CSS 스타일링 (기존과 동일) custom_css = """ /* 기존 다크모드 자동 변경 AI 상품 소싱 분석 시스템 CSS */ :root { --primary-color: #FB7F0D; --secondary-color: #ff9a8b; --accent-color: #FF6B6B; --background-color: #FFFFFF; --card-bg: #ffffff; --input-bg: #ffffff; --text-color: #334155; --text-secondary: #64748b; --border-color: #dddddd; --border-light: #e5e5e5; --table-even-bg: #f3f3f3; --table-hover-bg: #f0f0f0; --shadow: 0 8px 30px rgba(251, 127, 13, 0.08); --shadow-light: 0 2px 4px rgba(0, 0, 0, 0.1); --border-radius: 18px; } @media (prefers-color-scheme: dark) { :root { --background-color: #1a1a1a; --card-bg: #2d2d2d; --input-bg: #2d2d2d; --text-color: #e5e5e5; --text-secondary: #a1a1aa; --border-color: #404040; --border-light: #525252; --table-even-bg: #333333; --table-hover-bg: #404040; --shadow: 0 8px 30px rgba(0, 0, 0, 0.3); --shadow-light: 0 2px 4px rgba(0, 0, 0, 0.2); } } body { font-family: 'Pretendard', 'Noto Sans KR', -apple-system, BlinkMacSystemFont, sans-serif; background-color: var(--background-color) !important; color: var(--text-color) !important; line-height: 1.6; margin: 0; padding: 0; transition: background-color 0.3s ease, color 0.3s ease; } .gradio-container { width: 100%; margin: 0 auto; padding: 20px; background-color: var(--background-color) !important; } .custom-frame { background-color: var(--card-bg) !important; border: 1px solid var(--border-light) !important; border-radius: var(--border-radius); padding: 20px; margin: 10px 0; box-shadow: var(--shadow) !important; color: var(--text-color) !important; } .custom-button { border-radius: 30px !important; background: var(--primary-color) !important; color: white !important; font-size: 18px !important; padding: 10px 20px !important; border: none; box-shadow: 0 4px 8px rgba(251, 127, 13, 0.25); transition: transform 0.3s ease; height: 45px !important; width: 100% !important; } .custom-button:hover { transform: translateY(-2px); box-shadow: 0 6px 12px rgba(251, 127, 13, 0.3); } .export-button { background: linear-gradient(135deg, #28a745, #20c997) !important; color: white !important; border-radius: 25px !important; height: 50px !important; font-size: 17px !important; font-weight: bold !important; width: 100% !important; margin-top: 20px !important; } .section-title { display: flex; align-items: center; font-size: 20px; font-weight: 700; color: var(--text-color) !important; margin-bottom: 10px; padding-bottom: 5px; border-bottom: 2px solid var(--primary-color); font-family: 'Pretendard', 'Noto Sans KR', -apple-system, BlinkMacSystemFont, sans-serif; } .section-title img, .section-title i { margin-right: 10px; font-size: 20px; color: var(--primary-color); } .gr-input, .gr-text-input, .gr-sample-inputs, input[type="text"], input[type="number"], textarea, select { border-radius: var(--border-radius) !important; border: 1px solid var(--border-color) !important; padding: 12px !important; box-shadow: inset 0 1px 3px rgba(0, 0, 0, 0.05) !important; transition: all 0.3s ease !important; background-color: var(--input-bg) !important; color: var(--text-color) !important; } .gr-input:focus, .gr-text-input:focus, input[type="text"]:focus, textarea:focus, select:focus { border-color: var(--primary-color) !important; outline: none !important; box-shadow: 0 0 0 2px rgba(251, 127, 13, 0.2) !important; } .fade-in { animation: fadeIn 0.5s ease-out; } @keyframes fadeIn { from { opacity: 0; transform: translateY(10px); } to { opacity: 1; transform: translateY(0); } } """ with gr.Blocks( css=custom_css, title="🛒 AI 상품 소싱 분석기 v3.2 (더미 데이터 제거)", theme=gr.themes.Default(primary_hue="orange", secondary_hue="orange") ) as interface: # 폰트 및 아이콘 로드 gr.HTML(""" """) # 세션별 상태 변수 keywords_data_state = gr.State() export_data_state = gr.State({}) # === UI 컴포넌트들 === with gr.Column(elem_classes="custom-frame fade-in"): gr.HTML('
1단계: 메인 키워드 입력
') keyword_input = gr.Textbox( label="상품 메인키워드", placeholder="예: 슬리퍼, 무선이어폰, 핸드크림", value="", elem_id="keyword_input" ) collect_data_btn = gr.Button("1단계: 상품 데이터 수집하기", elem_classes="custom-button", size="lg") with gr.Column(elem_classes="custom-frame fade-in"): gr.HTML('
2단계: 수집된 키워드 목록
') keywords_result = gr.HTML() with gr.Column(elem_classes="custom-frame fade-in"): gr.HTML('
3단계: 분석할 키워드 선택
') analysis_keyword_input = gr.Textbox( label="분석할 키워드", placeholder="위 목록에서 원하는 키워드를 입력하세요 (예: 통굽 슬리퍼)", value="", elem_id="analysis_keyword_input" ) analyze_keyword_btn = gr.Button("키워드 심충분석 하기", elem_classes="custom-button", size="lg") with gr.Column(elem_classes="custom-frame fade-in"): gr.HTML('
키워드 심충분석
') analysis_result = gr.HTML(label="키워드 심충분석") with gr.Column(elem_classes="custom-frame fade-in"): gr.HTML('
분석 결과 출력
') gr.HTML("""

실제 데이터 출력 버전

• 분석된 데이터를 파일로 출력됩니다

""") export_btn = gr.Button("📊 분석결과 출력하기", elem_classes="export-button", size="lg") export_result = gr.HTML() download_file = gr.File(label="다운로드", visible=False) # ===== 이벤트 핸들러 ===== def on_collect_data(keyword): if not keyword.strip(): return ("
키워드를 입력해주세요.
", None) # 로딩 상태 표시 yield (create_loading_animation(), None) # 원격 API 호출 result_html, result_data = call_collect_data_api(keyword) yield (result_html, result_data) def on_analyze_keyword(analysis_keyword, base_keyword, keywords_data): if not analysis_keyword.strip(): return "
분석할 키워드를 입력해주세요.
", {} # 로딩 상태 표시 yield create_loading_animation(), {} # 강화된 API 호출 (더미 데이터 제거) html_result, enhanced_export_data = call_analyze_keyword_api_enhanced( analysis_keyword, base_keyword, keywords_data ) yield html_result, enhanced_export_data def on_export_results(export_data): """강화된 분석 결과 출력 핸들러 (더미 데이터 제거)""" try: logger.info(f"📊 입력 export_data: {type(export_data)}") if isinstance(export_data, dict): logger.info(f"📋 export_data 키들: {list(export_data.keys())}") # 강화된 출력 함수 호출 (더미 데이터 제거) zip_path, message = export_analysis_results_enhanced(export_data) if zip_path: success_html = f"""

출력 완료!

{message}
데이터출력:

아래 다운로드 버튼을 클릭하여 파일을 저장하세요.
⏰ 한국시간 기준으로 파일명이 생성됩니다.

""" return success_html, gr.update(value=zip_path, visible=True) else: error_html = f"""

출력 실패

{message}

🔍 디버깅 정보:
""" logger.error("❌ 강화된 출력 실패") return error_html, gr.update(visible=False) except Exception as e: logger.error(f"❌ 강화된 출력 핸들러 오류: {e}") import traceback logger.error(f"스택 트레이스:\n{traceback.format_exc()}") error_html = f"""

시스템 오류

강화된 출력 중 시스템 오류가 발생했습니다:

{type(e).__name__}: {str(e)}

💡 실제 분석 결과가 있어야만 파일이 생성됩니다.

""" return error_html, gr.update(visible=False) # ===== 이벤트 연결 ===== collect_data_btn.click( fn=on_collect_data, inputs=[keyword_input], outputs=[keywords_result, keywords_data_state], api_name="on_collect_data" ) analyze_keyword_btn.click( fn=on_analyze_keyword, inputs=[analysis_keyword_input, keyword_input, keywords_data_state], outputs=[analysis_result, export_data_state], api_name="on_analyze_keyword" ) export_btn.click( fn=on_export_results, inputs=[export_data_state], outputs=[export_result, download_file], api_name="on_export_results" ) return interface # ===== 메인 실행 ===== if __name__ == "__main__": # pytz 모듈 설치 확인 try: import pytz logger.info("✅ pytz 모듈 로드 성공 - 한국시간 지원") except ImportError: logger.info("시스템 시간을 사용합니다.") # 앱 실행 app = create_interface() app.launch(server_name="0.0.0.0", server_port=7860, share=True)