import os import gradio as gr from duckduckgo_search import DDGS import sqlite3 import json import requests from typing import List, Dict, Optional import time from bs4 import BeautifulSoup import urllib.parse #specialtoken=os.getenv("SPECIALTOKEN")+"deepseek-r1-0528" specialtoken=os.getenv("SPECIALTOKEN")+"models/"+"openai-large"#"openai-roblox" #plants=['Turmeric', 'Aloe Vera', 'Neem', 'Tulsi', 'Ashwagandha', 'Ginger', 'Basil', 'Peppermint', 'Lavender', 'Eucalyptus', 'Chamomile', 'Sandalwood', 'Giloy', 'Haritaki', 'Brahmi', 'Gotu Kola', 'Holy Basil', 'Fenugreek', 'Licorice', 'Fennel', 'Cinnamon', 'Clove', 'Black Pepper', 'Cardamom', 'Neem', 'Indian Gooseberry', 'Saffron', 'Thyme', 'Valerian', 'Marigold', 'Ginseng', 'Dandelion', 'Hibiscus', 'Milk Thistle', 'Magnolia', "St. John's Wort", 'Yarrow', 'Calendula', 'Coriander', 'Senna', 'Echinacea', 'Moringa', 'Plantain', 'Amla', 'Shatavari', 'Peppermint', 'Chamomile', 'Gotu Kola', 'Ashoka', 'Arnica', 'Burdock Root', "Cat's Claw", "Devil's Claw", 'Elderberry', 'Feverfew', 'Ginkgo Biloba', 'Goldenseal', 'Hawthorn', 'Kava', 'Lemon Balm', 'Marshmallow Root', 'Nettle', 'Olive Leaf', 'Passionflower', 'Red Clover', 'Reishi Mushroom', 'Rhodiola', 'Sage', 'Saw Palmetto', 'Slippery Elm', 'Stinging Nettle', 'Witch Hazel', 'Yellow Dock', 'Ashitaba', 'Bael', 'Bacopa', 'Cumin', 'Guduchi', 'Jamun', 'Jatamansi', 'Karela', 'Gudmar', 'Schisandra', 'Baikal Skullcap', 'Mullein', 'Chrysanthemum', 'Catuaba', 'Dong Quai', 'Jiaogulan', 'Muira Puama', 'Catnip', 'Olive'] #plants = ["Echinacea", "Ginkgo biloba", "Turmeric"] PROMPT_TEMPLATE = """Extract detailed information about a plant from any reliable source. Provide the information in a JSON object with the following keys: ["Name", "Scientific Name", "Alternate Names", "Description", "Plant Family", "Origin", "Growth Habitat", "Active Components", "Treatable Conditions", "Preparation Methods", "Dosage", "Duration", "Contraindications", "Side Effects", "Interactions", "Part Used", "Harvesting Time", "Storage Tips", "Images", "Related Videos", "Sources"]. Use only string values for each key. If any information is missing, set the value as an empty string (""). Format the output as a valid JSON object with proper syntax. Example: For the plant 'Name', provide the information accordingly. --- **Plant Name:** {plant_name} Additional source of information about that plant: '''{content}''' Please generate the JSON output for the above plant, ensuring it adheres to the specified keys and format. """ def fetch_page_content(url: str): """Get webpage content with error handling""" try: response = requests.get(url, timeout=10) soup = BeautifulSoup(response.text, 'html.parser') # Remove unwanted elements for element in soup(['script', 'style', 'header', 'footer', 'nav']): element.decompose() text = soup.get_text(separator='\n', strip=True) for each in ["Page not available","403 Forbidden"]: if each in text: return "No information found!" return text[:3500] # Limit to 3.5k characters except Exception as e: return f"Error fetching page: {str(e)}" def search_full_plant_information(plant_name:str): """ """ query = f"{plant_name} plant medicinal uses scientific information site:.edu OR site:.gov OR site:.org" search_results="" for attempt in range(3): # Retry up to 3 times try: search_results = DDGS().text(keywords=query, max_results=5) except Exception as e: if "Ratelimit" in str(e): try: wait_time = 2 ** attempt # Exponential backoff print(f"Rate limit hit, retrying in {wait_time} seconds...") time.sleep(wait_time) except Exception as e: pass if search_results: content="" for result in search_results: #content+=requests.get(f"{specialtoken}/Analyze this:{result['body']}").text content+=result['body']+" " content+=fetch_page_content(result['href'])+" " time.sleep(2) prompt = PROMPT_TEMPLATE.format(plant_name=plant_name, content=content) #encodeURI prompt=urllib.parse.quote(prompt) response = requests.get(f"{specialtoken}/{prompt}") #print (response.text) return response.text else: #URI encoded: content="Get any information from any source about:"+plant_name prompt = PROMPT_TEMPLATE.format(plant_name=plant_name, content=content) #encodeURI prompt=urllib.parse.quote(prompt) response = requests.get(f"{specialtoken}/{prompt}") print(prompt) return response.text return f"No data Found for {plant_name}!" DB_NAME="plants.db" def save_to_db(plant_data: Dict) -> bool: """Save processed plant data to database""" try: conn = sqlite3.connect(DB_NAME) cursor = conn.cursor() # Convert arrays to strings if they exist for field in ["Alternate Names", "Active Components", "Treatable Conditions", "Preparation Methods", "Contraindications", "Side Effects", "Interactions"]: if field in plant_data: if isinstance(plant_data[field], list): plant_data[field] = ", ".join(plant_data[field]) elif not isinstance(plant_data[field], str): plant_data[field] = str(plant_data[field]) columns = [] values = [] for key, value in plant_data.items(): if key.lower() == "error": # Skip error field continue columns.append(key.lower().replace(" ", "_")) values.append(str(value) if value else None) columns_str = ", ".join(columns) placeholders = ", ".join(["?"] * len(columns)) cursor.execute( f"INSERT INTO plants ({columns_str}) VALUES ({placeholders})", values ) conn.commit() conn.close() return True except Exception as e: print(f"Database save error: {e}") return False def process_plants(plants_array: List[str]) -> str: """Main processing pipeline""" results = [] for plant in plants_array: plant = plant.strip() if not plant: continue print(f"Processing {plant}...") plant_data = search_full_plant_information(plant) if plant_data: pass #save_success = save_to_db(plant_data) #plant_data["Database_Save_Success"] = save_success #results.append(plant_data) time.sleep(2) # Rate limiting return print(results) def split_and_search(text:str): all_data="" plants=text.split(",") for each in plants: sp=search_full_plant_information(each.strip()).replace('```json','').replace('```','').split('---')[0]+",\n" #.split('**Sponsored**')[0].split('**Sponsor**')[0] #yield sp if len(sp) < 10: emptydata="{"+f''' "Name": "No data for {plant_name}", "Scientific Name": "", "Alternate Names": "", "Description": "", "Plant Family": "", "Origin": "", "Growth Habitat": "", "Active Components": "", "Treatable Conditions": "", "Preparation Methods": "", "Dosage": "", "Duration": "", "Contraindications": "", "Side Effects": "", "Interactions": "", "Part Used": "", "Harvesting Time": "", "Storage Tips": "", "Images": "", "Related Videos": "", "Sources": ""'''+"}" sp=emptydata#f"No data Found for {plant_name}!" all_data+=sp time.sleep(1) return all_data #For View: def get_all_plants() -> List[Dict]: """Retrieve all plants from database""" try: conn = sqlite3.connect(DB_NAME) conn.row_factory = sqlite3.Row cursor = conn.cursor() cursor.execute("SELECT `_rowid_`,* FROM plants ORDER BY `_rowid_` DESC") plants = [dict(row) for row in cursor.fetchall()] conn.close() return plants except Exception as e: print(f"Database retrieval error: {e}") return [{"Error": "Failed to retrieve data from database"}] #use it here : #process_plants(plants) #or use interface: with gr.Blocks(title="AI-Powered Medicinal Plants Database") as app: gr.Markdown("# 🌿 AI-Powered Medicinal Plants Database") with gr.Tab("Fetch & Process Plants"): gr.Markdown("### Enter plant names (comma separated)") with gr.Row(): plant_input = gr.Textbox(label="Plant Names", placeholder="e.g., Neem, Peppermint, Aloe Vera") fetch_btn = gr.Button("Process Plants", variant="primary") output_area = gr.Textbox(label="AI-Processed Results", lines=12, interactive=False) #json_output = gr.JSON(label="AI-Processed Results") fetch_btn.click( fn=split_and_search, #fn=lambda x: process_plants([p.strip() for p in x.split(",")]), inputs=plant_input, outputs=output_area #json_output ) with gr.Tab("View Database"): gr.Markdown("### Stored Plant Information") with gr.Row(): refresh_btn = gr.Button("Refresh Data", variant="secondary") clear_db = gr.Button("Clear Database", variant="stop") db_table = gr.Dataframe( headers=["id", "name", "scientific_name", "description"], datatype=["number", "str", "str", "str"], col_count=(4, "fixed"), interactive=True ) refresh_btn.click( fn=get_all_plants, outputs=db_table ) if __name__ == "__main__": app.launch(debug=True, share=False)