import os import re import requests from datetime import datetime from dotenv import load_dotenv from translation_helper import get_greeting, normalize_language_code, translate_text, SUPPORTED_LANGUAGES load_dotenv() # Environment variables MANDI_API_URL = os.getenv("MANDI_API_URL") PEST_PAGE_URL = os.getenv("PEST_PAGE_URL") PHISHING_API_URL = os.getenv("PHISHING_API_URL") # Only URL, no auth SENTIMENT_API_URL = "https://theabeerrai-sentiment.hf.space/analyze/" SCHEME_API_URL = "https://theabeerrai-schemeai.hf.space/schemes" # --- Utilities --- def _mentions_payal(text: str) -> bool: """Return True if text contains Payal's name in English, Hindi or Telugu.""" if not text: return False return bool(re.search(r'\b(payal|पायल|పాయల్)\b', text, flags=re.IGNORECASE)) def _payal_message(body: str, caller_query: str = "", is_new_session: bool = False, language: str = "hi") -> dict: """ Wrap a plain body into Payal's friendly message format. Only greet if it's a new session (no previous conversation). """ prefix = "" if is_new_session: lang_code = normalize_language_code(language) prefix = get_greeting(lang_code) return {"message": f"{prefix}{body}", "assistant": "Payal"} # --- Weather Function --- def get_weather(lat: float, lon: float, days: int = 3, caller_query: str = "", is_new_session: bool = False, language: str = "hi"): try: url = ( f"https://api.open-meteo.com/v1/forecast?" f"latitude={lat}&longitude={lon}&daily=temperature_2m_max,temperature_2m_min,rain_sum" f"&forecast_days={days}&timezone=auto" ) r = requests.get(url, timeout=10) r.raise_for_status() data = r.json() daily = data.get("daily", {}) if not daily: message = translate_text("Weather information is not available. Please try again later.", language, "en") return _payal_message(message, caller_query, is_new_session, language) summary_lines = [] for i in range(len(daily.get("temperature_2m_max", []))): summary_lines.append( f"Day {i+1}: Max {daily['temperature_2m_max'][i]}°C, " f"Min {daily['temperature_2m_min'][i]}°C, Rain {daily['rain_sum'][i]}mm." ) summary = " ".join(summary_lines) user_text_en = f"Here is the weather summary for the next {days} days: {summary}" user_text = translate_text(user_text_en, language, "en") return _payal_message(user_text, caller_query, is_new_session, language) except Exception as e: error_msg = translate_text(f"Error fetching weather: {e}", language, "en") return _payal_message(error_msg, caller_query, is_new_session, language) # --- Mandi Price --- def get_mandi_price(district_name: str, market_name: str, crop: str, caller_query: str = "", is_new_session: bool = False, language: str = "hi"): try: current_date = datetime.now().strftime("%Y-%m-%d") payload = { "District_Name": district_name, "Market_Name": market_name, "Crop": crop, "Reported_Date": current_date } response = requests.post(MANDI_API_URL, json=payload, timeout=10) response.raise_for_status() data = response.json() price = data.get("Predicted Modal Price (₹/Quintal)") if price is not None: price_rounded = round(price, 2) user_text_en = ( f"Estimated mandi price for {crop} in {market_name} ({district_name}) " f"is ₹{price_rounded} per quintal." ) user_text = translate_text(user_text_en, language, "en") return { "predicted_price": price_rounded, **_payal_message(user_text, caller_query, is_new_session, language) } else: error_msg = translate_text("Estimated price not found in mandi API response.", language, "en") return _payal_message(error_msg, caller_query, is_new_session, language) except Exception as e: error_msg = translate_text(f"Mandi API error: {e}", language, "en") return _payal_message(error_msg, caller_query, is_new_session, language) # --- Pest Detection --- def get_pest_page(farmer_id: str = "", caller_query: str = "", is_new_session: bool = False, language: str = "hi"): if farmer_id: body_en = f"Farmer ID {farmer_id}, please go to the Pest Detection page so I can provide information about pests." else: body_en = "Please go to the Pest Detection page — there you can find information about pests and diseases." body = translate_text(body_en, language, "en") return _payal_message(body, caller_query, is_new_session, language) # --- Phishing URL Checker --- def check_phishing_url(url: str, caller_query: str = "", is_new_session: bool = False, language: str = "hi"): if not PHISHING_API_URL: error_msg = translate_text("Sorry, the link checking API is not set up.", language, "en") return _payal_message(error_msg, caller_query, is_new_session, language) try: payload = {"url": url} r = requests.post(PHISHING_API_URL, json=payload, timeout=10) r.raise_for_status() data = r.json() result = data.get("prediction") if result is None: result = data.get("result") if result is None: result = data.get("label") probability = data.get("probability") or (data.get("data", {}).get("probability")) # Prediction logic: 0 = safe/legit, 1 = phishing/dangerous is_safe = False # Default to unsafe if we can't determine if result is not None: # Handle numeric values (0 or 1) try: result_int = int(result) if result_int == 0: is_safe = True # 0 = legit website (safe) elif result_int == 1: is_safe = False # 1 = phishing (dangerous) except (ValueError, TypeError): # Handle string/boolean values result_str = str(result).lower() if result_str in ["safe", "legit", "legitimate", "0", "false", "good"]: is_safe = True elif result_str in ["phishing", "dangerous", "unsafe", "1", "true", "bad"]: is_safe = False safe_text_en = "safe" if is_safe else "dangerous" safe_text = translate_text(safe_text_en, language, "en") message_text_en = ( f"I have checked this link: {url}. This link appears to be {safe_text_en}." ) if probability: confidence = round(float(probability) * 100, 1) message_text_en += f" (~{confidence}% confidence)" message_text = translate_text(message_text_en, language, "en") return {"url": url, "result": safe_text, "raw": data, **_payal_message(message_text, caller_query, is_new_session, language)} except Exception as e: error_msg = translate_text(f"Error checking link: {e}", language, "en") return _payal_message(error_msg, caller_query, is_new_session, language) # --- Sentiment Analysis --- def analyze_sentiment(query: str, uid: str = "farmer", caller_query: str = "", is_new_session: bool = False, language: str = "hi"): try: payload = {"query": query, "uid": uid} r = requests.post(SENTIMENT_API_URL, json=payload, timeout=10) r.raise_for_status() data = r.json() sentiment = data.get("sentiment") or data.get("result") or "unknown" message_text_en = f"I have checked your sentiment: this message shows '{sentiment}' sentiment." message_text = translate_text(message_text_en, language, "en") return {"sentiment": sentiment, **_payal_message(message_text, caller_query, is_new_session, language)} except Exception as e: error_msg = translate_text(f"Sentiment analysis error: {e}", language, "en") return _payal_message(error_msg, caller_query, is_new_session, language) # --- Government Schemes --- def get_government_schemes(text: str, uid: str = "", caller_query: str = "", is_new_session: bool = False, language: str = "hi"): try: # Send as query parameters instead of JSON body params = { "query": text, "uid": uid if uid else "farmer" } r = requests.get(SCHEME_API_URL, params=params, timeout=10) r.raise_for_status() data = r.json() schemes = data.get("schemes") or data.get("result") or "No schemes found" message_text_en = f"I have checked government schemes for you: {schemes}" message_text = translate_text(message_text_en, language, "en") return {"schemes": schemes, **_payal_message(message_text, caller_query, is_new_session, language)} except Exception as e: error_msg = translate_text(f"Error fetching government schemes: {e}", language, "en") return _payal_message(error_msg, caller_query, is_new_session, language)