fetch place_ids from google

places.py

#!/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)