Spaces:
Sleeping
Sleeping
| 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) | |