""" Dosage recommendation routes — prefetch + recommendation + health. """ from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks, Response from sqlalchemy.orm import Session from sqlalchemy import func from pydantic import BaseModel from typing import List, Optional import time import json as _json from src.utils.logger import get_logger from src.database.connection import get_db_dependency from src.database.models import Medicine, DosageCache from src.nlp.enhanced_extractor import get_enhanced_extractor logger = get_logger(__name__) router = APIRouter(prefix="/api/v1/drugs/dosage") # ========== Pydantic Models ========== class PrefetchRequest(BaseModel): """Request model for medicine prefetch""" medications: List[str] class BatchDosageRequest(BaseModel): """Request model for batch dosage recommendations""" age: Optional[int] = None gender: Optional[str] = None medications: List[str] class DosageRecommendationItem(BaseModel): """Individual medicine dosage recommendation""" medicine_name: str composition: str quantity: str dosage: str frequency: str preference: str duration: str warnings: Optional[List[str]] = [] confidence_score: Optional[float] = None found_in_database: Optional[bool] = True # ========== Helpers ========== def _find_medicine(db: Session, name: str): """Look up a medicine by product name (exact then fuzzy). MySQL utf8mb4 collation is case-insensitive, so no need for lower().""" med = db.query(Medicine).filter( Medicine.product_name == name.strip() ).first() if not med: med = db.query(Medicine).filter( Medicine.product_name.ilike(f"%{name}%") ).order_by(Medicine.product_name).first() return med def _get_age_dosage(age_dosages: dict, age: int) -> dict: """Pick the age-appropriate dosage entry from age_group_dosages.""" if not age_dosages: return {"dosage": "Consult doctor", "notes": "No age-specific data available"} if age <= 2: return age_dosages.get("infant_0_2", {"dosage": "Consult doctor", "notes": ""}) elif age <= 12: return age_dosages.get("child_2_12", {"dosage": "Consult doctor", "notes": ""}) elif age <= 18: return age_dosages.get("adolescent_12_18", {"dosage": "Consult doctor", "notes": ""}) elif age <= 65: return age_dosages.get("adult_18_65", {"dosage": "Consult doctor", "notes": ""}) else: return age_dosages.get("elderly_65_plus", {"dosage": "Consult doctor", "notes": ""}) def _prefetch_medicines(medications: List[str]): """Background task: look up unknown medicines via Gemini and save to DB.""" from src.database.connection import SessionLocal db = SessionLocal() try: unknown_names = [] for name in medications: found = db.query(Medicine).filter( func.lower(Medicine.product_name) == name.lower().strip() ).first() if not found: found = db.query(Medicine).filter( Medicine.product_name.ilike(f"%{name}%") ).order_by(Medicine.product_name).first() if not found: unknown_names.append(name) if not unknown_names: logger.info(f"Prefetch: all {len(medications)} medicines already in DB") return logger.info(f"Prefetch: {len(unknown_names)} unknown medicines, calling Gemini") from src.nlp.gemini_extractor import get_gemini_extractor gemini = get_gemini_extractor() if not gemini.is_available(): logger.warning("Prefetch: Gemini extractor not available") return gemini_lookups = gemini.batch_lookup_medicines(unknown_names) for name in unknown_names: lookup = gemini_lookups.get(name) if not lookup or lookup.get('confidence', 0) < 0.5: logger.warning(f"Prefetch: no usable Gemini result for '{name}'") continue try: age_dosages = lookup.get('age_group_dosages', {}) age_dosages_str = _json.dumps(age_dosages) if age_dosages else None new_medicine = Medicine( product_name=name.strip(), salt_composition=lookup.get('composition'), product_form=lookup.get('product_form'), medicine_type=lookup.get('medicine_type'), primary_use=lookup.get('primary_use'), how_to_use=lookup.get('how_to_use'), common_side_effect=lookup.get('common_side_effects'), benefits=lookup.get('benefits'), how_it_works=lookup.get('how_it_works'), safety_advise=lookup.get('safety_advice'), symptoms=lookup.get('symptoms'), alcohol_interaction=lookup.get('alcohol_interaction'), pregnancy_interaction=lookup.get('pregnancy_interaction'), lactation_interaction=lookup.get('lactation_interaction'), fact_box=age_dosages_str, description=lookup.get('allergens', ''), source_file='gemini_ai', ) db.add(new_medicine) db.commit() # Also cache the Gemini dosage data dosage_cache = DosageCache( medicine_id=new_medicine.id, quantity=lookup.get('quantity', 'As prescribed'), dosage_code=lookup.get('dosage', 'BD'), frequency=lookup.get('frequency', 'Twice daily'), meal_preference=lookup.get('meal_preference', 'As advised'), duration=lookup.get('duration', 'As prescribed'), dose_morning=float(lookup.get('dose_morning', 0)), dose_afternoon=float(lookup.get('dose_afternoon', 0)), dose_night=float(lookup.get('dose_night', 0)), age_group_dosages=age_dosages_str, generation_method='gemini', confidence_score=lookup.get('confidence', 0.85), ) db.add(dosage_cache) db.commit() logger.info(f"Prefetch: saved '{name}' to DB (id={new_medicine.id}) with dosage cache") except Exception as e: db.rollback() logger.warning(f"Prefetch: failed to store '{name}': {e}") except Exception as e: logger.error(f"Prefetch background task failed: {e}") finally: db.close() # ========== Endpoints ========== @router.post("/prefetch", status_code=204) async def prefetch_medicines( request: PrefetchRequest, background_tasks: BackgroundTasks, ): """ Prefetch medicines: checks DB and triggers Gemini lookup for unknown medicines in the background. Returns 204 immediately. Call this as soon as medicine names are available, then call /recommendation to get the full response. """ background_tasks.add_task(_prefetch_medicines, request.medications) return Response(status_code=204) @router.post("/recommendation") async def get_batch_dosage_recommendation( request: BatchDosageRequest, include_metadata: bool = False, include_details: bool = False, ai_enhanced: bool = False, db: Session = Depends(get_db_dependency), ): """ BATCH Dosage Recommendation API Query Parameters: - include_metadata: include validation warnings, confidence scores, metadata - include_details: include extended medical info (benefits, side_effects, safety_advice, allergens, interactions, age_group_dosages) - ai_enhanced: enable Gemini AI fallback for low-confidence extractions """ start_time = time.time() try: results = [] not_found = [] low_confidence_medicines = [] extractor = get_enhanced_extractor() age_str = f"{request.age}yo" if request.age else "no-age" gender_str = request.gender or "no-gender" logger.info( f"Processing {len(request.medications)} medications for " f"{age_str} {gender_str} | Metadata: {include_metadata}" ) # ── Phase 1: DB lookup for all medicines ── db_results = {} unknown_names = [] for medicine_name in request.medications: medicine = _find_medicine(db, medicine_name) db_results[medicine_name] = medicine if not medicine: unknown_names.append(medicine_name) # ── Phase 2: Batch Gemini lookup for all unknown medicines ── gemini_lookups = {} if ai_enhanced and unknown_names: try: from src.nlp.gemini_extractor import get_gemini_extractor gemini = get_gemini_extractor() if gemini.is_available(): gemini_lookups = gemini.batch_lookup_medicines(unknown_names) logger.info( f"Gemini batch lookup returned " f"{sum(1 for v in gemini_lookups.values() if v)} results " f"for {len(unknown_names)} unknown medicines" ) except Exception as e: logger.warning(f"Gemini batch lookup failed: {e}") # ── Phase 3: Build results for each medication ── for medicine_name in request.medications: try: medicine = db_results.get(medicine_name) if not medicine: lookup = gemini_lookups.get(medicine_name) if lookup and lookup.get('confidence', 0) >= 0.5: try: age_dosages = lookup.get('age_group_dosages', {}) age_dosages_str = _json.dumps(age_dosages) if age_dosages else None new_medicine = Medicine( product_name=medicine_name.strip(), salt_composition=lookup.get('composition'), product_form=lookup.get('product_form'), medicine_type=lookup.get('medicine_type'), primary_use=lookup.get('primary_use'), how_to_use=lookup.get('how_to_use'), common_side_effect=lookup.get('common_side_effects'), benefits=lookup.get('benefits'), how_it_works=lookup.get('how_it_works'), safety_advise=lookup.get('safety_advice'), symptoms=lookup.get('symptoms'), alcohol_interaction=lookup.get('alcohol_interaction'), pregnancy_interaction=lookup.get('pregnancy_interaction'), lactation_interaction=lookup.get('lactation_interaction'), fact_box=age_dosages_str, description=lookup.get('allergens', ''), source_file='gemini_ai', ) db.add(new_medicine) db.commit() db.refresh(new_medicine) # Also cache the Gemini dosage data dosage_cache = DosageCache( medicine_id=new_medicine.id, quantity=lookup.get('quantity', 'As prescribed'), dosage_code=lookup.get('dosage', 'BD'), frequency=lookup.get('frequency', 'Twice daily'), meal_preference=lookup.get('meal_preference', 'As advised'), duration=lookup.get('duration', 'As prescribed'), dose_morning=float(lookup.get('dose_morning', 0)), dose_afternoon=float(lookup.get('dose_afternoon', 0)), dose_night=float(lookup.get('dose_night', 0)), age_group_dosages=age_dosages_str, generation_method='gemini', confidence_score=lookup.get('confidence', 0.85), ) db.add(dosage_cache) db.commit() logger.info(f"Gemini created DB record for: {medicine_name} (id={new_medicine.id}) with dosage cache") result = { "medicine_name": new_medicine.product_name, "product_form": lookup.get('product_form', ''), "composition": new_medicine.salt_composition or "Consult physician", "symptoms": lookup.get('symptoms', ''), "quantity": lookup.get('quantity', 'As prescribed'), "dosage": lookup.get('dosage', 'BD'), "frequency": lookup.get('frequency', 'Twice daily'), "dose_schedule": lookup.get('dose_schedule', '1-0-1'), "preference": lookup.get('meal_preference', 'As advised'), "duration": lookup.get('duration', 'As prescribed'), "age_group_dosages": age_dosages, } if request.age is not None: result["age_specific_dosage"] = _get_age_dosage(age_dosages, request.age) if include_metadata: result["warnings"] = ["AI-generated: not from verified database"] result["confidence_score"] = lookup.get('confidence', 0.7) result["found_in_database"] = False result["ai_generated"] = True result["extraction_method"] = "gemini_lookup" if include_details: result["product_id"] = new_medicine.product_id or '' result["marketer"] = new_medicine.marketer or '' result["medicine_type"] = new_medicine.medicine_type or '' result["introduction"] = new_medicine.introduction or '' result["description"] = new_medicine.description or '' result["how_to_use"] = new_medicine.how_to_use or '' result["safety_advise"] = new_medicine.safety_advise or '' result["if_miss"] = new_medicine.if_miss or '' result["packaging_detail"] = new_medicine.packaging_detail or '' result["package"] = new_medicine.package or '' result["qty"] = new_medicine.qty or '' result["mrp"] = new_medicine.mrp result["prescription_req"] = new_medicine.prescription_req or '' result["fact_box"] = new_medicine.fact_box or '' result["primary_use"] = new_medicine.primary_use or '' result["storage"] = new_medicine.storage or '' result["use_of"] = new_medicine.use_of or '' result["benefits"] = new_medicine.benefits or '' result["how_it_works"] = new_medicine.how_it_works or '' result["side_effects"] = new_medicine.common_side_effect or '' result["interactions"] = { "alcohol": new_medicine.alcohol_interaction or '', "pregnancy": new_medicine.pregnancy_interaction or '', "lactation": new_medicine.lactation_interaction or '', "driving": new_medicine.driving_interaction or '', "kidney": new_medicine.kidney_interaction or '', "liver": new_medicine.liver_interaction or '', } result["interaction"] = new_medicine.interaction or '' result["manufacturer_address"] = new_medicine.manufacturer_address or '' result["country_of_origin"] = new_medicine.country_of_origin or '' result["q_a"] = new_medicine.q_a or '' result["manufacturer_details"] = new_medicine.manufacturer_details or '' result["marketer_details"] = new_medicine.marketer_details or '' result["expiration"] = new_medicine.expirtation or '' result["reference"] = new_medicine.reference or '' results.append(result) logger.info(f"AI-generated result for: {medicine_name}") continue except Exception as e: logger.warning(f"Failed to store Gemini result for {medicine_name}: {e}") db.rollback() # Fallback: not found not_found.append(medicine_name) logger.warning(f"Medicine not found: '{medicine_name}'") result = { "medicine_name": medicine_name, "product_form": "", "composition": "Consult physician", "quantity": "Consult physician", "dosage": "Consult physician", "frequency": "Consult physician", "preference": "As advised", "duration": "As prescribed", } if include_metadata: result["warnings"] = ["Medicine not found in database"] result["confidence_score"] = 0.0 result["found_in_database"] = False results.append(result) continue # ── Check DosageCache first ── cached = db.query(DosageCache).filter( DosageCache.medicine_id == medicine.id ).first() if cached and cached.confidence_score and cached.confidence_score >= 0.80: # Use cached dosage data (Gemini is the source of truth) extraction_method = f"cached_{cached.generation_method or 'unknown'}" result = { "medicine_name": medicine.product_name, "product_form": medicine.product_form or "", "composition": medicine.salt_composition or "Composition not available", "symptoms": medicine.symptoms or "", "quantity": cached.quantity or "As prescribed", "dosage": cached.dosage_code or "OD", "frequency": cached.frequency or "Once daily", "dose_schedule": cached.dose_schedule, "preference": cached.meal_preference or "As advised", "duration": cached.duration or "As prescribed", } # Age-specific dosages from cache age_dosages = {} if cached.age_group_dosages: try: age_dosages = _json.loads(cached.age_group_dosages) except (ValueError, TypeError): pass if not age_dosages and medicine.fact_box: try: age_dosages = _json.loads(medicine.fact_box) if isinstance(medicine.fact_box, str) else medicine.fact_box if not isinstance(age_dosages, dict): age_dosages = {} except (ValueError, TypeError): age_dosages = {} result["age_group_dosages"] = age_dosages if request.age is not None: result["age_specific_dosage"] = _get_age_dosage(age_dosages, request.age) if include_metadata: result["warnings"] = [] result["confidence_score"] = cached.confidence_score result["found_in_database"] = True result["needs_verification"] = cached.confidence_score < 0.80 result["extraction_method"] = extraction_method if medicine.source_file == 'gemini_ai': result["ai_generated"] = True logger.info( f"Cache hit for {medicine.product_name} " f"({extraction_method}, confidence={cached.confidence_score:.2f})" ) else: # No usable cache — run extraction as fallback extracted = extractor.extract_enhanced( medicine, enable_validation=True, use_ai_fallback=ai_enhanced, ) if extracted['confidence_score'] < 0.80: low_confidence_medicines.append({ 'name': medicine.product_name, 'confidence': extracted['confidence_score'], }) logger.warning( f"Low confidence ({extracted['confidence_score']:.2f}): " f"{medicine.product_name}" ) if extracted.get('warnings'): for warning in extracted['warnings']: logger.warning(f"{medicine.product_name}: {warning}") # Derive meal-slot schedule from dosage code dose_morning = extracted.get('dose_morning', 0) dose_afternoon = extracted.get('dose_afternoon', 0) dose_night = extracted.get('dose_night', 0) if not any([dose_morning, dose_afternoon, dose_night]): schedule_map = {'OD': (1,0,0), 'BD': (1,0,1), 'TDS': (1,1,1), 'QDS': (1,1,1), 'OW': (1,0,0), 'BW': (1,0,0), 'OM': (1,0,0), 'OY': (1,0,0), 'ONCE': (1,0,0)} dose_morning, dose_afternoon, dose_night = schedule_map.get( extracted.get('dosage', 'OD'), (1, 0, 0) ) def _fmt_dose(v): return str(int(v)) if v == int(v) else str(v) dose_schedule = f"{_fmt_dose(dose_morning)}-{_fmt_dose(dose_afternoon)}-{_fmt_dose(dose_night)}" age_dosages = extracted.get('age_group_dosages', {}) if not age_dosages and medicine.fact_box: try: age_dosages = _json.loads(medicine.fact_box) if isinstance(medicine.fact_box, str) else medicine.fact_box if not isinstance(age_dosages, dict): age_dosages = {} except (ValueError, TypeError): age_dosages = {} result = { "medicine_name": medicine.product_name, "product_form": medicine.product_form or "", "composition": medicine.salt_composition or "Composition not available", "symptoms": medicine.symptoms or "", "quantity": extracted['quantity'], "dosage": extracted['dosage'], "frequency": extracted['frequency'], "dose_schedule": dose_schedule, "preference": extracted['meal_preference'], "duration": extracted['duration'], "age_group_dosages": age_dosages, } if request.age is not None: result["age_specific_dosage"] = _get_age_dosage(age_dosages, request.age) if include_metadata: result["warnings"] = extracted.get('warnings', []) result["confidence_score"] = extracted['confidence_score'] result["found_in_database"] = True result["needs_verification"] = extracted['needs_verification'] result["extraction_method"] = extracted.get('extraction_method', 'regex') if medicine.source_file == 'gemini_ai': result["ai_generated"] = True if include_details: result["product_id"] = medicine.product_id or '' result["marketer"] = medicine.marketer or '' result["medicine_type"] = medicine.medicine_type or '' result["introduction"] = medicine.introduction or '' result["description"] = medicine.description or '' result["how_to_use"] = medicine.how_to_use or '' result["safety_advise"] = medicine.safety_advise or '' result["if_miss"] = medicine.if_miss or '' result["packaging_detail"] = medicine.packaging_detail or '' result["package"] = medicine.package or '' result["qty"] = medicine.qty or '' result["mrp"] = medicine.mrp result["prescription_req"] = medicine.prescription_req or '' result["fact_box"] = medicine.fact_box or '' result["primary_use"] = medicine.primary_use or '' result["storage"] = medicine.storage or '' result["use_of"] = medicine.use_of or '' result["benefits"] = medicine.benefits or '' result["how_it_works"] = medicine.how_it_works or '' result["side_effects"] = medicine.common_side_effect or '' result["interactions"] = { "alcohol": medicine.alcohol_interaction or '', "pregnancy": medicine.pregnancy_interaction or '', "lactation": medicine.lactation_interaction or '', "driving": medicine.driving_interaction or '', "kidney": medicine.kidney_interaction or '', "liver": medicine.liver_interaction or '', } result["interaction"] = medicine.interaction or '' result["manufacturer_address"] = medicine.manufacturer_address or '' result["country_of_origin"] = medicine.country_of_origin or '' result["q_a"] = medicine.q_a or '' result["manufacturer_details"] = medicine.manufacturer_details or '' result["marketer_details"] = medicine.marketer_details or '' result["expiration"] = medicine.expirtation or '' result["reference"] = medicine.reference or '' results.append(result) except Exception as e: logger.error(f"Error processing medicine '{medicine_name}': {e}") result = { "medicine_name": medicine_name, "product_form": "", "composition": "Error retrieving data", "quantity": "Consult physician", "dosage": "Consult physician", "frequency": "Consult physician", "preference": "As advised", "duration": "As prescribed", } if include_metadata: result["warnings"] = [f"Error: {str(e)}"] result["confidence_score"] = 0.0 result["found_in_database"] = False results.append(result) # ── Phase 4: Drug-Drug Interaction Check ── drug_interactions = [] if len(request.medications) >= 2: try: from src.nlp.ddi_checker import get_ddi_checker ddi_checker = get_ddi_checker() resolved_medicines = [] for medicine_name in request.medications: med_db = db_results.get(medicine_name) if med_db: resolved_medicines.append({ "product_name": med_db.product_name, "salt_composition": med_db.salt_composition or "", "interaction": med_db.interaction or "", "found_in_db": True, }) else: lookup = gemini_lookups.get(medicine_name) resolved_medicines.append({ "product_name": medicine_name, "salt_composition": lookup.get("composition", "") if lookup else "", "interaction": "", "found_in_db": False, }) drug_interactions = ddi_checker.check_interactions( resolved_medicines, use_gemini_fallback=ai_enhanced, db_session=db, ) logger.info( f"DDI check: {len(drug_interactions)} pairs checked, " f"{sum(1 for d in drug_interactions if d.get('severity') not in ('none', 'unknown'))} interactions found" ) except Exception as e: logger.error(f"DDI check failed: {e}") response_time = (time.time() - start_time) * 1000 response = { "data": results, "error": None, } if len(request.medications) >= 2: response["drug_interactions"] = drug_interactions if include_metadata: response["metadata"] = { "total_processed": len(request.medications), "found_in_database": len(request.medications) - len(not_found), "not_found_in_database": len(not_found), "not_found_medicines": not_found, "low_confidence_count": len(low_confidence_medicines), "low_confidence_medicines": low_confidence_medicines, "response_time_ms": round(response_time, 2), "patient_age": request.age if request.age is not None else "not provided", "patient_gender": request.gender or "not provided", "ai_enhanced": ai_enhanced, "ddi_checks_performed": len(drug_interactions), "ddi_interactions_found": sum( 1 for d in drug_interactions if d.get("severity") not in ("none", "unknown") ), } logger.info( f"BATCH COMPLETED: {len(request.medications)} medicines | " f"Found: {len(request.medications) - len(not_found)} | " f"Not found: {len(not_found)} | " f"Low confidence: {len(low_confidence_medicines)} | " f"Time: {response_time:.2f}ms" ) return response except Exception as e: logger.error(f"Batch dosage error: {e}") raise HTTPException( status_code=500, detail={ "data": None, "error": { "code": 500, "message": f"Error processing batch dosage: {str(e)}", "status": "error", }, }, ) @router.get("/health") async def health_check(db: Session = Depends(get_db_dependency)): """Check database connectivity""" try: total_medicines = db.query(func.count(Medicine.id)).scalar() return { "status": "healthy", "database": "connected", "total_medicines": total_medicines, } except Exception as e: return { "status": "unhealthy", "database": "disconnected", "error": str(e), }