Created
August 31, 2023 23:09
-
-
Save hernamesbarbara/2fcd94cdeb4178dd06e6361780a61235 to your computer and use it in GitHub Desktop.
fetch place_ids from google
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python3 | |
import os | |
import requests | |
import pandas as pd | |
ENDPOINTS = { | |
"/search": {"url": "https://maps.googleapis.com/maps/api/place/findplacefromtext/json", "params": ["input", "inputtype", "fields"]}, | |
"/details": {"url": "https://maps.googleapis.com/maps/api/place/details/json", "params": ["place_id", "fields"]} | |
} | |
def get(endpoint, data, **kwargs): | |
url = ENDPOINTS[endpoint]['url'] | |
params = dict.fromkeys(ENDPOINTS[endpoint]['params']) | |
for key, value in kwargs.items(): | |
if key in params: | |
params[key] = value | |
for key, value in data.items(): | |
if key in params: | |
params[key] = value | |
params['key'] = GOOGLE_MAPS_API_KEY | |
# return url, params | |
return requests.get(url, params=params) | |
def process_place_details(details_dict): | |
fields = ['name', 'url', 'website', 'formatted_address', 'place_id'] | |
place = {field: details_dict.get(field) for field in fields} | |
return place | |
with open ( os.path.expanduser('~/.google_maps'), 'r') as f: | |
GOOGLE_MAPS_API_KEY = f.readline().strip() | |
def load_api(url, api_key=GOOGLE_MAPS_API_KEY): | |
return f"{url}&key={api_key}" | |
def lookup_place_id(name_of_place): | |
r = get('/search', data={'input': name_of_place, 'inputtype': 'textquery'}) | |
if not r or r.status_code != 200: | |
return None | |
if r.json()['candidates'][0]['place_id']: | |
return r.json()['candidates'][0]['place_id'] | |
def lookup_place_details(place_id): | |
r = get('/details', data={'place_id': place_id}) | |
if not r or r.status_code != 200: | |
return None | |
return process_place_details(r.json()['result']) | |
def process_row(row): | |
place_dict = row.to_dict() | |
try: | |
print(f"looking up place_id for {place_dict['Title']}") | |
place_id = lookup_place_id(place_dict['Title']) | |
except: | |
place_id = None | |
if place_id: | |
try: | |
place_details = lookup_place_details(place_id) | |
place_dict.update(place_details) | |
except: | |
place_details = {} | |
return place_dict | |
df = pd.read_csv("combined.csv") | |
results = [] | |
for i, (idx, row) in enumerate(df.iterrows()): | |
print(f"Processing row {i+1} of {len(df)}") | |
try: | |
place = process_row(row) | |
except Exception as e: | |
print(e) | |
place = {'name': row['Title']} | |
results.append(place) | |
colorder = [ | |
'name', | |
'url', | |
'website', | |
'Vibe', | |
'formatted_address', | |
'place_id' | |
] | |
df_out = pd.DataFrame(results) | |
df_out = df_out[colorder] | |
df_out.to_csv("places.csv", index=False) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment