129 lines
4.9 KiB
Python
129 lines
4.9 KiB
Python
# app/repositories/similar_mission_repository.py
|
|
"""
|
|
HealthSync AI 유사 사용자 미션 데이터 리포지토리
|
|
"""
|
|
from typing import Dict, Any, Optional, List
|
|
from app.repositories.queries.similar_mission_queries import SimilarMissionQueries
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class SimilarMissionRepository:
|
|
"""유사 사용자 미션 데이터 DB 조회 리포지토리"""
|
|
|
|
@staticmethod
|
|
def _get_db():
|
|
"""simple_db를 lazy import로 가져오기 (순환 import 방지)"""
|
|
from app.utils.database_utils import simple_db
|
|
return simple_db
|
|
|
|
@staticmethod
|
|
async def get_recent_mission_completions(user_ids: List[int]) -> List[Dict[str, Any]]:
|
|
"""유사 사용자들의 최근 24시간 미션 완료 이력 조회"""
|
|
try:
|
|
simple_db = SimilarMissionRepository._get_db()
|
|
|
|
if not user_ids:
|
|
return []
|
|
|
|
result = await simple_db.execute_query(
|
|
SimilarMissionQueries.GET_RECENT_MISSION_COMPLETIONS,
|
|
{"user_ids": user_ids}
|
|
)
|
|
|
|
logger.info(f"최근 미션 완료 이력 조회 성공 - user_count: {len(user_ids)}, "
|
|
f"completion_count: {len(result)}")
|
|
return result if result else []
|
|
|
|
except Exception as e:
|
|
logger.error(f"최근 미션 완료 이력 조회 실패 - user_ids: {user_ids}, error: {str(e)}")
|
|
raise Exception(f"최근 미션 완료 이력 조회 실패: {str(e)}")
|
|
|
|
@staticmethod
|
|
async def get_user_health_for_vector(user_id: int) -> Optional[Dict[str, Any]]:
|
|
"""벡터 생성을 위한 사용자 건강 정보 조회"""
|
|
try:
|
|
simple_db = SimilarMissionRepository._get_db()
|
|
result = await simple_db.execute_query(
|
|
SimilarMissionQueries.GET_USER_HEALTH_FOR_VECTOR,
|
|
{"user_id": user_id}
|
|
)
|
|
|
|
if result and len(result) > 0:
|
|
logger.info(f"사용자 건강 정보 조회 성공 - user_id: {user_id}")
|
|
return result[0]
|
|
|
|
logger.warning(f"사용자 건강 정보 없음 - user_id: {user_id}")
|
|
return None
|
|
|
|
except Exception as e:
|
|
logger.error(f"사용자 건강 정보 조회 실패 - user_id: {user_id}, error: {str(e)}")
|
|
raise Exception(f"사용자 건강 정보 조회 실패: {str(e)}")
|
|
|
|
@staticmethod
|
|
async def get_occupation_name(occupation_code: str) -> Optional[str]:
|
|
"""직업 코드로 직업명 조회"""
|
|
try:
|
|
simple_db = SimilarMissionRepository._get_db()
|
|
result = await simple_db.execute_query(
|
|
SimilarMissionQueries.GET_OCCUPATION_NAME,
|
|
{"occupation_code": occupation_code}
|
|
)
|
|
|
|
if result and len(result) > 0:
|
|
return result[0]["occupation_name"]
|
|
|
|
# 기본 직업명 매핑
|
|
occupation_mapping = {
|
|
"OFF001": "사무직",
|
|
"MED001": "의료진",
|
|
"EDU001": "교육직",
|
|
"ENG001": "엔지니어",
|
|
"SRV001": "서비스직"
|
|
}
|
|
return occupation_mapping.get(occupation_code, "기타")
|
|
|
|
except Exception as e:
|
|
logger.error(f"직업명 조회 실패 - occupation_code: {occupation_code}, error: {str(e)}")
|
|
return "기타"
|
|
|
|
@staticmethod
|
|
async def get_users_basic_info(user_ids: List[int]) -> List[Dict[str, Any]]:
|
|
"""여러 사용자의 기본 정보 조회"""
|
|
try:
|
|
simple_db = SimilarMissionRepository._get_db()
|
|
|
|
if not user_ids:
|
|
return []
|
|
|
|
result = await simple_db.execute_query(
|
|
SimilarMissionQueries.GET_USERS_BASIC_INFO,
|
|
{"user_ids": user_ids}
|
|
)
|
|
|
|
logger.info(f"사용자 기본 정보 조회 성공 - user_count: {len(user_ids)}, "
|
|
f"found_count: {len(result)}")
|
|
return result if result else []
|
|
|
|
except Exception as e:
|
|
logger.error(f"사용자 기본 정보 조회 실패 - user_ids: {user_ids}, error: {str(e)}")
|
|
raise Exception(f"사용자 기본 정보 조회 실패: {str(e)}")
|
|
|
|
|
|
|
|
@staticmethod
|
|
async def get_all_users_for_vector() -> List[Dict[str, Any]]:
|
|
"""벡터 처리를 위한 모든 사용자 데이터 조회"""
|
|
try:
|
|
simple_db = SimilarMissionRepository._get_db()
|
|
result = await simple_db.execute_query(
|
|
SimilarMissionQueries.GET_ALL_USERS_FOR_VECTOR
|
|
)
|
|
|
|
logger.info(f"전체 사용자 벡터 데이터 조회 성공 - count: {len(result)}")
|
|
return result if result else []
|
|
|
|
except Exception as e:
|
|
logger.error(f"전체 사용자 벡터 데이터 조회 실패 - error: {str(e)}")
|
|
raise Exception(f"전체 사용자 벡터 데이터 조회 실패: {str(e)}") |