app.py

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

#make token for every part of calls to detect issues faster in testing or in beta release:
#fast-api

GEMMA_TOKEN=os.environ.get("NEBIUS_API_KEY_GEMMA")
#To be used for the main requests
DEEPSEEKv3_TOKEN=os.environ.get("NEBIUS_API_KEY_DEEPSEEK")
DEEPSEEKv3_FAST_TOKEN=os.environ.get("NEBIUS_API_KEY_DEEPSEEK_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 []




def analyze_symptoms_from_text_ai(text_input: str) -> str:
    """
    Analyze user-provided text to extract medical symptoms in CSV format.
    
    Args:
        text_input: Raw text description of health condition
        
    Returns:
        str: Comma-separated list of symptoms in CSV format
        
    Example:
        >>> analyze_symptoms_from_text_ai("I have headache and nausea")
        'headache,nausea'
    """
    
    client = OpenAI(
        base_url="https://api.studio.nebius.com/v1/",
        api_key=GEMMA_TOKEN
    )
    
    try:
        response = client.chat.completions.create(
            model="google/gemma-2-2b-it",
            max_tokens=512,
            temperature=0.5,
            top_p=0.9,
            extra_body={"top_k": 50},
            messages=[
                {
                    "role": "system",
                    "content": """You are a medical symptom extractor. 
                    Analyze the text and return ONLY a comma-separated list of symptoms in CSV format.
                    Example input: "I have headache and feel nauseous"
                    Example output: headache,nausea
                    Return "" , IF NO SYMTPOMS OR DIESEASE IN THE TEXT"""
                },
                {
                    "role": "user",
                    "content": f"Analyze this text for symptoms and list them in CSV format: {text_input}"
                }
            ]
        )
        
        # Extract and clean the response
        try:
            symptoms_csv = response.choices[0].message.content.strip()
            return symptoms_csv
        except Exception as e:
            print ("Error:",e,"\nType 'response': ",type(response),"\n",response) 
        
        
    except Exception as e:
        print(f"Error analyzing symptoms: {str(e)}")
        return ""
 
#@tool
def full_treatment_answer_ai(user_input: str) -> str:
    """
    Searches for plant treatments based on user symptoms using Nebius API.
    
    Args:
        user_input: User's symptom description
        
    Returns:
        str: Complete treatment plan with plant recommendations and details
    """
    # Initialize Nebius client
    client = OpenAI(
        base_url="https://api.studio.nebius.com/v1/",
        api_key=DEEPSEEKv3_TOKEN
    )
    
    # Step 1: Extract symptoms from user input
    text_symptoms = user_input#analyze_symptoms_from_text_ai(user_input)
    symptoms_list = text_symptoms.split(",")
    symptoms = get_unique_symptoms(symptoms_list)
    
    if not symptoms:
        return "Could not identify any specific symptoms. Please describe your condition in more detail."
    
    # Step 2: Find relevant plants for each symptom
    all_related_plants = []
    for symptom in symptoms:
        related_plants = lookup_symptom_and_plants(symptom)
        all_related_plants.extend(related_plants)
    
    # Step 3: Prepare the prompt for Nebius API
    plants_info = ""
    if all_related_plants:
        plants_info = "Here are some plants that might be relevant:\n"
        for plant in all_related_plants[:6]:  # Limit to top 6 plants
            plants_info += f"""\nPlant: {plant['name']}
{plant['description']}
Cures: {plant['treatable_conditions']}
Dosage: {plant['dosage']}\n"""
    else:
        plants_info = "I dont know specific plants for these symptoms. 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 if image is available]

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

    # Step 4: Call Nebius API for treatment recommendation
    try:
        response = client.chat.completions.create(
            model="deepseek-ai/DeepSeek-V3-0324-fast",
            max_tokens=1024,  # Increased for detailed responses
            temperature=0.3,
            top_p=0.95,
            messages=[
                {
                    "role": "system",
                    "content": """You are a professional botanist assistant specializing in medicinal plants. 
                    Provide accurate, science-backed recommendations for plant-based treatments. 
                    Include dosage, preparation methods, and safety considerations."""
                },
                {
                    "role": "user",
                    "content": prompt
                }
            ]
        )
        
        final_result = response.choices[0].message.content
        
        # Step 5: Append detailed plant information if available
        if all_related_plants:
            final_result += "\n---\nMore Details for Recommended Plants:\n"
            for plant in all_related_plants[:3]:  # Show details for top 3 plants
                final_result += f"""
**Plant Name**: {plant['name']}
**Scientific Name**: {plant['scientific_name']}
**Other Names**: {plant['alternate_names']}
**Description**: {plant['description']}
**Treatable Conditions**: {plant['treatable_conditions']}
**Preparation**: {plant['preparation_methods']}
**Dosage**: {plant['dosage']}
**Side Effects**: {plant['side_effects']}
**Contraindications**: {plant['contraindications']}
**Sources**: {plant['sources']}
"""
        
        return final_result
        
    except Exception as e:
        print(f"Error generating treatment plan: {str(e)}")
        return "Sorry, I couldn't generate a treatment plan at this time. Please try again later."

#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 """
    response_text=analyze_symptoms_from_text_ai(text_input)
    if response_text and len(response_text)>2:
        if "error" not in response_text.lower():
            return response_text
    return ""

class BotanistAssistant:
    def __init__(self, api_endpoint: str = "https://api.studio.nebius.com/v1/"): #https://api.studio.nebius.com/v1/chat/completions
        self.api_endpoint = api_endpoint
        self.system_message = "You are a botanist assistant that extracts and structures information about medicinal plants."
        self.client = OpenAI(
            base_url="https://api.studio.nebius.com/v1/",
            api_key=DEEPSEEKv3_FAST_TOKEN
        )
    
    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 remove duplicates of 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": "Analyzes text to extract medical symptoms and returns them in CSV format.",
            "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": "full_treatment_answer_ai",
            "description": "Generates complete plant-based treatment plans for given symptoms, including dosage, preparation methods, and safety information.",
            "api": {
                "name": "full_treatment_answer_ai",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "user_input": {
                            "type": "string",
                            "description": "User's description of symptoms or health condition"
                        }
                    },
                    "required": ["user_input"]
                }
            }
        }
    ]
    
    def _call_tool(self, tool_name: str, parameters: Dict[str, Any]) -> Any:
        """Executes the specified tool with given parameters."""
        if tool_name == "full_treatment_answer_ai":
            return self.full_treatment_answer_ai(**parameters)
        elif tool_name == "analyze_symptoms_from_text_ai":
            return self.analyze_symptoms_from_text_ai(**parameters)
        else:
            raise ValueError(f"Unknown tool: {tool_name}")
            
    def _call_assistant_api(self, messages: List[Dict[str, str]]) -> Dict[str, Any]:
        """Makes the API call to Nebius assistant service with no tool support.""" #TODO tool
        
        try:
            response = self.client.chat.completions.create(
            model="deepseek-ai/DeepSeek-V3-0324-fast",
            max_tokens=1024,
            temperature=0.3,
            top_p=0.9,
            extra_body={"top_k": 50},
            messages=messages)
            return  response.choices[0].message.content #TODO if not working
            #return response.to_json()
            
        except Exception as e:
            print(f"Error: {str(e)}")
            return {"error": str(e)}
   

#assistant = BotanistAssistant()
# Simple query
# Treatment query (will automatically use tools)
#for response in assistant.respond("I have migraines and trouble sleeping", []):
#    print(response)
    
    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 api_response:
            yield api_response
        '''
        if "choices" in api_response:
            for choice in api_response["choices"][0]:
                if "message" in choice and "content" in choice["message"]:
                    yield choice["message"]["content"]
        '''        

def create_app(api_endpoint: str = "https://api.studio.nebius.com/v1/") -> 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 = "https://api.studio.nebius.com/v1/" # Replace with your actual endpoint
    app = create_app()
    app.launch(
        server_name="0.0.0.0",
        server_port=7860,
        share=False,
        favicon_path="🌿"  # Optional: Add path to plant icon
    )