File size: 2,606 Bytes
c1cbefd 5f52f06 c1cbefd |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 |
import datetime
import requests
from typing import Dict, Any, Optional
from utils.config import config
class ContextEnrichmentService:
"""Service for enriching AI context with current data"""
def __init__(self):
# Access config attributes properly
self.openweather_api_key = getattr(config, 'openweather_api_key', None)
self.tavily_api_key = getattr(config, 'tavily_api_key', None)
def get_current_context(self, user_query: str = "") -> Dict[str, Any]:
"""Get current context including time, weather, and recent news"""
context = {
"current_time": self._get_current_time(),
"weather": self._get_weather_summary("New York"), # Default location
"recent_news": self._get_recent_news(user_query) if user_query else []
}
return context
def _get_current_time(self) -> str:
"""Get current date and time"""
now = datetime.datetime.now()
return now.strftime("%A, %B %d, %Y at %I:%M %p")
def _get_weather_summary(self, city: str = "New York") -> Optional[str]:
"""Get weather summary for a city"""
if not self.openweather_api_key:
return "Weather data not configured"
try:
url = f"http://api.openweathermap.org/data/2.5/weather"
params = {
'q': city,
'appid': self.openweather_api_key,
'units': 'metric'
}
response = requests.get(url, params=params, timeout=5)
if response.status_code == 200:
data = response.json()
return f"{data['weather'][0]['description']}, {data['main']['temp']}°C in {data['name']}"
except Exception:
pass
return "Clear skies"
def _get_recent_news(self, query: str) -> list:
"""Get recent news related to query"""
if not self.tavily_api_key:
return []
try:
url = "https://api.tavily.com/search"
headers = {"Content-Type": "application/json"}
data = {
"query": query,
"api_key": self.tavily_api_key,
"max_results": 3
}
response = requests.post(url, json=data, headers=headers, timeout=10)
if response.status_code == 200:
result = response.json()
return result.get("results", [])
except Exception:
pass
return []
# Global instance
context_service = ContextEnrichmentService()
|