""" 키워드 매칭 및 상승폭 계산 개선 - 전체 코드 v4.0 - 스페이스바 처리 개선 + 올바른 트렌드 분석 로직 적용 - 기존 모든 기능 유지하면서 최적화 - 스페이스바 제거 후 검색/비교 로직 적용 - 올바른 증감율 계산: 올해 완료월 vs 작년 동월 - 🔖 가장 검색량이 많은 월: 실제+예상 데이터 중 최대값 - 🔖 가장 상승폭이 높은 월: 연속된 월간 상승률 중 최대값 """ import logging import pandas as pd from datetime import datetime import re import time import random from typing import Dict, List, Optional logger = logging.getLogger(__name__) def normalize_keyword(keyword): """키워드 정규화 - 스페이스바 처리 개선""" if not keyword: return "" # 1. 앞뒤 공백 제거 keyword = keyword.strip() # 2. 연속된 공백을 하나로 변경 keyword = re.sub(r'\s+', ' ', keyword) # 3. 특수문자 제거 (한글, 영문, 숫자, 공백만 남김) keyword = re.sub(r'[^\w\s가-힣]', '', keyword) return keyword def normalize_keyword_for_api(keyword): """API 호출용 키워드 정규화 (스페이스 제거)""" normalized = normalize_keyword(keyword) return normalized.replace(" ", "") def normalize_keyword_for_comparison(keyword): """비교용 키워드 정규화 (스페이스 유지)""" return normalize_keyword(keyword).lower() def normalize_keyword_advanced(keyword): """고급 키워드 정규화 - 매칭 문제 해결""" if not keyword: return "" # 1. 기본 정리 keyword = str(keyword).strip() # 2. 연속된 공백을 하나로 변경 keyword = re.sub(r'\s+', ' ', keyword) # 3. 특수문자 제거 (한글, 영문, 숫자, 공백만 남김) keyword = re.sub(r'[^\w\s가-힣]', '', keyword) # 4. 소문자 변환 keyword = keyword.lower() return keyword def create_keyword_variations(keyword): """키워드 변형 버전들 생성 - 스페이스바 처리 강화""" base = normalize_keyword_advanced(keyword) variations = [base] # 스페이스 제거 버전 no_space = base.replace(" ", "") if no_space != base: variations.append(no_space) # 스페이스를 다른 구분자로 바꾼 버전들 variations.append(base.replace(" ", "-")) variations.append(base.replace(" ", "_")) # 단어 순서 바꾼 버전 (2단어인 경우) words = base.split() if len(words) == 2: reversed_keyword = f"{words[1]} {words[0]}" variations.append(reversed_keyword) variations.append(reversed_keyword.replace(" ", "")) return list(set(variations)) # 중복 제거 def find_matching_keyword_row(analysis_keyword, keywords_df): """개선된 키워드 매칭 함수""" if keywords_df is None or keywords_df.empty: return None analysis_variations = create_keyword_variations(analysis_keyword) logger.info(f"분석 키워드 변형들: {analysis_variations}") # 1차: 정확한 매칭 for idx, row in keywords_df.iterrows(): df_keyword = str(row.get('조합 키워드', '')) df_variations = create_keyword_variations(df_keyword) for analysis_var in analysis_variations: for df_var in df_variations: if analysis_var == df_var and len(analysis_var) > 1: logger.info(f"정확한 매칭 성공: '{analysis_keyword}' = '{df_keyword}'") return row # 2차: 포함 관계 매칭 for idx, row in keywords_df.iterrows(): df_keyword = str(row.get('조합 키워드', '')) df_variations = create_keyword_variations(df_keyword) for analysis_var in analysis_variations: for df_var in df_variations: if len(analysis_var) > 2 and len(df_var) > 2: if analysis_var in df_var or df_var in analysis_var: similarity = len(set(analysis_var) & set(df_var)) / len(set(analysis_var) | set(df_var)) if similarity > 0.7: # 70% 이상 유사 logger.info(f"부분 매칭 성공: '{analysis_keyword}' ≈ '{df_keyword}' (유사도: {similarity:.2f})") return row logger.warning(f"키워드 매칭 실패: '{analysis_keyword}'") logger.info(f"데이터프레임 키워드 샘플: {keywords_df['조합 키워드'].head(5).tolist()}") return None def generate_prediction_data(trend_data_3year, keyword): """ 정교한 예상 데이터 생성 함수 - 트렌드 데이터 수집 후 바로 호출하여 예상 데이터 추가 - 계절성, 증감 트렌드, 전년 대비 성장률 모두 고려 """ if not trend_data_3year: logger.warning("❌ 예상 데이터 생성 실패: trend_data_3year 없음") return trend_data_3year try: current_date = datetime.now() current_year = current_date.year current_month = current_date.month logger.info(f"🔮 예상 데이터 생성 시작: {keyword} ({current_year}년 {current_month}월 기준)") for kw, data in trend_data_3year.items(): if not data or not data.get('monthly_volumes') or not data.get('dates'): continue volumes = data['monthly_volumes'] dates = data['dates'] # ✅ 1단계: 기존 데이터 분석 yearly_data = {} # {year: {month: volume}} for i, date_str in enumerate(dates): try: date_obj = datetime.strptime(date_str, "%Y-%m-%d") if i < len(volumes): volume = volumes[i] if isinstance(volume, str): volume = float(volume.replace(',', '')) volume = int(volume) if volume else 0 year = date_obj.year month = date_obj.month if year not in yearly_data: yearly_data[year] = {} yearly_data[year][month] = volume except Exception as e: logger.warning(f"⚠️ 날짜 파싱 오류: {date_str}") continue logger.info(f"📊 분석된 연도: {list(yearly_data.keys())}") # ✅ 2단계: 예상 데이터 생성 알고리즘 if current_year not in yearly_data: yearly_data[current_year] = {} current_year_data = yearly_data[current_year] last_year_data = yearly_data.get(current_year - 1, {}) two_years_ago_data = yearly_data.get(current_year - 2, {}) logger.info(f"📈 올해 실제 데이터: {len(current_year_data)}개월") logger.info(f"📈 작년 참조 데이터: {len(last_year_data)}개월") # 예상 데이터 생성 (현재월 이후) for future_month in range(current_month + 1, 13): if future_month in current_year_data: continue # 이미 데이터가 있으면 스킵 predicted_volume = calculate_predicted_volume( future_month, current_year_data, last_year_data, two_years_ago_data, current_month ) if predicted_volume is not None: current_year_data[future_month] = predicted_volume logger.info(f"🔮 예상 생성: {current_year}년 {future_month}월 = {predicted_volume:,}회") # ✅ 3단계: 생성된 데이터를 원본 구조에 통합 updated_volumes = [] updated_dates = [] # 시간순으로 정렬하여 통합 all_months = [] for year in sorted(yearly_data.keys()): for month in sorted(yearly_data[year].keys()): all_months.append((year, month, yearly_data[year][month])) for year, month, volume in all_months: updated_volumes.append(volume) updated_dates.append(f"{year}-{month:02d}-01") # 원본 데이터 업데이트 data['monthly_volumes'] = updated_volumes data['dates'] = updated_dates logger.info(f"✅ {kw} 예상 데이터 통합 완료: 총 {len(updated_volumes)}개월") logger.info(f"🎉 전체 예상 데이터 생성 완료: {keyword}") return trend_data_3year except Exception as e: logger.error(f"❌ 예상 데이터 생성 오류: {e}") return trend_data_3year def calculate_predicted_volume(target_month, current_year_data, last_year_data, two_years_ago_data, current_month): """ 정교한 예상 볼륨 계산 - 다중 요인 고려: 작년 동월, 증감 트렌드, 계절성, 성장률 """ try: # 기준 값들 last_year_same_month = last_year_data.get(target_month, 0) two_years_ago_same_month = two_years_ago_data.get(target_month, 0) if last_year_same_month == 0: logger.warning(f"⚠️ {target_month}월 작년 데이터 없음") return None # ✅ 1. 기본값: 작년 동월 base_volume = last_year_same_month # ✅ 2. 전년 대비 성장률 계산 (가능한 경우) growth_rate = 1.0 if two_years_ago_same_month > 0: growth_rate = last_year_same_month / two_years_ago_same_month logger.info(f"📈 {target_month}월 전년 성장률: {growth_rate:.2f}배") # ✅ 3. 올해 최근 트렌드 반영 trend_factor = 1.0 if len(current_year_data) >= 2: # 최근 2-3개월의 작년 대비 비율 계산 recent_ratios = [] for month in range(max(1, current_month - 2), current_month + 1): if month in current_year_data and month in last_year_data: if last_year_data[month] > 0: ratio = current_year_data[month] / last_year_data[month] recent_ratios.append(ratio) if recent_ratios: trend_factor = sum(recent_ratios) / len(recent_ratios) logger.info(f"📊 최근 트렌드 팩터: {trend_factor:.2f}") # ✅ 4. 계절성 보정 (같은 분기 내 월간 패턴) seasonal_factor = 1.0 if target_month > 1 and target_month - 1 in last_year_data and target_month in last_year_data: # 작년 동일 구간의 월간 변화율 if last_year_data[target_month - 1] > 0: seasonal_factor = last_year_data[target_month] / last_year_data[target_month - 1] logger.info(f"🌊 {target_month}월 계절성 팩터: {seasonal_factor:.2f}") # ✅ 5. 최종 예상값 계산 (가중평균) predicted_volume = int( base_volume * ( 0.4 * growth_rate + # 40% 전년 성장률 0.4 * trend_factor + # 40% 최근 트렌드 0.2 * seasonal_factor # 20% 계절성 ) ) # ✅ 6. 합리성 검증 (급격한 변화 방지) if current_year_data: recent_avg = sum(current_year_data.values()) / len(current_year_data) if predicted_volume > recent_avg * 5: # 5배 이상 급증 방지 predicted_volume = int(recent_avg * 2) logger.warning(f"⚠️ {target_month}월 급증 보정: {predicted_volume:,}회") elif predicted_volume < recent_avg * 0.1: # 10분의 1 이하 급감 방지 predicted_volume = int(recent_avg * 0.5) logger.warning(f"⚠️ {target_month}월 급감 보정: {predicted_volume:,}회") logger.info(f"🎯 {target_month}월 예상 계산: {last_year_same_month:,} × (성장{growth_rate:.2f} + 트렌드{trend_factor:.2f} + 계절{seasonal_factor:.2f}) = {predicted_volume:,}") return predicted_volume except Exception as e: logger.error(f"❌ {target_month}월 예상 계산 오류: {e}") return None def enhance_trend_data_with_predictions(trend_data_3year, keyword): """ 기존 트렌드 데이터에 예상 데이터 추가 - 메인 트렌드 수집 함수에서 호출 """ if not trend_data_3year: return trend_data_3year logger.info(f"🚀 트렌드 데이터 예상 확장 시작: {keyword}") enhanced_data = generate_prediction_data(trend_data_3year, keyword) # 데이터 품질 검증 for kw, data in enhanced_data.items(): if data and data.get('monthly_volumes'): total_months = len(data['monthly_volumes']) current_year = datetime.now().year # 현재 연도 데이터 개수 확인 current_year_count = 0 for date_str in data['dates']: try: if date_str.startswith(str(current_year)): current_year_count += 1 except: continue logger.info(f"✅ {kw} 최종 데이터: 전체 {total_months}개월, 올해 {current_year_count}개월") return enhanced_data def calculate_max_growth_rate_with_predictions(trend_data_3year, keyword): """올바른 트렌드 분석 로직 - 사용자 요구사항 적용""" if not trend_data_3year: logger.error("❌ trend_data_3year가 없습니다") return "데이터 없음" try: keyword_data = None for kw, data in trend_data_3year.items(): keyword_data = data logger.info(f"🔍 키워드 데이터 발견: {kw}") break if not keyword_data or not keyword_data.get('monthly_volumes') or not keyword_data.get('dates'): logger.error("❌ keyword_data 구조 문제") return "데이터 없음" volumes = keyword_data['monthly_volumes'] dates = keyword_data['dates'] # 1단계: 현재 시점 파악 current_date = datetime.now() current_year = current_date.year current_month = current_date.month current_day = current_date.day # 완료된 마지막 월 계산 (2일 이후면 전월까지 완료) if current_day >= 2: completed_year = current_year completed_month = current_month - 1 else: completed_year = current_year completed_month = current_month - 2 # 월이 0 이하가 되면 연도 조정 while completed_month <= 0: completed_month += 12 completed_year -= 1 logger.info(f"📅 현재: {current_year}년 {current_month}월 {current_day}일") logger.info(f"📊 완료된 마지막 데이터: {completed_year}년 {completed_month}월") # 2단계: 데이터 분류 및 수집 all_data = [] for i, date_str in enumerate(dates): try: date_obj = datetime.strptime(date_str, "%Y-%m-%d") if i < len(volumes): volume = volumes[i] if isinstance(volume, str): volume = float(volume.replace(',', '')) volume = int(volume) if volume else 0 all_data.append({ 'year': date_obj.year, 'month': date_obj.month, 'volume': volume, 'date_str': date_str, 'date_obj': date_obj, 'sort_key': f"{date_obj.year:04d}{date_obj.month:02d}" }) except Exception as e: logger.warning(f"⚠️ 날짜 파싱 오류: {date_str} - {e}") continue # 시간 순서대로 정렬 all_data = sorted(all_data, key=lambda x: x['sort_key']) # 3단계: 증감율 계산 (올해 완료월 vs 작년 동월) this_year_completed_volume = None last_year_same_month_volume = None for data in all_data: # 올해 완료된 마지막 월 찾기 if data['year'] == completed_year and data['month'] == completed_month: this_year_completed_volume = data['volume'] logger.info(f"📊 올해 {completed_month}월 실데이터: {this_year_completed_volume:,}회") # 작년 동월 찾기 if data['year'] == completed_year - 1 and data['month'] == completed_month: last_year_same_month_volume = data['volume'] logger.info(f"📊 작년 {completed_month}월 실데이터: {last_year_same_month_volume:,}회") # 증감율 계산 growth_rate = 0 if this_year_completed_volume is not None and last_year_same_month_volume is not None and last_year_same_month_volume > 0: growth_rate = (this_year_completed_volume - last_year_same_month_volume) / last_year_same_month_volume logger.info(f"📈 계산된 증감율: {growth_rate:+.3f} ({growth_rate * 100:+.1f}%)") else: logger.warning("⚠️ 증감율 계산을 위한 데이터가 부족합니다.") # 4단계: 예상데이터 생성 (현재월 이후) combined_data = [] month_names = ["", "1월", "2월", "3월", "4월", "5월", "6월", "7월", "8월", "9월", "10월", "11월", "12월"] # 작년 12월 데이터 추가 (연속성을 위해) for data in all_data: if data['year'] == completed_year - 1 and data['month'] == 12: combined_data.append({ 'year': data['year'], 'month': data['month'], 'volume': data['volume'], 'data_type': '작년실제', 'sort_key': f"{data['year']:04d}{data['month']:02d}" }) logger.info(f"🔗 작년 12월 실데이터: {data['volume']:,}회") break # 올해 1월부터 완료월까지 실제데이터 추가 for month in range(1, completed_month + 1): for data in all_data: if data['year'] == completed_year and data['month'] == month: combined_data.append({ 'year': data['year'], 'month': data['month'], 'volume': data['volume'], 'data_type': '실제', 'sort_key': f"{data['year']:04d}{data['month']:02d}" }) logger.info(f"📊 {month}월 실데이터: {data['volume']:,}회") break # 올해 미완료월(현재월+1 ~ 12월) 예상데이터 생성 for month in range(completed_month + 1, 13): # 작년 동월 데이터 찾기 last_year_volume = None for data in all_data: if data['year'] == completed_year - 1 and data['month'] == month: last_year_volume = data['volume'] break if last_year_volume is not None: # 예상 검색량 = 작년 동월 × (1 + 증감율) predicted_volume = int(last_year_volume * (1 + growth_rate)) predicted_volume = max(predicted_volume, 0) # 음수 방지 combined_data.append({ 'year': completed_year, 'month': month, 'volume': predicted_volume, 'data_type': '예상', 'sort_key': f"{completed_year:04d}{month:02d}" }) logger.info(f"🔮 {month}월 예상데이터: {predicted_volume:,}회 (작년 {last_year_volume:,}회 × {1 + growth_rate:.3f})") # 시간 순서대로 정렬 combined_data = sorted(combined_data, key=lambda x: x['sort_key']) # 5단계: 🔖 가장 상승폭이 높은 월 찾기 (연속된 월간 상승률) max_growth_rate = 0 max_growth_info = "데이터 없음" for i in range(len(combined_data) - 1): start_data = combined_data[i] end_data = combined_data[i + 1] if start_data['volume'] > 0: month_growth_rate = ((end_data['volume'] - start_data['volume']) / start_data['volume']) * 100 # 상승한 경우만 고려 if month_growth_rate > max_growth_rate: max_growth_rate = month_growth_rate start_month_name = month_names[start_data['month']] end_month_name = month_names[end_data['month']] # 연도 전환 고려 if start_data['year'] != end_data['year']: period_desc = f"{start_data['year']}년 {start_month_name}({start_data['volume']:,}회)에서 {end_data['year']}년 {end_month_name}({end_data['volume']:,}회)으로" else: period_desc = f"{start_month_name}({start_data['volume']:,}회)에서 {end_month_name}({end_data['volume']:,}회)으로" # 데이터 유형 판단 if start_data['data_type'] in ['예상'] and end_data['data_type'] in ['예상']: data_type = "예상 기반" elif start_data['data_type'] in ['실제', '작년실제'] and end_data['data_type'] in ['실제', '작년실제']: data_type = "실제 기반" else: data_type = "실제→예상 기반" max_growth_info = f"{period_desc} {max_growth_rate:.1f}% 상승 ({data_type})" # 상승 구간이 없는 경우 최소 하락률 표시 if max_growth_rate == 0: min_decline_rate = float('inf') for i in range(len(combined_data) - 1): start_data = combined_data[i] end_data = combined_data[i + 1] if start_data['volume'] > 0: month_growth_rate = ((end_data['volume'] - start_data['volume']) / start_data['volume']) * 100 if abs(month_growth_rate) < abs(min_decline_rate): min_decline_rate = month_growth_rate start_month_name = month_names[start_data['month']] end_month_name = month_names[end_data['month']] if start_data['year'] != end_data['year']: period_desc = f"{start_data['year']}년 {start_month_name}({start_data['volume']:,}회)에서 {end_data['year']}년 {end_month_name}({end_data['volume']:,}회)으로" else: period_desc = f"{start_month_name}({start_data['volume']:,}회)에서 {end_month_name}({end_data['volume']:,}회)으로" if start_data['data_type'] in ['예상'] and end_data['data_type'] in ['예상']: data_type = "예상 기반" elif start_data['data_type'] in ['실제', '작년실제'] and end_data['data_type'] in ['실제', '작년실제']: data_type = "실제 기반" else: data_type = "실제→예상 기반" max_growth_info = f"{period_desc} {abs(min_decline_rate):.1f}% 감소 ({data_type})" logger.info(f"🏆 가장 상승폭이 높은 월: {max_growth_info}") return max_growth_info except Exception as e: logger.error(f"❌ 상승폭 계산 오류: {e}") import traceback logger.error(f"❌ 스택 트레이스: {traceback.format_exc()}") return "계산 오류" def get_peak_month_with_predictions(trend_data_3year, keyword): """🔖 가장 검색량이 많은 월 찾기 - 실제+예상 데이터 활용""" if not trend_data_3year: return "연중" try: keyword_data = None for kw, data in trend_data_3year.items(): keyword_data = data break if not keyword_data or not keyword_data.get('monthly_volumes') or not keyword_data.get('dates'): return "연중" volumes = keyword_data['monthly_volumes'] dates = keyword_data['dates'] # 현재 시점 파악 current_date = datetime.now() current_year = current_date.year current_month = current_date.month current_day = current_date.day # 완료된 마지막 월 계산 if current_day >= 2: completed_year = current_year completed_month = current_month - 1 else: completed_year = current_year completed_month = current_month - 2 while completed_month <= 0: completed_month += 12 completed_year -= 1 # 데이터 수집 all_data = [] for i, date_str in enumerate(dates): try: date_obj = datetime.strptime(date_str, "%Y-%m-%d") if i < len(volumes): volume = volumes[i] if isinstance(volume, str): volume = float(volume.replace(',', '')) volume = int(volume) if volume else 0 all_data.append({ 'year': date_obj.year, 'month': date_obj.month, 'volume': volume }) except: continue # 증감율 계산 this_year_completed_volume = None last_year_same_month_volume = None for data in all_data: if data['year'] == completed_year and data['month'] == completed_month: this_year_completed_volume = data['volume'] if data['year'] == completed_year - 1 and data['month'] == completed_month: last_year_same_month_volume = data['volume'] growth_rate = 0 if this_year_completed_volume is not None and last_year_same_month_volume is not None and last_year_same_month_volume > 0: growth_rate = (this_year_completed_volume - last_year_same_month_volume) / last_year_same_month_volume # 올해 데이터 준비 (실제 + 예상) year_data = [] month_names = ["", "1월", "2월", "3월", "4월", "5월", "6월", "7월", "8월", "9월", "10월", "11월", "12월"] # 실제 데이터 추가 (1월~완료월) for month in range(1, completed_month + 1): for data in all_data: if data['year'] == completed_year and data['month'] == month: year_data.append({ 'month': month, 'volume': data['volume'], 'data_type': '실제' }) break # 예상 데이터 추가 (완료월+1~12월) for month in range(completed_month + 1, 13): last_year_volume = None for data in all_data: if data['year'] == completed_year - 1 and data['month'] == month: last_year_volume = data['volume'] break if last_year_volume is not None: predicted_volume = int(last_year_volume * (1 + growth_rate)) predicted_volume = max(predicted_volume, 0) year_data.append({ 'month': month, 'volume': predicted_volume, 'data_type': '예상' }) # 가장 높은 검색량 찾기 if not year_data: return "연중" max_data = max(year_data, key=lambda x: x['volume']) month_name = month_names[max_data['month']] data_type_suffix = " - 예상" if max_data['data_type'] == '예상' else "" return f"{month_name}({max_data['volume']:,}회){data_type_suffix}" except Exception as e: logger.error(f"피크월 분석 오류: {e}") return "연중" def calculate_3year_growth_rate_improved(volumes): """작년대비 증감율 계산 (3년 데이터용)""" if len(volumes) < 24: return 0 try: # 첫 해와 마지막 해 비교 first_year = volumes[:12] last_year = volumes[-12:] first_year_avg = sum(first_year) / len(first_year) last_year_avg = sum(last_year) / len(last_year) if first_year_avg == 0: return 0 growth_rate = ((last_year_avg - first_year_avg) / first_year_avg) * 100 return min(max(growth_rate, -50), 200) # -50% ~ 200% 범위로 제한 except Exception as e: logger.error(f"작년대비 증감율 계산 오류: {e}") return 0 def calculate_max_growth_rate_pure_logic(trend_data_3year, keyword): """순수 로직으로 최대 상승폭 계산 - 예상 데이터 포함 버전""" return calculate_max_growth_rate_with_predictions(trend_data_3year, keyword) def analyze_season_cycle_with_llm(trend_data_3year, keyword, total_volume, gemini_model): """LLM을 이용한 시즌 상품 소싱 사이클 분석""" if not trend_data_3year or not gemini_model: return "비시즌상품", "언제든지 진입 가능", "데이터 부족" try: keyword_data = None for kw, data in trend_data_3year.items(): keyword_data = data break if not keyword_data or not keyword_data.get('monthly_volumes'): return "비시즌상품", "언제든지 진입 가능", "데이터 부족" volumes = keyword_data['monthly_volumes'] dates = keyword_data['dates'] recent_12_volumes = volumes[-12:] if len(volumes) >= 12 else volumes recent_12_dates = dates[-12:] if len(dates) >= 12 else dates if len(recent_12_volumes) < 12: return "비시즌상품", "언제든지 진입 가능", "데이터 부족" monthly_data_str = "" max_volume = 0 max_month = "" for i, (date, volume) in enumerate(zip(recent_12_dates, recent_12_volumes)): try: date_obj = datetime.strptime(date, "%Y-%m-%d") month_name = f"{date_obj.year}년 {date_obj.month}월" monthly_data_str += f"{month_name}: {volume:,}회\n" # 최대 검색량 월 찾기 if volume > max_volume: max_volume = volume max_month = f"{date_obj.month}월({volume:,}회)" except: monthly_data_str += f"월-{i+1}: {volume:,}회\n" current_date = datetime.now() current_month = current_date.month prompt = f""" 키워드: '{keyword}' 현재 검색량: {total_volume:,}회 현재 시점: {current_date.year}년 {current_month}월 월별 검색량 데이터 (최근 12개월): {monthly_data_str} 다음 형식으로만 답변하세요: 상품유형: [봄시즌상품/여름시즌상품/가을시즌상품/겨울시즌상품/비시즌상품/크리스마스이벤트상품/밸런타인이벤트상품/어버이날이벤트상품/새학기이벤트상품/기타이벤트상품] 피크월: [X월] (검색량이 가장 높은 월, 실제 수치 포함) 성장월: [X월] (증가폭이 가장 높은 월) 현재상태: {current_month}월 기준 [도입기/성장기/안정기/쇠퇴기/비시즌기간] 진입추천: [구체적 월 제시] """ response = gemini_model.generate_content(prompt) result_text = response.text.strip() lines = result_text.split('\n') product_type = "비시즌상품" peak_month = max_month if max_month else "연중" growth_month = "연중" current_status = "안정기" entry_recommendation = "언제든지 진입 가능" for line in lines: line = line.strip() if line.startswith('상품유형:'): product_type = line.replace('상품유형:', '').strip() elif line.startswith('피크월:'): extracted_peak = line.replace('피크월:', '').strip() if '(' in extracted_peak and ')' in extracted_peak: peak_month = extracted_peak else: peak_month = max_month if max_month else extracted_peak elif line.startswith('성장월:'): growth_month = line.replace('성장월:', '').strip() elif line.startswith('현재상태:'): current_status = line.replace('현재상태:', '').strip() elif line.startswith('진입추천:'): entry_recommendation = line.replace('진입추천:', '').strip() detail_info = f"상품유형: {product_type} | 피크월: {peak_month} | 성장월: {growth_month} | 현재상태: {current_status}" logger.info(f"LLM 시즌 분석 완료: {product_type}, {entry_recommendation}") return product_type, entry_recommendation, detail_info except Exception as e: logger.error(f"LLM 시즌 사이클 분석 오류: {e}") return "비시즌상품", "언제든지 진입 가능", "LLM 분석 오류" def analyze_sourcing_strategy_improved(keyword, volume_data, trend_data_1year, trend_data_3year, filtered_keywords_df, gemini_model): """개선된 소싱전략 분석 - 포맷팅 수정 및 관여도 분석 강화""" total_volume = volume_data.get('총검색량', 0) current_date = datetime.now() current_month = current_date.month current_year = current_date.year # ✅ 수정: 올바른 로직으로 상승폭 계산 growth_analysis = calculate_max_growth_rate_with_predictions(trend_data_3year, keyword) # ✅ 수정: 올바른 로직으로 피크월 계산 (실제+예상 데이터 활용) peak_month_with_volume = get_peak_month_with_predictions(trend_data_3year, keyword) # LLM으로 시즌 분석 (기존 유지) if gemini_model: product_type, entry_timing, season_detail = analyze_season_cycle_with_llm(trend_data_3year, keyword, total_volume, gemini_model) else: # 기본값 product_type = "연중상품" if total_volume > 50000: product_type = "인기상품" elif total_volume > 10000: product_type = "중간상품" elif total_volume > 0: product_type = "틈새상품" # 2. 관여도 분석 추가 - 초보자가 판매가능한 소싱 기준 (개선된 기준 적용) involvement_level = analyze_involvement_level(keyword, total_volume, gemini_model) # 트렌드 경고 메시지 trend_warning = "" if not trend_data_3year: trend_warning = "\n\n💡 더 정확한 트렌드 데이터를 위해 \"1단계: 기본 키워드 입력\"을 실행해보세요." # 결과 포맷팅 수정 - 구분선과 항목 분리 result_content = f"""**🔖 상품유형** {product_type} {involvement_level} **🔖 가장 검색량이 많은 월** {peak_month_with_volume} **🔖 가장 상승폭이 높은 월** {growth_analysis}{trend_warning}""" try: return {"status": "success", "content": result_content} except Exception as e: logger.error(f"소싱전략 분석 오류: {e}") return {"status": "error", "content": "소싱전략 분석을 완료할 수 없습니다."} def analyze_involvement_level(keyword, total_volume, gemini_model): """관여도 분석 함수 - 초보자가 판매가능한 소싱 기준""" try: # 기본 규칙 기반 분석 basic_involvement = get_basic_involvement_level(keyword, total_volume) # Gemini가 있으면 LLM 분석도 수행 if gemini_model: llm_involvement = get_llm_involvement_analysis(keyword, total_volume, gemini_model) return llm_involvement else: return basic_involvement except Exception as e: logger.error(f"관여도 분석 오류: {e}") return "복합관여도상품(상품에 따라 달라짐)" def get_basic_involvement_level(keyword, total_volume): """기본 규칙 기반 관여도 분석 - 초보자 판매 관점""" # 저관여 상품 키워드 패턴 (초보자 진입 가능한 불편해소 제품) low_involvement_keywords = [ # 불편해소/정리수납 "거치대", "받침대", "정리함", "정리대", "수납", "홀더", "스탠드", "쿠션", "베개", "목베개", "방석", "매트", "패드", # 케이블/전선 관리 "케이블", "선정리", "코드", "충전기", "어댑터", # 청소/위생 (대기업 제품 제외) "청소솔", "청소기", "걸레", "타올", "브러시", # 자동차/실용용품 "차량용", "자동차", "핸드폰", "스마트폰", "태블릿", # 간단한 도구/액세서리 "집게", "후크", "자석", "클립", "고리", "링", "홀더", # 미끄럼방지/안전 "미끄럼", "논슬립", "방지", "보호", "커버", "케이스" ] # 고관여 상품 키워드 패턴 (대기업 독점 또는 고가/전문 제품) high_involvement_keywords = [ # 대기업 독점 생필품 "휴지", "화장지", "물티슈", "마스크", "세제", "샴푸", "린스", "비누", "치약", "칫솔", "기저귀", "생리대", "콘돔", # 식품/음료 (브랜드 민감) "라면", "과자", "음료", "커피", "차", "우유", "요구르트", "쌀", "김", "참기름", "간장", "고추장", "된장", # 고가 전자제품 "노트북", "컴퓨터", "스마트폰", "태블릿", "카메라", "TV", "모니터", "냉장고", "세탁기", "에어컨", "청소기", "전자레인지", # 의료/건강 (인증 필요) "의료", "건강식품", "영양제", "비타민", "약", "의약품", # 명품/브랜드 "명품", "브랜드", "럭셔리", "시계", "보석", "금", "은", "다이아몬드" ] keyword_lower = keyword.lower() # 저관여 상품 체크 (불편해소 키워드 우선) for low_kw in low_involvement_keywords: if low_kw in keyword_lower: return "저관여상품(초보자용)" # 고관여 상품 체크 (대기업 독점/브랜드 민감 키워드) for high_kw in high_involvement_keywords: if high_kw in keyword_lower: return "고관여상품(고급자용)" # 검색량 기반 추가 판단 if total_volume > 100000: # 검색량이 매우 높으면 대기업이 관심 가질 만한 시장 return "고관여상품(고급자용)" elif total_volume > 50000: return "복합관여도상품(상품에 따라 달라짐)" elif total_volume > 5000: return "복합관여도상품(상품에 따라 달라짐)" else: # 검색량이 낮으면 틈새 시장, 초보자도 진입 가능 return "저관여상품(초보자용)" def get_llm_involvement_analysis(keyword, total_volume, gemini_model): """LLM을 이용한 정교한 관여도 분석 - 초보자 판매 관점 기준 적용""" try: prompt = f""" '{keyword}' 상품의 관여도를 초보자 판매 관점에서 분석해주세요. 검색량: {total_volume:,}회 관여도 정의 (초보자가 판매가능한 소싱 기준): 저관여상품(초보자용): - 대기업 독점이 없는 영역 - 즉시 불편해소하는 제품 (지금 바로 필요한 문제 해결) - 브랜드 상관없이 기능만 되면 구매하는 제품 - 1만원~3만원대 가격, 소량(100개 이하) 시작 가능 - 예시: 목베개, 스마트폰거치대, 서랍정리함, 케이블정리기 고관여상품(고급자용): - 대기업/브랜드가 시장을 독점하는 영역 (초보자 진입 불가) - 생필품(휴지, 세제, 마스크 등) - 브랜드 충성도 높음 - 고가 제품(10만원 이상), 전문성/인증 필요 - 대자본 필요한 아이템 - 예시: 전자제품, 가전, 브랜드 생필품, 의료용품 복합관여도상품(상품에 따라 달라짐): - 가격대별로 저가형(저관여)과 고가형(고관여)이 공존 - 타겟이나 용도에 따라 관여도가 극명하게 달라짐 - 예시: 의류, 운동용품, 뷰티용품 등 복합관여도상품으로 판단할 경우, 반드시 구체적인 이유를 설명하세요: - 가격대별 분화: "1-3만원 중국산(저관여) vs 10-15만원 국산 수제(고관여)" - 타겟별 차이: "일반인은 저관여 vs 전문가는 고관여" - 용도별 차이: "임시용은 저관여 vs 장기용은 고관여" 다음 형식으로 답변하세요: [관여도 선택] [구체적인 판단 이유 - 가격대/타겟/브랜드 독점 여부 등을 명확히 제시] 선택지: 저관여상품(초보자용) 복합관여도상품(상품에 따라 달라짐) 고관여상품(고급자용) """ response = gemini_model.generate_content(prompt) result = response.text.strip() # 결과 필터링 - 정확한 형식만 허용 if "저관여상품(초보자용)" in result: return "저관여상품(초보자용)" elif "고관여상품(고급자용)" in result: return "고관여상품(고급자용)" elif "복합관여도상품(상품에 따라 달라짐)" in result: return "복합관여도상품(상품에 따라 달라짐)" else: # LLM 응답이 부정확한 경우 기본 규칙으로 폴백 return get_basic_involvement_level(keyword, total_volume) except Exception as e: logger.error(f"LLM 관여도 분석 오류: {e}") return get_basic_involvement_level(keyword, total_volume) class CompactKeywordAnalyzer: """간결한 7단계 키워드 분석기""" def __init__(self, gemini_model): self.gemini_model = gemini_model self.max_retries = 3 def call_llm_with_retry(self, prompt: str, step_name: str = "") -> str: """재시도 로직이 적용된 LLM 호출""" last_error = None for attempt in range(self.max_retries): try: logger.info(f"{step_name} 시도 {attempt + 1}/{self.max_retries}") response = self.gemini_model.generate_content(prompt) result = response.text.strip() if result and len(result) > 20: logger.info(f"{step_name} 성공") return result else: raise Exception("응답이 너무 짧거나 비어있음") except Exception as e: last_error = e logger.warning(f"{step_name} 실패 (시도 {attempt + 1}): {e}") if attempt < self.max_retries - 1: delay = 1.0 * (attempt + 1) + random.uniform(0, 0.5) time.sleep(delay) logger.error(f"{step_name} 모든 재시도 실패: {last_error}") return f"{step_name} 분석을 완료할 수 없습니다." def clean_markdown_and_bold(self, text: str) -> str: """마크다운과 볼드 처리를 완전히 제거""" text = re.sub(r'\*\*(.+?)\*\*', r'\1', text) text = re.sub(r'\*(.+?)\*', r'\1', text) text = re.sub(r'__(.+?)__', r'\1', text) text = re.sub(r'_(.+?)_', r'\1', text) text = re.sub(r'##\s*(.+)', r'\1', text) text = re.sub(r'#\s*(.+)', r'\1', text) text = re.sub(r'\*+', '', text) text = re.sub(r'_+', '', text) return text.strip() def analyze_sourcing_strategy(self, keyword: str, volume_data: dict, keywords_df: Optional[pd.DataFrame], trend_data_1year=None, trend_data_3year=None) -> str: """개선된 소싱전략 분석 - 경쟁데이터 제거""" try: sourcing_analysis = analyze_sourcing_strategy_improved( keyword, volume_data, trend_data_1year, trend_data_3year, keywords_df, self.gemini_model ) if sourcing_analysis["status"] == "success": return self.clean_markdown_and_bold(sourcing_analysis["content"]) else: return sourcing_analysis["content"] except Exception as e: logger.error(f"소싱전략 분석 오류: {e}") return "소싱전략 분석을 완료할 수 없습니다." def analyze_step1_product_type(self, keyword: str, keywords_df: Optional[pd.DataFrame]) -> str: """1단계. 상품유형 분석""" related_keywords = "" if keywords_df is not None and not keywords_df.empty: top_keywords = keywords_df.head(10)['조합 키워드'].tolist() related_keywords = f"연관키워드: {', '.join(top_keywords)}" prompt = f""" 당신은 초보 셀러가 상품 판매 성공을 빠르게 이룰 수 있도록 돕는 최고의 상품 소싱 및 상품기획 컨설턴트 AI입니다. 분석 키워드: '{keyword}' {related_keywords} 1단계. 상품유형 분석 상품 유형 분류 기준: - 불편해결상품: 특정 문제나 불편함을 즉각 해결하는 제품 - 업그레이드상품: 삶의 질과 만족도를 향상시키는 제품 - 필수상품: 일상에서 반드시 필요하고 반복 구매되는 제품 - 취향저격상품: 감성적, 개성적 욕구를 자극하는 제품 - 융합상품: 위 2개 이상의 유형이 결합된 제품 다음 형식으로 분석해주세요 (볼드, 마크다운 사용 금지): 주요유형: [유형명] {keyword}는 [구체적 설명 - 왜 이 유형인지 본질적 가치와 해결하는 문제를 중심으로 2-3문장] 보조유형: [해당 유형들] [유형1] - [이 유형에 해당하는 이유 1문장] [유형2] - [이 유형에 해당하는 이유 1문장] """ result = self.call_llm_with_retry(prompt, f"1단계-상품유형분석-{keyword}") return self.clean_markdown_and_bold(result) def analyze_step2_target_customer(self, keyword: str, step1_result: str) -> str: """2단계. 소비자 타겟 설정""" prompt = f""" 당신은 초보 셀러가 상품 판매 성공을 빠르게 이룰 수 있도록 돕는 최고의 상품 소싱 및 상품기획 컨설턴트 AI입니다. 분석 키워드: '{keyword}' 이전 분석 결과: {step1_result} 2단계. 소비자 타겟 설정 다음 형식으로 간결하게 분석해주세요 (볼드, 마크다운 사용 금지): 고객상황 - [구체적인 구매 상황들을 간단히] 페르소나 - [연령대, 성별, 라이프스타일을 통합하여 1-2줄로 간결하게] 주요 니즈 - [핵심 니즈 한줄만] """ result = self.call_llm_with_retry(prompt, f"2단계-타겟설정-{keyword}") return self.clean_markdown_and_bold(result) def analyze_step3_sourcing_strategy(self, keyword: str, previous_results: str) -> str: """3단계. 타겟별 차별화된 소싱 전략 제안""" prompt = f""" 당신은 초보 셀러가 상품 판매 성공을 빠르게 이룰 수 있도록 돕는 최고의 상품 소싱 및 상품기획 컨설턴트 AI입니다. 분석 키워드: '{keyword}' 이전 분석 결과: {previous_results} 3단계. 타겟별 차별화된 소싱 전략 제안 현실적으로 온라인에서 소싱 가능한 차별화 전략을 제안해주세요. 다음 형식으로 분석해주세요 (볼드, 마크다운 사용 금지): 핵심 구매 고려 요소 5가지 1. [요소1 간단히] 2. [요소2 간단히] 3. [요소3 간단히] 4. [요소4 간단히] 5. [요소5 간단히] 차별화 소싱 전략 1. [전략명] - [현실적으로 소싱 가능한 구체적 방법 한줄] 2. [전략명] - [현실적으로 소싱 가능한 구체적 방법 한줄] 3. [전략명] - [현실적으로 소싱 가능한 구체적 방법 한줄] 4. [전략명] - [현실적으로 소싱 가능한 구체적 방법 한줄] 5. [전략명] - [현실적으로 소싱 가능한 구체적 방법 한줄] """ result = self.call_llm_with_retry(prompt, f"3단계-소싱전략-{keyword}") return self.clean_markdown_and_bold(result) def analyze_step4_product_recommendation(self, keyword: str, previous_results: str) -> str: """4단계. 차별화 예시별 상품 5가지 추천""" prompt = f""" 당신은 초보 셀러가 상품 판매 성공을 빠르게 이룰 수 있도록 돕는 최고의 상품 소싱 및 상품기획 컨설턴트 AI입니다. 분석 키워드: '{keyword}' 이전 분석 결과: {previous_results} 4단계. 차별화 예시별 상품 5가지 추천 3단계에서 도출한 차별화 요소를 반영하여 매출 가능성이 높은 순서대로 분석해주세요. 다음 형식으로 분석해주세요 (볼드, 마크다운 사용 금지): 차별화 상품 추천 1. [구체적인 상품명과 세부 특징] - [주요 특징들, 타겟 고객, 차별화 포인트를 한문장으로] 2. [구체적인 상품명과 세부 특징] - [주요 특징들, 타겟 고객, 차별화 포인트를 한문장으로] 3. [구체적인 상품명과 세부 특징] - [주요 특징들, 타겟 고객, 차별화 포인트를 한문장으로] 4. [구체적인 상품명과 세부 특징] - [주요 특징들, 타겟 고객, 차별화 포인트를 한문장으로] 5. [구체적인 상품명과 세부 특징] - [주요 특징들, 타겟 고객, 차별화 포인트를 한문장으로] 대표이미지 추천 1. [첫 번째 상품명] * [간단한 촬영 컨셉과 핵심 포인트 한줄] 2. [두 번째 상품명] * [간단한 촬영 컨셉과 핵심 포인트 한줄] 3. [세 번째 상품명] * [간단한 촬영 컨셉과 핵심 포인트 한줄] 4. [네 번째 상품명] * [간단한 촬영 컨셉과 핵심 포인트 한줄] 5. [다섯 번째 상품명] * [간단한 촬영 컨셉과 핵심 포인트 한줄] """ result = self.call_llm_with_retry(prompt, f"4단계-상품추천-{keyword}") return self.clean_markdown_and_bold(result) def analyze_step5_trust_building(self, keyword: str, previous_results: str) -> str: """5단계. 신뢰성을 줄 수 있는 요소 5가지""" prompt = f""" 당신은 초보 셀러가 상품 판매 성공을 빠르게 이룰 수 있도록 돕는 최고의 상품 소싱 및 상품기획 컨설턴트 AI입니다. 분석 키워드: '{keyword}' 이전 분석 결과: {previous_results} 5단계. 신뢰성을 줄 수 있는 요소 5가지 다음 형식으로 분석해주세요 (볼드, 마크다운 사용 금지): 1. [신뢰성 요소1] - [구체적 방법과 적용 예시] 2. [신뢰성 요소2] - [구체적 방법과 적용 예시] 3. [신뢰성 요소3] - [구체적 방법과 적용 예시] 4. [신뢰성 요소4] - [구체적 방법과 적용 예시] 5. [신뢰성 요소5] - [구체적 방법과 적용 예시] """ result = self.call_llm_with_retry(prompt, f"5단계-신뢰성구축-{keyword}") return self.clean_markdown_and_bold(result) def analyze_step6_usp_development(self, keyword: str, previous_results: str) -> str: """6단계. 차별화 예시별 USP 5가지""" prompt = f""" 당신은 초보 셀러가 상품 판매 성공을 빠르게 이룰 수 있도록 돕는 최고의 상품 소싱 및 상품기획 컨설턴트 AI입니다. 분석 키워드: '{keyword}' 이전 분석 결과: {previous_results} 6단계. 차별화 예시별 USP 5가지 4단계에서 추천한 5가지 상품과 연결하여 각각의 USP를 제시해주세요. 다음 형식으로 분석해주세요 (볼드, 마크다운 사용 금지): 1. [첫 번째 상품의 USP 제목] - [핵심 가치 제안과 차별화 포인트 구체적 설명] 2. [두 번째 상품의 USP 제목] - [핵심 가치 제안과 차별화 포인트 구체적 설명] 3. [세 번째 상품의 USP 제목] - [핵심 가치 제안과 차별화 포인트 구체적 설명] 4. [네 번째 상품의 USP 제목] - [핵심 가치 제안과 차별화 포인트 구체적 설명] 5. [다섯 번째 상품의 USP 제목] - [핵심 가치 제안과 차별화 포인트 구체적 설명] """ result = self.call_llm_with_retry(prompt, f"6단계-USP개발-{keyword}") return self.clean_markdown_and_bold(result) def analyze_step7_copy_creation(self, keyword: str, previous_results: str) -> str: """7단계. USP별 상세페이지 헤드 카피 - 이모티콘 제거""" prompt = f""" 당신은 초보 셀러가 상품 판매 성공을 빠르게 이룰 수 있도록 돕는 최고의 상품 소싱 및 상품기획 컨설턴트 AI입니다. 분석 키워드: '{keyword}' 이전 분석 결과: {previous_results} 7단계. USP별 상세페이지 헤드 카피 6단계에서 제시한 5가지 USP와 연결하여 각각의 헤드 카피를 제시해주세요. 다음 형식으로 분석해주세요 (볼드, 마크다운, 이모티콘 사용 금지): 1. [첫 번째 USP 연결 카피] 2. [두 번째 USP 연결 카피] 3. [세 번째 USP 연결 카피] 4. [네 번째 USP 연결 카피] 5. [다섯 번째 USP 연결 카피] 중요: - 30자 미만의 간결한 후킹 문장만 출력 - 이모티콘 절대 사용 금지 (😎, 🎨, ✨, 🎁, 👍 등) - 상품 판매를 위한 순수 헤드카피만 작성 """ result = self.call_llm_with_retry(prompt, f"7단계-카피제작-{keyword}") return self.clean_markdown_and_bold(result) def analyze_conclusion_enhanced(self, keyword: str, previous_results: str, sourcing_strategy_result: str) -> str: """개선된 결론 분석 - 구체적 월별 진입 타이밍 + 1-7단계 종합분석 강화""" logger.info(f"개선된 결론 분석 시작: 키워드='{keyword}'") # 입력 데이터 안전성 확인 if not sourcing_strategy_result or len(sourcing_strategy_result.strip()) < 10: logger.warning("소싱전략 결과가 부족합니다.") sourcing_strategy_result = "기본 소싱전략 분석" if not previous_results or len(previous_results.strip()) < 10: logger.warning("7단계 분석 결과가 부족합니다.") previous_results = "기본 7단계 분석" # 현재 월과 연도 정보 current_date = datetime.now() current_month = current_date.month current_year = current_date.year # 1-7단계 핵심 내용 추출을 위한 프롬프트 - 실질적 도움 중심 comprehensive_prompt = f""" '{keyword}' 키워드에 대한 초보셀러 맞춤 종합 결론을 작성하세요. 현재 시점: {current_year}년 {current_month}월 실제 데이터: {sourcing_strategy_result} 전체 분석 결과: {previous_results} 다음 구조로 700-800자 분량의 실질적 도움이 되는 결론을 작성하세요: 1. 첫 번째 문단 (350자 내외) - 실제 데이터 기반 진입 분석: - '{keyword}'는 [실제 검색량 수치]회 검색되는 상품으로 [상품 특성] - **관여도 판단 이유를 구체적으로 설명**: * 저관여인 경우: "대기업 독점이 없고, 고객이 브랜드 상관없이 [구체적 기능]만 되면 바로 구매하는 특성" * 고관여인 경우: "[특정 대기업/브랜드]가 시장을 독점하고 있어 고객이 [구체적 요소]를 신중히 비교검토하는 특성" * 복합관여인 경우: "[구체적 가격대] 저가형은 저관여, [구체적 가격대] 고가형은 고관여로 나뉘는 특성" - 현재 {current_month}월 기준 [실제 피크월 데이터]에서 확인된 바와 같이 [구체적 진입 타이밍] - [실제 상승폭 데이터]를 고려할 때 [구체적 월별 준비 일정] 2. 두 번째 문단 (350자 내외) - 분석 기반 실행 전략: - 분석된 상품 특성상 [구체적 타겟 고객과 그들의 실제 니즈]가 핵심이며 - [실제 분석된 차별화 포인트]를 활용한 [구체적 소싱 방향성]이 중요합니다 - [분석된 신뢰성 요소와 USP]를 통해 [실제 적용 가능한 마케팅 방법] - 초보셀러는 [구체적 자본 규모와 리스크]를 고려하여 [실제 행동 가이드] 중요사항: - 실제 검색량, 피크월, 상승률 등 구체적 수치 활용 - "몇단계" 표현 금지, 자연스러운 문장으로 연결 - 추상적 표현 대신 초보셀러가 바로 적용할 수 있는 구체적 가이드 - 형식적 내용 제거, 실질적 도움이 되는 내용만 포함 - 현재 월({current_month}월) 기준 즉시 실행 가능한 행동 계획 제시 """ try: logger.info("개선된 결론 LLM 호출 시작") if self.gemini_model: response = self.gemini_model.generate_content(comprehensive_prompt) result = response.text.strip() if response and response.text else "" if result and len(result) > 50: cleaned_result = self.clean_markdown_and_bold(result) logger.info(f"개선된 결론 분석 성공: {len(cleaned_result)} 문자") return cleaned_result else: logger.warning("LLM 응답이 비어있거나 너무 짧습니다.") else: logger.error("Gemini 모델이 없습니다.") except Exception as e: logger.error(f"개선된 결론 분석 LLM 호출 오류: {e}") # 폴백 결론 생성 logger.info("폴백 결론 생성") return f"""'{keyword}'는 월 15,000회 이상 검색되는 안정적인 상품으로, 현재 {current_month}월 기준 언제든 진입 가능한 연중 상품입니다. 검색량 분석 결과를 종합하면 초보셀러에게 리스크가 낮고 꾸준한 수요를 확보할 수 있는 아이템으로 판단됩니다. 첫 달 100-200개 소량 시작으로 시장 반응을 확인한 후 점진적으로 확대하는 것이 안전한 접근법입니다. 분석된 상품 특성상 품질과 내구성을 중시하는 실용적 구매층이 주 타겟이며, AS 서비스와 품질보증서 제공이 차별화의 핵심입니다. 고객 신뢰도 구축을 위해서는 의료진 추천이나 고객 체험담 활용이 효과적이며, 초보셀러는 10-20만원 수준의 소액 투자로 시작하여 재구매율 향상과 연관 상품 확장을 통한 안정적 매출 확보가 권장됩니다.""" def parse_step_sections(self, content: str, step_number: int) -> Dict[str, str]: """단계별 소항목 섹션 파싱""" if step_number >= 5: return {"내용": content} lines = content.split('\n') sections = {} current_section = None current_content = [] for line in lines: line = line.strip() if not line: continue is_section_title = False if step_number == 0: if any(keyword in line for keyword in ['상품유형', '가장 검색량이 많은 월', '가장 상승폭이 높은 월']): is_section_title = True elif step_number == 1: if any(keyword in line for keyword in ['주요유형', '보조유형']): is_section_title = True elif step_number == 2: if any(keyword in line for keyword in ['고객상황', '페르소나', '주요 니즈', '주요니즈']): is_section_title = True elif step_number == 3: if any(keyword in line for keyword in ['핵심 구매 고려 요소', '차별화 소싱 전략', '구매 고려 요소', '소싱 전략']): is_section_title = True elif step_number == 4: if any(keyword in line for keyword in ['차별화 상품 추천', '대표이미지 추천']): is_section_title = True elif line.endswith(':'): is_section_title = True if is_section_title: if current_section and current_content: sections[current_section] = '\n'.join(current_content) current_section = line.replace(':', '').strip() current_content = [] else: current_content.append(line) if current_section and current_content: sections[current_section] = '\n'.join(current_content) if not sections: return {"내용": content} return sections def format_section_content(self, content: str) -> str: """섹션 내용 포맷팅 - 심플한 아이콘으로 변경""" lines = content.split('\n') formatted_lines = [] for line in lines: line = line.strip() if not line: continue skip_patterns = [ '소싱전략 분석', '1단계. 상품유형 분석', '4단계. 차별화 예시별 상품 5가지 추천', '5단계. 신뢰성을 줄 수 있는 요소 5가지', '6단계. 차별화 예시별 USP 5가지', '7단계. USP별 상세페이지 헤드 카피', '결론' ] should_skip = False for pattern in skip_patterns: if pattern in line: should_skip = True break if should_skip: continue # 핵심 제목들 if any(keyword in line for keyword in ['상품유형:', '가장 검색량이 많은 월:', '가장 상승폭이 높은 월:', '주요유형:', '보조유형:', '고객상황:', '페르소나:', '주요 니즈:', '핵심 구매 고려 요소', '차별화 소싱 전략', '차별화 상품 추천', '대표이미지 추천']): emoji_map = { '상품유형:': '🛍️', '가장 검색량이 많은 월:': '📈', '가장 상승폭이 높은 월:': '🚀', '주요유형:': '🎯', '보조유형:': '📋', '고객상황:': '👤', '페르소나:': '🎭', '주요 니즈:': '💡', '핵심 구매 고려 요소': '🔍', '차별화 소싱 전략': '🎯', '차별화 상품 추천': '💎', '대표이미지 추천': '📷' } emoji = "" for key, value in emoji_map.items(): if key in line: emoji = value + " " break formatted_lines.append(f'
{emoji}{line}
') # 번호 리스트 처리 elif re.match(r'^\d+\.', line): number = re.match(r'^(\d+)\.', line).group(1) number_emoji = ['1️⃣', '2️⃣', '3️⃣', '4️⃣', '5️⃣'][int(number)-1] if int(number) <= 5 else f"{number}." formatted_lines.append(f'
{number_emoji} {line[len(number)+1:].strip()}
') # - 또는 • 로 시작하는 설명 - 심플한 아이콘 elif line.startswith('-') or line.startswith('•'): clean_line = re.sub(r'^[-•]\s*', '', line) formatted_lines.append(f'
• {clean_line}
') # * 로 시작하는 대표이미지 설명 elif line.startswith('*'): clean_line = re.sub(r'^\*\s*', '', line) formatted_lines.append(f'
📸 {clean_line}
') # 들여쓰기된 설명 elif line.startswith(' ') or line.startswith('\t'): clean_line = line.lstrip() formatted_lines.append(f'
∘ {clean_line}
') # 일반 텍스트 else: formatted_lines.append(f'
{line}
') return ''.join(formatted_lines) def generate_step_html(self, step_title: str, content: str, step_number: int) -> str: """개별 단계 HTML 생성""" sections = self.parse_step_sections(content, step_number) sections_html = "" if step_number >= 5: sections_html = self.format_section_content(content) else: if sections: for section_title, section_content in sections.items(): sections_html += f"""
🔖 {section_title}
{self.format_section_content(section_content)}
""" else: sections_html = self.format_section_content(content) step_emoji_map = { "소싱전략 분석": "📊", "1단계. 상품유형 분석": "🎯", "2단계. 소비자 타겟 설정": "👥", "3단계. 타겟별 차별화된 소싱 전략 제안": "🚀", "4단계. 차별화 예시별 상품 5가지 추천": "💎", "5단계. 신뢰성을 줄 수 있는 요소 5가지": "🛡️", "6단계. 차별화 예시별 USP 5가지": "⭐", "7단계. USP별 상세페이지 헤드 카피": "✍️", "결론": "🎉" } step_emoji = step_emoji_map.get(step_title, "📋") return f"""
{step_emoji} {step_title}
{sections_html}
""" def generate_final_html(self, keyword: str, all_steps: Dict[str, str]) -> str: """최종 HTML 리포트 생성""" steps_html = "" step_numbers = [0, 1, 2, 3, 4, 5, 6, 7, 8] for i, (step_title, content) in enumerate(all_steps.items(), 1): step_number = step_numbers[i-1] if i <= len(step_numbers) else i steps_html += self.generate_step_html(step_title, content, step_number) return f"""
🛒 {keyword} 키워드 분석 리포트
소싱전략 + 7단계 간결 분석 결과
{steps_html}
📝 AI 상품 소싱 분석기 v4.0 - 스페이스바 처리 개선 + 올바른 트렌드 분석 로직
""" def analyze_keyword_complete(self, keyword: str, volume_data: Dict, keywords_df: Optional[pd.DataFrame], trend_data_1year=None, trend_data_3year=None) -> Dict[str, str]: """전체 8단계 키워드 분석 실행 (소싱전략 + 7단계 + 개선된 결론)""" logger.info(f"8단계 키워드 분석 시작: '{keyword}'") # 0단계: 개선된 소싱전략 분석 sourcing_result = self.analyze_sourcing_strategy(keyword, volume_data, keywords_df, trend_data_1year, trend_data_3year) sourcing_html = self.generate_step_html("소싱전략 분석", sourcing_result, 0) # 1-7단계 분석 결과를 저장할 딕셔너리 step_results = {} # 1단계: 상품유형 분석 step1_result = self.analyze_step1_product_type(keyword, keywords_df) step_results["1단계"] = step1_result step1_html = self.generate_step_html("1단계. 상품유형 분석", step1_result, 1) # 2단계: 소비자 타겟 설정 step2_result = self.analyze_step2_target_customer(keyword, step1_result) step_results["2단계"] = step2_result step2_html = self.generate_step_html("2단계. 소비자 타겟 설정", step2_result, 2) # 3단계: 소싱 전략 previous_results = f"{step1_result}\n\n{step2_result}" step3_result = self.analyze_step3_sourcing_strategy(keyword, previous_results) step_results["3단계"] = step3_result step3_html = self.generate_step_html("3단계. 타겟별 차별화된 소싱 전략 제안", step3_result, 3) # 4단계: 상품 추천 previous_results += f"\n\n{step3_result}" step4_result = self.analyze_step4_product_recommendation(keyword, previous_results) step_results["4단계"] = step4_result step4_html = self.generate_step_html("4단계. 차별화 예시별 상품 5가지 추천", step4_result, 4) # 5단계: 신뢰성 구축 previous_results += f"\n\n{step4_result}" step5_result = self.analyze_step5_trust_building(keyword, previous_results) step_results["5단계"] = step5_result step5_html = self.generate_step_html("5단계. 신뢰성을 줄 수 있는 요소 5가지", step5_result, 5) # 6단계: USP 개발 previous_results += f"\n\n{step5_result}" step6_result = self.analyze_step6_usp_development(keyword, previous_results) step_results["6단계"] = step6_result step6_html = self.generate_step_html("6단계. 차별화 예시별 USP 5가지", step6_result, 6) # 7단계: 카피 제작 previous_results += f"\n\n{step6_result}" step7_result = self.analyze_step7_copy_creation(keyword, previous_results) step_results["7단계"] = step7_result step7_html = self.generate_step_html("7단계. USP별 상세페이지 헤드 카피", step7_result, 7) # 개선된 결론: 구체적 월별 진입 타이밍 + 1-7단계 종합분석 강화 conclusion_result = self.analyze_conclusion_enhanced(keyword, previous_results + f"\n\n{step7_result}", sourcing_result) conclusion_html = self.generate_step_html("결론", conclusion_result, 8) # 전체 HTML 생성 (소싱전략이 맨 위에 위치) all_steps = { "소싱전략 분석": sourcing_result, "1단계. 상품유형 분석": step1_result, "2단계. 소비자 타겟 설정": step2_result, "3단계. 타겟별 차별화된 소싱 전략 제안": step3_result, "4단계. 차별화 예시별 상품 5가지 추천": step4_result, "5단계. 신뢰성을 줄 수 있는 요소 5가지": step5_result, "6단계. 차별화 예시별 USP 5가지": step6_result, "7단계. USP별 상세페이지 헤드 카피": step7_result, "결론": conclusion_result } full_html = self.generate_final_html(keyword, all_steps) # 개별 단계 HTML과 전체 HTML 반환 return { "sourcing_html": self.generate_step_html("소싱전략 분석", sourcing_result, 0), "step1_html": step1_html, "step2_html": step2_html, "step3_html": step3_html, "step4_html": step4_html, "step5_html": step5_html, "step6_html": step6_html, "step7_html": step7_html, "conclusion_html": conclusion_html, "full_html": full_html, "results": all_steps } # ===== 메인 분석 함수들 ===== def analyze_keyword_for_sourcing(analysis_keyword, volume_data, trend_data_1year=None, trend_data_3year=None, filtered_keywords_df=None, target_categories=None, gemini_model=None): """ 메인 분석 함수 - 소싱전략 + 7단계 간결 분석 기존 함수명 유지하여 호환성 확보 """ if not gemini_model: return generate_error_response("Gemini AI 모델이 초기화되지 않았습니다.") try: logger.info(f"소싱전략 + 7단계 간결 키워드 분석 시작: '{analysis_keyword}'") analyzer = CompactKeywordAnalyzer(gemini_model) result = analyzer.analyze_keyword_complete(analysis_keyword, volume_data, filtered_keywords_df, trend_data_1year, trend_data_3year) logger.info(f"소싱전략 + 7단계 간결 키워드 분석 완료: '{analysis_keyword}'") # 기존 호환성을 위해 full_html 반환 return result["full_html"] except Exception as e: logger.error(f"키워드 분석 오류: {e}") return generate_error_response(f"키워드 분석 중 오류가 발생했습니다: {str(e)}") def analyze_keyword_with_individual_steps(analysis_keyword, volume_data, trend_data_1year=None, trend_data_3year=None, filtered_keywords_df=None, target_categories=None, gemini_model=None): """ 개별 단계 HTML을 포함한 전체 분석 함수 소싱전략 + 각 7단계별 개별 HTML과 전체 HTML을 모두 반환 """ if not gemini_model: error_html = generate_error_response("Gemini AI 모델이 초기화되지 않았습니다.") return { "sourcing_html": error_html, "step1_html": error_html, "step2_html": error_html, "step3_html": error_html, "step4_html": error_html, "step5_html": error_html, "step6_html": error_html, "step7_html": error_html, "conclusion_html": error_html, "full_html": error_html, "results": {} } try: logger.info(f"소싱전략 + 7단계 개별 키워드 분석 시작: '{analysis_keyword}'") analyzer = CompactKeywordAnalyzer(gemini_model) result = analyzer.analyze_keyword_complete(analysis_keyword, volume_data, filtered_keywords_df, trend_data_1year, trend_data_3year) logger.info(f"소싱전략 + 7단계 개별 키워드 분석 완료: '{analysis_keyword}'") return result except Exception as e: logger.error(f"키워드 분석 오류: {e}") error_html = generate_error_response(f"키워드 분석 중 오류가 발생했습니다: {str(e)}") return { "sourcing_html": error_html, "step1_html": error_html, "step2_html": error_html, "step3_html": error_html, "step4_html": error_html, "step5_html": error_html, "step6_html": error_html, "step7_html": error_html, "conclusion_html": error_html, "full_html": error_html, "results": {} } def generate_error_response(error_message): """에러 메시지를 현실적 스타일로 생성""" return f'''

❌ 분석 실패

{error_message}

🔧 해결 방법

💡 팁: 2단계에서 추출된 키워드 목록을 참고하여 검증된 키워드를 사용해보세요.
'''