app/ai_agent/maps_tool.py

import os
import requests
from langchain.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Type

class LocationInput(BaseModel):
    location: str = Field(description="La localisation pour la recherche d'établissements médicaux")
    keyword: str = Field(default="hospital|pharmacy", description="Type d'établissement: hôpital, pharmacie, etc.")

class GoogleMapsTool(BaseTool):
    name : str = "google_maps_search"
    description : str = "Recherche des établissements médicaux près d'une localisation using Google Maps API"
    args_schema: Type[BaseModel] = LocationInput

    def _run(self, location: str, keyword: str = "hospital|pharmacy"):
        api_key = os.getenv("GOOGLE_MAPS_API_KEY")
        geocode_url = f"https://maps.googleapis.com/maps/api/geocode/json?address={location}&key={api_key}"
        geocode_data = requests.get(geocode_url).json()
        if not geocode_data['results']:
            return "Désolé, je n'ai pas pu trouver cette localisation."
        coords = geocode_data['results'][0]['geometry']['location']
        lat, lng = coords['lat'], coords['lng']
        places_url = f"https://maps.googleapis.com/maps/api/place/nearbysearch/json?location={lat},{lng}&radius=5000&keyword={keyword}&type=hospital|pharmacy&key={api_key}"
        places_data = requests.get(places_url).json()
        results = []
        for place in places_data.get('results', [])[:5]:
            is_open = "Ouvert" if place.get('opening_hours', {}).get('open_now', False) else "Fermé"
            details_url = f"https://maps.googleapis.com/maps/api/place/details/json?place_id={place['place_id']}&fields=name,formatted_phone_number,opening_hours,formatted_address&key={api_key}"
            details_data = requests.get(details_url).json()
            place_info = details_data.get('result', {})
            results.append({
                'name': place_info.get('name', 'Nom non disponible'),
                'address': place_info.get('formatted_address', 'Adresse non disponible'),
                'phone': place_info.get('formatted_phone_number', 'Téléphone non disponible'),
                'status': is_open,
                'rating': place.get('rating', 'Non noté'),
                'types': place.get('types', [])
            })
        return results

    def _arun(self, location: str, keyword: str):
        raise NotImplementedError("Async non supporté")