Spaces:
Paused
Paused
| from fastapi import FastAPI, HTTPException, Depends, Request, BackgroundTasks | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from fastapi.responses import StreamingResponse, JSONResponse | |
| from pydantic import BaseModel | |
| import os | |
| import json | |
| import time | |
| import io | |
| import requests | |
| from typing import Optional, List, Dict, Any | |
| import gtts | |
| # Import NewsAgent class | |
| from main import NewsAgent | |
| app = FastAPI( | |
| title="NewsAI API", | |
| description="A FastAPI backend for a location-specific news agent that provides news based on pincode and preferred language.", | |
| version="1.0.0" | |
| ) | |
| # Add CORS middleware to allow frontend to communicate with the backend | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=["*"], # Replace with specific origins in production | |
| allow_credentials=True, | |
| allow_methods=["*"], | |
| allow_headers=["*"], | |
| ) | |
| # Initialize the NewsAgent | |
| news_agent = NewsAgent() | |
| # Session storage for user conversations | |
| user_sessions = {} | |
| # Supported languages for translation | |
| SUPPORTED_LANGUAGES = { | |
| "en": "English", | |
| "hi": "Hindi", | |
| "bn": "Bengali", | |
| "te": "Telugu", | |
| "ta": "Tamil", | |
| "mr": "Marathi", | |
| "gu": "Gujarati", | |
| "kn": "Kannada", | |
| "ml": "Malayalam", | |
| "pa": "Punjabi", | |
| "or": "Odia", | |
| "as": "Assamese", | |
| "fr": "French", | |
| "de": "German", | |
| "es": "Spanish", | |
| "zh-CN": "Chinese (Simplified)", | |
| "ja": "Japanese", | |
| "ko": "Korean", | |
| "ar": "Arabic", | |
| "ru": "Russian" | |
| } | |
| # Request/response models | |
| class ChatRequest(BaseModel): | |
| message: str | |
| session_id: str | |
| pincode: Optional[str] = None | |
| language: str = "en" | |
| class PincodeRequest(BaseModel): | |
| pincode: str | |
| class TextToSpeechRequest(BaseModel): | |
| text: str | |
| lang: str = "en" | |
| class TranslateRequest(BaseModel): | |
| text: str | |
| target_language: str | |
| class NewsResponse(BaseModel): | |
| response: str | |
| audio_url: Optional[str] = None | |
| translated: bool = False | |
| # Helper function to get or create a session | |
| def get_session(session_id: str) -> Dict: | |
| """Get or create a user session by ID.""" | |
| if session_id not in user_sessions: | |
| user_sessions[session_id] = {"location": None, "language": "en", "history": []} | |
| return user_sessions[session_id] | |
| # Helper function to get location from pincode | |
| def get_location_from_pincode(pincode: str) -> Optional[str]: | |
| """Get location (city/state) from Indian pincode.""" | |
| try: | |
| # First try India Post API | |
| url = f"https://api.postalpincode.in/pincode/{pincode}" | |
| response = requests.get(url, timeout=5,verify=False) | |
| data = response.json() | |
| if data and data[0]["Status"] == "Success": | |
| post_office = data[0]["PostOffice"][0] | |
| district = post_office["District"] | |
| state = post_office["State"] | |
| return f"{district}, {state}" | |
| # Fallback to pincode map | |
| pincode_map = { | |
| "11": "Delhi", | |
| "12": "Haryana", | |
| "13": "Haryana", | |
| "14": "Punjab", | |
| "15": "Punjab", | |
| "16": "Punjab", | |
| "17": "Himachal Pradesh", | |
| "18": "Jammu & Kashmir", | |
| "19": "Jammu & Kashmir", | |
| "20": "Uttar Pradesh", | |
| "21": "Uttar Pradesh", | |
| "22": "Uttar Pradesh", | |
| "23": "Uttar Pradesh", | |
| "24": "Uttar Pradesh", | |
| "25": "Uttar Pradesh", | |
| "26": "Uttar Pradesh", | |
| "27": "Uttar Pradesh", | |
| "28": "Uttar Pradesh", | |
| "30": "Rajasthan", | |
| "31": "Rajasthan", | |
| "32": "Rajasthan", | |
| "33": "Rajasthan", | |
| "34": "Rajasthan", | |
| "36": "Gujarat", | |
| "37": "Gujarat", | |
| "38": "Gujarat", | |
| "39": "Gujarat", | |
| "40": "Maharashtra", | |
| "41": "Maharashtra", | |
| "42": "Maharashtra", | |
| "43": "Maharashtra", | |
| "44": "Maharashtra", | |
| "45": "Madhya Pradesh", | |
| "46": "Madhya Pradesh", | |
| "47": "Madhya Pradesh", | |
| "48": "Madhya Pradesh", | |
| "49": "Chhattisgarh", | |
| "50": "Andhra Pradesh", | |
| "51": "Andhra Pradesh", | |
| "52": "Telangana", | |
| "53": "Telangana", | |
| "56": "Karnataka", | |
| "57": "Karnataka", | |
| "58": "Karnataka", | |
| "59": "Karnataka", | |
| "60": "Tamil Nadu", | |
| "61": "Tamil Nadu", | |
| "62": "Tamil Nadu", | |
| "63": "Tamil Nadu", | |
| "64": "Tamil Nadu", | |
| "67": "Kerala", | |
| "68": "Kerala", | |
| "69": "Kerala", | |
| "70": "West Bengal", | |
| "71": "West Bengal", | |
| "72": "West Bengal", | |
| "73": "West Bengal", | |
| "74": "West Bengal", | |
| "75": "Odisha", | |
| "76": "Odisha", | |
| "77": "Odisha", | |
| "78": "Assam", | |
| "79": "North East India", | |
| "80": "Bihar", | |
| "81": "Bihar", | |
| "82": "Bihar", | |
| "83": "Jharkhand", | |
| "84": "Jharkhand", | |
| "85": "Jharkhand" | |
| } | |
| # Get state from first 2 digits | |
| state = pincode_map.get(pincode[:2], "Unknown") | |
| return state | |
| except Exception as e: | |
| print(f"Error getting location from pincode: {e}") | |
| return None | |
| # Routes | |
| async def root(): | |
| """Health check endpoint""" | |
| return {"status": "online", "message": "NewsAI API is running"} | |
| async def get_languages(): | |
| """Get list of supported languages""" | |
| return {"languages": SUPPORTED_LANGUAGES} | |
| async def lookup_pincode(request: PincodeRequest): | |
| """Look up location from pincode""" | |
| location = get_location_from_pincode(request.pincode) | |
| if not location: | |
| raise HTTPException(status_code=404, detail="Could not find location for this pincode") | |
| return {"pincode": request.pincode, "location": location} | |
| async def chat(request: ChatRequest): | |
| """Process a chat message and return a response in the requested language""" | |
| session = get_session(request.session_id) | |
| # Update language preference | |
| if request.language: | |
| session["language"] = request.language | |
| # Update location if pincode provided | |
| if request.pincode: | |
| location = get_location_from_pincode(request.pincode) | |
| if location: | |
| session["location"] = location | |
| # Process the query | |
| query = request.message | |
| # If location is set, include it in the query for location-specific news | |
| if session["location"] and "news" in query.lower() and session["location"].lower() not in query.lower(): | |
| query = f"{query} in {session['location']}" | |
| # Process the query | |
| response = news_agent.process_query(query) | |
| # Translate response if needed | |
| original_response = response | |
| if session["language"] != "en": | |
| try: | |
| # Use the translate_text method from NewsAgent | |
| translation_input = json.dumps({"text": response, "lang": session["language"]}) | |
| translated_response = news_agent.translate_text(translation_input) | |
| # Extract translated text from response format "Translated text: {text}" | |
| if "Translated text: " in translated_response: | |
| response = translated_response.replace("Translated text: ", "") | |
| else: | |
| response = translated_response | |
| except Exception as e: | |
| print(f"Translation error: {e}") | |
| # Keep original response if translation fails | |
| # Store in session history | |
| session["history"].append({"role": "user", "content": request.message}) | |
| session["history"].append({"role": "assistant", "content": response}) | |
| return { | |
| "response": response, | |
| "original_response": original_response if session["language"] != "en" else None, | |
| "language": session["language"], | |
| "location": session["location"] | |
| } | |
| async def translate_text(request: TranslateRequest): | |
| """Translate text to the specified language""" | |
| try: | |
| translation_input = json.dumps({"text": request.text, "lang": request.target_language}) | |
| translated_text = news_agent.translate_text(translation_input) | |
| # Extract translated text from response format "Translated text: {text}" | |
| if "Translated text: " in translated_text: | |
| translated_text = translated_text.replace("Translated text: ", "") | |
| return {"translated_text": translated_text, "language": request.target_language} | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=f"Translation error: {str(e)}") | |
| async def text_to_speech(request: TextToSpeechRequest): | |
| """Convert text to speech and return audio file""" | |
| try: | |
| # Generate speech | |
| tts = gtts.gTTS(text=request.text, lang=request.lang, slow=False) | |
| # Save to in-memory file | |
| audio_io = io.BytesIO() | |
| tts.write_to_fp(audio_io) | |
| audio_io.seek(0) | |
| # Return audio file | |
| return StreamingResponse( | |
| audio_io, | |
| media_type="audio/mpeg", | |
| headers={"Content-Disposition": "attachment; filename=speech.mp3"} | |
| ) | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=f"Error generating speech: {str(e)}") | |
| async def get_location_news(location: str, count: int = 5, language: str = "en"): | |
| """Fetch news for a specific location and optionally translate it""" | |
| try: | |
| # Fetch news | |
| news = news_agent.fetch_city_news(f"{location}, {count}") | |
| # Translate if needed | |
| if language != "en": | |
| translation_input = json.dumps({"text": news, "lang": language}) | |
| translated_news = news_agent.translate_text(translation_input) | |
| # Extract translated text | |
| if "Translated text: " in translated_news: | |
| news = translated_news.replace("Translated text: ", "") | |
| else: | |
| news = translated_news | |
| return {"news": news, "language": language} | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=f"Error fetching news: {str(e)}") | |