from typing import List, Dict, Optional from langchain.tools import StructuredTool import json # Now import from payments module from src.payments.agent import process_purchase_request # Load datasets with open("data/supplements.json", "r") as f: SUPPLEMENTS = json.load(f) with open("data/prescription_drugs.json", "r") as f: PRESCRIPTION_DRUGS = json.load(f) # Combine both datasets for unified searching ALL_PRODUCTS = {**SUPPLEMENTS, **PRESCRIPTION_DRUGS} # Supplement search and retrieval functions def search_supplements_by_condition(query: str) -> List[Dict]: """ Search for supplements that are recommended for a specific health condition, well-being goal, or enhancement purpose. Args: query: The condition, goal, or purpose to search supplements for (e.g., 'hair_loss', 'cognitive_function', 'muscle_growth') Returns: A list of supplements that match the query """ matching_supplements = [] query = query.lower() for supp_id, supp_data in SUPPLEMENTS.items(): # Check conditions conditions = [c.lower() for c in supp_data.get("conditions", [])] # Check benefits/enhancements benefits = [b.lower() for b in supp_data.get("benefits", [])] # Check categories categories = [cat.lower() for cat in supp_data.get("categories", [])] # Match against all fields if (query in conditions or query in benefits or query in categories or query in supp_data["name"].lower() or query in supp_data["description"].lower()): matching_supplements.append({ "id": supp_id, "name": supp_data["name"], "description": supp_data["description"], "link": supp_data["affiliate_link"], "conditions": supp_data.get("conditions", []), "benefits": supp_data.get("benefits", []), "categories": supp_data.get("categories", []), "disclaimers": supp_data.get("disclaimers", []) }) return matching_supplements def search_supplements_by_name(name: str) -> Optional[Dict]: """ Search for a supplement by its name. Args: name: The name of the supplement to search for (e.g., 'Vitamin D3', 'Magnesium') Returns: Details of the supplement if found, None otherwise """ name = name.lower() for supp_id, supp_data in SUPPLEMENTS.items(): if name in supp_data["name"].lower(): return { "id": supp_id, "name": supp_data["name"], "description": supp_data["description"], "link": supp_data["affiliate_link"], "conditions": supp_data.get("conditions", []), "benefits": supp_data.get("benefits", []), "categories": supp_data.get("categories", []), "disclaimers": supp_data.get("disclaimers", []) } return None def get_supplement_details(supplement_id: str) -> Optional[Dict]: """ Get detailed information about a specific supplement. Args: supplement_id: The ID of the supplement to retrieve Returns: Detailed information about the supplement or None if not found """ if supplement_id in SUPPLEMENTS: supp_data = SUPPLEMENTS[supplement_id] return { "id": supplement_id, "name": supp_data["name"], "description": supp_data["description"], "link": supp_data["affiliate_link"], "conditions": supp_data.get("conditions", []), "disclaimers": supp_data.get("disclaimers", []) } return None def list_all_supplements() -> List[Dict]: """ List all available supplements in the database. Returns: A list of all supplements with their basic information """ return [ { "id": supp_id, "name": supp_data["name"], "description": supp_data["description"], "conditions": supp_data.get("conditions", []) } for supp_id, supp_data in SUPPLEMENTS.items() ] # Prescription drug search and retrieval functions def search_prescription_drugs_by_condition(query: str) -> List[Dict]: """ Search for prescription drugs that are used for a specific health condition, well-being goal, or therapeutic purpose. Args: query: The condition, goal, or purpose to search drugs for (e.g., 'hypertension', 'depression', 'hair_loss') Returns: A list of prescription drugs that match the query """ matching_drugs = [] query = query.lower() for drug_id, drug_data in PRESCRIPTION_DRUGS.items(): # Check conditions conditions = [c.lower() for c in drug_data.get("conditions", [])] # Check benefits benefits = [b.lower() for b in drug_data.get("benefits", [])] # Check categories categories = [cat.lower() for cat in drug_data.get("categories", [])] # Match against all fields if (query in conditions or query in benefits or query in categories or query in drug_data["name"].lower() or query in drug_data["description"].lower()): matching_drugs.append({ "id": drug_id, "name": drug_data["name"], "description": drug_data["description"], "link": drug_data["affiliate_link"], "conditions": drug_data.get("conditions", []), "benefits": drug_data.get("benefits", []), "categories": drug_data.get("categories", []), "disclaimers": drug_data.get("disclaimers", []), "requires_prescription": True }) return matching_drugs def search_prescription_drugs_by_name(name: str) -> Optional[Dict]: """ Search for a prescription drug by its name. Args: name: The name of the drug to search for (e.g., 'Metformin', 'Atorvastatin') Returns: Details of the prescription drug if found, None otherwise """ name = name.lower() for drug_id, drug_data in PRESCRIPTION_DRUGS.items(): if name in drug_data["name"].lower(): return { "id": drug_id, "name": drug_data["name"], "description": drug_data["description"], "link": drug_data["affiliate_link"], "conditions": drug_data.get("conditions", []), "benefits": drug_data.get("benefits", []), "categories": drug_data.get("categories", []), "disclaimers": drug_data.get("disclaimers", []), "requires_prescription": True } return None def get_prescription_drug_details(drug_id: str) -> Optional[Dict]: """ Get detailed information about a specific prescription drug. Args: drug_id: The ID of the prescription drug to retrieve Returns: Detailed information about the drug or None if not found """ if drug_id in PRESCRIPTION_DRUGS: drug_data = PRESCRIPTION_DRUGS[drug_id] return { "id": drug_id, "name": drug_data["name"], "description": drug_data["description"], "link": drug_data["affiliate_link"], "conditions": drug_data.get("conditions", []), "benefits": drug_data.get("benefits", []), "categories": drug_data.get("categories", []), "disclaimers": drug_data.get("disclaimers", []), "requires_prescription": True } return None def list_all_prescription_drugs() -> List[Dict]: """ List all available prescription drugs in the database. Returns: A list of all prescription drugs with their basic information """ return [ { "id": drug_id, "name": drug_data["name"], "description": drug_data["description"], "conditions": drug_data.get("conditions", []), "requires_prescription": True } for drug_id, drug_data in PRESCRIPTION_DRUGS.items() ] # Combined search functions def search_all_products_by_condition(query: str) -> List[Dict]: """ Search for both supplements and prescription drugs that are recommended for a specific health condition, well-being goal, or enhancement purpose. Args: query: The condition, goal, or purpose to search for (e.g., 'hair_loss', 'cognitive_function', 'depression') Returns: A list of supplements and prescription drugs that match the query """ supplements = search_supplements_by_condition(query) prescription_drugs = search_prescription_drugs_by_condition(query) return supplements + prescription_drugs def search_all_products_by_name(name: str) -> List[Dict]: """ Search for both supplements and prescription drugs by name. Args: name: The name to search for (e.g., 'Vitamin D3', 'Metformin') Returns: List of products matching the name query """ results = [] supplement = search_supplements_by_name(name) if supplement: results.append(supplement) prescription = search_prescription_drugs_by_name(name) if prescription: results.append(prescription) return results def get_product_details(product_id: str) -> Optional[Dict]: """ Get detailed information about a specific product (supplement or prescription drug). Args: product_id: The ID of the product to retrieve Returns: Detailed information about the product or None if not found """ supplement = get_supplement_details(product_id) if supplement: return supplement prescription = get_prescription_drug_details(product_id) if prescription: return prescription return None def list_all_products() -> List[Dict]: """ List all available products (supplements and prescription drugs) in the database. Returns: A list of all products with their basic information """ supplements = list_all_supplements() prescription_drugs = list_all_prescription_drugs() return supplements + prescription_drugs def create_payment_link(request: str) -> str: """ Create a payment link for products and consultations based on a natural language request. Args: request: A string describing what products and quantities to purchase (e.g., "I want to buy 1 tretinoin and 1 consultation") Returns: The payment link URL """ # Process the request and get payment link response = process_purchase_request(request) #response['messages'][0].content return response # Create and export Langchain tools def get_tools(): """Return all the tools needed for the agent""" return [ StructuredTool.from_function(search_supplements_by_condition), StructuredTool.from_function(search_prescription_drugs_by_condition), StructuredTool.from_function(search_all_products_by_condition), StructuredTool.from_function(search_supplements_by_name), StructuredTool.from_function(search_prescription_drugs_by_name), StructuredTool.from_function(search_all_products_by_name), StructuredTool.from_function(get_supplement_details), StructuredTool.from_function(get_prescription_drug_details), StructuredTool.from_function(get_product_details), StructuredTool.from_function(list_all_supplements), StructuredTool.from_function(list_all_prescription_drugs), StructuredTool.from_function(list_all_products), StructuredTool.from_function(create_payment_link) ] # Export data for use in main.py def get_supplements(): return SUPPLEMENTS def get_prescription_drugs(): return PRESCRIPTION_DRUGS def get_available_products() -> Dict: """ Get a list of all available products in the catalog. Returns: A dictionary containing available products categorized by type """ # Load the products catalog with open("src/payments/products.json", "r") as f: products_catalog = json.load(f) # Separate into supplements, prescription drugs, and services supplements = [] prescription_drugs = [] services = [] for product_id in products_catalog.keys(): if product_id == "NP_consultation": services.append(product_id) # Add consultation as a service elif product_id in PRESCRIPTION_DRUGS: prescription_drugs.append(product_id) elif product_id in SUPPLEMENTS: supplements.append(product_id) return { "supplements": supplements, "prescription_drugs": prescription_drugs, "services": services, "all_products": list(products_catalog.keys()) } if __name__ == "__main__": sample_request = '1 Finasteride consultation, 1 Minoxidil consultation' result = process_purchase_request(sample_request) print(result) breakpoint() asd = 1 # You can test with other examples # result2 = process_purchase_request("I want to buy 3 items of product_i and 2 of product_k") # print(result2)