Spaces:
Sleeping
Sleeping
| from fastapi import FastAPI, Query | |
| from pydantic import BaseModel | |
| import os | |
| from chatbot import crop_advice, fertilizer_advice, fertilizer_advice_from_query, soil_health_advice, weather_advice | |
| from utils import ( | |
| get_weather, | |
| get_mandi_price, | |
| get_pest_page, | |
| check_phishing_url, | |
| analyze_sentiment, | |
| get_government_schemes | |
| ) | |
| from firebase_helper import get_or_create_user, get_session_id, add_to_history, get_conversation_history, start_new_chat | |
| from translation_helper import detect_language, normalize_language_code, translate_to_english, translate_text | |
| from intent_router import analyze_intent, extract_coordinates, extract_url, extract_crop_name | |
| app = FastAPI(title="Farmer Advisory Chatbot API", version="1.1") | |
| # --- Request Schemas --- | |
| class CropRequest(BaseModel): | |
| query: str | |
| lat: float | |
| lon: float | |
| uid: str = "" | |
| session_id: str = "" | |
| language: str = "" # Auto-detect if not provided | |
| class FertilizerRequest(BaseModel): | |
| query: str | |
| uid: str = "" | |
| session_id: str = "" | |
| language: str = "" # Auto-detect if not provided | |
| class SoilRequest(BaseModel): | |
| soil_info: str | |
| uid: str = "" | |
| session_id: str = "" | |
| language: str = "" # Auto-detect if not provided | |
| class WeatherRequest(BaseModel): | |
| lat: float | |
| lon: float | |
| days: int = 3 | |
| uid: str = "" | |
| session_id: str = "" | |
| language: str = "" # Auto-detect if not provided | |
| class PhishingRequest(BaseModel): | |
| url: str | |
| uid: str = "" | |
| session_id: str = "" | |
| language: str = "" # Auto-detect if not provided | |
| class SentimentRequest(BaseModel): | |
| query: str | |
| uid: str = "farmer" | |
| session_id: str = "" | |
| language: str = "" # Auto-detect if not provided | |
| class SchemeRequest(BaseModel): | |
| text: str | |
| uid: str = "" | |
| session_id: str = "" | |
| language: str = "" # Auto-detect if not provided | |
| class UnifiedChatRequest(BaseModel): | |
| query: str | |
| uid: str = "" | |
| session_id: str = "" | |
| language: str = "" # Auto-detect if not provided | |
| lat: float = 0.0 # Optional, will be extracted from query if not provided | |
| lon: float = 0.0 # Optional, will be extracted from query if not provided | |
| # --- Endpoints --- | |
| def crop_endpoint(req: CropRequest): | |
| # Handle user and session | |
| uid = req.uid | |
| session_id = req.session_id | |
| # Always create/get session if uid is provided | |
| if uid: | |
| # Get or create user (without name/email) | |
| get_or_create_user(uid, "", "", "farmer") | |
| # Automatically get or create session (first time creates new, subsequent requests reuse) | |
| session_id = get_session_id(uid, session_id) | |
| else: | |
| session_id = "" # No uid means no session tracking | |
| # Detect or use provided language | |
| language = req.language if req.language else detect_language(req.query) | |
| language = normalize_language_code(language) | |
| # Get response with context (no soil/local_info) | |
| result = crop_advice(req.query, req.lat, req.lon, "", "", uid, session_id, language) | |
| # Store in history | |
| if uid and session_id: | |
| add_to_history(uid, session_id, req.query, result) | |
| return {"response": result, "session_id": session_id, "uid": uid, "language": language} | |
| def fertilizer_endpoint(req: FertilizerRequest): | |
| # Handle user and session | |
| uid = req.uid | |
| session_id = req.session_id | |
| # Always create/get session if uid is provided | |
| if uid: | |
| # Get or create user (without name/email) | |
| get_or_create_user(uid, "", "", "farmer") | |
| # Automatically get or create session (first time creates new, subsequent requests reuse) | |
| session_id = get_session_id(uid, session_id) | |
| else: | |
| session_id = "" # No uid means no session tracking | |
| # Detect or use provided language | |
| language = req.language if req.language else detect_language(req.query) | |
| language = normalize_language_code(language) | |
| # Translate query to English to extract crop, stage, soil_test info | |
| query_english = translate_to_english(req.query, language) if language != "en" else req.query | |
| # Extract crop, stage, and soil_test from query using AI or simple parsing | |
| # For now, pass the query directly to the chatbot which will handle it | |
| # The chatbot function will need to be updated to parse the query | |
| # Get response with context - pass query directly | |
| result = fertilizer_advice_from_query(req.query, query_english, uid, session_id, language) | |
| # Store in history | |
| if uid and session_id: | |
| add_to_history(uid, session_id, req.query, result) | |
| return {"response": result, "session_id": session_id, "uid": uid, "language": language} | |
| def soil_endpoint(req: SoilRequest): | |
| # Handle user and session | |
| uid = req.uid | |
| session_id = req.session_id | |
| # Always create/get session if uid is provided | |
| if uid: | |
| # Get or create user (without name/email) | |
| get_or_create_user(uid, "", "", "farmer") | |
| # Automatically get or create session (first time creates new, subsequent requests reuse) | |
| session_id = get_session_id(uid, session_id) | |
| else: | |
| session_id = "" # No uid means no session tracking | |
| # Detect or use provided language | |
| language = req.language if req.language else detect_language(req.soil_info) | |
| language = normalize_language_code(language) | |
| # Get response with context | |
| result = soil_health_advice(req.soil_info, uid, session_id, language) | |
| # Store in history | |
| if uid and session_id: | |
| add_to_history(uid, session_id, f"Soil health advice for: {req.soil_info}", result) | |
| return {"response": result, "session_id": session_id, "uid": uid, "language": language} | |
| def weather_endpoint(req: WeatherRequest): | |
| # Handle user and session | |
| uid = req.uid | |
| session_id = req.session_id | |
| # Always create/get session if uid is provided | |
| if uid: | |
| # Get or create user (without name/email) | |
| get_or_create_user(uid, "", "", "farmer") | |
| # Automatically get or create session (first time creates new, subsequent requests reuse) | |
| session_id = get_session_id(uid, session_id) | |
| else: | |
| session_id = "" # No uid means no session tracking | |
| # Check if it's a new session (before storing current conversation) | |
| is_new_session = False | |
| if uid and session_id: | |
| history = get_conversation_history(uid, session_id, limit=1) | |
| is_new_session = len(history) == 0 | |
| # Detect or use provided language | |
| language = req.language if req.language else "hi" # Default to Hindi for weather | |
| language = normalize_language_code(language) | |
| summary = get_weather(req.lat, req.lon, req.days, caller_query="", is_new_session=is_new_session, language=language) | |
| # Extract summary text if it's a dict | |
| summary_text = summary.get("message", str(summary)) if isinstance(summary, dict) else str(summary) | |
| # Get response with context | |
| advice = weather_advice(req.lat, req.lon, summary_text, uid, session_id, language) | |
| # Store in history | |
| if uid and session_id: | |
| user_query = f"Weather advice for location: {req.lat}, {req.lon}" | |
| add_to_history(uid, session_id, user_query, advice) | |
| return {"weather_summary": summary, "response": advice, "session_id": session_id, "uid": uid, "language": language} | |
| def pest_endpoint( | |
| farmer_id: str = Query("", description="Farmer ID"), | |
| uid: str = Query("", description="User ID"), | |
| session_id: str = Query("", description="Session ID"), | |
| language: str = Query("", description="Language code (hi, en, te, etc.)") | |
| ): | |
| # Always create/get session if uid is provided | |
| if uid: | |
| get_or_create_user(uid, "", "", "farmer") | |
| # Automatically get or create session (first time creates new, subsequent requests reuse) | |
| session_id = get_session_id(uid, session_id) | |
| else: | |
| session_id = "" # No uid means no session tracking | |
| # Check if it's a new session (before storing current conversation) | |
| is_new_session = False | |
| if uid and session_id: | |
| history = get_conversation_history(uid, session_id, limit=1) | |
| is_new_session = len(history) == 0 | |
| # Use provided language or default to Hindi | |
| language = language if language else "hi" | |
| language = normalize_language_code(language) | |
| pest_message = get_pest_page(farmer_id, caller_query="", is_new_session=is_new_session, language=language) | |
| # Store in history if uid provided | |
| if uid and session_id: | |
| user_query = f"Pest detection request for farmer_id: {farmer_id}" | |
| result_text = pest_message.get("message", str(pest_message)) if isinstance(pest_message, dict) else str(pest_message) | |
| add_to_history(uid, session_id, user_query, result_text) | |
| return {"response": pest_message, "session_id": session_id, "uid": uid, "language": language} | |
| def phishing_endpoint(req: PhishingRequest): | |
| """Check if a given URL is safe or phishing.""" | |
| # Handle user and session | |
| uid = req.uid | |
| session_id = req.session_id | |
| # Always create/get session if uid is provided | |
| if uid: | |
| get_or_create_user(uid, "", "", "farmer") | |
| # Automatically get or create session (first time creates new, subsequent requests reuse) | |
| session_id = get_session_id(uid, session_id) | |
| else: | |
| session_id = "" # No uid means no session tracking | |
| # Check if it's a new session (before storing current conversation) | |
| is_new_session = False | |
| if uid and session_id: | |
| history = get_conversation_history(uid, session_id, limit=1) | |
| is_new_session = len(history) == 0 | |
| # Detect or use provided language (default to Hindi) | |
| language = req.language if req.language else "hi" | |
| language = normalize_language_code(language) | |
| result = check_phishing_url(req.url, caller_query="", is_new_session=is_new_session, language=language) | |
| # Store in history if uid provided | |
| if uid and session_id: | |
| user_query = f"Check URL for phishing: {req.url}" | |
| result_text = result.get("message", str(result)) if isinstance(result, dict) else str(result) | |
| add_to_history(uid, session_id, user_query, result_text) | |
| return {"phishing_check": result, "session_id": session_id, "uid": uid, "language": language} | |
| # --- 📝 NEW: Sentiment Analysis --- | |
| def sentiment_endpoint(req: SentimentRequest): | |
| """Analyze sentiment of given text.""" | |
| uid = req.uid | |
| session_id = req.session_id | |
| # Always create/get session if uid is provided | |
| if uid: | |
| get_or_create_user(uid, "", "", "farmer") | |
| # Automatically get or create session (first time creates new, subsequent requests reuse) | |
| session_id = get_session_id(uid, session_id) | |
| else: | |
| session_id = "" # No uid means no session tracking | |
| # Check if it's a new session (before storing current conversation) | |
| is_new_session = False | |
| if uid and session_id: | |
| history = get_conversation_history(uid, session_id, limit=1) | |
| is_new_session = len(history) == 0 | |
| # Detect or use provided language | |
| language = req.language if req.language else detect_language(req.query) | |
| language = normalize_language_code(language) | |
| result = analyze_sentiment(req.query, uid, caller_query="", is_new_session=is_new_session, language=language) | |
| # Store in history if uid provided | |
| if uid and session_id: | |
| result_text = result.get("message", str(result)) if isinstance(result, dict) else str(result) | |
| add_to_history(uid, session_id, req.query, result_text) | |
| return {"sentiment_analysis": result, "session_id": session_id, "uid": uid, "language": language} | |
| # --- 🏛️ NEW: Government Scheme Info --- | |
| def schemes_endpoint(req: SchemeRequest): | |
| """Get government schemes related to the farmer's query.""" | |
| uid = req.uid | |
| session_id = req.session_id | |
| # Always create/get session if uid is provided | |
| if uid: | |
| get_or_create_user(uid, "", "", "farmer") | |
| # Automatically get or create session (first time creates new, subsequent requests reuse) | |
| session_id = get_session_id(uid, session_id) | |
| else: | |
| session_id = "" # No uid means no session tracking | |
| # Check if it's a new session (before storing current conversation) | |
| is_new_session = False | |
| if uid and session_id: | |
| history = get_conversation_history(uid, session_id, limit=1) | |
| is_new_session = len(history) == 0 | |
| # Detect or use provided language | |
| language = req.language if req.language else detect_language(req.text) | |
| language = normalize_language_code(language) | |
| result = get_government_schemes(req.text, uid=uid, caller_query="", is_new_session=is_new_session, language=language) | |
| # Store in history if uid provided | |
| if uid and session_id: | |
| result_text = result.get("message", str(result)) if isinstance(result, dict) else str(result) | |
| add_to_history(uid, session_id, req.text, result_text) | |
| return {"government_schemes": result, "session_id": session_id, "uid": uid, "language": language} | |
| # --- 📜 NEW: Get Conversation History --- | |
| def get_history_endpoint(uid: str = Query(..., description="User ID"), session_id: str = Query("", description="Session ID (optional)")): | |
| """Get conversation history for a user session.""" | |
| if not uid: | |
| return {"error": "uid is required"} | |
| # If no session_id provided, get the current active session | |
| if not session_id: | |
| session_id = get_session_id(uid, None) | |
| history = get_conversation_history(uid, session_id if session_id else None, limit=50) | |
| return {"uid": uid, "session_id": session_id, "history": history} | |
| # --- 🆕 NEW: Start New Chat --- | |
| def start_new_chat_endpoint(uid: str = Query(..., description="User ID")): | |
| """Explicitly start a new chat session. This creates a fresh session for the user.""" | |
| if not uid: | |
| return {"error": "uid is required"} | |
| # Ensure user exists | |
| get_or_create_user(uid, "", "", "farmer") | |
| # Create a new session | |
| new_session_id = start_new_chat(uid) | |
| return { | |
| "message": "New chat session started", | |
| "session_id": new_session_id, | |
| "uid": uid | |
| } | |
| # --- 🤖 NEW: Unified Chat Endpoint --- | |
| def unified_chat_endpoint(req: UnifiedChatRequest): | |
| """ | |
| Unified chatbot endpoint that analyzes user query and routes to appropriate API. | |
| Users can just send text and the system automatically determines the intent and calls the right endpoint. | |
| """ | |
| # Handle user and session | |
| uid = req.uid | |
| session_id = req.session_id | |
| # Always create/get session if uid is provided | |
| if uid: | |
| get_or_create_user(uid, "", "", "farmer") | |
| session_id = get_session_id(uid, session_id) | |
| else: | |
| session_id = "" | |
| # Detect or use provided language | |
| language = req.language if req.language else detect_language(req.query) | |
| language = normalize_language_code(language) | |
| # Analyze intent | |
| intent_analysis = analyze_intent(req.query, language) | |
| intent = intent_analysis.get("intent", "general") | |
| params = intent_analysis.get("params", {}) | |
| confidence = intent_analysis.get("confidence", 0.5) | |
| # Use provided lat/lon if available, otherwise use extracted | |
| lat = req.lat if req.lat != 0.0 else (params.get("lat") or 28.6139) # Default to Delhi | |
| lon = req.lon if req.lon != 0.0 else (params.get("lon") or 77.2090) # Default to Delhi | |
| result = None | |
| endpoint_used = None | |
| try: | |
| # Route to appropriate endpoint based on intent | |
| if intent == "crop": | |
| endpoint_used = "crop" | |
| result = crop_advice(req.query, lat, lon, "", "", uid, session_id, language) | |
| elif intent == "fertilizer": | |
| endpoint_used = "fertilizer" | |
| query_english = translate_to_english(req.query, language) if language != "en" else req.query | |
| result = fertilizer_advice_from_query(req.query, query_english, uid, session_id, language) | |
| elif intent == "soil": | |
| endpoint_used = "soil" | |
| result = soil_health_advice(req.query, uid, session_id, language) | |
| elif intent == "weather": | |
| endpoint_used = "weather" | |
| days = params.get("days") or 3 | |
| is_new_session = False | |
| if uid and session_id: | |
| history = get_conversation_history(uid, session_id, limit=1) | |
| is_new_session = len(history) == 0 | |
| summary = get_weather(lat, lon, days, caller_query="", is_new_session=is_new_session, language=language) | |
| summary_text = summary.get("message", str(summary)) if isinstance(summary, dict) else str(summary) | |
| result = weather_advice(lat, lon, summary_text, uid, session_id, language) | |
| elif intent == "phishing": | |
| endpoint_used = "phishing" | |
| url = params.get("url") or extract_url(req.query) | |
| if not url: | |
| result = translate_text("No URL found in your query. Please provide a URL to check.", language, "en") | |
| else: | |
| is_new_session = False | |
| if uid and session_id: | |
| history = get_conversation_history(uid, session_id, limit=1) | |
| is_new_session = len(history) == 0 | |
| phishing_result = check_phishing_url(url, caller_query="", is_new_session=is_new_session, language=language) | |
| result = phishing_result.get("message", str(phishing_result)) if isinstance(phishing_result, dict) else str(phishing_result) | |
| elif intent == "sentiment": | |
| endpoint_used = "sentiment" | |
| is_new_session = False | |
| if uid and session_id: | |
| history = get_conversation_history(uid, session_id, limit=1) | |
| is_new_session = len(history) == 0 | |
| sentiment_result = analyze_sentiment(req.query, uid, caller_query="", is_new_session=is_new_session, language=language) | |
| result = sentiment_result.get("message", str(sentiment_result)) if isinstance(sentiment_result, dict) else str(sentiment_result) | |
| elif intent == "schemes": | |
| endpoint_used = "schemes" | |
| is_new_session = False | |
| if uid and session_id: | |
| history = get_conversation_history(uid, session_id, limit=1) | |
| is_new_session = len(history) == 0 | |
| schemes_result = get_government_schemes(req.query, uid=uid, caller_query="", is_new_session=is_new_session, language=language) | |
| result = schemes_result.get("message", str(schemes_result)) if isinstance(schemes_result, dict) else str(schemes_result) | |
| elif intent == "pest": | |
| endpoint_used = "pest" | |
| is_new_session = False | |
| if uid and session_id: | |
| history = get_conversation_history(uid, session_id, limit=1) | |
| is_new_session = len(history) == 0 | |
| pest_result = get_pest_page("", caller_query="", is_new_session=is_new_session, language=language) | |
| result = pest_result.get("message", str(pest_result)) if isinstance(pest_result, dict) else str(pest_result) | |
| else: # general or unclear intent | |
| endpoint_used = "general" | |
| # Use crop advice as default for general farming questions | |
| if lat != 28.6139 or lon != 77.2090: # If coordinates were extracted | |
| result = crop_advice(req.query, lat, lon, "", "", uid, session_id, language) | |
| else: | |
| # General response using AI | |
| import google.generativeai as genai | |
| from firebase_helper import format_history_for_prompt | |
| genai.configure(api_key=os.getenv("GOOGLE_API_KEY")) | |
| model = genai.GenerativeModel("gemini-2.5-flash") | |
| query_english = translate_to_english(req.query, language) if language != "en" else req.query | |
| history_context = "" | |
| is_new_session = False | |
| if uid: | |
| history = get_conversation_history(uid, session_id, limit=5) | |
| is_new_session = len(history) == 0 | |
| history_context = format_history_for_prompt(history) | |
| from translation_helper import get_greeting, SUPPORTED_LANGUAGES | |
| greeting_text = get_greeting(language) if is_new_session else "" | |
| lang_name = SUPPORTED_LANGUAGES.get(language, "Hindi") | |
| greeting_instruction = "" | |
| if is_new_session: | |
| greeting_instruction = f"Start your response with a friendly greeting in {lang_name}: '{greeting_text}' (Hello! I am Payal — ) and then continue with your answer in {lang_name}." | |
| prompt = f""" | |
| You are "Payal" — a helpful female farmer advisory assistant for Indian farmers. | |
| Answer the farmer's question in a friendly and helpful manner. | |
| {greeting_instruction} | |
| {history_context} | |
| Question: "{query_english}" | |
| Provide a helpful answer in {lang_name} language. If you need more information (like location for crop advice), politely ask for it. | |
| """ | |
| result = model.generate_content(prompt).text.strip() | |
| if language != "en": | |
| from translation_helper import translate_text | |
| result = translate_text(result, language, "en") | |
| # Store in history | |
| if uid and session_id: | |
| add_to_history(uid, session_id, req.query, result) | |
| return { | |
| "response": result, | |
| "session_id": session_id, | |
| "uid": uid, | |
| "language": language, | |
| "intent": intent, | |
| "endpoint_used": endpoint_used, | |
| "confidence": confidence, | |
| "extracted_params": params | |
| } | |
| except Exception as e: | |
| import traceback | |
| traceback.print_exc() | |
| return { | |
| "error": f"Error processing request: {str(e)}", | |
| "session_id": session_id, | |
| "uid": uid, | |
| "language": language, | |
| "intent": intent | |
| } | |