# app/services/health_service.py """ HealthSync AI 건강 데이터 서비스 """ import asyncio from typing import Dict, Any, Optional from app.services.base_service import BaseService from app.utils.claude_client import ClaudeClient from app.config.prompts import get_health_diagnosis_prompt from app.repositories.health_repository import HealthRepository from app.exceptions import ( UserNotFoundException, HealthDataNotFoundException, DatabaseException, ClaudeAPIException ) class HealthService(BaseService): """건강 데이터 조회 서비스""" def __init__(self): super().__init__() self.claude_client = ClaudeClient() self.health_repository = HealthRepository() async def get_latest_health_checkup(self, user_id: int) -> Dict[str, Any]: """사용자의 최신 건강검진 데이터 조회""" try: self.log_operation("get_latest_health_checkup", user_id=user_id) # 실제 DB에서 데이터 조회 health_data = await self.health_repository.get_latest_health_checkup_by_user_id(user_id) if not health_data: raise HealthDataNotFoundException(user_id) # BMI 계산 (DB에 없는 경우) if not health_data.get("bmi") and health_data.get("height") and health_data.get("weight"): height_m = health_data["height"] / 100 health_data["bmi"] = round(health_data["weight"] / (height_m ** 2), 1) self.log_operation("get_latest_health_checkup_success", user_id=user_id, checkup_year=health_data.get("reference_year")) return health_data except (UserNotFoundException, HealthDataNotFoundException): # 사용자 정의 예외는 그대로 전파 raise except Exception as e: self.logger.error(f"건강검진 데이터 조회 실패 - user_id: {user_id}, error: {str(e)}") raise DatabaseException(f"건강검진 데이터 조회 중 오류가 발생했습니다: {str(e)}") async def get_user_basic_info(self, user_id: int) -> Dict[str, Any]: """사용자 기본 정보 조회 (직업, 나이 등)""" try: self.log_operation("get_user_basic_info", user_id=user_id) # 실제 DB에서 데이터 조회 user_data = await self.health_repository.get_user_basic_info_by_id(user_id) if not user_data: raise UserNotFoundException(user_id) # 생년월일에서 나이 계산 (DB 쿼리에서 계산하지만 추가 검증) if not user_data.get("age") and user_data.get("birth_date"): from datetime import datetime birth_date = user_data["birth_date"] if isinstance(birth_date, str): birth_date = datetime.strptime(birth_date, "%Y-%m-%d").date() today = datetime.now().date() user_data["age"] = today.year - birth_date.year - ( (today.month, today.day) < (birth_date.month, birth_date.day)) self.log_operation("get_user_basic_info_success", user_id=user_id, occupation=user_data.get("occupation")) return user_data except UserNotFoundException: # 사용자 정의 예외는 그대로 전파 raise except Exception as e: self.logger.error(f"사용자 기본정보 조회 실패 - user_id: {user_id}, error: {str(e)}") raise DatabaseException(f"사용자 정보 조회 중 오류가 발생했습니다: {str(e)}") async def get_combined_user_health_data(self, user_id: int) -> Dict[str, Any]: """사용자 기본정보와 건강검진 데이터를 결합하여 반환""" try: # 병렬로 데이터 조회 user_data, health_data = await asyncio.gather( self.get_user_basic_info(user_id), self.get_latest_health_checkup(user_id) ) # 데이터 결합 combined_data = { "user_info": user_data, "health_data": health_data } self.log_operation("get_combined_user_health_data", user_id=user_id, has_user_data=bool(user_data), has_health_data=bool(health_data)) return combined_data except (UserNotFoundException, HealthDataNotFoundException, DatabaseException): # 사용자 정의 예외는 그대로 전파 raise except Exception as e: self.logger.error(f"사용자 종합 데이터 조회 실패 - user_id: {user_id}, error: {str(e)}") raise DatabaseException(f"사용자 종합 데이터 조회 중 오류가 발생했습니다: {str(e)}") async def get_three_sentence_diagnosis(self, user_id: int) -> str: """사용자 건강검진 데이터 기반 3줄 요약 진단""" try: self.log_operation("get_three_sentence_diagnosis_start", user_id=user_id) # 1. 사용자 데이터 조회 combined_data = await self.get_combined_user_health_data(user_id) user_data = combined_data["user_info"] health_data = combined_data["health_data"] # 2. 프롬프트 생성 prompt = self._build_diagnosis_prompt(user_data, health_data) # 3. Claude API 호출 claude_response = await self.claude_client.call_claude_api(prompt) # 4. 응답 정제 (앞뒤 공백 제거) diagnosis = claude_response.strip() self.log_operation("get_three_sentence_diagnosis_success", user_id=user_id, diagnosis_length=len(diagnosis)) return diagnosis except (UserNotFoundException, HealthDataNotFoundException, DatabaseException): # 사용자 정의 예외는 그대로 전파 raise except Exception as e: # Claude API 오류 self.logger.error(f"3줄 요약 진단 실패 - user_id: {user_id}, error: {str(e)}") raise ClaudeAPIException(f"건강 진단 분석 중 오류가 발생했습니다: {str(e)}") async def get_combined_user_health_data(self, user_id: int) -> Dict[str, Any]: """사용자 기본정보와 건강검진 데이터를 결합하여 반환""" try: # 병렬로 데이터 조회 user_data, health_data = await asyncio.gather( self.get_user_basic_info(user_id), self.get_latest_health_checkup(user_id) ) # 데이터 결합 combined_data = { "user_info": user_data, "health_data": health_data } self.log_operation("get_combined_user_health_data", user_id=user_id, has_user_data=bool(user_data), has_health_data=bool(health_data)) return combined_data except Exception as e: self.logger.error(f"사용자 종합 데이터 조회 실패 - user_id: {user_id}, error: {str(e)}") raise Exception(f"사용자 종합 데이터 조회 실패: {str(e)}") async def get_three_sentence_diagnosis(self, user_id: int) -> str: """사용자 건강검진 데이터 기반 3줄 요약 진단""" try: self.log_operation("get_three_sentence_diagnosis_start", user_id=user_id) # 1. 사용자 데이터 조회 combined_data = await self.get_combined_user_health_data(user_id) user_data = combined_data["user_info"] health_data = combined_data["health_data"] # 2. 프롬프트 생성 prompt = self._build_diagnosis_prompt(user_data, health_data) # 3. Claude API 호출 claude_response = await self.claude_client.call_claude_api(prompt) # 4. 응답 정제 (앞뒤 공백 제거) diagnosis = claude_response.strip() self.log_operation("get_three_sentence_diagnosis_success", user_id=user_id, diagnosis_length=len(diagnosis)) return diagnosis except Exception as e: self.logger.error(f"3줄 요약 진단 실패 - user_id: {user_id}, error: {str(e)}") raise Exception(f"건강 진단 분석 실패: {str(e)}") def _build_diagnosis_prompt(self, user_data: Dict[str, Any], health_data: Dict[str, Any]) -> str: """사용자 데이터를 기반으로 진단 프롬프트 생성""" try: prompt_template = get_health_diagnosis_prompt() # BMI 계산 bmi = health_data.get("bmi", 0) if not bmi and health_data.get("height") and health_data.get("weight"): height_m = health_data["height"] / 100 bmi = round(health_data["weight"] / (height_m ** 2), 1) # 프롬프트에 데이터 매핑 formatted_prompt = prompt_template.format( # 사용자 기본 정보 occupation=user_data.get("occupation", "정보 없음"), name=user_data.get("name", "정보 없음"), age=user_data.get("age", health_data.get("age", "정보 없음")), # 신체 정보 height=health_data.get("height", "정보 없음"), weight=health_data.get("weight", "정보 없음"), bmi=bmi if bmi else "정보 없음", waist_circumference=health_data.get("waist_circumference", "정보 없음"), # 혈압 및 혈당 systolic_bp=health_data.get("systolic_bp", "정보 없음"), diastolic_bp=health_data.get("diastolic_bp", "정보 없음"), fasting_glucose=health_data.get("fasting_glucose", "정보 없음"), # 콜레스테롤 total_cholesterol=health_data.get("total_cholesterol", "정보 없음"), hdl_cholesterol=health_data.get("hdl_cholesterol", "정보 없음"), ldl_cholesterol=health_data.get("ldl_cholesterol", "정보 없음"), triglyceride=health_data.get("triglyceride", "정보 없음"), # 기타 혈액 검사 hemoglobin=health_data.get("hemoglobin", "정보 없음"), serum_creatinine=health_data.get("serum_creatinine", "정보 없음"), ast=health_data.get("ast", "정보 없음"), alt=health_data.get("alt", "정보 없음"), gamma_gtp=health_data.get("gamma_gtp", "정보 없음"), urine_protein=health_data.get("urine_protein", "정보 없음"), # 감각기관 visual_acuity_left=health_data.get("visual_acuity_left", "정보 없음"), visual_acuity_right=health_data.get("visual_acuity_right", "정보 없음"), hearing_left=health_data.get("hearing_left", "정보 없음"), hearing_right=health_data.get("hearing_right", "정보 없음"), # 생활습관 smoking_status=self._convert_smoking_status(health_data.get("smoking_status")), drinking_status=self._convert_drinking_status(health_data.get("drinking_status")) ) self.log_operation("build_diagnosis_prompt", user_id="N/A", prompt_length=len(formatted_prompt)) return formatted_prompt except Exception as e: self.logger.error(f"진단 프롬프트 생성 실패: {str(e)}") raise Exception(f"진단 프롬프트 생성 실패: {str(e)}") def _convert_smoking_status(self, status: int) -> str: """흡연 상태 코드를 텍스트로 변환""" smoking_map = { 0: "비흡연", 1: "과거 흡연", 2: "현재 흡연" } return smoking_map.get(status, "정보 없음") def _convert_drinking_status(self, status: int) -> str: """음주 상태 코드를 텍스트로 변환""" drinking_map = { 0: "비음주", 1: "음주" } return drinking_map.get(status, "정보 없음")