# app/utils/data_utils.py import json import hashlib from datetime import datetime from typing import Dict, List, Any, Optional def create_store_hash(store_id: str, store_name: str, region: str) -> str: """ 가게의 고유 해시를 생성합니다. Args: store_id: 가게 ID store_name: 가게명 region: 지역 Returns: 생성된 해시값 """ combined = f"{store_id}_{store_name}_{region}" return hashlib.md5(combined.encode('utf-8')).hexdigest() def combine_store_and_reviews(store_info: Dict[str, Any], reviews: List[Dict[str, Any]]) -> str: """ 가게 정보와 리뷰를 결합하여 JSON 문자열을 생성합니다. Args: store_info: 가게 정보 reviews: 리뷰 목록 Returns: 결합된 JSON 문자열 """ combined_data = { "store_info": store_info, "reviews": reviews, "review_summary": generate_review_summary(reviews), "combined_at": datetime.now().isoformat() } return json.dumps(combined_data, ensure_ascii=False, separators=(',', ':')) def generate_review_summary(reviews: List[Dict[str, Any]]) -> Dict[str, Any]: """ 리뷰 목록에서 요약 정보를 생성합니다. Args: reviews: 리뷰 목록 Returns: 리뷰 요약 정보 """ if not reviews: return { "total_reviews": 0, "average_rating": 0, "rating_distribution": {}, "total_likes": 0, "common_keywords": [] } # 기본 통계 계산 total_reviews = len(reviews) total_rating = sum(review.get('rating', 0) for review in reviews) average_rating = total_rating / total_reviews if total_reviews > 0 else 0 total_likes = sum(review.get('likes', 0) for review in reviews) # 별점 분포 계산 rating_distribution = {} for review in reviews: rating = review.get('rating', 0) rating_distribution[rating] = rating_distribution.get(rating, 0) + 1 # 공통 키워드 추출 (배지 기준) keyword_count = {} for review in reviews: badges = review.get('badges', []) for badge in badges: keyword_count[badge] = keyword_count.get(badge, 0) + 1 # 상위 10개 키워드 common_keywords = sorted(keyword_count.items(), key=lambda x: x[1], reverse=True)[:10] common_keywords = [keyword for keyword, count in common_keywords] return { "total_reviews": total_reviews, "average_rating": round(average_rating, 2), "rating_distribution": rating_distribution, "total_likes": total_likes, "common_keywords": common_keywords } def extract_text_for_embedding(store_info: Dict[str, Any], reviews: List[Dict[str, Any]]) -> str: """ Vector DB 임베딩을 위한 텍스트를 추출합니다. Args: store_info: 가게 정보 reviews: 리뷰 목록 Returns: 임베딩용 텍스트 """ # 가게 기본 정보 store_text = f"가게명: {store_info.get('place_name', '')}\n" store_text += f"카테고리: {store_info.get('category_name', '')}\n" store_text += f"주소: {store_info.get('address_name', '')}\n" # 리뷰 내용 요약 review_contents = [] review_keywords = [] for review in reviews[:20]: # 최근 20개 리뷰만 사용 content = review.get('content', '').strip() if content: review_contents.append(content) badges = review.get('badges', []) review_keywords.extend(badges) # 리뷰 텍스트 조합 if review_contents: store_text += f"리뷰 내용: {' '.join(review_contents[:10])}\n" # 최대 10개 리뷰 # 키워드 조합 if review_keywords: unique_keywords = list(set(review_keywords)) store_text += f"키워드: {', '.join(unique_keywords[:15])}\n" # 최대 15개 키워드 return store_text.strip() def create_metadata(store_info: Dict[str, Any], food_category: str, region: str) -> Dict[str, Any]: """ Vector DB용 메타데이터를 생성합니다. Args: store_info: 가게 정보 food_category: 음식 카테고리 region: 지역 Returns: 메타데이터 딕셔너리 """ return { "store_id": store_info.get('id', ''), "store_name": store_info.get('place_name', ''), "food_category": food_category, "region": region, "category_name": store_info.get('category_name', ''), "address": store_info.get('address_name', ''), "phone": store_info.get('phone', ''), "place_url": store_info.get('place_url', ''), "x": store_info.get('x', ''), # 좌표를 개별 키로 분리 "y": store_info.get('y', ''), # 좌표를 개별 키로 분리 "last_updated": datetime.now().isoformat() } def is_duplicate_store(metadata1: Dict[str, Any], metadata2: Dict[str, Any]) -> bool: """ 두 가게가 중복인지 확인합니다. Args: metadata1: 첫 번째 가게 메타데이터 metadata2: 두 번째 가게 메타데이터 Returns: 중복 여부 """ # Store ID 기준 확인 if metadata1.get('store_id') and metadata2.get('store_id'): return metadata1['store_id'] == metadata2['store_id'] # 가게명 + 주소 기준 확인 name1 = metadata1.get('store_name', '').strip() name2 = metadata2.get('store_name', '').strip() addr1 = metadata1.get('address', '').strip() addr2 = metadata2.get('address', '').strip() if name1 and name2 and addr1 and addr2: return name1 == name2 and addr1 == addr2 return False