""" 트렌드 분석 모듈 v2.16 - 정교한 이중 API 호출 및 역산 로직 구현 - 이중 트렌드 API 호출: 일별 + 월별 데이터 - 일별 데이터로 전월 정확한 검색량 역산 - 전월 기준으로 3년 모든 월 검색량 역산 - 작년 동월 기반 미래 3개월 예상 - 최종 단계에서 10% 감소 조정 적용 """ import urllib.request import json import time import logging from datetime import datetime, timedelta import calendar import api_utils # 로깅 설정 logger = logging.getLogger(__name__) # ===== 핵심 개선 함수들 ===== def get_complete_month(): """완성된 마지막 월 계산 - 단순화된 로직""" current_date = datetime.now() current_day = current_date.day current_year = current_date.year current_month = current_date.month # 3일 이후에만 전월을 완성된 것으로 간주 if current_day >= 3: 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 return completed_year, completed_month def get_daily_trend_data(keywords, max_retries=3): """1차 호출: 일별 트렌드 데이터 (전월 정확 계산용)""" for retry_attempt in range(max_retries): try: # 데이터랩 API 설정 datalab_config = api_utils.get_next_datalab_api_config() if not datalab_config: logger.warning("데이터랩 API 키가 설정되지 않았습니다.") return None client_id = datalab_config["CLIENT_ID"] client_secret = datalab_config["CLIENT_SECRET"] # 완성된 마지막 월 계산 completed_year, completed_month = get_complete_month() # 일별 데이터 기간: 전월 1일 ~ 어제 current_date = datetime.now() yesterday = current_date - timedelta(days=1) start_date = f"{completed_year:04d}-{completed_month:02d}-01" end_date = yesterday.strftime("%Y-%m-%d") logger.info(f"📞 1차 호출 (일별): {start_date} ~ {end_date}") # 키워드 그룹 생성 keywordGroups = [] for kw in keywords[:5]: keywordGroups.append({ 'groupName': kw, 'keywords': [kw] }) # API 요청 body_dict = { 'startDate': start_date, 'endDate': end_date, 'timeUnit': 'date', # 일별! 'keywordGroups': keywordGroups } url = "https://openapi.naver.com/v1/datalab/search" body = json.dumps(body_dict, ensure_ascii=False) request = urllib.request.Request(url) request.add_header("X-Naver-Client-Id", client_id) request.add_header("X-Naver-Client-Secret", client_secret) request.add_header("Content-Type", "application/json") response = urllib.request.urlopen(request, data=body.encode("utf-8"), timeout=15) rescode = response.getcode() if rescode == 200: response_body = response.read() response_json = json.loads(response_body) logger.info(f"일별 트렌드 데이터 조회 성공") return response_json else: logger.error(f"일별 API 오류: 상태코드 {rescode}") if retry_attempt < max_retries - 1: time.sleep(2 * (retry_attempt + 1)) continue return None except Exception as e: logger.error(f"일별 트렌드 조회 오류 (시도 {retry_attempt + 1}): {e}") if retry_attempt < max_retries - 1: time.sleep(2 * (retry_attempt + 1)) continue return None return None def get_monthly_trend_data(keywords, max_retries=3): """2차 호출: 월별 트렌드 데이터 (3년 전체 + 예상용)""" for retry_attempt in range(max_retries): try: # 데이터랩 API 설정 datalab_config = api_utils.get_next_datalab_api_config() if not datalab_config: logger.warning("데이터랩 API 키가 설정되지 않았습니다.") return None client_id = datalab_config["CLIENT_ID"] client_secret = datalab_config["CLIENT_SECRET"] # 완성된 마지막 월 계산 completed_year, completed_month = get_complete_month() # 월별 데이터 기간: 3년 전 1월 ~ 완성된 마지막 월 start_year = completed_year - 3 start_date = f"{start_year:04d}-01-01" end_date = f"{completed_year:04d}-{completed_month:02d}-01" logger.info(f"📞 2차 호출 (월별): {start_date} ~ {end_date}") # 키워드 그룹 생성 keywordGroups = [] for kw in keywords[:5]: keywordGroups.append({ 'groupName': kw, 'keywords': [kw] }) # API 요청 body_dict = { 'startDate': start_date, 'endDate': end_date, 'timeUnit': 'month', # 월별! 'keywordGroups': keywordGroups } url = "https://openapi.naver.com/v1/datalab/search" body = json.dumps(body_dict, ensure_ascii=False) request = urllib.request.Request(url) request.add_header("X-Naver-Client-Id", client_id) request.add_header("X-Naver-Client-Secret", client_secret) request.add_header("Content-Type", "application/json") response = urllib.request.urlopen(request, data=body.encode("utf-8"), timeout=15) rescode = response.getcode() if rescode == 200: response_body = response.read() response_json = json.loads(response_body) logger.info(f"월별 트렌드 데이터 조회 성공") return response_json else: logger.error(f"월별 API 오류: 상태코드 {rescode}") if retry_attempt < max_retries - 1: time.sleep(2 * (retry_attempt + 1)) continue return None except Exception as e: logger.error(f"월별 트렌드 조회 오류 (시도 {retry_attempt + 1}): {e}") if retry_attempt < max_retries - 1: time.sleep(2 * (retry_attempt + 1)) continue return None return None def calculate_previous_month_from_daily(current_volume, daily_data): """일별 트렌드로 전월 정확한 검색량 역산""" if not daily_data or "results" not in daily_data: logger.warning("일별 데이터가 없어 전월 계산을 건너뜁니다.") return current_volume try: completed_year, completed_month = get_complete_month() # 전월 일수 계산 prev_month_days = calendar.monthrange(completed_year, completed_month)[1] for result in daily_data["results"]: keyword = result["title"] if not result["data"]: continue # 최근 30일과 전월 데이터 분리 recent_30_ratios = [] # 최근 30일 (현재 검색량 기준) prev_month_ratios = [] # 전월 데이터 for data_point in result["data"]: try: date_obj = datetime.strptime(data_point["period"], "%Y-%m-%d") ratio = data_point["ratio"] # 전월 데이터 if date_obj.year == completed_year and date_obj.month == completed_month: prev_month_ratios.append(ratio) # 최근 30일 (현재 검색량 기준 구간) current_date = datetime.now() if (current_date - date_obj).days <= 30: recent_30_ratios.append(ratio) except: continue if not recent_30_ratios or not prev_month_ratios: logger.warning(f"'{keyword}' 비교 데이터가 부족합니다.") continue # 평균 비율 계산 recent_30_avg = sum(recent_30_ratios) / len(recent_30_ratios) prev_month_avg = sum(prev_month_ratios) / len(prev_month_ratios) if recent_30_avg == 0: continue # 전월 검색량 역산 # 현재 검색량 = 최근 30일 평균 기준 # 전월 검색량 = (전월 평균 / 최근 30일 평균) × 현재 검색량 × (전월 일수 / 30일) prev_month_volume = int( (prev_month_avg / recent_30_avg) * current_volume * (prev_month_days / 30) ) logger.info(f"'{keyword}' 전월 {completed_year}.{completed_month:02d} 역산 검색량: {prev_month_volume:,}회") logger.info(f" - 최근 30일 평균 비율: {recent_30_avg:.1f}%") logger.info(f" - 전월 평균 비율: {prev_month_avg:.1f}%") logger.info(f" - 전월 일수 보정: {prev_month_days}일") return prev_month_volume except Exception as e: logger.error(f"전월 역산 계산 오류: {e}") return current_volume return current_volume def calculate_all_months_from_previous(prev_month_volume, monthly_data, completed_year, completed_month): """전월을 기준으로 모든 월 검색량 역산""" if not monthly_data or "results" not in monthly_data: logger.warning("월별 데이터가 없어 역산 계산을 건너뜁니다.") return [], [] monthly_volumes = [] dates = [] try: for result in monthly_data["results"]: keyword = result["title"] if not result["data"]: continue # 전월(기준월) 비율 찾기 base_ratio = None for data_point in result["data"]: try: date_obj = datetime.strptime(data_point["period"], "%Y-%m-%d") if date_obj.year == completed_year and date_obj.month == completed_month: base_ratio = data_point["ratio"] break except: continue if base_ratio is None or base_ratio == 0: logger.warning(f"'{keyword}' 기준월 비율을 찾을 수 없습니다.") continue logger.info(f"'{keyword}' 기준월 {completed_year}.{completed_month:02d} 비율: {base_ratio}% (검색량: {prev_month_volume:,}회)") # 모든 월 검색량 역산 for data_point in result["data"]: try: date_obj = datetime.strptime(data_point["period"], "%Y-%m-%d") ratio = data_point["ratio"] # 해당 월의 일수 가져오기 month_days = calendar.monthrange(date_obj.year, date_obj.month)[1] base_month_days = calendar.monthrange(completed_year, completed_month)[1] # 역산 계산: (해당월 비율 / 기준월 비율) × 기준월 검색량 × (해당월 일수 / 기준월 일수) calculated_volume = int( (ratio / base_ratio) * prev_month_volume * (month_days / base_month_days) ) calculated_volume = max(calculated_volume, 0) # 음수 방지 monthly_volumes.append(calculated_volume) dates.append(data_point["period"]) except: continue logger.info(f"'{keyword}' 전체 월별 검색량 역산 완료: {len(monthly_volumes)}개월") break # 첫 번째 키워드만 처리 except Exception as e: logger.error(f"월별 역산 계산 오류: {e}") return [], [] return monthly_volumes, dates def generate_future_from_growth_rate(monthly_volumes, dates, completed_year, completed_month): """증감율 기반 미래 3개월 예상 생성""" if len(monthly_volumes) < 12: logger.warning("미래 예측을 위한 충분한 데이터가 없습니다.") return [], [] try: # 작년 동월들과 올해 완성된 월들 비교하여 증감율 계산 this_year_volumes = [] last_year_volumes = [] for i, date_str in enumerate(dates): try: date_obj = datetime.strptime(date_str, "%Y-%m-%d") # 올해 완성된 데이터 (1월 ~ 완성된 월) if date_obj.year == completed_year and date_obj.month <= completed_month: this_year_volumes.append(monthly_volumes[i]) # 작년 동일 기간 데이터 if date_obj.year == completed_year - 1 and date_obj.month <= completed_month: last_year_volumes.append(monthly_volumes[i]) except: continue # 증감율 계산 if len(this_year_volumes) >= 3 and len(last_year_volumes) >= 3: this_year_avg = sum(this_year_volumes) / len(this_year_volumes) last_year_avg = sum(last_year_volumes) / len(last_year_volumes) if last_year_avg > 0: growth_rate = (this_year_avg - last_year_avg) / last_year_avg # 증감율 범위 제한 (-50% ~ +100%) growth_rate = max(-0.5, min(growth_rate, 1.0)) else: growth_rate = 0 else: growth_rate = 0 logger.info(f"계산된 증감율: {growth_rate*100:+.1f}%") # 미래 3개월 예상 predicted_volumes = [] predicted_dates = [] for month_offset in range(1, 4): # 1, 2, 3개월 후 pred_year = completed_year pred_month = completed_month + month_offset while pred_month > 12: pred_month -= 12 pred_year += 1 # 작년 동월 데이터 찾기 last_year_pred_year = pred_year - 1 last_year_pred_month = pred_month last_year_volume = None for i, date_str in enumerate(dates): try: date_obj = datetime.strptime(date_str, "%Y-%m-%d") if date_obj.year == last_year_pred_year and date_obj.month == last_year_pred_month: last_year_volume = monthly_volumes[i] break except: continue # 예상 검색량 = 작년 동월 × (1 + 증감율) if last_year_volume is not None: predicted_volume = int(last_year_volume * (1 + growth_rate)) predicted_volume = max(predicted_volume, 0) # 음수 방지 predicted_volumes.append(predicted_volume) predicted_dates.append(f"{pred_year:04d}-{pred_month:02d}-01") logger.info(f"예상 {pred_year}.{pred_month:02d}: 작년 동월 {last_year_volume:,}회 → 예상 {predicted_volume:,}회") return predicted_volumes, predicted_dates except Exception as e: logger.error(f"미래 예측 생성 오류: {e}") return [], [] def apply_final_10_percent_reduction(monthly_data): """최종 단계: 모든 결과에 10% 감소 적용""" adjusted_data = {} try: for keyword, data in monthly_data.items(): adjusted_volumes = [] for volume in data["monthly_volumes"]: if volume >= 10: adjusted_volume = int(volume * 0.9) # 10% 감소 else: adjusted_volume = volume # 10 미만은 그대로 adjusted_volumes.append(adjusted_volume) # 데이터 복사 및 검색량 조정 adjusted_data[keyword] = data.copy() adjusted_data[keyword]["monthly_volumes"] = adjusted_volumes # 현재 검색량도 조정 if data["current_volume"] >= 10: adjusted_data[keyword]["current_volume"] = int(data["current_volume"] * 0.9) logger.info("최종 10% 감소 조정 완료") except Exception as e: logger.error(f"10% 감소 조정 오류: {e}") return monthly_data return adjusted_data # ===== 메인 함수들 (기존 호환성 유지) ===== def get_naver_trend_data_v5(keywords, period="1year", max_retries=3): """개선된 네이버 데이터랩 API 호출 - 이중 호출 구현""" if period == "1year": # 이중 API 호출 daily_data = get_daily_trend_data(keywords, max_retries) monthly_data = get_monthly_trend_data(keywords, max_retries) # 결과 반환 (기존 코드와 호환성 유지를 위해 월별 데이터 우선 반환) return { 'daily_data': daily_data, 'monthly_data': monthly_data, 'results': monthly_data['results'] if monthly_data else [] } else: # 3year # 3년 데이터는 월별만 호출 (기존 방식 유지) monthly_data = get_monthly_trend_data(keywords, max_retries) return monthly_data def calculate_monthly_volumes_v7(keywords, current_volumes, trend_data, period="1year"): """개선된 월별 검색량 계산 - 정교한 역산 로직 적용""" monthly_data = {} # 트렌드 데이터 확인 if isinstance(trend_data, dict) and 'daily_data' in trend_data and 'monthly_data' in trend_data: # 새로운 이중 데이터 구조 daily_data = trend_data['daily_data'] monthly_data_api = trend_data['monthly_data'] else: # 기존 단일 데이터 구조 (3year 등) daily_data = None monthly_data_api = trend_data if not monthly_data_api or "results" not in monthly_data_api: logger.warning("월별 트렌드 데이터가 없어 계산을 건너뜁니다.") return monthly_data logger.info(f"개선된 월별 검색량 계산 시작: {len(monthly_data_api['results'])}개 키워드") for result in monthly_data_api["results"]: keyword = result["title"] api_keyword = keyword.replace(" ", "") # 현재 검색량 volume_data = current_volumes.get(api_keyword, {"총검색량": 0}) current_volume = volume_data["총검색량"] if current_volume == 0: logger.warning(f"'{keyword}' 현재 검색량이 0이므로 계산을 건너뜁니다.") continue logger.info(f"'{keyword}' 처리 시작 - 현재 검색량: {current_volume:,}회") if period == "1year" and daily_data: # 🔥 새로운 정교한 로직 적용 completed_year, completed_month = get_complete_month() # 1단계: 일별 데이터로 전월 정확한 검색량 계산 prev_month_volume = calculate_previous_month_from_daily(current_volume, daily_data) # 2단계: 전월을 기준으로 모든 월 검색량 역산 monthly_volumes, dates = calculate_all_months_from_previous( prev_month_volume, monthly_data_api, completed_year, completed_month ) if not monthly_volumes: logger.warning(f"'{keyword}' 월별 검색량 계산 실패") continue # 3단계: 미래 3개월 예상 생성 predicted_volumes, predicted_dates = generate_future_from_growth_rate( monthly_volumes, dates, completed_year, completed_month ) # 실제 + 예상 데이터 결합 - 최근 12개월 + 향후 3개월 = 총 15개월 # 최근 12개월만 실제 데이터로 제한 recent_12_months = monthly_volumes[-12:] if len(monthly_volumes) >= 12 else monthly_volumes recent_12_dates = dates[-12:] if len(dates) >= 12 else dates all_volumes = recent_12_months + predicted_volumes all_dates = recent_12_dates + predicted_dates # 증감율 계산 (예상 3개월) growth_rate = calculate_future_3month_growth_rate(all_volumes, all_dates) monthly_data[keyword] = { "monthly_volumes": all_volumes, # 10% 감소 적용 전 - 총 15개월 (12개월 실제 + 3개월 예상) "dates": all_dates, "current_volume": current_volume, # 10% 감소 적용 전 "growth_rate": growth_rate, "volume_per_percent": prev_month_volume / 100 if prev_month_volume > 0 else 0, "current_ratio": 100, "actual_count": len(recent_12_months), # 실제 12개월 "predicted_count": len(predicted_volumes) # 예상 3개월 } else: # 기존 방식 (3year 등) if not result["data"]: continue current_ratio = result["data"][-1]["ratio"] if current_ratio == 0: continue volume_per_percent = current_volume / current_ratio monthly_volumes = [] dates = [] for data_point in result["data"]: ratio = data_point["ratio"] period_date = data_point["period"] estimated_volume = int(volume_per_percent * ratio) monthly_volumes.append(estimated_volume) dates.append(period_date) growth_rate = calculate_3year_growth_rate_improved(monthly_volumes) monthly_data[keyword] = { "monthly_volumes": monthly_volumes, # 10% 감소 적용 전 "dates": dates, "current_volume": current_volume, # 10% 감소 적용 전 "growth_rate": growth_rate, "volume_per_percent": volume_per_percent, "current_ratio": current_ratio, "actual_count": len(monthly_volumes), "predicted_count": 0 } logger.info(f"'{keyword}' 계산 완료 - 검색량 데이터 {len(monthly_data[keyword]['monthly_volumes'])}개") # 🔥 4단계: 최종 10% 감소 적용 final_data = apply_final_10_percent_reduction(monthly_data) logger.info("개선된 월별 검색량 계산 완료 (10% 감소 적용됨)") return final_data # ===== 증감율 계산 함수들 (기존 유지) ===== def calculate_future_3month_growth_rate(volumes, dates): """예상 3개월 증감율 계산""" if len(volumes) < 4: return 0 try: completed_year, completed_month = get_complete_month() # 기준월 데이터 찾기 base_month_volume = None for i, date_str in enumerate(dates): try: date_obj = datetime.strptime(date_str, "%Y-%m-%d") if date_obj.year == completed_year and date_obj.month == completed_month: base_month_volume = volumes[i] break except: continue if base_month_volume is None: return 0 # 향후 3개월 예상 데이터 찾기 future_volumes = [] for month_offset in range(1, 4): pred_year = completed_year pred_month = completed_month + month_offset while pred_month > 12: pred_month -= 12 pred_year += 1 for i, date_str in enumerate(dates): try: date_obj = datetime.strptime(date_str, "%Y-%m-%d") if date_obj.year == pred_year and date_obj.month == pred_month: future_volumes.append(volumes[i]) break except: continue if len(future_volumes) < 3: return 0 # 증감율 계산 future_average = sum(future_volumes) / len(future_volumes) if base_month_volume > 0: growth_rate = ((future_average - base_month_volume) / base_month_volume) * 100 return min(max(growth_rate, -50), 100) return 0 except Exception as e: logger.error(f"예상 3개월 증감율 계산 오류: {e}") return 0 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) except Exception as e: logger.error(f"3년 증감율 계산 오류: {e}") return 0 def calculate_correct_growth_rate(volumes, dates): """작년 대비 증감율 계산""" if len(volumes) < 13: return 0 try: completed_year, completed_month = get_complete_month() this_year_volume = None last_year_volume = None for i, date_str in enumerate(dates): try: date_obj = datetime.strptime(date_str, "%Y-%m-%d") if date_obj.year == completed_year and date_obj.month == completed_month: this_year_volume = volumes[i] if date_obj.year == completed_year - 1 and date_obj.month == completed_month: last_year_volume = volumes[i] except: continue if this_year_volume is not None and last_year_volume is not None and last_year_volume > 0: growth_rate = ((this_year_volume - last_year_volume) / last_year_volume) * 100 return min(max(growth_rate, -50), 100) return 0 except Exception as e: logger.error(f"작년 대비 증감율 계산 오류: {e}") return 0 def generate_future_predictions_correct(volumes, dates, growth_rate): """미래 예측 생성 (호환성 유지)""" return generate_future_from_growth_rate(volumes, dates, *get_complete_month()) # ===== 차트 생성 함수들 ===== def create_enhanced_current_chart(volume_data, keyword): """향상된 현재 검색량 정보 차트 - PC vs 모바일 비율 포함""" total_vol = volume_data['총검색량'] pc_vol = volume_data['PC검색량'] mobile_vol = volume_data['모바일검색량'] # 검색량 수준 평가 if total_vol >= 100000: level_text = "높음 🔥" level_color = "#dc3545" elif total_vol >= 10000: level_text = "중간 📊" level_color = "#ffc107" elif total_vol > 0: level_text = "낮음 📉" level_color = "#6c757d" else: level_text = "데이터 없음 ⚠️" level_color = "#6c757d" # PC vs 모바일 비율 if total_vol > 0: pc_ratio = (pc_vol / total_vol) * 100 mobile_ratio = (mobile_vol / total_vol) * 100 else: pc_ratio = mobile_ratio = 0 return f"""
📊 트렌드 분석 시스템: 네이버 데이터랩 기반 정확한 검색량 분석
오류: {str(e)}
오류: {str(e)}
오류: {error_msg}