""" 카테고리 분석 모듈 - 상품의 카테고리 분석 기능 제공 (개선버전) - 1년/3년 트렌드 모두 분석 - 너비 100% 적용 - 3년 기준 성장률 계산 """ import pandas as pd import time import re import random from collections import Counter, defaultdict import text_utils import product_search import keyword_search import logging # 로깅 설정 logger = logging.getLogger(__name__) logger.setLevel(logging.INFO) formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') handler = logging.StreamHandler() handler.setFormatter(formatter) logger.addHandler(handler) # 마지막 키워드 분석 결과를 저장할 전역 변수 _last_keyword_results = [] def get_last_keyword_results(): """마지막으로 분석된 키워드 결과 반환""" global _last_keyword_results return _last_keyword_results def exponential_backoff_sleep(retry_count, base_delay=0.3, max_delay=5.0): """지수 백오프 방식의 대기 시간 계산""" delay = min(base_delay * (2 ** retry_count), max_delay) # 약간의 랜덤성 추가 (지터) jitter = random.uniform(0, 0.5) * delay time.sleep(delay + jitter) def analyze_product_categories(main_keyword, product_name, category_filter=None): """ 메인 키워드와 상품명으로 카테고리 분석을 수행 Args: main_keyword (str): 메인 검색 키워드 product_name (str): 분석할 상품명 category_filter (str, optional): 카테고리 필터 Returns: dict: 분석 결과 """ logger.info(f"카테고리 분석 시작: 메인 키워드={main_keyword}, 상품명={product_name}") # 1단계: 메인 키워드로 100개 상품 가져오기 (10개씩 10페이지) all_products = [] for page in range(1, 11): result = product_search.fetch_products_by_keyword(main_keyword, page=page, display=10) if result["products"]: all_products.extend(result["products"]) exponential_backoff_sleep(0) # API 레이트 리밋 방지 if not all_products: return { "status": "error", "message": "상품을 가져오지 못했습니다.", "main_keyword": main_keyword, "product_name": product_name, "total_count": 0, "products": [], "categories": [], "analysis": None } # 2단계: 상품명에서 키워드 추출 (개선: 더 정확한 키워드 추출) product_keywords = [] # 공백과 쉼표를 기준으로 자연스럽게 분리 words = re.split(r'[,\s]+', product_name) for word in words: word = word.strip() if word and len(word) >= 2: # 최소 2글자 이상인 단어만 # 중복 제거 if word not in product_keywords: product_keywords.append(word) logger.info(f"상품명에서 추출한 키워드: {product_keywords}") # 3단계: 상품 카테고리 분석 category_counter = Counter() products_by_category = defaultdict(list) for product in all_products: category = product["카테고리"] category_counter[category] += 1 products_by_category[category].append(product) # 카테고리 필터 적용 if category_filter and category_filter != "전체 보기": # 카테고리에서 괄호 부분 제거 category_filter_clean = category_filter.split(" (")[0] if " (" in category_filter else category_filter filtered_categories = {} for cat, count in category_counter.items(): # 상품 카테고리에서도 괄호 있으면 제거 cat_clean = cat if " (" in cat_clean: cat_clean = cat_clean.split(" (")[0] if category_filter_clean.lower() in cat_clean.lower(): filtered_categories[cat] = count category_counter = Counter(filtered_categories) # 4단계: 키워드 검색량 조회 all_keywords = [main_keyword] + product_keywords search_volumes = keyword_search.fetch_all_search_volumes(all_keywords) # 5단계: 카테고리별 매칭 상태 분석 category_matching = [] # 정렬된 카테고리 목록 (출현 빈도순) sorted_categories = [cat for cat, _ in category_counter.most_common()] for category in sorted_categories: products_in_category = products_by_category[category] count = len(products_in_category) # 이 카테고리에 속한 상품들 중 10개만 가져옴 sample_products = products_in_category[:100] category_matching.append({ "카테고리": category, "상품수": count, "매칭상품": sample_products }) # 6단계: 검색량 정보 추가 및 결과 정리 keyword_info = [] for kw in all_keywords: volume = search_volumes.get(kw, {"PC검색량": 0, "모바일검색량": 0, "총검색량": 0}) keyword_info.append({ "키워드": kw, "PC검색량": volume.get("PC검색량", 0), "모바일검색량": volume.get("모바일검색량", 0), "총검색량": volume.get("총검색량", 0), "검색량구간": text_utils.get_search_volume_range(volume.get("총검색량", 0)) }) # 결과 반환 return { "status": "success", "message": "분석이 완료되었습니다.", "main_keyword": main_keyword, "product_name": product_name, "total_count": len(all_products), "products": all_products, "categories": sorted_categories, "category_counter": dict(category_counter), "category_matching": category_matching, "keyword_info": keyword_info } def analyze_keywords_by_category(keywords, selected_category, df_all=None): """ 입력된 키워드 목록과 선택된 카테고리로 분석을 수행하는 함수 """ import re if not keywords or not selected_category: return "키워드와 카테고리를 모두 선택해주세요." # 카테고리에서 카운트 정보 제거 (예: "패션의류 (10)" -> "패션의류") selected_category_clean = selected_category is_overall_view = False # '전체 보기'인지 여부 플래그 if " (" in selected_category and selected_category != "전체 보기": selected_category_clean = selected_category.split(" (")[0] elif selected_category == "전체 보기": selected_category_clean = "" # 전체 카테고리 분석용 is_overall_view = True # 키워드 리스트 처리 (최대 20개) if isinstance(keywords, str): # 쉼표나 엔터로 분리 keywords_list = [k.strip() for k in re.split(r'[,\n]+', keywords) if k.strip()] # 20개로 제한 keywords_list = keywords_list[:20] else: keywords_list = keywords[:20] if not keywords_list: return "분석할 키워드가 없습니다." logger.info(f"카테고리 분석 시작: {len(keywords_list)}개 키워드, 선택 카테고리: '{selected_category_clean if not is_overall_view else '전체 보기'}'") # 개선된 HTML 결과 - 너비 100% 적용 result_html = f'''
분석 결과 요약
분석 키워드 ({len(keywords_list)}개)
{''.join([f'{k}' for k in keywords_list])}
선택된 카테고리
{selected_category}
카테고리 일치 분석 결과
키워드
분석 키워드
매칭된 카테고리
''' # 키워드 배치 처리 준비 (5개씩 묶음) batch_size = 5 batches = [] for i in range(0, len(keywords_list), batch_size): batches.append(keywords_list[i:i + batch_size]) logger.info(f"총 {len(batches)}개 배치로 {len(keywords_list)}개 키워드 처리") # 각 배치 처리 match_results = {} # 각 키워드별 매칭 정보 저장 (match_count, total_count) batch_categories_info = {} # 각 키워드별로 추출된 카테고리 목록 및 비율 저장 for batch_idx, batch in enumerate(batches): logger.info(f"배치 {batch_idx+1}/{len(batches)} 처리 중...") # 각 키워드 처리 for keyword in batch: max_retries = 3 retry_count = 0 api_keyword = keyword.replace(" ", "") # 공백 제거 current_keyword_match_count = 0 current_keyword_total_products = 0 current_keyword_categories_found = [] # 비율과 함께 저장될 카테고리 문자열 리스트 while retry_count < max_retries: try: # 네이버 API 호출 products = product_search.fetch_naver_shopping_data_for_analysis(api_keyword, count=100) # 상위 10개 상품 if products: current_keyword_total_products = len(products) categories_counter_for_keyword = Counter() # 현재 키워드의 상품들 카테고리 분포 for product in products: product_category_full = product.get("category", "") or product.get("카테고리", "") if product_category_full: categories_counter_for_keyword[product_category_full] += 1 # 실제 매칭 여부 카운트 (is_overall_view가 아닐 때만) if not is_overall_view and selected_category_clean: product_category_for_match = product_category_full if " (" in product_category_for_match: # 상품 카테고리 이름에서 count 제거 product_category_for_match = product_category_for_match.split(" (")[0] sel_lower = selected_category_clean.lower() prod_lower = product_category_for_match.lower() if sel_lower in prod_lower or prod_lower in sel_lower: current_keyword_match_count += 1 if is_overall_view: # '전체 보기'일 경우, 모든 상품이 매칭된 것으로 간주 current_keyword_match_count = current_keyword_total_products # 카테고리별 비율 계산 및 저장 category_percentages = [] for cat, count in categories_counter_for_keyword.most_common(): # 빈도 높은 순으로 percentage = (count / current_keyword_total_products) * 100 if current_keyword_total_products > 0 else 0 category_percentages.append((cat, percentage)) # category_percentages.sort(key=lambda x: x[1], reverse=True) # 이미 most_common으로 정렬됨 for cat, percentage in category_percentages: current_keyword_categories_found.append(f"{cat} ({percentage:.0f}%)") logger.info(f" - '{keyword}' 처리 완료: {current_keyword_match_count}/{current_keyword_total_products} 일치") break # 성공했으므로 재시도 루프 종료 else: logger.warning(f" - '{keyword}' API 결과 없음 (시도 {retry_count+1}/{max_retries})") retry_count += 1 exponential_backoff_sleep(retry_count) except Exception as e: logger.error(f" - '{keyword}' 처리 중 오류: {e} (시도 {retry_count+1}/{max_retries})") retry_count += 1 exponential_backoff_sleep(retry_count) # 결과 저장 if retry_count >= max_retries and current_keyword_total_products == 0: # 최종 실패 match_results[keyword] = { "match_count": 0, "total_count": 0, "error": True } batch_categories_info[keyword] = ["오류 발생"] logger.error(f" - '{keyword}' 최대 재시도 후 실패") else: match_results[keyword] = { "match_count": current_keyword_match_count, "total_count": current_keyword_total_products, "error": False } batch_categories_info[keyword] = current_keyword_categories_found if current_keyword_categories_found else ["카테고리 정보 없음"] # API 레이트 리밋 방지 - 지수 백오프 사용 exponential_backoff_sleep(0) logger.info(f"전체 {len(keywords_list)}개 키워드 중 {len(match_results)}개 처리 완료") # 결과를 HTML로 변환 for keyword in keywords_list: result = match_results.get(keyword, {"match_count": 0, "total_count": 0, "error": True}) # 수정된 부분: keyword_status 결정 로직 변경 # 선택된 카테고리와 하나라도 매칭되면 "O", 아니면 "X" # "전체 보기" 선택 시에는 항상 "O" (모든 상품이 매칭된 것으로 간주했으므로) if result.get("error"): keyword_status = "오류" # 에러 발생 시 status_color = "red" elif is_overall_view: # '전체 보기'의 경우 keyword_status = "O" status_color = "#009879" # Green else: # 특정 카테고리 선택 시 keyword_status = "O" if result["match_count"] > 0 else "X" status_color = "#009879" if keyword_status == "O" else "red" # 매칭된 카테고리 정보 categories_html_list = batch_categories_info.get(keyword, ["정보 없음"]) categories_html = "
".join(categories_html_list) if result.get("error", False): result_html += f'''
{keyword}
{keyword_status}
분석 중 오류 발생
''' else: match_count_display = result["match_count"] total_count_display = result["total_count"] result_html += f'''
{keyword}{match_count_display}/{total_count_display}
{keyword_status}
{categories_html}
''' result_html += '
' # .analysis-result, .result-container 닫기 return result_html def analyze_product_terms(product_name, main_keyword=""): """ 상품명에서 추출한 키워드들을 분석하여 카테고리 항목 제공 (1년, 3년 트렌드 모두 분석) Args: product_name (str): 분석할 상품명 main_keyword (str): 메인 키워드 (optional) Returns: tuple: (HTML 형식의 결과 테이블, 키워드 분석 결과 리스트, 트렌드 분석 결과) """ global _last_keyword_results # 함수 시작 부분에 global 선언 # 전처리: 상품명 앞뒤 공백 제거 및 유효성 확인 product_name = product_name.strip() if product_name else "" if not product_name: return "상품명이 비어있습니다. 유효한 상품명을 입력해주세요.", [], None # 디버깅용 로그 logger.info(f"분석 시작 - 상품명: '{product_name}', 메인 키워드: '{main_keyword}'") # 상품명에서 키워드를 더 자연스럽게 분리 (공백과 쉼표 기준으로 분리) # 수정된 부분: 정규표현식 패턴 조정 및 예외처리 추가 try: words = [] # 먼저 쉼표로 분리 comma_parts = product_name.split(',') for part in comma_parts: # 각 부분을 공백으로 분리 space_parts = part.split() words.extend([word.strip() for word in space_parts if word.strip()]) # 중복 제거 및 1글자 이상 키워드만 유지 words = list(set([word for word in words if len(word) >= 1])) logger.info(f"상품명에서 추출한 원본 키워드 (총 {len(words)}개): {words}") # 키워드가 하나도 추출되지 않았다면 원본 상품명 사용 if not words: words = [product_name] logger.warning(f"키워드 추출 실패, 원본 상품명 사용: '{product_name}'") except Exception as e: logger.error(f"키워드 추출 중 오류 발생: {e}") # 오류 발생 시 원본 상품명을 그대로 사용 words = [product_name] # 메인 키워드 처리 if not main_keyword: # 메인 키워드가 없는 경우, 오징어 관련 키워드 찾기 (기존 로직) for word in words: if "오징어" in word: main_keyword = "오징어" break # 원본 키워드 목록 저장 keywords = [] for word in words: # 숫자, 영문 등을 포함한 모든 단어 허용 if word and word != main_keyword: # 메인 키워드가 있고, 단어에 메인 키워드가 없으면 조합 if main_keyword and main_keyword not in word: # 조합 키워드 생성 (자연스러운 형태로) combined = f"{word} {main_keyword}" if combined not in keywords: keywords.append(combined) # 원래 키워드도 따로 추가 (개선: 단일 키워드도 유지) if word not in keywords: keywords.append(word) # 메인 키워드도 단독으로 추가 if main_keyword and main_keyword not in keywords: keywords.append(main_keyword) if not keywords: return "상품명에서 키워드를 추출할 수 없습니다.", [], None logger.info(f"분석할 최종 키워드 목록 (총 {len(keywords)}개): {keywords}") # 추출된 키워드를 배치로 나누기 (배치당 5개씩) batch_size = 5 keyword_batches = [] for i in range(0, len(keywords), batch_size): keyword_batches.append(keywords[i:i + batch_size]) logger.info(f"총 {len(keyword_batches)}개 배치로 {len(keywords)}개 키워드 처리") # 키워드 분석 결과 저장 keyword_results = [] # 각 배치 처리 for batch_idx, batch in enumerate(keyword_batches): logger.info(f"배치 {batch_idx+1}/{len(keyword_batches)} 처리 중...") # 상품 검색 배치 처리 batch_products = {} for keyword in batch: # API 호출용 키워드 (공백 제거) api_keyword = keyword.replace(" ", "") # 최대 3번 재시도 max_retries = 3 retry_count = 0 while retry_count < max_retries: try: # 키워드로 상품 검색 products = product_search.fetch_naver_shopping_data_for_analysis(api_keyword, count=100) if products: batch_products[keyword] = products logger.info(f" - '{keyword}' 상품 검색 성공: {len(products)}개") break # 성공했으므로 루프 종료 else: logger.warning(f" - '{keyword}' 상품 없음 (시도 {retry_count+1}/{max_retries})") retry_count += 1 exponential_backoff_sleep(retry_count) except Exception as e: logger.error(f" - '{keyword}' 상품 검색 중 오류: {e} (시도 {retry_count+1}/{max_retries})") retry_count += 1 exponential_backoff_sleep(retry_count) # 최대 재시도 후에도 실패한 경우 로그 기록 if retry_count >= max_retries and keyword not in batch_products: logger.error(f" - '{keyword}' 최대 재시도 후 실패") # 검색량 조회 배치 처리 api_keywords = [kw.replace(" ", "") for kw in batch] volumes = keyword_search.fetch_all_search_volumes(api_keywords) # 각 키워드 처리 for keyword in batch: if keyword in batch_products: products = batch_products[keyword] # 개선: 카테고리 항목과 함께 카테고리별 점유율 계산 category_counter = Counter() for product in products: category = product.get("category", "") or product.get("카테고리", "") if category: category_counter[category] += 1 # 카테고리와 점유율 계산 total_products = len(products) categories_with_percentage = [] for category, count in category_counter.most_common(): percentage = (count / total_products) * 100 if total_products > 0 else 0 categories_with_percentage.append(f"{category}({percentage:.0f}%)") # 검색량 조회 (API 호출용 키워드 사용) api_keyword = keyword.replace(" ", "") volume_data = volumes.get(api_keyword, {"PC검색량": 0, "모바일검색량": 0, "총검색량": 0}) # 결과 저장 (카테고리 항목과 카운트 정보 포함) keyword_results.append({ "키워드": keyword, # UI 표시용 키워드 (공백 포함) "PC검색량": volume_data.get("PC검색량", 0), "모바일검색량": volume_data.get("모바일검색량", 0), "총검색량": volume_data.get("총검색량", 0), "검색량구간": text_utils.get_search_volume_range(volume_data.get("총검색량", 0)), "카테고리항목": "\n".join(categories_with_percentage) if categories_with_percentage else "-", "카테고리정보": dict(category_counter) # 원본 카테고리 카운터 저장 (요약용) }) logger.info(f" - '{keyword}' 분석 완료: 카테고리 항목 {len(category_counter)}개, 검색량 {volume_data.get('총검색량', 0)}") # 최종 결과 요약 출력 logger.info(f"키워드 분석 완료: 총 {len(keywords)}개 중 {len(keyword_results)}개 성공") # 결과를 검색량 기준으로 내림차순 정렬 (높은 것이 먼저 나오도록) keyword_results = sorted(keyword_results, key=lambda x: x["총검색량"], reverse=True) # 추천 카테고리 계산 recommended_categories = Counter() for result in keyword_results: for category, count in result.get("카테고리정보", {}).items(): recommended_categories[category] += count # 추천 카테고리 상위 3개 선택 top_categories = recommended_categories.most_common(3) # 총 상품 수 계산 total_products_count = sum(recommended_categories.values()) # 카테고리별 점유율 계산 top_categories_with_percentage = [] for category, count in top_categories: percentage = (count / total_products_count) * 100 if total_products_count > 0 else 0 top_categories_with_percentage.append({ "카테고리": category, "개수": count, "점유율": f"{percentage:.0f}%" }) # 1년, 3년 트렌드 분석 실행 trend_results = {"1year": None, "3year": None} trend_html = "" if keyword_results: try: # 트렌드 분석 모듈 import import trend_analysis # 상위 5개 키워드로 1년, 3년 트렌드 분석 top_keywords = [result["키워드"] for result in keyword_results[:5]] logger.info(f"1년, 3년 트렌드 분석 시작: {top_keywords}") # 1년 트렌드 분석 trend_result_1year = trend_analysis.get_trend_data(top_keywords, "1year") if trend_result_1year["status"] == "success": trend_results["1year"] = trend_result_1year # 3년 트렌드 분석 trend_result_3year = trend_analysis.get_trend_data(top_keywords, "3year") if trend_result_3year["status"] == "success": trend_results["3year"] = trend_result_3year # 트렌드 분석 HTML 생성 (1년, 3년 모두) if trend_results["1year"] or trend_results["3year"]: trend_html = f'''
🔍 검색량 트렌드 분석
''' for period, result in trend_results.items(): if result and result["status"] == "success": period_text = "최근 1년" if period == "1year" else "최근 3년" # 트렌드 인사이트 추출 insights = trend_analysis.analyze_trend_insights(result["trend_data"]) trend_html += f'''
📊 {period_text} 주요 인사이트
''' for keyword, insight in insights.items(): growth_icon = "📈" if insight['growth_rate'] > 0 else "📉" if insight['growth_rate'] < 0 else "📊" growth_color = "#28a745" if insight['growth_rate'] > 0 else "#dc3545" if insight['growth_rate'] < 0 else "#6c757d" trend_html += f'''
{keyword} {growth_icon}
🏆 최고점: {insight['max_volume']:,} ({insight['max_period']})
📊 전체 평균: {insight['total_avg']:,}
📈 성장률: {insight['growth_rate']:+.1f}%
''' trend_html += '''
''' trend_html += result["graph_html"] trend_html += '''
''' trend_html += '''
''' logger.info(f"1년, 3년 트렌드 분석 완료") else: logger.warning(f"트렌드 분석 실패") trend_html = '''
🔍 검색량 트렌드 분석
⚠️ 트렌드 분석을 실행할 수 없습니다. 나중에 다시 시도해주세요.
''' except Exception as e: logger.error(f"트렌드 분석 중 오류 발생: {e}") trend_html = '''
🔍 검색량 트렌드 분석
❌ 트렌드 분석 중 오류가 발생했습니다.
''' # 결과를 HTML 테이블로 변환 - 너비 100% 적용 html = f'''
상품명 키워드 분석 결과
분석 요약

{len(keyword_results)}개 키워드 분석

메인 키워드: {main_keyword if main_keyword else '없음'}

추천 카테고리
''' # 추천 카테고리 목록 추가 for idx, cat_info in enumerate(top_categories_with_percentage, 1): html += f'''
추천 카테고리 {idx} : {cat_info['카테고리']}({cat_info['점유율']})
''' html += '''
''' for idx, result in enumerate(keyword_results): # 카테고리 항목 준비 (줄바꿈을
로 변환) category_items = result.get("카테고리항목", "-").replace("\n", "
") html += f''' ''' html += '''
순번 키워드 PC검색량 모바일검색량 총검색량 검색량구간 카테고리항목
{idx + 1} {result["키워드"]} {result["PC검색량"]:,} {result["모바일검색량"]:,} {result["총검색량"]:,} {result["검색량구간"]} {category_items}
''' # 트렌드 분석 HTML 추가 html += trend_html # 분석 결과가 없는 경우 안내 메시지 if not keyword_results: html += '''

표시할 키워드가 없습니다. 다른 상품명을 입력해보세요.

''' # 분석 결과를 전역 변수에 저장 (다운로드용) _last_keyword_results = keyword_results # HTML과 함께 키워드 분석 결과 및 트렌드 결과도 함께 반환 return html, keyword_results, trend_results def collect_categories_per_keyword(keywords, max_products=10): """ 키워드마다 상품 n개를 호출하여 카테고리 집합 반환 Args: keywords (list): 키워드 목록 max_products (int): 키워드당 검색할 최대 상품 수 Returns: dict: 키워드별 카테고리 집합을 담은 사전 {키워드: {카테고리1, 카테고리2, ...}} """ logger.info(f"키워드별 카테고리 수집 시작: {len(keywords)}개 키워드") keyword_category_map = {} # 배치 처리를 위한 준비 batch_size = 5 batches = [] for i in range(0, len(keywords), batch_size): batches.append(keywords[i:i + batch_size]) logger.info(f"총 {len(batches)}개 배치로 {len(keywords)}개 키워드 처리") # 각 배치 처리 for batch_idx, batch in enumerate(batches): logger.info(f"배치 {batch_idx+1}/{len(batches)} 처리 중...") for keyword in batch: # API 호출용 키워드 (공백 제거) api_keyword = keyword.replace(" ", "") max_retries = 3 retry_count = 0 while retry_count < max_retries: try: # 키워드로 상품 검색 products = product_search.fetch_naver_shopping_data_for_analysis(api_keyword, count=max_products) if products: # 카테고리 추출 categories = set() for product in products: # 두 가지 키 모두 시도 (category와 카테고리) category = product.get("category", "") or product.get("카테고리", "") if category: categories.add(category) keyword_category_map[keyword] = categories logger.info(f" - '{keyword}' 카테고리 수집 완료: {len(categories)}개") break # 성공했으므로 루프 종료 else: logger.warning(f" - '{keyword}' 상품 검색 결과 없음 (시도 {retry_count+1}/{max_retries})") retry_count += 1 exponential_backoff_sleep(retry_count) except Exception as e: logger.error(f" - '{keyword}' 처리 중 오류: {e} (시도 {retry_count+1}/{max_retries})") retry_count += 1 exponential_backoff_sleep(retry_count) if retry_count >= max_retries: logger.error(f" - '{keyword}' 최대 재시도 후 실패") keyword_category_map[keyword] = set() # API 레이트 리밋 방지 (안정적인 지연으로 변경) exponential_backoff_sleep(0) # 초기 지연 적용 logger.info(f"키워드별 카테고리 수집 완료: {len(keyword_category_map)}개") return keyword_category_map