import os import pandas as pd from dotenv import load_dotenv import requests from geopy.distance import geodesic load_dotenv() def find_place(input_text, api_key): # Define the base URL and parameters for the request base_url = "https://maps.googleapis.com/maps/api/place/findplacefromtext/json" params = { "fields": "formatted_address,name,rating,opening_hours,geometry", "input": input_text, "inputtype": "textquery", "key": api_key # Replace with your actual API key } # Make the GET request response = requests.get(base_url, params=params) # Check if the request was successful if response.status_code == 200: # Return the response content (JSON) return response.json() else: # Return an error message if the request failed return f"Request failed with status code: {response.status_code}" def get_zip_code(address): base_url = "https://maps.googleapis.com/maps/api/geocode/json" params = { "address": address, "key": "AIzaSyDXCm-fwucr6R93CHa9Bb4j5qArup5D4kY" } response = requests.get(base_url, params=params) if response.status_code == 200: data = response.json() if data['status'] == 'OK': # Extract the postal code from the address components for component in data['results'][0]['address_components']: if 'postal_code' in component['types']: return component['long_name'] return None else: return f"API request returned status: {data['status']}" else: return f"Request failed with status code: {response.status_code}" def get_distances_for_multiple_destinations(origin, destinations, api_key): destinations_param = '|'.join(destinations) base_url = "https://maps.googleapis.com/maps/api/distancematrix/json" params = { "origins": origin, "destinations": destinations_param, "key": api_key } response = requests.get(base_url, params=params) if response.status_code == 200: data = response.json() if data['status'] == 'OK': results = [] for element in data['rows'][0]['elements']: if element['status'] == 'OK': distance = element['distance']['text'] results.append(distance) else: results.append("Distance calculation failed") return results else: return ["API request returned error: " + data['status']] else: return [f"Request failed with status code: {response.status_code}"] class LocationDataHandler: def __init__(self, data_frame): # self.data_file = data_file self.api_key = os.getenv('GOOGLE_MAPS_API_KEY') self.data_frame = data_frame def get_latlong_from_address(self, address): try: result = find_place(address, self.api_key) lat = result['candidates'][0]['geometry']['location']['lat'] lng = result['candidates'][0]['geometry']['location']['lng'] return lat, lng except Exception as e: print(f"Error retrieving location for {address}: {e}") return None def calculate_distance(self, ref_point, latitude, longitude): try: location = (latitude, longitude) return geodesic(ref_point, location).kilometers except Exception as e: print(f"Error calculating distance: {e}") return None def filter_by_address(self, address, max_distance_km=30): ref_point = self.get_latlong_from_address(address) if not ref_point: return None def distance_calculation(row): return self.calculate_distance(ref_point, row['Latitude'], row['Longitude']) # Calculate distance for each row self.data_frame['Distance'] = self.data_frame.apply(distance_calculation, axis=1) # Select top 10 within the maximum distance df = self.data_frame[self.data_frame['Distance'] <= max_distance_km].sort_values(by='Distance').head(10).reset_index(drop=True) # Calculate Actual Distance Using GMaps API destinations = df.apply(lambda row: f"{row['Latitude']},{row['Longitude']}", axis=1).tolist() distances = get_distances_for_multiple_destinations(f"{ref_point[0]},{ref_point[1]}", destinations, self.api_key) # Add the distances back to the DataFrame df['Distance'] = distances return df