# -*- coding: utf-8 -*- """ AI 상품 소싱 분석 시스템 v2.9 - 출력 기능 추가 + 멀티사용자 안전 - 연관검색어 엑셀 출력 - 키워드 심충분석 HTML 출력 - 압축파일로 결과 다운로드 - Gemini API 키 통합 관리 - 한국시간 적용 - 멀티 사용자 안전: gr.State 사용으로 세션별 데이터 관리 """ import gradio as gr import pandas as pd import os import logging import google.generativeai as genai from datetime import datetime, timedelta import pytz # 한국시간 적용을 위한 추가 import time import re from collections import Counter import zipfile import tempfile # 로깅 설정 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # 모듈 임포트 import api_utils import text_utils import keyword_search import product_search import keyword_processor import export_utils import keyword_analysis import trend_analysis_v2 # ===== Gemini API 설정 ===== def setup_gemini_model(): """Gemini 모델 초기화 - api_utils에서 관리""" try: # api_utils에서 Gemini 모델 가져오기 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 # Gemini 모델 초기화 gemini_model = setup_gemini_model() # ===== 한국시간 관련 함수 ===== 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") # ===== 출력 전용 상태 변수 제거 (멀티 사용자 안전을 위해 gr.State 사용) ===== # export_state 전역 변수 제거 - 멀티 사용자 환경에서 데이터 혼합 문제 해결 # ===== 연관검색어 분석 기능 ===== def analyze_related_keywords(keyword): """연관검색어 분석 - 네이버 상품 40개를 기반으로 복합키워드 추출""" logger.info(f"연관검색어 분석 시작: '{keyword}'") try: # 1단계: 네이버 상품 40개 추출 api_keyword = keyword.replace(" ", "") products_data = [] # 40개 상품을 가져오기 위해 여러 페이지 호출 for page in range(1, 5): # 4페이지 * 10개 = 40개 result = product_search.fetch_products_by_keyword(api_keyword, page=page, display=10) if result["status"] == "success" and result["products"]: products_data.extend(result["products"]) else: break time.sleep(0.3) # API 레이트 리밋 방지 if not products_data: return { "status": "error", "message": f"'{keyword}' 키워드로 상품을 찾을 수 없습니다.", "keywords_df": pd.DataFrame() } # 실제 가져온 상품 수 제한 products_data = products_data[:40] logger.info(f"상품 추출 완료: {len(products_data)}개") # 2단계: 상품명에서 키워드 추출 (스페이스바로 분류) all_words = [] for product in products_data: title = product.get("상품명", "") # 공백과 쉼표로 분리 words = re.split(r'[,\s]+', title) all_words.extend([word.strip() for word in words if word.strip() and len(word.strip()) >= 1]) # 중복 제거 unique_words = list(set(all_words)) logger.info(f"추출된 단어 수: {len(unique_words)}개") # 3단계: 분석할 키워드를 앞뒤로 붙여서 복합키워드 생성 compound_keywords = [] main_keyword = keyword.strip() for word in unique_words: if word != main_keyword and len(word) >= 2: # 단일 글자 제외 # 앞에 붙이기 front_compound = f"{word} {main_keyword}" compound_keywords.append(front_compound) # 뒤에 붙이기 back_compound = f"{main_keyword} {word}" compound_keywords.append(back_compound) # 중복 제거 compound_keywords = list(set(compound_keywords)) logger.info(f"생성된 복합키워드 수: {len(compound_keywords)}개") # 4단계: 검색량 추출 api_keywords = [kw.replace(" ", "") for kw in compound_keywords] search_volumes = keyword_search.fetch_all_search_volumes(api_keywords) # 5단계: 앞뒤 키워드 중 높은 것 선택, 낮은 것 제거 keyword_pairs = {} # {base_word: {"front": front_kw, "back": back_kw, "front_vol": vol, "back_vol": vol}} for word in unique_words: if word != main_keyword and len(word) >= 2: front_kw = f"{word} {main_keyword}" back_kw = f"{main_keyword} {word}" front_api = front_kw.replace(" ", "") back_api = back_kw.replace(" ", "") front_vol = search_volumes.get(front_api, {}).get("총검색량", 0) back_vol = search_volumes.get(back_api, {}).get("총검색량", 0) keyword_pairs[word] = { "front": front_kw, "back": back_kw, "front_vol": front_vol, "back_vol": back_vol } # 6단계: 높은 검색량의 키워드만 선택 final_keywords = [] for word, data in keyword_pairs.items(): if data["front_vol"] > data["back_vol"]: selected_kw = data["front"] selected_vol = data["front_vol"] selected_api = selected_kw.replace(" ", "") elif data["back_vol"] > data["front_vol"]: selected_kw = data["back"] selected_vol = data["back_vol"] selected_api = selected_kw.replace(" ", "") elif data["front_vol"] == data["back_vol"] and data["front_vol"] > 0: # 같은 검색량이면 자연스러운 순서 선택 (일반적으로 뒤에 붙이는 것이 자연스러움) selected_kw = data["back"] selected_vol = data["back_vol"] selected_api = selected_kw.replace(" ", "") else: # 둘 다 0이면 제외 continue if selected_vol > 0: # 검색량이 있는 것만 포함 vol_data = search_volumes.get(selected_api, {}) final_keywords.append({ "연관 키워드": selected_kw, "PC검색량": vol_data.get("PC검색량", 0), "모바일검색량": vol_data.get("모바일검색량", 0), "총검색량": selected_vol, "검색량구간": text_utils.get_search_volume_range(selected_vol) }) # 검색량 기준으로 내림차순 정렬 final_keywords = sorted(final_keywords, key=lambda x: x["총검색량"], reverse=True) # DataFrame 생성 df_keywords = pd.DataFrame(final_keywords) logger.info(f"연관검색어 분석 완료: {len(final_keywords)}개 키워드") return { "status": "success", "message": f"'{keyword}' 연관검색어 {len(final_keywords)}개를 찾았습니다.", "keywords_df": df_keywords, "total_products": len(products_data) } except Exception as e: logger.error(f"연관검색어 분석 오류: {e}") return { "status": "error", "message": f"연관검색어 분석 중 오류가 발생했습니다: {str(e)}", "keywords_df": pd.DataFrame() } # ===== 로딩 애니메이션 ===== def create_loading_animation(): """로딩 애니메이션 HTML""" return """

분석 중입니다...

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

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

❌ 분석 오류

{error_message}

해결 방법:

''' # ===== 메인 키워드 분석 함수 ===== def safe_keyword_analysis(analysis_keyword, base_keyword, keywords_data): """에러 방지를 위한 안전한 키워드 분석 - 세션별 데이터 반환""" # 입력값 검증 if not analysis_keyword or not analysis_keyword.strip(): return generate_error_response("분석할 키워드를 입력해주세요."), {} analysis_keyword = analysis_keyword.strip() try: # 검색량 조회 - 에러 방지 api_keyword = keyword_analysis.normalize_keyword_for_api(analysis_keyword) search_volumes = keyword_search.fetch_all_search_volumes([api_keyword]) volume_data = search_volumes.get(api_keyword, {"PC검색량": 0, "모바일검색량": 0, "총검색량": 0}) # 검색량이 0이거나 키워드가 존재하지 않는 경우 처리 if volume_data['총검색량'] == 0: logger.warning(f"'{analysis_keyword}' 키워드의 검색량이 0이거나 존재하지 않습니다.") error_result = f"""

⚠️ 키워드 분석 불가

'{analysis_keyword}' 키워드는 검색량이 없거나 올바르지 않은 키워드입니다.

💡 권장사항

""" return error_result, {} logger.info(f"'{analysis_keyword}' 현재 검색량: {volume_data['총검색량']:,}") # 트렌드 분석 시도 monthly_data_1year = {} monthly_data_3year = {} trend_available = False try: # 데이터랩 API 키 확인 datalab_config = api_utils.get_next_datalab_api_config() if datalab_config and not datalab_config["CLIENT_ID"].startswith("YOUR_"): logger.info("데이터랩 API 키가 설정되어 있어 1년, 3년 트렌드 분석을 시도합니다.") # 최적화된 API 함수 사용 # 1년 트렌드 데이터 trend_data_1year = trend_analysis_v2.get_naver_trend_data_v5([analysis_keyword], "1year", max_retries=3) if trend_data_1year: current_volumes = {api_keyword: volume_data} monthly_data_1year = trend_analysis_v2.calculate_monthly_volumes_v7([analysis_keyword], current_volumes, trend_data_1year, "1year") # 3년 트렌드 데이터 trend_data_3year = trend_analysis_v2.get_naver_trend_data_v5([analysis_keyword], "3year", max_retries=3) if trend_data_3year: current_volumes = {api_keyword: volume_data} monthly_data_3year = trend_analysis_v2.calculate_monthly_volumes_v7([analysis_keyword], current_volumes, trend_data_3year, "3year") # 3년 데이터가 없는 경우 1년 데이터로 확장 if not monthly_data_3year and monthly_data_1year: logger.info("3년 데이터가 없어 1년 데이터를 기반으로 3년 차트 생성") keyword = analysis_keyword if keyword in monthly_data_1year: data_1y = monthly_data_1year[keyword] # 3년 분량의 날짜 생성 (24개월 추가) extended_dates = [] extended_volumes = [] # 기존 1년 데이터 이전에 24개월 추가 (모두 0으로) start_date = datetime.strptime(data_1y["dates"][0], "%Y-%m-%d") for i in range(24, 0, -1): prev_date = start_date - timedelta(days=30 * i) extended_dates.append(prev_date.strftime("%Y-%m-%d")) extended_volumes.append(0) # 기존 1년 데이터 추가 (예상 데이터 제외) actual_count = data_1y.get("actual_count", len(data_1y["dates"])) extended_dates.extend(data_1y["dates"][:actual_count]) extended_volumes.extend(data_1y["monthly_volumes"][:actual_count]) monthly_data_3year = { keyword: { "monthly_volumes": extended_volumes, "dates": extended_dates, "current_volume": data_1y["current_volume"], "growth_rate": trend_analysis_v2.calculate_3year_growth_rate_improved(extended_volumes), "volume_per_percent": data_1y["volume_per_percent"], "current_ratio": data_1y["current_ratio"], "actual_count": len(extended_volumes), "predicted_count": 0 } } if monthly_data_1year or monthly_data_3year: trend_available = True logger.info("트렌드 분석 성공") else: logger.info("트렌드 데이터 처리 실패") else: logger.info("데이터랩 API 키가 설정되지 않음") except Exception as e: logger.info(f"트렌드 분석 건너뜀: {str(e)[:100]}") # 키워드 데이터 준비 step2_keywords_df = keywords_data.get("keywords_df") if keywords_data else None filtered_keywords_df = step2_keywords_df # 단순히 원본 데이터 사용 target_categories = [] # 빈 리스트 # === 📈 검색량 트렌드 분석 섹션 === if trend_available and (monthly_data_1year or monthly_data_3year): try: trend_chart = trend_analysis_v2.create_trend_chart_v7(monthly_data_1year, monthly_data_3year) except Exception as e: logger.warning(f"트렌드 차트 생성 실패, 기본 차트 사용: {e}") trend_chart = trend_analysis_v2.create_enhanced_current_chart(volume_data, analysis_keyword) else: trend_chart = trend_analysis_v2.create_enhanced_current_chart(volume_data, analysis_keyword) # 트렌드 섹션 trend_section = f"""

📈 검색량 트렌드 분석

{trend_chart}
""" # === 🎯 키워드 분석 섹션 (AI 분석) === # api_utils에서 Gemini 모델 가져오기 current_gemini_model = api_utils.get_gemini_model() keyword_analysis_html = keyword_analysis.analyze_keyword_for_sourcing( analysis_keyword, volume_data, monthly_data_1year, monthly_data_3year, filtered_keywords_df, target_categories, current_gemini_model ) keyword_analysis_section = f"""

🎯 키워드 분석

{keyword_analysis_html}
""" # 경고 섹션 (필요한 경우) warning_section = "" if not trend_available: warning_section = f"""
⚠️
일부 기능 제한
트렌드 분석에 제한이 있습니다. 현재 검색량 분석과 AI 추천은 정상 제공됩니다.
완전한 월 데이터 기준으로 분석하기 위해 최신 완료된 월까지만 표시됩니다.
""" # 최종 결과 조합 final_result = warning_section + trend_section + keyword_analysis_section # 세션별 출력용 상태 데이터 반환 (멀티 사용자 안전) session_export_data = { "main_keyword": base_keyword, "analysis_keyword": analysis_keyword, "main_keywords_df": keywords_data.get("keywords_df") if keywords_data else None, "related_keywords_df": None, # 여기서는 연관검색어 분석하지 않음 "analysis_html": final_result } return final_result, session_export_data except Exception as e: logger.error(f"키워드 분석 중 전체 오류: {e}") error_result = generate_error_response(f"키워드 분석 중 오류가 발생했습니다: {str(e)}") return error_result, {} # ===== 2단계: 상품 데이터 기반 키워드 추출 ===== def extract_keywords_from_products(keyword): """네이버 쇼핑에서 실제 상품 데이터를 수집하고 모든 키워드 표시""" logger.info(f"상품 키워드 추출 시작: 키워드='{keyword}'") api_keyword = keyword_analysis.normalize_keyword_for_api(keyword) search_results = product_search.fetch_naver_shopping_data( keyword, korean_only=True, apply_main_keyword=True, exclude_zero_volume=True ) if not search_results.get("product_list"): return { "status": "error", "message": "상품 데이터를 가져올 수 없습니다.", "products": [], "keywords": [] } processed_results = keyword_processor.process_search_results( search_results, keyword, exclude_zero_volume=True ) df_keywords = processed_results["keywords_df"] df_products = processed_results["products_df"] if df_keywords.empty: return { "status": "error", "message": "추출된 키워드가 없습니다.", "products": [], "keywords": [] } logger.info(f"키워드 추출 완료: 총 {len(df_keywords)}개 키워드") return { "status": "success", "message": "키워드 추출 완료", "products": df_products, "keywords_df": df_keywords, "categories": processed_results["categories"] } # ===== 파일 출력 함수들 ===== 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: 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 main_keywords_df is not None and not main_keywords_df.empty: 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 col_num in [1, 2, 3]: # PC검색량, 모바일검색량, 총검색량 컬럼 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)) # 두 번째 시트: 분석키워드 연관검색어 if related_keywords_df is not None and not related_keywords_df.empty: 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 col_num in [1, 2, 3]: # PC검색량, 모바일검색량, 총검색량 컬럼 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"엑셀 파일 생성 완료: {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 상품 소싱 분석 시스템 v2.9

{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) 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") if html_path and os.path.exists(html_path): zipf.write(html_path, f"{filename_base}.html") logger.info(f"HTML 파일 압축 추가: {filename_base}.html") logger.info(f"압축 파일 생성 완료: {zip_path}") return zip_path except Exception as e: logger.error(f"압축 파일 생성 오류: {e}") return None def export_analysis_results(export_data): """분석 결과 출력 메인 함수 - 세션별 데이터 처리""" try: # 출력할 데이터 확인 if not export_data or not isinstance(export_data, dict): return None, "분석 데이터가 없습니다. 먼저 키워드 심충분석을 실행해주세요." analysis_keyword = export_data.get("analysis_keyword", "") analysis_html = export_data.get("analysis_html", "") main_keyword = export_data.get("main_keyword", "") main_keywords_df = export_data.get("main_keywords_df") related_keywords_df = export_data.get("related_keywords_df") if not analysis_keyword: return None, "분석할 키워드가 설정되지 않았습니다. 먼저 키워드 분석을 실행해주세요." if not analysis_html: return None, "분석 결과가 없습니다. 먼저 키워드 심충분석을 실행해주세요." # 파일명 생성 (한국시간 적용) filename_base = create_timestamp_filename(analysis_keyword) logger.info(f"출력 파일명: {filename_base}") # 엑셀 파일 생성 excel_path = None if main_keywords_df is not None or related_keywords_df is not None: excel_path = export_to_excel( main_keyword, main_keywords_df, analysis_keyword, related_keywords_df, filename_base ) # HTML 파일 생성 html_path = export_to_html(analysis_html, filename_base) # 압축 파일 생성 if excel_path or html_path: zip_path = create_zip_file(excel_path, html_path, filename_base) if zip_path: return zip_path, f"✅ 분석 결과가 성공적으로 출력되었습니다!\n파일명: {filename_base}.zip" else: return None, "압축 파일 생성에 실패했습니다." else: return None, "출력할 파일이 없습니다." except Exception as e: logger.error(f"분석 결과 출력 오류: {e}") return None, f"출력 중 오류가 발생했습니다: {str(e)}" # ===== 그라디오 인터페이스 ===== def create_interface(): # CSS 파일 로드 try: with open('style.css', 'r', encoding='utf-8') as f: custom_css = f.read() with open('keyword_analysis_report.css', 'r', encoding='utf-8') as f: keyword_css = f.read() custom_css += "\n" + keyword_css 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; } .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; } """ with gr.Blocks( css=custom_css, title="🛒 AI 상품 소싱 분석기 v2.9", theme=gr.themes.Default(primary_hue="orange", secondary_hue="orange") ) as interface: # 폰트 및 아이콘 로드 gr.HTML(""" """) # 세션별 상태 변수 (멀티 사용자 안전) keywords_data_state = gr.State() export_data_state = gr.State({}) # === 1단계: 메인 키워드 입력 === 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") # === 2단계: 수집된 키워드 목록 === with gr.Column(elem_classes="custom-frame fade-in"): gr.HTML('
2단계: 수집된 키워드 목록
') keywords_result = gr.HTML() # === 3단계: 분석할 키워드 선택 === 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('
분석 결과 출력
') 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) result = extract_keywords_from_products(keyword) if result["status"] == "error": yield (f"
{result['message']}
", None) return keywords_df = result["keywords_df"] html_table = export_utils.create_table_without_checkboxes(keywords_df) success_html = f"""

✅ 네이버 데이터 수집 완료!

• 실제 상품 {len(result['products'])}개 분석
• 추출된 키워드: {len(keywords_df)}개
• 아래 목록에서 원하는 키워드를 선택하여 분석하세요

📊 전체 키워드 목록
{html_table} """ yield (success_html, result) def on_analyze_keyword(analysis_keyword, base_keyword, keywords_data): if not analysis_keyword.strip(): return "
분석할 키워드를 입력해주세요.
", {} # 로딩 상태 표시 yield create_loading_animation(), {} # 연관검색어 분석 먼저 실행 related_result = analyze_related_keywords(analysis_keyword) # 실제 키워드 분석 실행 (세션별 데이터 반환) keyword_result, session_export_data = safe_keyword_analysis(analysis_keyword, base_keyword, keywords_data) # 연관검색어 분석 결과를 세션 데이터에 추가 if related_result["status"] == "success" and not related_result["keywords_df"].empty: session_export_data["related_keywords_df"] = related_result["keywords_df"] # 연관검색어 분석 결과 HTML 생성 if related_result["status"] == "success" and not related_result["keywords_df"].empty: df_keywords = related_result["keywords_df"] related_table = export_utils.create_table_without_checkboxes(df_keywords) related_html = f"""

🔗 연관검색어 분석

🔗 연관검색어 분석 완료!

• 분석 기준 상품: {related_result['total_products']}개
• 발견된 연관검색어: {len(df_keywords)}개
• 메인 키워드와 결합된 복합키워드만 표시됩니다

{related_table}
""" # 세션 데이터의 analysis_html을 업데이트 session_export_data["analysis_html"] = related_html + session_export_data["analysis_html"] else: related_html = f"""

🔗 연관검색어 분석

'{analysis_keyword}' 키워드의 연관검색어를 찾을 수 없습니다.
""" # 세션 데이터의 analysis_html을 업데이트 session_export_data["analysis_html"] = related_html + session_export_data["analysis_html"] # 최종 결과 조합 final_result = related_html + keyword_result yield final_result, session_export_data def on_export_results(export_data): """분석 결과 출력 핸들러 - 세션별 데이터 처리""" try: zip_path, message = export_analysis_results(export_data) if zip_path: # 성공 메시지와 함께 다운로드 파일 제공 success_html = f"""

출력 완료!

{message}
포함 파일:
• 📊 엑셀 파일: 메인키워드 조합키워드 + 연관검색어 데이터
• 🌐 HTML 파일: 키워드 심충분석 결과 (그래프 포함)

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

""" return success_html, gr.update(value=zip_path, visible=True) else: # 실패 메시지 error_html = f"""

출력 실패

{message}

""" return error_html, gr.update(visible=False) except Exception as e: logger.error(f"출력 핸들러 오류: {e}") error_html = f"""

시스템 오류

출력 중 시스템 오류가 발생했습니다: {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] ) analyze_keyword_btn.click( fn=on_analyze_keyword, inputs=[analysis_keyword_input, keyword_input, keywords_data_state], outputs=[analysis_result, export_data_state] ) export_btn.click( fn=on_export_results, inputs=[export_data_state], outputs=[export_result, download_file] ) return interface # ===== API 설정 확인 함수 ===== def check_datalab_api_config(): """네이버 데이터랩 API 설정 확인""" logger.info("=== 네이버 데이터랩 API 설정 확인 ===") datalab_config = api_utils.get_next_datalab_api_config() if not datalab_config: logger.warning("❌ 데이터랩 API 키가 설정되지 않았습니다.") logger.info("트렌드 분석 기능이 비활성화됩니다.") return False client_id = datalab_config["CLIENT_ID"] client_secret = datalab_config["CLIENT_SECRET"] logger.info(f"총 {len(api_utils.NAVER_DATALAB_CONFIGS)}개의 데이터랩 API 설정 사용 중") logger.info(f"현재 선택된 API:") logger.info(f" CLIENT_ID: {client_id[:8]}***{client_id[-4:] if len(client_id) > 12 else '***'}") logger.info(f" CLIENT_SECRET: {client_secret[:4]}***{client_secret[-2:] if len(client_secret) > 6 else '***'}") # 기본값 체크 if client_id.startswith("YOUR_"): logger.error("❌ CLIENT_ID가 기본값으로 설정되어 있습니다!") return False if client_secret.startswith("YOUR_"): logger.error("❌ CLIENT_SECRET이 기본값으로 설정되어 있습니다!") return False # 길이 체크 if len(client_id) < 10: logger.warning("⚠️ CLIENT_ID가 짧습니다. 올바른 키인지 확인해주세요.") if len(client_secret) < 5: logger.warning("⚠️ CLIENT_SECRET이 짧습니다. 올바른 키인지 확인해주세요.") logger.info("✅ 데이터랩 API 키 형식 검증 완료") return True def check_gemini_api_config(): """Gemini API 설정 확인""" logger.info("=== Gemini API 설정 확인 ===") is_valid, message = api_utils.validate_gemini_config() if is_valid: logger.info(f"✅ {message}") # 첫 번째 사용 가능한 키 테스트 test_key = api_utils.get_next_gemini_api_key() if test_key: logger.info(f"현재 사용 중인 Gemini API 키: {test_key[:8]}***{test_key[-4:]}") return True else: logger.warning(f"❌ {message}") logger.info("AI 분석 기능이 제한될 수 있습니다.") return False # ===== 메인 실행 ===== if __name__ == "__main__": # pytz 모듈 설치 확인 try: import pytz logger.info("✅ pytz 모듈 로드 성공 - 한국시간 지원") except ImportError: logger.warning("⚠️ pytz 모듈이 설치되지 않음 - pip install pytz 실행 필요") logger.info("시스템 시간을 사용합니다.") # API 설정 초기화 api_utils.initialize_api_configs() logger.info("===== 상품 소싱 분석 시스템 v2.9 (출력기능 추가 + 한국시간 + 멀티사용자 안전) 시작 =====") # 네이버 데이터랩 API 설정 확인 datalab_available = check_datalab_api_config() # Gemini API 설정 확인 gemini_available = check_gemini_api_config() # 필요한 패키지 안내 print("📦 필요한 패키지:") print(" pip install gradio google-generativeai pandas requests xlsxwriter markdown plotly pytz") print() # API 키 설정 안내 if not gemini_available: print("⚠️ GEMINI_API_KEY 또는 GOOGLE_API_KEY 환경변수를 설정하세요.") print(" export GEMINI_API_KEY='your-api-key'") print(" 또는") print(" export GOOGLE_API_KEY='your-api-key'") print() if not datalab_available: print("⚠️ 네이버 데이터랩 API 트렌드 분석을 위해서는:") print(" 1. 네이버 개발자센터(https://developers.naver.com)에서 애플리케이션 등록") print(" 2. '데이터랩(검색어 트렌드)' API 추가") print(" 3. 발급받은 CLIENT_ID와 CLIENT_SECRET을 api_utils.py의 NAVER_DATALAB_CONFIGS에 설정") print(" 4. 현재는 현재 검색량 정보만 표시됩니다.") print() else: print("✅ 데이터랩 API 설정 완료 - 1년, 3년 트렌드 분석이 가능합니다!") print() if gemini_available: print("✅ Gemini API 설정 완료 - AI 분석이 가능합니다!") print() print("🛡️ v2.9 멀티사용자 안전 개선사항:") print(" • 전역 변수 export_state 완전 제거") print(" • gr.State({}) 사용으로 각 사용자별 세션 데이터 완전 분리") print(" • safe_keyword_analysis() 함수에서 세션별 데이터 반환") print(" • export_analysis_results() 함수에서 세션별 데이터 처리") print(" • 이벤트 핸들러에서 export_data_state 세션 상태 관리") print(" • 허깅페이스 스페이스 등 멀티사용자 환경에서 안전한 동시 사용 보장") print() print("🚀 기존 v2.9 기능:") print(" • 연관검색어 엑셀 출력") print(" • 키워드 심충분석 HTML 출력") print(" • 압축파일로 결과 다운로드") print(" • Gemini API 키 통합 관리") print(" • 한국시간 적용") print() # 앱 실행 app = create_interface() app.launch(server_name="0.0.0.0", server_port=7860, share=True)