Spaces:
Sleeping
Sleeping
File size: 2,422 Bytes
411a994 |
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 |
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é")
|