app.py

from typing import List, Tuple, Dict, Any, Generator
import sqlite3 
import urllib
import requests
import os
import gradio as gr
from smolagents import tool
import time

#specialtoken=os.getenv("SPECIALTOKEN")+"deepseek-r1-0528"
specialtoken=os.getenv("SPECIALTOKEN")+"openai"#"openai-large"#"openai-roblox"
fasttoken=os.getenv("SPECIALTOKEN")+"models/"+"openai-fast"

#unused for getting all symptoms from plants db
def get_all_treatable_conditions()->List:
    #Gets all the symptoms:
    conn = sqlite3.connect('plants.db')
    # Create a cursor object
    cursor = conn.cursor()
    # Execute a query to retrieve data
    cursor.execute("SELECT treatable_conditions FROM 'plants'")
    # Fetch all results
    rows = cursor.fetchall()
    # Print the results
    treatable_conditions=[]
    for row in rows:
        treatable_conditions.append(row)
    # Close the connection
    conn.close()
    return treatable_conditions

#Unused for writing db
def write_symptoms_into_db(data_list):
    """Initialize SQLite database"""
    try:
        conn = sqlite3.connect('plants.db')
        c = conn.cursor()
        c.execute('''CREATE TABLE IF NOT EXISTS symptoms (name TEXT)''')
        conn.commit()
        for each in data_list:
            insert_sql = f'''INSERT INTO 'symptoms' (name) VALUES ("{each}")'''
            c.execute(insert_sql)
            conn.commit()
        print("Database created successfully!")
        conn.close()
    except sqlite3.Error as e:
        print(f"An error occurred: {e}")

#@tool
def get_unique_symptoms(list_text:List[str])->List[str]:
    """Processes and deduplicates symptom descriptions into a normalized list of unique symptoms.

    Performs comprehensive text processing to:
    - Extract individual symptoms from complex descriptions
    - Normalize formatting (removes common connecting words and punctuation)
    - Deduplicate symptoms while preserving original meaning
    - Handle multiple input formats (strings and tuples)

    Args:
        list_text: List of symptom descriptions in various formats.
                   Each element can be:
                   - String: "fever and headache" 
                   - Tuple: ("been dizzy and nauseous",)
                   Example: ["fatigue, nausea", "headache and fever"]

    Returns:
        List of unique, alphabetically sorted symptom terms in lowercase.
        Returns empty list if:
        - Input is empty
        - No valid strings found
        Example: ['bleed', 'fever', 'headache']

    Processing Details:
        1. Text normalization:
           - Removes connecting words ("and", "also", "like", etc.)
           - Replaces punctuation with spaces
           - Converts to lowercase
        2. Special cases:
           - Handles tuple inputs by extracting first element
           - Skips non-string/non-tuple elements with warning
        3. Deduplication:
           - Uses set operations for uniqueness
           - Returns sorted list for consistency

    Examples:
        >>> get_unique_symptoms(["fever and headache", "bleed"])
        ['bleed', 'fever', 'headache']

        >>> get_unique_symptoms([("been dizzy",), "nausea"])
        ['dizzy', 'nausea']

        >>> get_unique_symptoms([123, None])
        No Correct DataType
        []

    Edge Cases:
        - Empty strings are filtered out
        - Single-word symptoms preserved
        - Mixed punctuation handled
        - Warning printed for invalid types
    """

    all_symptoms = []
    
    for text in list_text:
        # Handle potential errors in input text.  Crucial for robustness.
        if type(text)==tuple:
            text=text[0]
            symptoms = text.replace(" and ", " ").replace(" been ", " ").replace(" also ", " ").replace(" like ", " ").replace(" due ", " ").replace(" a ", " ").replace(" as ", " ").replace(" an ", " ").replace(",", " ").replace(".", " ").replace(";", " ").replace("(", " ").replace(")", " ").split()
            symptoms = [symptom.strip() for symptom in symptoms if symptom.strip()]  # Remove extra whitespace and empty strings
            all_symptoms.extend(symptoms)
        elif type(text)==str and len(text)>1:
            symptoms = text.replace(" and ", " ").replace(" been ", " ").replace(" also ", " ").replace(" like ", " ").replace(" due ", " ").replace(" a ", " ").replace(" as ", " ").replace(" an ", " ").replace(",", " ").replace(".", " ").replace(";", " ").replace("(", " ").replace(")", " ").split()
            symptoms = [symptom.strip() for symptom in symptoms if symptom.strip()]  # Remove extra whitespace and empty strings
            all_symptoms.extend(symptoms)
        else:
            print ("No Correct DataType or 1 charater text O.o")
        #if not isinstance(text, str) or not text:
        #continue

    unique_symptoms = sorted(list(set(all_symptoms)))  # Use set to get unique items, then sort for consistency
    
    return unique_symptoms

#@tool
def lookup_symptom_and_plants(symptom_input:str)->List:
    """Search for medicinal plants that can treat a given symptom by querying a SQLite database.
    
    This function performs a case-insensitive search in the database for:
    1. First looking for an exact or partial match of the symptom
    2. If not found, searches for individual words from the symptom input
    3. Returns all plants that list the matched symptom in their treatable conditions
    
    Args:
        symptom_input (str): The symptom to search for (e.g., "headache", "stomach pain").
                            Can be a single symptom or multiple words. Leading/trailing
                            whitespace is automatically trimmed.

    Returns:
        List[dict]: A list of plant dictionaries containing all columns from the 'plants' table
                   where the symptom appears in treatable_conditions. Each dictionary represents
                   one plant with column names as keys. Returns empty list if:
                   - Symptom not found
                   - No plants treat the symptom
                   - Database error occurs

    Raises:
        sqlite3.Error: If there's a database connection or query error (handled internally,
                      returns empty list but prints error to console)

    Notes:
        - The database connection is opened and closed within this function
        - Uses LIKE queries with wildcards for flexible matching
        - Treatable conditions are expected to be stored as comma-separated values
        - Case-insensitive matching is performed by converting to lowercase
        - The function will attempt to match individual words if full phrase not found

    Example:
        >>> lookup_symptom_and_plants("headache")
        [{'name': 'Neem', 'scientific_name': 'Azadirachta indica', 'alternate_names': 'Indian Lilac, Margosa Tree', 'description': 'Neem is a tropical evergreen tree known for its extensive medicinal properties. Native to the Indian subcontinent, it has been used for thousands of years in traditional medicine systems like Ayurveda and Unani. Various parts of the tree including fruits, seeds, oil, leaves, roots, and bark have therapeutic benefits.', 'plant_family': 'Meliaceae', 'origin': 'Indian subcontinent', 'growth_habitat': 'Tropical and subtropical regions, often found in dry and arid soils', 'active_components': 'Azadirachtin, Nimbin, Nimbidin, Sodium nimbidate, Quercetin', 'treatable_conditions': 'Skin diseases, infections, fever, diabetes, dental issues, inflammation, malaria, digestive disorders', 'preparation_methods': 'Leaves and bark can be dried and powdered; oil extracted from seeds; decoctions and infusions made from leaves or bark', 'dosage': 'Varies depending on preparation and condition; oils generally used topically, leaf powder doses range from 500 mg to 2 grams daily when taken orally', 'duration': 'Treatment duration depends on condition, often several weeks to months for chronic ailments', 'contraindications': 'Pregnant and breastfeeding women advised to avoid internal consumption; caution in people with liver or kidney disease', 'side_effects': 'Possible allergic reactions, nausea, diarrhea if consumed in excess', 'interactions': 'May interact with blood sugar lowering medications and immunosuppressants', 'part_used': 'Leaves, seeds, bark, roots, oil, fruits', 'harvesting_time': 'Leaves and fruits commonly harvested in summer; seeds collected when fruits mature', 'storage_tips': 'Store dried parts in airtight containers away from direct sunlight; oils kept in cool, dark places', 'images': '', 'related_videos': '', 'sources': 'https://www.ncbi.nlm.nih.gov/pmc/articles/PMC3769010/, https://www.who.int/medicines/areas/traditional/overview/en/'}]
        
        >>> lookup_symptom_and_plants("unknown symptom")
        []
    """
    symptom_lower = symptom_input.strip().lower()
    try:
        conn = sqlite3.connect('plants.db')
        conn.row_factory = sqlite3.Row
        c = conn.cursor()

        # Check if the symptom exists in symptoms table (case-insensitive)
        c.execute(f'''SELECT name FROM "symptoms" WHERE LOWER("name") LIKE "%{symptom_lower}%"''')
        result = c.fetchone()
        #if result:
        #    result=result[0]

        if not result:
            #get any 1 symptom only in case the text is not showing.
            symptoms=symptom_lower.split(" ")
            for each in symptoms:
                c.execute(f'''SELECT name FROM "symptoms" WHERE LOWER("name") LIKE "%{each}%"''')
                result = c.fetchone()
                if result:
                    print(f"Symptom '{symptom_input}' finally found in the database.")
                    #result=result[0]
                    symptom_lower=each
                    break
            if not result:
                print(f"Symptom '{symptom_input}' not found in the database.")
                conn.close()
                return []

        # If symptom exists, search in plants table for related plants
        # Assuming 'TreatableConditions' is a comma-separated string
        query = f"""SELECT * FROM plants WHERE LOWER(treatable_conditions) LIKE "%{symptom_lower}%" """
        c.execute(query)
        plants_rows = c.fetchall()

        plants_list = []
        for row in plants_rows:
            # Convert row to dict for easier handling
            plant_dict = {key: row[key] for key in row.keys()}
            plants_list.append(plant_dict)

        conn.close()
        return plants_list

    except sqlite3.Error as e:
        print(f"Database error: {e}")
        return []


#@tool
def analyze_symptoms_from_text_ai(user_input:str)->str:
    """Analyze user-provided text to extract medical symptoms in CSV format.
    
    This function sends the user's text description to an AI service which identifies
    and returns potential medical symptoms in a comma-separated string format.
    
    Args:
        user_input (str): The text containing symptom descriptions to be analyzed.
                          Example: "I feel some pain in head and I feel dizzy"
        
    Returns:
        str: A comma-separated string of identified symptoms.
             Example: "pain in head,dizzy"
             
    Note:
        This function requires an internet connection as it makes an API call to
        an external AI service. The service URL is expected to be available in
        the 'fasttoken' variable.
        
    Example:
        >>> analyze_symptoms_from_text_ai("I have a headache and nausea")
        'headache,nausea'
    """
    #user_input="""i feel some pain in head and i feel dizzy""" #user input
    #text_symptoms='pain in head,dizzy' #returned output
    prompt='''Analyze this text for possible symptoms and list them in only csv format by comma in 1 line from the following text : """{user_input}""" Return "" if no symptoms found.'''
    prompt=urllib.parse.quote(prompt)
    response = requests.get(f"{fasttoken}/{prompt}")
    text_symptoms=response.text
    
    return text_symptoms
 
#@tool
def full_treatment_answer_ai(user_input:str)->str:
    """
    Searches a database for treatable conditions based on user-input and getting from it symptoms and plants needed for the treatment.

    Args:
        user_input (str): A string representing the user's full symptoms.

    Returns:
        str: Ready Full treatment plan with plants answer about the symptoms  (and with side effects of the plants if available).
    """
    all_related_plants=[]
    """
    old Return:
        A list of dictionaries, where each dictionary represents a treatable condition
        and its associated symptoms. Returns an empty list if no matching conditions are found.
        Returns None if there's an error connecting to or querying the database.
    """
    #EXAMPLE:
    '''[{'name': 'Neem',
     'scientific_name': 'Azadirachta indica',
     'alternate_names': 'Indian Lilac, Margosa Tree',
     'description': 'Neem is a tropical evergreen tree known for its extensive medicinal properties. Native to the Indian subcontinent, it has been used for thousands of years in traditional medicine systems like Ayurveda and Unani. Various parts of the tree including fruits, seeds, oil, leaves, roots, and bark have therapeutic benefits.',
     'plant_family': 'Meliaceae',
     'origin': 'Indian subcontinent',
     'growth_habitat': 'Tropical and subtropical regions, often found in dry and arid soils',
     'active_components': 'Azadirachtin, Nimbin, Nimbidin, Sodium nimbidate, Quercetin',
     'treatable_conditions': 'Skin diseases, infections, fever, diabetes, dental issues, inflammation, malaria, digestive disorders',
     'preparation_methods': 'Leaves and bark can be dried and powdered; oil extracted from seeds; decoctions and infusions made from leaves or bark',
     'dosage': 'Varies depending on preparation and condition; oils generally used topically, leaf powder doses range from 500 mg to 2 grams daily when taken orally',
     'duration': 'Treatment duration depends on condition, often several weeks to months for chronic ailments',
     'contraindications': 'Pregnant and breastfeeding women advised to avoid internal consumption; caution in people with liver or kidney disease',
     'side_effects': 'Possible allergic reactions, nausea, diarrhea if consumed in excess',
     'interactions': 'May interact with blood sugar lowering medications and immunosuppressants',
     'part_used': 'Leaves, seeds, bark, roots, oil, fruits',
     'harvesting_time': 'Leaves and fruits commonly harvested in summer; seeds collected when fruits mature',
     'storage_tips': 'Store dried parts in airtight containers away from direct sunlight; oils kept in cool, dark places',
     'images': '',
     'related_videos': '',
     'sources': 'https://www.ncbi.nlm.nih.gov/pmc/articles/PMC3769010/, https://www.who.int/medicines/areas/traditional/overview/en/'}]'''
    text_symptoms=user_input#analyze_symptoms_from_text_ai(user_input)
    symptoms_list=text_symptoms.split(",")
    symptoms=get_unique_symptoms(symptoms_list)
    
    for symptom in symptoms: 
        related_plants = lookup_symptom_and_plants(symptom)
        for each in related_plants:
            all_related_plants.append(each)
    plants_info=""
    if len(all_related_plants)>0:
        plants_info="Here are some plants that might be relevant:\n"
        for each in all_related_plants[:6]: #GET top 7 if available
            plants_info+="""Plant:{each['name']}\n{each['description']}\nCures:{each['treatable_conditions']}\nDosage:{each['dosage']}.\n """
    else:
        plants_info="No Plants found , please get useful plants from anywhere or any other sources."
    prompt=f"""I have these symptoms: {",".join(symptoms)}. 

{plants_info}

Please analyze my symptoms and recommend the most appropriate plant remedy. 
Consider the symptoms, when to use each plant, dosage, and how to use it.
Provide your recommendation in this format:

**Recommended Plant**: [plant name]
**Reason**: [why this plant is good for these symptoms]
**Dosage**: [recommended dosage]
**Instructions**: [how to use it]
**Image**: [mention the image is available if applicable]

If multiple plants could work well, you may recommend up to 3 options."""

    #filter the names and description, treatments only and send it in post to AI
    prompt=urllib.parse.quote(prompt)
    response = requests.get(f"{fasttoken}/{prompt}")
    final_result=response.text
    
    #Get the answer and send it to user then post all data for each plant in a good form for user to view

    if len(all_related_plants)>0:
        final_result+="\n---\nMore Details for Plants:\n"
        for each in all_related_plants[:4]:
            final_result+=f"""\n**Plant Name**: {each['name']}
**Other Names**: {each['scientific_name']} , {each['alternate_names']}
**Description**: {each['description']}
**Plant Family**: {each['plant_family']}, **Origin**: {each['origin']}
**Treatment Conditions**: {each['treatable_conditions']}
**Preparation Methods**: {each['preparation_methods']}
**Side Effects**: {each['side_effects']}    
**Storage Tips**: {each['storage_tips']}
**Info. Sources**: {each['sources']}
"""
    return final_result

#example:
# user_input="""i feel some pain in head and i feel dizzy""" #user input
# prompt='''Analyze the possible symptoms and list them in only csv format by comma in 1 line,from the following text : """{user_input}"""'''
# output='pain in head,dizzy'
# symptoms=output.split(",")
# get_unique_symptoms(symptoms)


#Filter all conditions:
#all_treatment_conditions=get_all_treatable_conditions()
#unique_symptoms=get_unique_symptoms(all_treatment_conditions)
#write_symptoms_into_db(unique_symptoms)
#related_plants = lookup_symptom_and_plants("fever")

'''
user_input = "fever"
related_plants = lookup_symptom_and_plants(user_input)
if related_plants:
    print(f"Plants related to '{user_input}':")
    for plant in related_plants:
        print(f"- {plant['name']}")
else:
    print(f"No plants found for symptom '{user_input}'.")
'''
def is_symtoms_intext_ai(text_input:str)->str: #used instead of: analyze_symptoms_from_text_ai()
    """Gives Symptoms or "" if no symptoms """
    prompt='''Analyze this text for possible symptoms and list them in only csv format by comma in 1 line from the following text : """{user_input}"""
    ---
    Return "" , IF NO SYMTPOMS OR DIESEASE IN THE TEXT
    '''
    prompt=urllib.parse.quote(prompt)
    response = requests.get(f"{fasttoken}/{prompt}")
    response_text=response.text
    if len(response_text)>2:
        if "error" not in response_text.lower():
            return response_text
    return ""

class BotanistAssistant:
    def __init__(self, api_endpoint: str):
        self.api_endpoint = api_endpoint
        self.system_message = "You are a botanist assistant that extracts and structures information about medicinal plants."
        
    def _build_chat_history(self, history: List[Tuple[str, str]]) -> List[Dict[str, str]]:
        """Formats chat history into API-compatible message format."""
        messages = [{"role": "system", "content": self.system_message}]
        for user_msg, bot_msg in history:
            if user_msg:
                messages.append({"role": "user", "content": user_msg})
            if bot_msg:
                messages.append({"role": "assistant", "content": bot_msg})
        return messages
    
    def _get_tools_schema(self) -> List[Dict[str, Any]]:
        """Returns the complete tools schema for plant medicine analysis."""
        return [
        {
            "name": "get_unique_symptoms",
            "description": "Extracts and deduplicates symptoms from text input. Handles natural language processing to identify individual symptoms from complex descriptions.",
            "api": {
                "name": "get_unique_symptoms",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "list_text": {
                            "type": "array",
                            "items": {"type": "string"},
                            "description": "List of symptom descriptions (strings or tuples)"
                        }
                    },
                    "required": ["list_text"]
                }
            }
        },
        {
            "name": "lookup_symptom_and_plants",
            "description": "Finds medicinal plants associated with specific symptoms from the database. Returns matching plants with their treatment properties.",
            "api": {
                "name": "lookup_symptom_and_plants",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "symptom_input": {
                            "type": "string",
                            "description": "Individual symptom to search for plant treatments"
                        }
                    },
                    "required": ["symptom_input"]
                }
            }
        },
        {
            "name": "analyze_symptoms_from_text_ai",
            "description": "AI-powered symptom analysis that interprets natural language descriptions of health conditions and extracts medically relevant symptoms.",
            "api": {
                "name": "analyze_symptoms_from_text_ai",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "text_input": {
                            "type": "string",
                            "description": "Raw text description of health condition"
                        }
                    },
                    "required": ["text_input"]
                }
            }
        },
        {
            "name": "search_for_treatment_answer_ai",
            "description": "Comprehensive treatment finder that takes symptoms, analyzes them, and returns complete plant-based treatment plans with dosage instructions.",
            "api": {
                "name": "search_for_treatment_answer_ai",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "list_symptoms": {
                            "type": "array",
                            "items": {"type": "string"},
                            "description": "List of identified symptoms"
                        }
                    },
                    "required": ["list_symptoms"]
                }
            }
        }
    ]
    
    def _call_assistant_api(self, messages: List[Dict[str, str]]) -> Dict[str, Any]:
        """Makes the API call to the assistant service."""
        payload = {
            "model": "openai",
            "messages": messages,
            #"tools": self._get_tools_schema(), TOOLS REMOVED 
            #"tool_choice": "auto" #TODO fix tools 
        }
        
        try:
            response = requests.post(
                self.api_endpoint,
                json=payload,
                headers={"Content-Type": "application/json"},
                timeout=30  # 30-second timeout
            )
            response.raise_for_status()
            return response.json() #-_-
        except requests.exceptions.RequestException as e:
            print(f"API request failed: {str(e)}")
            return {"error": str(e)}
    
    def respond(
        self,
        message: str,
        history: List[Tuple[str, str]],
        system_message: str = None
    ) -> Generator[str, None, None]:
        """Handles the chat response generation."""
        # Update system message if provided
        if system_message:
            self.system_message = system_message
            
        # Build API payload
        messages = self._build_chat_history(history)
        messages.append({"role": "user", "content": message})
        
        
        # Get API response
        api_response = self._call_assistant_api(messages)
        
        # Process response
        if "error" in api_response:
            yield "Error: Could not connect to the assistant service. Please try again later."
            return
        
        
        treatment_response=""
        symtoms_intext=is_symtoms_intext_ai(message)
        if symtoms_intext:
            time.sleep(1.6)
            treatment_response=full_treatment_answer_ai(symtoms_intext)
            #yield treatment_response
        
        if len(treatment_response):
            treatment_response="\n**I have found that this may help you alot:**\n"+treatment_response
            yield treatment_response
        
        # Get treatment information
        #treatment_response = search_for_treatment_answer_ai(message)
        #yield treatment_response  # First yield the treatment info
        
        # Stream additional assistant responses if available
        if "choices" in api_response:
            for choice in api_response["choices"]:
                if "message" in choice and "content" in choice["message"]:
                    yield choice["message"]["content"].split("**Sponsor**")[0].split("Sponsor")[0]

def create_app(api_endpoint: str) -> gr.Blocks:
    """Creates and configures the Gradio interface."""
    assistant = BotanistAssistant(api_endpoint)
    
    with gr.Blocks(title="Botanist Assistant", theme=gr.themes.Soft()) as demo:
        gr.Markdown("# 🌿 Natural Medical Assistant")
        gr.Markdown("Describe your symptoms to get natural plant-based treatment recommendations")
        
        with gr.Row():
            with gr.Column(scale=3):
                chat = gr.ChatInterface(
                    assistant.respond,
                    additional_inputs=[
                        gr.Textbox(
                            value=assistant.system_message,
                            label="System Role",
                            interactive=True
                        )
                    ]
                )
            with gr.Column(scale=1):
                gr.Markdown("### Common Symptoms")
                gr.Examples(
                    examples=[
                        ["I have headache and fever"],
                        ["got nausea, and injury caused bleeding"],
                        ["I feel insomnia and anxiety"],
                        ["I am suffering from ADHD"]
                    ],
                    inputs=chat.textbox,
                    label="Try these examples"
                )
                
        gr.Markdown("---")
        gr.Markdown("""> Note: Recommendations can work as real cure treatment but you can consider it as informational purposes only.
Also You can consult a Natrual healthcare professional before use.""")

    
    return demo

if __name__ == "__main__":
    API_ENDPOINT = specialtoken # Replace with your actual endpoint
    app = create_app(API_ENDPOINT)
    app.launch(
        server_name="0.0.0.0",
        server_port=7860,
        share=False,
        favicon_path="🌿"  # Optional: Add path to plant icon
    )