"""
동물약 추천 MVP 웹앱
- 반려동물 정보 + 증상 입력 → 약품 추천
- 실행: python pet_recommend_app.py
- 접속: http://localhost:5001
"""
from flask import Flask, render_template_string, request, jsonify
from db_setup import (
session, APDB, Inventory, ComponentCode, Symptoms, SymptomComponentMapping, DosageInfo,
SupplementaryProduct, UnifiedProductRecommendation, SymptomSupplementaryRecommendation, InventorySupplementary
)
from sqlalchemy import distinct, and_, or_
from concurrent.futures import ThreadPoolExecutor
import openai
import re
app = Flask(__name__)
# ============================================================
# OpenAI API 설정
# ============================================================
OPENAI_API_KEY = "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"
openai.api_key = OPENAI_API_KEY
# ============================================================
# MDR-1 유전자 변이 관련 설정
# ============================================================
# MDR-1 민감 성분 코드 (Avermectin 계열 등)
MDR1_SENSITIVE_COMPONENTS = {
'IC2010131', # 셀라멕틴 (Selamectin)
'IC2030110', # 이버멕틴+피란텔 (Ivermectin + Pyrantel)
'IC2030132', # 이미다클로프리드+목시덱틴 (Imidacloprid + Moxidectin)
}
# MDR-1 민감 견종 정보
MDR1_BREEDS = {
# 고위험군 (40%+)
'collie': {'name': '콜리', 'risk': 'high', 'frequency': '70%'},
'aussie': {'name': '오스트레일리안 셰퍼드', 'risk': 'high', 'frequency': '50%'},
'mini_aussie': {'name': '미니어처 오스트레일리안 셰퍼드', 'risk': 'high', 'frequency': '50%'},
'longhair_whippet': {'name': '롱헤어 휘핏', 'risk': 'high', 'frequency': '65%'},
'silken_windhound': {'name': '실켄 윈드하운드', 'risk': 'high', 'frequency': '30-50%'},
# 중위험군 (10-40%)
'sheltie': {'name': '셰틀랜드 쉽독', 'risk': 'medium', 'frequency': '15%'},
'english_shepherd': {'name': '잉글리쉬 셰퍼드', 'risk': 'medium', 'frequency': '15%'},
'whippet': {'name': '휘핏', 'risk': 'medium', 'frequency': '10-20%'},
'mcnab': {'name': '맥냅', 'risk': 'medium', 'frequency': '17%'},
'german_shepherd': {'name': '저먼 셰퍼드', 'risk': 'medium', 'frequency': '10%'},
'white_shepherd': {'name': '화이트 셰퍼드', 'risk': 'medium', 'frequency': '14%'},
# 저위험군 (10% 미만)
'old_english': {'name': '올드 잉글리쉬 쉽독', 'risk': 'low', 'frequency': '5%'},
'border_collie': {'name': '보더 콜리', 'risk': 'low', 'frequency': '2-5%'},
'chinook': {'name': '치눅', 'risk': 'low', 'frequency': '미확인'},
# 기타
'mix_herding': {'name': '믹스견 (목양견 계통)', 'risk': 'unknown', 'frequency': '변동'},
'mix_hound': {'name': '믹스견 (하운드 계통)', 'risk': 'unknown', 'frequency': '변동'},
'other': {'name': '기타 견종', 'risk': 'none', 'frequency': '-'},
'unknown': {'name': '모름', 'risk': 'unknown', 'frequency': '-'},
}
# ============================================================
# 제품군(카테고리) 정의
# ============================================================
PRODUCT_CATEGORIES = {
'heartworm': {
'name': '심장사상충제',
'icon': '💗',
'keywords': ['심장사상충', '사상충', 'dirofilaria', '하트웜'],
'description': '심장사상충 예방약'
},
'parasite': {
'name': '구충제',
'icon': '🐛',
'keywords': ['구충', '기생충', '내부구충', '촌충', '회충', '십이지장충', '선충', '벼룩', '진드기'],
'description': '내/외부 기생충 치료'
},
'skin': {
'name': '피부약',
'icon': '🧴',
'keywords': ['피부', '진균', '피부염', '알레르기', '가려움', '습진', '곰팡이', '아토피'],
'description': '피부질환 치료'
},
'ear': {
'name': '귀약',
'icon': '👂',
'keywords': ['귀', '외이도염', '귓병', '이염', '이진드기'],
'description': '귀질환 치료'
},
'eye': {
'name': '안약',
'icon': '👁️',
'keywords': ['안약', '눈', '결막염', '각막염', '안구'],
'description': '눈질환 치료'
},
'antibiotic': {
'name': '항생제',
'icon': '💊',
'keywords': ['항생', '감염', '세균', '항균'],
'description': '세균감염 치료'
},
'digestive': {
'name': '정장제',
'icon': '🍀',
'keywords': ['정장', '장내세균', '설사', '유산균', '프로바이오틱', '소화불량', '소화장애'],
'description': '소화기 건강'
},
'painkiller': {
'name': '소염진통제',
'icon': '💪',
'keywords': ['소염', '진통', '관절', '염증', '통증'],
'description': '통증/염증 완화'
},
'vomit': {
'name': '구토억제제',
'icon': '🤢',
'keywords': ['구토', '메스꺼움', '오심'],
'description': '구토 억제'
},
}
# ============================================================
# 체중 기반 용량 계산 함수
# ============================================================
def format_tablets(tablets):
"""소수점 정제 수를 사람이 읽기 쉬운 형태로 변환"""
if tablets is None:
return None
if tablets < 0.3:
return "1/4정"
elif tablets < 0.6:
return "1/2정"
elif tablets < 0.9:
return "3/4정"
elif tablets < 1.2:
return "1정"
elif tablets < 1.6:
return "1.5정"
elif tablets < 2.2:
return "2정"
elif tablets < 2.6:
return "2.5정"
elif tablets < 3.2:
return "3정"
else:
return f"{round(tablets)}정"
def get_tablets_numeric(tablets):
"""
format_tablets의 역방향 함수 - 정제 수를 숫자로 반환
1/4정 → 0.25, 1/2정 → 0.5, 3/4정 → 0.75, 1정 → 1.0, 1.5정 → 1.5 등
"""
if tablets < 0.3:
return 0.25 # 1/4정
elif tablets < 0.6:
return 0.5 # 1/2정
elif tablets < 0.9:
return 0.75 # 3/4정
elif tablets < 1.2:
return 1.0 # 1정
elif tablets < 1.6:
return 1.5 # 1.5정
elif tablets < 2.2:
return 2.0 # 2정
elif tablets < 2.6:
return 2.5 # 2.5정
elif tablets < 3.2:
return 3.0 # 3정
else:
return round(tablets)
def calculate_recommended_dosage(product_idx, weight_kg, animal_type):
"""
체중 기반 적정 용량 계산
Args:
product_idx: APDB.idx (제품 고유 ID)
weight_kg: 반려동물 체중 (kg)
animal_type: 'dog' 또는 'cat'
Returns:
dict: 계산된 용량 정보 또는 None
"""
if not weight_kg or weight_kg <= 0:
return None
try:
# 먼저 이 제품의 모든 DosageInfo 조회 (피펫형 여부 판단용)
all_dosage_infos = session.query(DosageInfo).filter(
DosageInfo.apdb_idx == product_idx,
or_(
DosageInfo.animal_type == animal_type,
DosageInfo.animal_type == 'all'
)
).all()
# 피펫형 제품 여부 판단 (DosageInfo가 하나라도 피펫이면 피펫형)
is_pipette = any(d.unit_type == '피펫' for d in all_dosage_infos)
# 체중 범위에 맞는 DosageInfo 찾기
dosage_info = None
for d in all_dosage_infos:
# 체중 범위가 지정된 경우
if d.weight_min_kg is not None and d.weight_max_kg is not None:
if d.weight_min_kg <= weight_kg <= d.weight_max_kg:
dosage_info = d
break
# 체중 범위가 없는 경우 (전체 적용)
elif d.weight_min_kg is None and d.weight_max_kg is None:
dosage_info = d
break
# 피펫형 제품인데 체중 범위에 맞지 않으면 필터링을 위해 특별 응답 반환
if is_pipette and not dosage_info:
return {
'calculated': False,
'is_pipette': True, # 피펫형이므로 필터링 대상
'message': '',
'details': {},
'warning': None
}
if not dosage_info:
return None
result = {
'calculated': False,
'is_pipette': is_pipette, # 피펫형 제품 여부 (체중 필터링에 사용)
'message': '',
'details': {},
'warning': None
}
# 케이스 1: kg당 용량이 있는 경우 (정제형 등)
if dosage_info.dose_per_kg:
total_dose = weight_kg * dosage_info.dose_per_kg
result['calculated'] = True
result['details'] = {
'total_dose': round(total_dose, 2),
'dose_unit': dosage_info.dose_unit,
'frequency': dosage_info.frequency,
'route': dosage_info.route
}
# 정제 수 계산 (정제/캡슐형인 경우)
if dosage_info.unit_dose and dosage_info.unit_type in ['정', '캡슐']:
tablets = total_dose / dosage_info.unit_dose
tablets_formatted = format_tablets(tablets)
# 실제 복용량 계산 (정제 수 × 1정당 함량)
tablets_numeric = get_tablets_numeric(tablets)
actual_dose = tablets_numeric * dosage_info.unit_dose
result['details']['tablets'] = round(tablets, 2)
result['details']['tablets_formatted'] = tablets_formatted
result['details']['unit_type'] = dosage_info.unit_type
result['details']['actual_dose'] = actual_dose # 실제 복용량
result['message'] = f"체중 {weight_kg}kg 기준: 1회 {tablets_formatted} ({round(actual_dose, 1)}{dosage_info.dose_unit})"
else:
result['message'] = f"체중 {weight_kg}kg 기준: 1회 {round(total_dose, 1)}{dosage_info.dose_unit}"
if dosage_info.frequency:
result['message'] += f", {dosage_info.frequency}"
# 케이스 2: 체중 범위별 고정 용량 (피펫형 등)
elif dosage_info.weight_min_kg and dosage_info.weight_max_kg:
result['calculated'] = True
result['details'] = {
'weight_min': dosage_info.weight_min_kg,
'weight_max': dosage_info.weight_max_kg,
'unit_dose': dosage_info.unit_dose,
'dose_unit': dosage_info.dose_unit,
'frequency': dosage_info.frequency,
'route': dosage_info.route
}
result['message'] = f"체중 {dosage_info.weight_min_kg}~{dosage_info.weight_max_kg}kg용"
if dosage_info.unit_dose and dosage_info.dose_unit:
result['message'] += f": {dosage_info.unit_dose}{dosage_info.dose_unit}"
if dosage_info.frequency:
result['message'] += f", {dosage_info.frequency}"
# 케이스 3: 기본 정보만 있는 경우
elif dosage_info.frequency or dosage_info.route:
result['calculated'] = True
parts = []
if dosage_info.frequency:
parts.append(dosage_info.frequency)
if dosage_info.route:
parts.append(f"{dosage_info.route} 투여")
result['message'] = ', '.join(parts)
return result
except Exception as e:
print(f"[용량 계산 오류] product_idx={product_idx}, error={e}")
return None
def generate_recommendation_reason(animal_type, symptom_descriptions, product_name, component_name, llm_pharm, efficacy_clean, component_code=None, symptom_codes=None):
"""GPT-4o-mini로 추천 이유 생성"""
try:
# 한글 동물 타입
animal_ko = '강아지' if animal_type == 'dog' else '고양이'
# 특수 케이스: 아시엔로정 + 소화기 증상 → 중증 세균성 장염 설명 추가
special_note = ""
digestive_symptoms = {'g01', 'g02', 'g03', 'g04', 'g05'} # 소화기 관련 증상 코드
if component_code == 'IA5010124' and symptom_codes: # 아시엔로정 (엔로플록사신)
if any(s in digestive_symptoms for s in symptom_codes):
special_note = "\n⚠️ 특별 참고: 이 제품은 일반 설사가 아닌 중증 세균성 장염에 사용하는 항생제입니다. 증상이 심하거나 점액/혈변이 있는 경우에 적합합니다."
# 복합 성분 여부 확인 (+ 기호로 연결된 경우)
is_complex_formula = '+' in component_name
# llm_pharm에서 컨텍스트 추출
if llm_pharm:
indication = llm_pharm.get('어떤질병에사용하나요?', '')
description = llm_pharm.get('자연어설명', '') or llm_pharm.get('LLM설명', '')
usage = llm_pharm.get('기간/용법', '')
caution = llm_pharm.get('check', '')
else:
indication = ''
description = efficacy_clean[:200] if efficacy_clean else ''
usage = ''
caution = ''
context = f"""
제품명: {product_name}
성분: {component_name}
적응증: {indication}
설명: {description}
용법: {usage}
"""
# 특별 지시 사항
special_instruction = ""
# 복합 성분 제품일 경우 각 성분 역할 설명 요청
if is_complex_formula:
special_instruction += "\n중요: 이 제품은 여러 성분이 함께 들어간 '복합제'입니다. 각 성분의 역할을 간단히 설명해주세요 (예: 'A성분은 세균을 억제하고, B성분은 염증을 가라앉힙니다'). 단, '성분이 결합되어 있다'는 표현은 화학적 오해를 줄 수 있으니 사용하지 마세요."
# 아시엔로정 소화기 증상일 경우 프롬프트에 특별 지시 추가
if component_code == 'IA5010124' and symptom_codes and any(s in digestive_symptoms for s in symptom_codes):
special_instruction += "\n주의: 이 제품은 항생제로, 중증 세균성 장염(심한 염증성 장질환)에 사용됩니다. 일반적인 설사가 아닌 세균 감염에 의한 장염에 효과적이라는 점을 반드시 언급해주세요."
# 복합 성분 제품은 더 긴 설명 허용
sentence_limit = "3-4문장" if is_complex_formula else "2문장 이내"
prompt = f"""반려인이 {animal_ko}의 증상으로 "{', '.join(symptom_descriptions)}"를 선택했습니다.
아래 제품 정보를 바탕으로, 왜 이 제품이 해당 증상에 적합한지 {sentence_limit}로 친절하고 따뜻하게 설명해주세요.
예상 질환명을 언급하고, 제품의 효능을 간단히 설명해주세요.
전문용어는 쉽게 풀어서 설명해주세요.
주의사항:
- "귀하", "고객님" 같은 딱딱한 표현 대신 "반려인님" 또는 생략하세요.
- 복합 성분 제품은 "A와 B가 결합되어" 대신 "A와 B가 함께 들어있어" 또는 "A 성분과 B 성분이 각각" 표현을 사용하세요.{special_instruction}
{context}
"""
# 복합 성분 제품은 더 긴 응답 허용
max_tok = 280 if is_complex_formula else 120
client = openai.OpenAI(api_key=OPENAI_API_KEY)
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
max_tokens=max_tok,
temperature=0.7
)
gpt_response = response.choices[0].message.content.strip()
# 특별 참고사항 추가 (아시엔로정 + 소화기 증상)
if special_note:
return gpt_response + special_note
return gpt_response
except Exception as e:
print(f"OpenAI API Error: {e}")
# fallback: DB의 자연어설명 사용
if llm_pharm and llm_pharm.get('자연어설명'):
fallback_msg = llm_pharm.get('자연어설명')
if special_note:
return fallback_msg + special_note
return fallback_msg
return "해당 증상에 효과적인 성분이 포함된 제품입니다."
# ============================================================
# HTML 템플릿 (단일 파일로 포함)
# ============================================================
HTML_TEMPLATE = '''
동물약 추천 - 애니팜
'''
# ============================================================
# 라우트
# ============================================================
def get_product_image(product):
"""제품 이미지 URL 반환 (없으면 동일 성분 제품에서 fallback)"""
# 1순위: 고도몰 CDN 이미지
if product.godoimage_url_f:
return product.godoimage_url_f
# 2순위: ani.0bin.in 이미지
if product.image_url1:
return product.image_url1
# 3순위: 동일 성분의 다른 제품에서 이미지 가져오기
if product.component_code:
fallback = session.query(APDB).filter(
APDB.component_code == product.component_code,
APDB.idx != product.idx,
(APDB.godoimage_url_f != None) | (APDB.image_url1 != None)
).first()
if fallback:
return fallback.godoimage_url_f or fallback.image_url1
return None
@app.route('/')
def index():
"""메인 페이지"""
# 증상 목록 조회 (카테고리별)
symptoms = session.query(Symptoms).order_by(Symptoms.prefix, Symptoms.symptom_code).all()
symptoms_by_category = {}
category_names = {
'a': '👁️ 눈',
'b': '🦷 구강/치아',
'c': '👂 귀',
'd': '🐾 피부/털',
'e': '🦵 다리/발',
'f': '🦴 뼈/관절',
'g': '🍽️ 소화기/배변'
}
for symptom in symptoms:
category = category_names.get(symptom.prefix, symptom.prefix_description or symptom.prefix)
if category not in symptoms_by_category:
symptoms_by_category[category] = []
symptoms_by_category[category].append({
'code': symptom.symptom_code,
'description': symptom.symptom_description
})
return render_template_string(HTML_TEMPLATE, symptoms_by_category=symptoms_by_category)
@app.route('/api/categories', methods=['GET'])
def get_categories():
"""카테고리별 제품 수 조회 API"""
try:
animal_type = request.args.get('animal_type', 'dog')
animal_ko = '개' if animal_type == 'dog' else '고양이'
# 재고가 있는 제품 조회
inventory_products = session.query(APDB).join(
Inventory, Inventory.apdb_id == APDB.idx
).filter(
Inventory.transaction_type == 'INBOUND'
).distinct().all()
# 카테고리별 제품 수 계산
categories = []
for cat_code, cat_info in PRODUCT_CATEGORIES.items():
count = 0
keywords = cat_info.get('keywords', [])
for product in inventory_products:
# 동물 타입 체크 (ComponentCode에서)
component = session.query(ComponentCode).filter(
ComponentCode.code == product.component_code
).first()
if component:
target_animals = component.target_animals or []
# JSONB 리스트 형태 처리
if isinstance(target_animals, list):
target_str = ', '.join(target_animals).lower()
else:
target_str = str(target_animals).lower()
if animal_ko not in target_str and '개, 고양이' not in target_str and '고양이, 개' not in target_str:
continue
# 효능효과에서 키워드 매칭
efficacy = (product.efficacy_effect or '').lower()
# HTML 태그 제거
import re
efficacy_clean = re.sub(r'<[^>]+>', '', efficacy)
if any(kw in efficacy_clean for kw in keywords):
count += 1
if count > 0:
categories.append({
'code': cat_code,
'name': cat_info['name'],
'icon': cat_info['icon'],
'count': count,
'description': cat_info['description']
})
# 제품 수 기준 정렬
categories.sort(key=lambda x: x['count'], reverse=True)
return jsonify({
'success': True,
'categories': categories
})
except Exception as e:
print(f"카테고리 조회 오류: {e}")
import traceback
traceback.print_exc()
return jsonify({
'success': False,
'message': str(e),
'categories': []
})
@app.route('/api/recommend', methods=['POST'])
def recommend():
"""추천 API - 증상-성분 매핑 테이블 활용"""
try:
data = request.get_json()
animal_type = data.get('animal_type') # 'dog' or 'cat'
breed = data.get('breed') # 견종 (MDR-1 체크용)
weight = data.get('weight')
pregnancy = data.get('pregnancy', 'none') # 임신/수유 여부
symptom_codes = data.get('symptoms', [])
if not animal_type or not symptom_codes:
return jsonify({
'success': False,
'message': '동물 종류와 증상을 선택해주세요.'
})
# 한글 변환
animal_ko = '개' if animal_type == 'dog' else '고양이'
# 1. 선택된 증상 정보 조회
matched_symptoms = session.query(Symptoms).filter(
Symptoms.symptom_code.in_(symptom_codes)
).all()
symptom_descriptions = [s.symptom_description for s in matched_symptoms]
# 2. 증상-성분 매핑에서 관련 성분 코드 조회
mapped_components = session.query(SymptomComponentMapping.component_code).filter(
SymptomComponentMapping.symptom_code.in_(symptom_codes)
).distinct().all()
mapped_component_codes = [m[0] for m in mapped_components]
# 매핑된 성분이 없으면 추천 불가
if not mapped_component_codes:
return jsonify({
'success': True,
'animal_type': animal_type,
'matched_symptoms': symptom_descriptions,
'recommendations': [],
'total_count': 0,
'message': '선택하신 증상에 맞는 약품 매핑이 아직 없습니다. 수의사와 상담을 권장합니다.'
})
# 3. 재고 제품 조회 (INBOUND) + 매핑된 성분 필터
inventory_products = session.query(APDB).join(
Inventory, Inventory.apdb_id == APDB.idx
).filter(
Inventory.transaction_type == 'INBOUND',
APDB.component_code.in_(mapped_component_codes)
).distinct().all()
# 4. 동물 종류에 맞는 제품 필터링 + 점수 계산
recommendations = []
for product in inventory_products:
# ComponentCode에서 target_animals 확인
comp = session.query(ComponentCode).filter(
ComponentCode.code == product.component_code
).first()
if not comp:
continue
target_animals = comp.target_animals or []
# 대상 동물 체크 (target_animals가 없으면 전체 사용 가능으로 간주)
if target_animals and animal_ko not in target_animals:
continue # 해당 동물에게 사용 불가
# 이 성분이 몇 개의 선택 증상과 매핑되는지 점수 계산
matching_symptom_count = session.query(SymptomComponentMapping).filter(
SymptomComponentMapping.component_code == product.component_code,
SymptomComponentMapping.symptom_code.in_(symptom_codes)
).count()
# 효능효과에서 키워드 추출
efficacy = product.efficacy_effect or ''
efficacy_clean = re.sub(r'<[^>]+>', '', efficacy).lower()
# 용도 간단 설명 추출
efficacy_short = ''
if '외이도염' in efficacy_clean or '귓병' in efficacy_clean or '귀' in efficacy_clean:
efficacy_short = '귀 치료제'
elif '구토' in efficacy_clean:
efficacy_short = '구토 억제제'
elif '피부' in efficacy_clean or '진균' in efficacy_clean or '피부염' in efficacy_clean:
efficacy_short = '피부 치료제'
elif '관절' in efficacy_clean or '소염' in efficacy_clean or '진통' in efficacy_clean:
efficacy_short = '소염진통제'
elif '구충' in efficacy_clean or '기생충' in efficacy_clean or '진드기' in efficacy_clean:
efficacy_short = '구충제'
elif '항생' in efficacy_clean or '감염' in efficacy_clean:
efficacy_short = '항생제'
elif '장' in efficacy_clean or '설사' in efficacy_clean or '정장' in efficacy_clean:
efficacy_short = '정장제'
recommendations.append({
'idx': product.idx, # 추천 제품 API 호출용
'name': product.product_name,
'product_idx': product.idx, # 용량 계산용
'component_code': product.component_code, # MDR-1 체크용
'component_name': comp.component_name_ko or product.component_name_ko or '',
'target_animals': ', '.join(target_animals) if target_animals else '전체',
'efficacy': efficacy_short,
'score': matching_symptom_count, # 매핑 점수
'image_url': get_product_image(product), # 고도몰 CDN 이미지 (fallback 포함)
'llm_pharm': product.llm_pharm, # AI 추천 이유 생성용
'efficacy_clean': efficacy_clean, # AI 추천 이유 생성용 fallback
'usage_guide': product.usage_guide # 상세 사용 안내 (희석 비율 등)
})
# 점수순 정렬 (높은 점수 = 더 많은 증상과 매핑)
recommendations.sort(key=lambda x: x['score'], reverse=True)
# 중복 제거 (동일 제품명)
seen_names = set()
unique_recommendations = []
for rec in recommendations:
if rec['name'] not in seen_names:
seen_names.add(rec['name'])
unique_recommendations.append(rec)
# 상위 5개만
unique_recommendations = unique_recommendations[:5]
# AI 추천 이유 생성 (병렬 처리)
def generate_reason_for_product(rec):
reason = generate_recommendation_reason(
animal_type=animal_type,
symptom_descriptions=symptom_descriptions,
product_name=rec['name'],
component_name=rec['component_name'],
llm_pharm=rec.get('llm_pharm'),
efficacy_clean=rec.get('efficacy_clean', ''),
component_code=rec.get('component_code'),
symptom_codes=symptom_codes
)
return {**rec, 'reason': reason}
with ThreadPoolExecutor(max_workers=5) as executor:
unique_recommendations = list(executor.map(generate_reason_for_product, unique_recommendations))
# MDR-1 체크 (강아지 + 견종 선택된 경우)
if animal_type == 'dog' and breed and breed != 'other':
breed_info = MDR1_BREEDS.get(breed, {})
breed_risk = breed_info.get('risk', 'none')
breed_name = breed_info.get('name', '')
for rec in unique_recommendations:
component_code = rec.get('component_code', '')
if component_code in MDR1_SENSITIVE_COMPONENTS:
if breed_risk == 'high':
rec['mdr1_warning'] = 'danger'
rec['mdr1_message'] = f"⚠️ {breed_name}은(는) MDR-1 유전자 변이 고위험 견종입니다. 이 약물(이버멕틴/셀라멕틴 계열)은 심각한 신경독성 부작용을 유발할 수 있습니다. 반드시 수의사 상담 후 사용하세요."
elif breed_risk == 'medium':
rec['mdr1_warning'] = 'caution'
rec['mdr1_message'] = f"⚠️ {breed_name}은(는) MDR-1 유전자 변이 중위험 견종입니다. 이 약물 사용 시 저용량부터 시작하고, 부작용 발생 시 즉시 수의사에게 연락하세요."
elif breed_risk == 'low':
rec['mdr1_warning'] = 'info'
rec['mdr1_message'] = f"ℹ️ {breed_name}은(는) MDR-1 저위험 견종이나, 개체별 차이가 있을 수 있습니다. 처음 사용 시 주의깊게 관찰하세요."
elif breed_risk == 'unknown':
rec['mdr1_warning'] = 'info'
rec['mdr1_message'] = "ℹ️ 견종에 따라 이 약물(Avermectin 계열)에 민감할 수 있습니다. 수의사 상담을 권장합니다."
# 체중 기반 용량 계산 + 피펫형 제품 필터링
weight_kg = None
try:
if weight:
weight_kg = float(weight)
except (ValueError, TypeError):
weight_kg = None
# 피펫형 제품 체중 필터링 적용
weight_filtered_recommendations = []
for rec in unique_recommendations:
if weight_kg and weight_kg > 0:
dosage_info = calculate_recommended_dosage(
product_idx=rec.get('product_idx'),
weight_kg=weight_kg,
animal_type=animal_type
)
# 피펫형 제품: 체중 범위에 맞지 않으면 제외
if dosage_info and dosage_info.get('is_pipette'):
if dosage_info.get('calculated'):
rec['dosage_info'] = dosage_info
weight_filtered_recommendations.append(rec)
# else: 체중 범위 밖이므로 제외 (추가하지 않음)
else:
# 피펫형이 아닌 제품은 그대로 유지
if dosage_info and dosage_info.get('calculated'):
rec['dosage_info'] = dosage_info
weight_filtered_recommendations.append(rec)
else:
# 체중 미입력 시 필터링 없이 모두 포함
weight_filtered_recommendations.append(rec)
unique_recommendations = weight_filtered_recommendations
# 응답에서 불필요한 필드 제거
for rec in unique_recommendations:
rec.pop('llm_pharm', None)
rec.pop('efficacy_clean', None)
rec.pop('component_code', None)
rec.pop('product_idx', None) # 용량 계산 후 제거
return jsonify({
'success': True,
'animal_type': animal_type,
'matched_symptoms': symptom_descriptions,
'recommendations': unique_recommendations,
'total_count': len(unique_recommendations),
'weight_kg': weight_kg, # 입력된 체중 반환
'pregnancy_status': pregnancy # 임신/수유 상태 반환
})
except Exception as e:
print(f"Error: {e}")
return jsonify({
'success': False,
'message': f'오류가 발생했습니다: {str(e)}'
})
@app.route('/api/recommend_by_category', methods=['POST'])
def recommend_by_category():
"""제품군 기반 추천 API"""
try:
data = request.get_json()
animal_type = data.get('animal_type') # 'dog' or 'cat'
breed = data.get('breed') # 견종 (MDR-1 체크용)
category = data.get('category') # 제품군 코드
pregnancy = data.get('pregnancy', 'none') # 임신/수유 여부
if not animal_type or not category:
return jsonify({
'success': False,
'message': '동물 종류와 제품군을 선택해주세요.'
})
animal_ko = '개' if animal_type == 'dog' else '고양이'
# 카테고리 정보
category_info = PRODUCT_CATEGORIES.get(category, {})
keywords = category_info.get('keywords', [])
category_name = category_info.get('name', category)
if not keywords:
return jsonify({
'success': False,
'message': '잘못된 제품군입니다.'
})
# 재고가 있는 제품 조회
inventory_products = session.query(APDB).join(
Inventory, Inventory.apdb_id == APDB.idx
).filter(
Inventory.transaction_type == 'INBOUND'
).distinct().all()
recommendations = []
import re
for product in inventory_products:
# 동물 타입 체크 (ComponentCode에서)
component = session.query(ComponentCode).filter(
ComponentCode.code == product.component_code
).first()
if component:
target_animals = component.target_animals or []
# JSONB 리스트 형태 처리
if isinstance(target_animals, list):
target_str = ', '.join(target_animals).lower()
else:
target_str = str(target_animals).lower()
if animal_ko not in target_str and '개, 고양이' not in target_str and '고양이, 개' not in target_str:
continue
# 효능효과에서 키워드 매칭
efficacy = (product.efficacy_effect or '').lower()
efficacy_clean = re.sub(r'<[^>]+>', '', efficacy)
if any(kw in efficacy_clean for kw in keywords):
# 효능효과 짧게 추출
efficacy_short = efficacy_clean[:100] + '...' if len(efficacy_clean) > 100 else efficacy_clean
# target_animals 문자열로 변환
target_animals_str = ', '.join(component.target_animals) if component and isinstance(component.target_animals, list) else (component.target_animals if component else '')
rec = {
'idx': product.idx, # 추천 제품 API 호출용
'name': product.product_name,
'product_idx': product.idx, # 용량 계산용
'component_name': component.component_name_ko if component else product.component_code,
'component_code': product.component_code,
'target_animals': target_animals_str,
'efficacy': efficacy_short,
'image_url': get_product_image(product),
'usage_guide': product.usage_guide # 상세 사용 안내 (희석 비율 등)
}
# MDR-1 체크 (강아지이고 견종이 선택된 경우)
if animal_type == 'dog' and breed and breed != 'other':
if product.component_code in MDR1_SENSITIVE_COMPONENTS:
breed_info = MDR1_BREEDS.get(breed, {})
risk = breed_info.get('risk', 'none')
if risk == 'high':
rec['mdr1_warning'] = 'danger'
rec['mdr1_message'] = f"⚠️ {breed_info.get('name', breed)}은(는) 이 약물에 심각한 신경독성 부작용이 발생할 수 있습니다. 반드시 수의사 상담 후 사용하세요."
elif risk == 'medium':
rec['mdr1_warning'] = 'caution'
rec['mdr1_message'] = f"⚠️ {breed_info.get('name', breed)}은(는) 이 약물에 민감할 수 있습니다. 저용량부터 시작하고 모니터링하세요."
elif risk == 'unknown':
rec['mdr1_warning'] = 'info'
rec['mdr1_message'] = "ℹ️ 견종에 따라 이 약물에 민감할 수 있습니다. 수의사 상담을 권장합니다."
recommendations.append(rec)
# 중복 제거 (제품명 기준)
seen_names = set()
unique_recommendations = []
for rec in recommendations:
if rec['name'] not in seen_names:
seen_names.add(rec['name'])
unique_recommendations.append(rec)
# 체중 기반 용량 계산 + 피펫형 제품 필터링
weight = data.get('weight')
weight_kg = None
try:
if weight:
weight_kg = float(weight)
except (ValueError, TypeError):
weight_kg = None
# 피펫형 제품 체중 필터링 적용
weight_filtered_recommendations = []
for rec in unique_recommendations:
if weight_kg and weight_kg > 0:
dosage_info = calculate_recommended_dosage(
product_idx=rec.get('product_idx'),
weight_kg=weight_kg,
animal_type=animal_type
)
# 피펫형 제품: 체중 범위에 맞지 않으면 제외
if dosage_info and dosage_info.get('is_pipette'):
if dosage_info.get('calculated'):
rec['dosage_info'] = dosage_info
weight_filtered_recommendations.append(rec)
# else: 체중 범위 밖이므로 제외 (추가하지 않음)
else:
# 피펫형이 아닌 제품은 그대로 유지
if dosage_info and dosage_info.get('calculated'):
rec['dosage_info'] = dosage_info
weight_filtered_recommendations.append(rec)
else:
# 체중 미입력 시 필터링 없이 모두 포함
weight_filtered_recommendations.append(rec)
unique_recommendations = weight_filtered_recommendations
# component_code, product_idx 제거
for rec in unique_recommendations:
rec.pop('component_code', None)
rec.pop('product_idx', None)
return jsonify({
'success': True,
'animal_type': animal_type,
'category_name': category_name,
'recommendations': unique_recommendations,
'total_count': len(unique_recommendations),
'weight_kg': weight_kg,
'pregnancy_status': pregnancy # 임신/수유 상태 반환
})
except Exception as e:
print(f"제품군 추천 오류: {e}")
import traceback
traceback.print_exc()
return jsonify({
'success': False,
'message': f'오류가 발생했습니다: {str(e)}'
})
# ============================================================
# 통합 추천 API (v2) - APDB + 보조제품
# ============================================================
def get_supplementary_image(supp):
"""보조제품 이미지 URL 반환"""
return supp.image_url or '/static/img/no-image.png'
def has_supplementary_stock(supplementary_id):
"""보조제품 재고 확인"""
inv = session.query(InventorySupplementary).filter(
InventorySupplementary.supplementary_id == supplementary_id,
InventorySupplementary.quantity > 0
).first()
return inv is not None
@app.route('/api/product//recommendations')
def get_product_recommendations(product_idx):
"""
특정 APDB 제품의 통합 추천 목록 조회
- 관련 APDB 제품 (약품 → 약품)
- 관련 보조제품 (약품 → 건기식/기구/케어)
"""
try:
result = {
"success": True,
"product_idx": product_idx,
"apdb_recommendations": [],
"supplementary_recommendations": []
}
# 1. APDB → APDB 추천
apdb_recs = session.query(UnifiedProductRecommendation).filter(
UnifiedProductRecommendation.source_type == 'apdb',
UnifiedProductRecommendation.source_id == product_idx,
UnifiedProductRecommendation.target_type == 'apdb',
UnifiedProductRecommendation.is_active == True
).order_by(UnifiedProductRecommendation.priority.desc()).all()
for rec in apdb_recs:
product = session.query(APDB).filter(APDB.idx == rec.target_id).first()
if not product:
continue
# 재고 확인
has_stock = session.query(Inventory).filter(
Inventory.apdb_id == product.idx,
Inventory.transaction_type == 'INBOUND'
).first() is not None
if has_stock:
result["apdb_recommendations"].append({
"idx": product.idx,
"name": product.product_name,
"image_url": get_product_image(product),
"relation_type": rec.relation_type,
"reason": rec.recommendation_reason,
"priority": rec.priority
})
# 2. APDB → Supplementary 추천
supp_recs = session.query(UnifiedProductRecommendation).filter(
UnifiedProductRecommendation.source_type == 'apdb',
UnifiedProductRecommendation.source_id == product_idx,
UnifiedProductRecommendation.target_type == 'supplementary',
UnifiedProductRecommendation.is_active == True
).order_by(UnifiedProductRecommendation.priority.desc()).all()
for rec in supp_recs:
supp = session.query(SupplementaryProduct).filter(
SupplementaryProduct.id == rec.target_id,
SupplementaryProduct.is_active == True
).first()
if supp and has_supplementary_stock(supp.id):
result["supplementary_recommendations"].append({
"id": supp.id,
"name": supp.product_name,
"brand": supp.brand,
"category": supp.category,
"sub_category": supp.sub_category,
"image_url": get_supplementary_image(supp),
"relation_type": rec.relation_type,
"reason": rec.recommendation_reason,
"priority": rec.priority,
"price": float(supp.price) if supp.price else None,
"efficacy_tags": supp.efficacy_tags
})
result["total_apdb"] = len(result["apdb_recommendations"])
result["total_supplementary"] = len(result["supplementary_recommendations"])
return jsonify(result)
except Exception as e:
print(f"통합 추천 조회 오류: {e}")
import traceback
traceback.print_exc()
return jsonify({
'success': False,
'message': f'오류가 발생했습니다: {str(e)}'
})
@app.route('/api/supplementary/')
def get_supplementary_detail(supp_id):
"""보조제품 상세 정보 조회"""
try:
supp = session.query(SupplementaryProduct).filter(
SupplementaryProduct.id == supp_id
).first()
if not supp:
return jsonify({
'success': False,
'message': '제품을 찾을 수 없습니다.'
})
# 재고 정보
inv = session.query(InventorySupplementary).filter(
InventorySupplementary.supplementary_id == supp_id
).first()
return jsonify({
'success': True,
'product': {
'id': supp.id,
'product_name': supp.product_name,
'product_name_en': supp.product_name_en,
'brand': supp.brand,
'manufacturer': supp.manufacturer,
'barcode': supp.barcode,
'product_code': supp.product_code,
'category': supp.category,
'sub_category': supp.sub_category,
'description': supp.description,
'usage_instructions': supp.usage_instructions,
'ingredients': supp.ingredients,
'main_ingredient': supp.main_ingredient,
'ingredient_details': supp.ingredient_details,
'efficacy': supp.efficacy,
'efficacy_tags': supp.efficacy_tags,
'target_animal': supp.target_animal,
'target_age': supp.target_age,
'target_size': supp.target_size,
'precautions': supp.precautions,
'warnings': supp.warnings,
'image_url': supp.image_url,
'image_url_2': supp.image_url_2,
'price': float(supp.price) if supp.price else None,
'unit': supp.unit,
'package_size': supp.package_size,
'external_url': supp.external_url,
'stock': inv.quantity if inv else 0
}
})
except Exception as e:
print(f"보조제품 상세 조회 오류: {e}")
return jsonify({
'success': False,
'message': f'오류가 발생했습니다: {str(e)}'
})
@app.route('/api/supplementary')
def list_supplementary_products():
"""보조제품 목록 조회 (카테고리별 필터링)"""
try:
category = request.args.get('category') # supplement, equipment, care
sub_category = request.args.get('sub_category')
target_animal = request.args.get('target_animal')
query = session.query(SupplementaryProduct).filter(
SupplementaryProduct.is_active == True
)
if category:
query = query.filter(SupplementaryProduct.category == category)
if sub_category:
query = query.filter(SupplementaryProduct.sub_category == sub_category)
if target_animal:
query = query.filter(
or_(
SupplementaryProduct.target_animal == target_animal,
SupplementaryProduct.target_animal == 'all'
)
)
products = query.order_by(SupplementaryProduct.product_name).all()
result = []
for supp in products:
if has_supplementary_stock(supp.id):
result.append({
'id': supp.id,
'product_name': supp.product_name,
'brand': supp.brand,
'category': supp.category,
'sub_category': supp.sub_category,
'main_ingredient': supp.main_ingredient,
'efficacy_tags': supp.efficacy_tags,
'target_animal': supp.target_animal,
'image_url': get_supplementary_image(supp),
'price': float(supp.price) if supp.price else None,
'package_size': supp.package_size
})
return jsonify({
'success': True,
'products': result,
'total': len(result)
})
except Exception as e:
print(f"보조제품 목록 조회 오류: {e}")
return jsonify({
'success': False,
'message': f'오류가 발생했습니다: {str(e)}'
})
@app.route('/api/symptoms/supplementary-recommendations', methods=['POST'])
def get_symptom_supplementary_recommendations():
"""증상 선택 시 보조제품 추천"""
try:
data = request.get_json()
symptom_codes = data.get('symptom_codes', [])
animal_type = data.get('animal_type') # dog, cat
if not symptom_codes:
return jsonify({
'success': False,
'message': '증상을 선택해주세요.'
})
# 증상 → 보조제품 매핑 조회
recommendations = session.query(SymptomSupplementaryRecommendation).filter(
SymptomSupplementaryRecommendation.symptom_code.in_(symptom_codes),
SymptomSupplementaryRecommendation.is_active == True
).order_by(SymptomSupplementaryRecommendation.priority.desc()).all()
result = []
seen_ids = set()
for rec in recommendations:
if rec.supplementary_id in seen_ids:
continue
supp = session.query(SupplementaryProduct).filter(
SupplementaryProduct.id == rec.supplementary_id,
SupplementaryProduct.is_active == True
).first()
if not supp:
continue
# 동물 타입 필터링
if animal_type and supp.target_animal:
if supp.target_animal != 'all' and supp.target_animal != animal_type:
continue
# 재고 확인
if not has_supplementary_stock(supp.id):
continue
seen_ids.add(rec.supplementary_id)
# 매칭된 증상 정보 조회
symptom = session.query(Symptoms).filter(
Symptoms.symptom_code == rec.symptom_code
).first()
result.append({
'id': supp.id,
'product_name': supp.product_name,
'brand': supp.brand,
'category': supp.category,
'sub_category': supp.sub_category,
'main_ingredient': supp.main_ingredient,
'efficacy_tags': supp.efficacy_tags,
'image_url': get_supplementary_image(supp),
'price': float(supp.price) if supp.price else None,
'relation_type': rec.relation_type,
'reason': rec.recommendation_reason,
'matched_symptom': symptom.description if symptom else rec.symptom_code
})
return jsonify({
'success': True,
'supplementary_recommendations': result,
'total': len(result)
})
except Exception as e:
print(f"증상 기반 보조제품 추천 오류: {e}")
import traceback
traceback.print_exc()
return jsonify({
'success': False,
'message': f'오류가 발생했습니다: {str(e)}'
})
# ============================================================
# 실행
# ============================================================
if __name__ == '__main__':
print("=" * 60)
print("🐾 동물약 추천 MVP 웹앱")
print("=" * 60)
print("접속 주소: http://localhost:7001")
print("종료: Ctrl+C")
print("=" * 60)
app.run(host='0.0.0.0', port=7001, debug=True)