{settings.APP_DESCRIPTION}
POST /find-reviews - 리뷰 검색 및 Vector DB 저장 (본인 가게 우선)
{{
"region": "서울특별시 강남구 역삼동",
"store_name": "맛있는 한식당"
}}
POST /build-vector - Vector DB 구축
{{
"region": "서울특별시 강남구 역삼동",
"store_name": "맛있는 한식당",
"force_rebuild": false
}}
POST /action-recommendation - 액션 추천 요청
{{
"store_id": "12345",
"context": "매출이 감소하고 있어서 개선이 필요합니다"
}}
"""
@app.post("/find-reviews", response_model=FindReviewsResponse)
async def find_reviews(
request: FindReviewsRequest,
vector_service: VectorService = Depends(get_vector_service),
restaurant_service: RestaurantService = Depends(get_restaurant_service),
review_service: ReviewService = Depends(get_review_service)
):
"""
지역과 가게명으로 리뷰를 찾아 Vector DB에 저장합니다.
🔥 본인 가게 리뷰는 반드시 포함됩니다. (수정된 버전)
"""
start_time = datetime.now()
logger.info(f"🔍 리뷰 검색 요청: {request.region} - {request.store_name}")
try:
# 1단계: 본인 가게 검색
logger.info("1단계: 본인 가게 검색 중... (최우선)")
target_restaurant = await restaurant_service.find_store_by_name_and_region(
request.region, request.store_name
)
if not target_restaurant:
logger.error(f"❌ 본인 가게를 찾을 수 없음: {request.store_name}")
raise HTTPException(
status_code=404,
detail=f"'{request.store_name}' 가게를 찾을 수 없습니다. 가게명과 지역을 정확히 입력해주세요."
)
logger.info(f"✅ 본인 가게 발견: {target_restaurant.place_name} (ID: {target_restaurant.id})")
# 2단계: 동종 업체 검색
logger.info("2단계: 동종 업체 검색 중...")
similar_restaurants = []
food_category = "기타" # 기본값
try:
food_category = extract_food_category(target_restaurant.category_name)
logger.info(f"추출된 음식 카테고리: {food_category}")
similar_restaurants = await restaurant_service.find_similar_stores(
request.region, food_category, settings.MAX_RESTAURANTS_PER_CATEGORY
)
logger.info(f"✅ 동종 업체 {len(similar_restaurants)}개 발견")
except Exception as e:
logger.warning(f"⚠️ 동종 업체 검색 실패 (본인 가게는 계속 진행): {e}")
# 3단계: 전체 가게 목록 구성 (본인 가게 우선 + 중복 제거)
logger.info("3단계: 전체 가게 목록 구성 중...")
# 본인 가게를 첫 번째로 배치
all_restaurants = [target_restaurant]
# 동종 업체 추가 (개선된 중복 제거)
for restaurant in similar_restaurants:
if not _is_duplicate_restaurant(target_restaurant, restaurant):
all_restaurants.append(restaurant)
logger.info(f"✅ 전체 가게 목록 구성 완료: {len(all_restaurants)}개 (본인 가게 포함)")
# 4단계: 전체 리뷰 수집 (본인 가게 우선 처리)
logger.info("4단계: 리뷰 수집 중... (본인 가게 우선)")
# 본인 가게 우선 처리를 위한 특별 로직
review_results = []
# 4-1: 본인 가게 리뷰 수집 (실패 시 전체 중단)
try:
logger.info("본인 가게 리뷰 우선 수집 중... (필수)")
target_store_info, target_reviews = await review_service.collect_store_reviews(
target_restaurant.id,
max_reviews=settings.MAX_REVIEWS_PER_RESTAURANT * 2 # 본인 가게는 더 많이
)
if not target_store_info or not target_reviews:
logger.error(f"❌ 본인 가게 리뷰 수집 실패: {target_restaurant.place_name}")
raise HTTPException(
status_code=422,
detail=f"본인 가게 '{target_restaurant.place_name}'의 리뷰를 수집할 수 없습니다."
)
# 본인 가게 결과를 첫 번째로 설정
review_results.append((target_restaurant.id, target_store_info, target_reviews))
logger.info(f"✅ 본인 가게 리뷰 수집 성공: {len(target_reviews)}개")
except HTTPException:
raise
except Exception as e:
logger.error(f"❌ 본인 가게 리뷰 수집 중 오류: {e}")
raise HTTPException(
status_code=500,
detail=f"본인 가게 리뷰 수집 중 오류가 발생했습니다: {str(e)}"
)
# 4-2: 동종 업체 리뷰 수집 (실패해도 본인 가게는 유지)
if len(all_restaurants) > 1: # 본인 가게 외에 다른 가게가 있는 경우
try:
logger.info(f"동종 업체 리뷰 수집 중... ({len(all_restaurants) - 1}개)")
# 본인 가게 제외한 동종 업체만 수집
similar_restaurants_only = all_restaurants[1:]
similar_results = await review_service.collect_multiple_stores_reviews(similar_restaurants_only)
# 동종 업체 결과 추가
review_results.extend(similar_results)
similar_reviews_count = sum(len(reviews) for _, _, reviews in similar_results)
logger.info(f"✅ 동종 업체 리뷰 수집 완료: {len(similar_results)}개 가게, {similar_reviews_count}개 리뷰")
except Exception as e:
logger.warning(f"⚠️ 동종 업체 리뷰 수집 실패 (본인 가게는 유지): {e}")
else:
logger.info("동종 업체가 없어 본인 가게 리뷰만 사용")
# 5단계: Vector DB 구축
logger.info("5단계: Vector DB 구축 중...")
try:
# 대상 가게 정보를 딕셔너리로 변환
target_store_info_dict = {
'id': target_restaurant.id,
'place_name': target_restaurant.place_name,
'category_name': target_restaurant.category_name,
'address_name': target_restaurant.address_name,
'phone': target_restaurant.phone,
'place_url': target_restaurant.place_url,
'x': target_restaurant.x,
'y': target_restaurant.y
}
# Vector DB에 저장
vector_result = await vector_service.build_vector_store(
target_store_info_dict, review_results, food_category, request.region
)
if not vector_result.get('success', False):
raise Exception(f"Vector DB 저장 실패: {vector_result.get('error', 'Unknown error')}")
logger.info("✅ Vector DB 구축 완료")
except Exception as e:
logger.error(f"❌ Vector DB 구축 실패: {e}")
raise HTTPException(
status_code=500,
detail=f"Vector DB 구축 중 오류가 발생했습니다: {str(e)}"
)
# 최종 검증: 본인 가게가 첫 번째에 있는지 확인
if not review_results or review_results[0][0] != target_restaurant.id:
logger.error("❌ 본인 가게가 첫 번째에 없음")
raise HTTPException(
status_code=500,
detail="본인 가게 리뷰 처리 순서 오류가 발생했습니다."
)
# 성공 응답
total_reviews = sum(len(reviews) for _, _, reviews in review_results)
execution_time = (datetime.now() - start_time).total_seconds()
return FindReviewsResponse(
success=True,
message=f"✅ 본인 가게 리뷰 포함 보장 완료! (총 {len(review_results)}개 가게, {total_reviews}개 리뷰)",
target_store=RestaurantInfo(
id=target_restaurant.id,
place_name=target_restaurant.place_name,
category_name=target_restaurant.category_name,
address_name=target_restaurant.address_name,
phone=target_restaurant.phone,
place_url=target_restaurant.place_url,
x=target_restaurant.x,
y=target_restaurant.y
),
total_stores=len(review_results),
total_reviews=total_reviews,
food_category=food_category,
region=request.region
)
except HTTPException:
raise
except Exception as e:
execution_time = (datetime.now() - start_time).total_seconds()
logger.error(f"❌ 전체 프로세스 실패: {e}")
raise HTTPException(
status_code=500,
detail=f"서비스 처리 중 예상치 못한 오류가 발생했습니다: {str(e)}"
)
def _is_duplicate_restaurant(restaurant1: RestaurantInfo, restaurant2: RestaurantInfo) -> bool:
"""
두 음식점이 중복인지 확인 (개선된 로직)
Args:
restaurant1: 첫 번째 음식점
restaurant2: 두 번째 음식점
Returns:
중복 여부
"""
# 1. ID 기준 확인
if restaurant1.id == restaurant2.id:
return True
# 2. place_url에서 추출한 store_id 기준 확인
store_id1 = _extract_store_id_from_place_url(restaurant1.place_url)
store_id2 = _extract_store_id_from_place_url(restaurant2.place_url)
if store_id1 and store_id2 and store_id1 == store_id2:
return True
# 3. restaurant.id와 place_url store_id 교차 확인
if restaurant1.id == store_id2 or restaurant2.id == store_id1:
return True
# 4. 이름 + 주소 기준 확인 (최후 방법)
if (restaurant1.place_name == restaurant2.place_name and
restaurant1.address_name == restaurant2.address_name):
return True
return False
def _extract_store_id_from_place_url(place_url: str) -> Optional[str]:
"""
카카오맵 URL에서 store_id를 추출합니다.
Args:
place_url: 카카오맵 장소 URL
Returns:
추출된 store_id 또는 None
"""
try:
if not place_url:
return None
import re
# URL 패턴: https://place.map.kakao.com/123456789
pattern = r'/(\d+)(?:\?|$|#)'
match = re.search(pattern, place_url)
if match:
return match.group(1)
else:
return None
except Exception:
return None
@app.post(
"/action-recommendation",
response_model=ActionRecommendationResponse,
summary="액션 추천 요청",
description="점주가 Claude AI에게 액션 추천을 요청합니다."
)
async def action_recommendation(
request: ActionRecommendationRequest,
claude_service: ClaudeService = Depends(get_claude_service),
vector_service: VectorService = Depends(get_vector_service)
):
"""🧠 Claude AI 액션 추천 API"""
try:
logger.info(f"액션 추천 요청: store_id={request.store_id}")
start_time = datetime.now()
# 1단계: Vector DB에서 컨텍스트 조회
try:
db_status = vector_service.get_db_status()
if db_status.get('total_documents', 0) == 0:
raise HTTPException(
status_code=404,
detail={
"success": False,
"error": "NO_VECTOR_DATA",
"message": "Vector DB에 데이터가 없습니다. 먼저 /build-vector API를 호출하여 데이터를 구축해주세요.",
"timestamp": datetime.now().isoformat()
}
)
# Vector DB에서 유사한 케이스 검색
context_data = vector_service.search_similar_cases(request.store_id, request.context)
except HTTPException:
raise
except Exception as e:
logger.error(f"Vector DB 조회 실패: {e}")
# Vector DB 조회 실패해도 일반적인 추천은 제공
context_data = None
# 2단계: Claude AI 호출 (프롬프트 구성부터 파싱까지 모두 포함)
try:
# 컨텍스트 구성
full_context = f"가게 ID: {request.store_id}\n점주 요청: {request.context}"
additional_context = context_data if context_data else None
# Claude AI 액션 추천 생성 (완전한 처리)
claude_response, parsed_response = await claude_service.generate_action_recommendations(
context=full_context,
additional_context=additional_context
)
if not claude_response:
raise Exception("Claude AI로부터 응답을 받지 못했습니다")
logger.info(f"Claude 응답 길이: {len(claude_response)} 문자")
json_parse_success = parsed_response is not None
except Exception as e:
logger.error(f"Claude AI 호출 실패: {e}")
raise HTTPException(
status_code=500,
detail=f"AI 추천 생성 중 오류: {str(e)}"
)
# 3단계: 응답 구성
claude_execution_time = (datetime.now() - start_time).total_seconds()
# 가게 정보 추출 (Vector DB에서)
store_name = request.store_id # 기본값
food_category = "기타" # 기본값
try:
store_context = vector_service.get_store_context(request.store_id)
if store_context:
store_name = store_context.get('store_name', request.store_id)
food_category = store_context.get('food_category', '기타')
except Exception as e:
logger.warning(f"가게 정보 추출 실패: {e}")
response = ActionRecommendationResponse(
success=True,
message=f"액션 추천이 완료되었습니다. (실행시간: {claude_execution_time:.1f}초, JSON 파싱: {'성공' if json_parse_success else '실패'})",
claude_input=full_context + (f"\n--- 동종 업체 분석 데이터 ---\n{additional_context}" if additional_context else ""),
claude_response=claude_response,
parsed_response=parsed_response,
store_name=store_name,
food_category=food_category,
similar_stores_count=len(context_data.split("---")) if context_data else 0,
execution_time=claude_execution_time,
json_parse_success=json_parse_success
)
logger.info(f"✅ 액션 추천 완료: Claude 응답 {len(claude_response) if claude_response else 0} 문자, JSON 파싱 {'성공' if json_parse_success else '실패'}, {claude_execution_time:.1f}초 소요")
return response
except HTTPException:
raise
except Exception as e:
logger.error(f"❌ 액션 추천 요청 실패: {str(e)}")
raise HTTPException(
status_code=500,
detail={
"success": False,
"error": "RECOMMENDATION_FAILED",
"message": f"액션 추천 중 오류가 발생했습니다: {str(e)}",
"timestamp": datetime.now().isoformat()
}
)
@app.get(
"/vector-status",
response_model=VectorDBStatusResponse,
summary="Vector DB 상태 조회",
description="Vector DB의 현재 상태를 조회합니다."
)
async def get_vector_db_status(vector_service: VectorService = Depends(get_vector_service)):
"""Vector DB 상태 조회 API"""
try:
status_info = vector_service.get_db_status()
status = VectorDBStatus(
collection_name=status_info['collection_name'],
total_documents=status_info['total_documents'],
total_stores=status_info['total_stores'],
db_path=status_info['db_path'],
last_updated=datetime.now().isoformat()
)
return VectorDBStatusResponse(
success=True,
status=status,
message="Vector DB 상태 조회가 완료되었습니다."
)
except Exception as e:
logger.error(f"Vector DB 상태 조회 실패: {e}")
raise HTTPException(
status_code=500,
detail={
"success": False,
"error": "STATUS_CHECK_FAILED",
"message": f"Vector DB 상태 조회 중 오류가 발생했습니다: {str(e)}",
"timestamp": datetime.now().isoformat()
}
)
@app.get("/health", summary="헬스 체크", description="API 서버 및 외부 서비스 상태를 확인합니다.")
async def health_check():
"""🏥 헬스체크 API"""
health_result = {
"status": "healthy",
"timestamp": datetime.now().isoformat(),
"services": {},
"app_info": {
"name": settings.APP_TITLE,
"version": settings.APP_VERSION,
"startup_completed": app_state["startup_completed"]
}
}
# 서비스별 헬스체크
services_to_check = [
("restaurant_service", "restaurant_api"),
("review_service", "review_api"),
("claude_service", "claude_ai"),
("vector_service", "vector_db")
]
healthy_count = 0
total_checks = len(services_to_check)
for service_key, health_key in services_to_check:
try:
service = app_state.get(service_key)
if service is None:
health_result["services"][health_key] = "not_initialized"
continue
# 서비스별 헬스체크 메서드 호출
if hasattr(service, 'health_check'):
status = await service.health_check()
else:
status = True # 헬스체크 메서드가 없으면 초기화됐다고 가정
# Vector DB의 경우 상세 정보 추가
if health_key == "vector_db" and status:
try:
db_status = service.get_db_status()
health_result["vector_db_info"] = {
"total_documents": db_status.get('total_documents', 0),
"total_stores": db_status.get('total_stores', 0),
"db_path": db_status.get('db_path', '')
}
except:
pass
health_result["services"][health_key] = "healthy" if status else "unhealthy"
if status:
healthy_count += 1
except Exception as e:
logger.warning(f"헬스체크 실패 - {service_key}: {e}")
health_result["services"][health_key] = f"error: {str(e)}"
# 전체 상태 결정
if healthy_count == total_checks:
health_result["status"] = "healthy"
elif healthy_count > 0:
health_result["status"] = "degraded"
else:
health_result["status"] = "unhealthy"
# 요약 정보
health_result["summary"] = {
"healthy_services": healthy_count,
"total_services": total_checks,
"health_percentage": round((healthy_count / total_checks) * 100, 1)
}
# 초기화 에러가 있으면 포함
if app_state["initialization_errors"]:
health_result["initialization_errors"] = app_state["initialization_errors"]
# 환경 정보
health_result["environment"] = {
"python_version": sys.version.split()[0],
"fastapi_version": fastapi.__version__,
"is_k8s": hasattr(settings, 'IS_K8S_ENV') and settings.IS_K8S_ENV,
"claude_model": settings.CLAUDE_MODEL
}
# HTTP 상태 코드 결정
if health_result["status"] == "healthy":
return health_result
elif health_result["status"] == "degraded":
return JSONResponse(status_code=200, content=health_result) # 부분 장애는 200
else:
return JSONResponse(status_code=503, content=health_result) # 전체 장애는 503
# 🔧 전역 예외 처리
@app.exception_handler(Exception)
async def global_exception_handler(request, exc):
"""전역 예외 처리"""
logger.error(f"Unhandled exception: {exc}")
return JSONResponse(
status_code=500,
content={
"error": "Internal server error",
"detail": str(exc) if settings.LOG_LEVEL.lower() == "debug" else "An unexpected error occurred",
"timestamp": datetime.now().isoformat(),
"path": str(request.url)
}
)
if __name__ == "__main__":
import uvicorn
print("🍽️ " + "="*60)
print(f" {settings.APP_TITLE} 서버 시작")
print("="*64)
print(f"📊 구성 정보:")
print(f" - Python 버전: {sys.version.split()[0]}")
print(f" - FastAPI 버전: {fastapi.__version__}")
print(f" - Vector DB Path: {settings.VECTOR_DB_PATH}")
print(f" - Claude Model: {settings.CLAUDE_MODEL}")
print(f" - 환경: {'Kubernetes' if hasattr(settings, 'IS_K8S_ENV') and settings.IS_K8S_ENV else 'Local'}")
print()
print(f"📚 문서:")
print(f" - Swagger UI: http://{settings.HOST}:{settings.PORT}/docs")
print(f" - ReDoc: http://{settings.HOST}:{settings.PORT}/redoc")
print(f" - 메인 페이지: http://{settings.HOST}:{settings.PORT}/")
print()
try:
uvicorn.run(
"app.main:app", # 🔧 문자열로 지정 (리로드 지원)
host=settings.HOST,
port=settings.PORT,
log_level=settings.LOG_LEVEL.lower(),
reload=False, # 프로덕션에서는 False
access_log=True,
loop="uvloop" if sys.platform != "win32" else "asyncio"
)
except KeyboardInterrupt:
print("\n🛑 서버가 사용자에 의해 중단되었습니다.")
except Exception as e:
print(f"\n❌ 서버 시작 실패: {e}")
import traceback
traceback.print_exc()
sys.exit(1)