195 lines
6.2 KiB
Python
195 lines
6.2 KiB
Python
# 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.0,
|
|
"rating_distribution": {},
|
|
"common_keywords": [],
|
|
"sentiment_summary": {
|
|
"positive": 0,
|
|
"neutral": 0,
|
|
"negative": 0
|
|
}
|
|
}
|
|
|
|
# 기본 통계
|
|
total_reviews = len(reviews)
|
|
ratings = [review.get('rating', 0) for review in reviews if review.get('rating', 0) > 0]
|
|
average_rating = sum(ratings) / len(ratings) if ratings else 0.0
|
|
|
|
# 별점 분포
|
|
rating_distribution = {}
|
|
for rating in ratings:
|
|
rating_distribution[str(rating)] = rating_distribution.get(str(rating), 0) + 1
|
|
|
|
# 키워드 추출 (badges 기반)
|
|
keyword_counts = {}
|
|
for review in reviews:
|
|
badges = review.get('badges', [])
|
|
for badge in badges:
|
|
keyword_counts[badge] = keyword_counts.get(badge, 0) + 1
|
|
|
|
# 상위 키워드 추출
|
|
common_keywords = sorted(keyword_counts.items(), key=lambda x: x[1], reverse=True)[:10]
|
|
common_keywords = [keyword for keyword, count in common_keywords]
|
|
|
|
# 감정 분석 (간단한 별점 기반)
|
|
sentiment_summary = {
|
|
"positive": len([r for r in ratings if r >= 4]),
|
|
"neutral": len([r for r in ratings if r == 3]),
|
|
"negative": len([r for r in ratings if r <= 2])
|
|
}
|
|
|
|
return {
|
|
"total_reviews": total_reviews,
|
|
"average_rating": round(average_rating, 2),
|
|
"rating_distribution": rating_distribution,
|
|
"common_keywords": common_keywords,
|
|
"sentiment_summary": sentiment_summary,
|
|
"has_recent_reviews": any(
|
|
review.get('date', '') >= datetime.now().strftime('%Y.%m.%d')
|
|
for review in reviews[-10:] # 최근 10개 리뷰 확인
|
|
)
|
|
}
|
|
|
|
def extract_text_for_embedding(store_info: Dict[str, Any], reviews: List[Dict[str, Any]]) -> str:
|
|
"""
|
|
임베딩을 위한 텍스트를 추출합니다.
|
|
|
|
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
|