{ "cells": [ { "cell_type": "code", "execution_count": 1, "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "/home/ec2-user/anaconda3/envs/python3/lib/python3.6/site-packages/psycopg2/__init__.py:144: UserWarning: The psycopg2 wheel package will be renamed from release 2.8; in order to keep installing from binary please use \"pip install psycopg2-binary\" instead. For details see: .\n", " \"\"\")\n" ] } ], "source": [ "%run ./utils.ipynb\n", "engine = get_rds_engine()" ] }, { "cell_type": "code", "execution_count": 2, "metadata": {}, "outputs": [], "source": [ "import requests\n", "from requests.utils import requote_uri\n", "import json\n", "import pandas as pd\n", "import ast\n", "\n", "def get_api_key():\n", " return get_secret(\"google_distance_matrix\") # don't mind the secret name, the API key it returns is enabeld for multiple APIs\n", "\n", "def get_geo_api_results(origin, destination, mode):\n", " api_key = get_api_key()\n", " api_base = \"https://maps.googleapis.com/maps/api/distancematrix/json\"\n", " api_url = f\"{api_base}?key={api_key}&origins={origin}&destinations={destination}&mode={mode}\"\n", " # Replace spaces and other crap to normalize URL\n", " api_url = requote_uri(api_url)\n", " headers = {'Content-Type': 'application/json',\n", " 'User-Agent': 'FanSifter Python Client',\n", " 'Accept': 'application/json'}\n", " response = requests.get(api_url, headers=headers)\n", " if response.status_code == 200:\n", " #print(response.json())\n", " return response.json()\n", " else:\n", " component_logger.info('[!] HTTP {0} calling [{1}]'.format(response.status_code, api_url))\n", " return None\n", "\n", "# use this to get geocoded response from address\n", "def get_geolocation(address):\n", " api_key = get_api_key()\n", " api_base = \"https://maps.googleapis.com/maps/api/geocode/json\"\n", " api_url = f\"{api_base}?key={api_key}&address={address}\"\n", " api_url = requote_uri(api_url)\n", " headers = {'Content-Type': 'application/json',\n", " 'User-Agent': 'FanSifter Python Client',\n", " 'Accept': 'application/json'}\n", " response = requests.get(api_url, headers=headers)\n", " if response.status_code == 200:\n", " #print(response.json())\n", " return response.json()\n", " else:\n", " component_logger.info('[!] HTTP {0} calling [{1}]'.format(response.status_code, api_url))\n", " return None\n", " #print(json.dumps(response, indent=4, sort_keys=True))\n", "\n", "# use this to get geocoded response from latitude longitude\n", "def get_geolocation_coord(lat,lon):\n", " api_key = get_api_key()\n", " api_base = \"https://maps.googleapis.com/maps/api/geocode/json\"\n", " api_url = f\"{api_base}?key={api_key}&latlng={lat},{lon}\"\n", " api_url = requote_uri(api_url)\n", " headers = {'Content-Type': 'application/json',\n", " 'User-Agent': 'FanSifter Python Client',\n", " 'Accept': 'application/json'}\n", " response = requests.get(api_url, headers=headers)\n", " if response.status_code == 200:\n", " #print(response.json())\n", " return response.json()\n", " else:\n", " component_logger.info('[!] HTTP {0} calling [{1}]'.format(response.status_code, api_url))\n", " return None\n", " #print(json.dumps(response, indent=4, sort_keys=True))\n", " \n", "def parse_geolocation_to_df(geolocation):\n", " address_components = geolocation['results'][0]['address_components']\n", " address_geometry = geolocation['results'][0]['geometry'][\"location\"]\n", "\n", " administrative_area_level_1 = [x[\"long_name\"] for x in address_components if x[\"types\"][0] == 'administrative_area_level_1' ]\n", " administrative_area_level_2 = [x[\"long_name\"] for x in address_components if x[\"types\"][0] == 'administrative_area_level_2' ]\n", " country = [x[\"long_name\"] for x in address_components if x[\"types\"][0] == 'country' ]\n", " locality = [x[\"long_name\"] for x in address_components if x[\"types\"][0] == 'locality' ]\n", "\n", " db_dict = {\n", " \"\"\n", " \"country\": country[0] if country else None,\n", " \"locality\":locality[0] if locality else None,\n", " \"administrative_area_level_1\": administrative_area_level_1[0] if administrative_area_level_1 else None,\n", " \"administrative_area_level_2\": administrative_area_level_2[0] if administrative_area_level_2 else None,\n", " \"longitude\": address_geometry[\"lng\"],\n", " \"latitude\": address_geometry[\"lat\"]\n", " }\n", " df = pd.DataFrame([db_dict])\n", " return df\n", "\n", "def get_distance_between_addresses(x, mode):\n", " \"\"\" Gets distance from origin and destination address.\n", " Return distance and duration, in text from and plain numeric values.\n", " Can have those modes: \"walking\", \"driving\", \"bicycling\", \"transit\"\n", "\n", " See more from docs: https://developers.google.com/maps/documentation/distance-matrix/intro#RequestParameters\n", "\n", " #TODO! When calling this, we can save money by avoiding doing duplicate calls.. e.g. in one dataset,\n", " I could reduce the amount of calls by 2.5x by taking unique addresses only\n", " \"\"\"\n", " user_location = None\n", " user_city = None\n", "\n", " try:\n", " # Address \"632\" is not really useful for us. Zip comes handy in those cases. Or Phone area code.\n", " if not (isNaN(x['userAddress']) or x['userAddress'] == 'NaN' or x['userAddress'] == ' ' or len(x['userAddress']) <= 3):\n", " user_location = 'userAddress'\n", " # Only for concatenating user address and city\n", " try:\n", " if not (isNaN(x['userCity']) or x['userCity'] == 'NaN' or x['userCity'] == ' '):\n", " user_city = 'userCity'\n", " except Exception as e:\n", " pass\n", " except Exception as e:\n", " pass\n", "\n", " if not user_location:\n", " try:\n", " if not (isNaN(x['userZip']) or x['userZip'] == 'NaN'):\n", " user_location = 'userZip'\n", " except Exception as e:\n", " pass\n", "\n", " if not user_location:\n", " try:\n", " if not (isNaN(x['userAddress']) or x['userAddress'] == 'NaN'):\n", " user_location = 'userAddress'\n", " except Exception as e:\n", " pass\n", "\n", " if not user_location:\n", " try:\n", " if not (isNaN(x['userPhoneLocation']) or x['userPhoneLocation'] == 'NaN'):\n", " user_location = 'userPhoneLocation'\n", " except Exception as e:\n", " pass\n", "\n", " if not user_location:\n", " return np.nan\n", "\n", " origin = x[[user_location]].tolist()[0]\n", " if user_city:\n", " origin = x[[user_location]].tolist()[0] + ' ' + x[[user_city]].tolist()[0]\n", " destination = x[['userEventLocation']].tolist()[0]\n", "\n", " if isinstance(origin, float):\n", " origin = str(int(origin))\n", "\n", " if isinstance(origin, int):\n", " origin = str(origin)\n", "\n", " #component_logger.info(f\"type {type(origin)} {origin}\")\n", " #component_logger.info(f\"type {type(destination)} {destination}\")\n", "\n", " \"\"\" Remove symbols from text, else it might cause invalid response in distnace matrix API.\"\"\"\n", " # TODO! replace with \"slugify\" library or a more robust regexp\n", " destination = re.sub(r'[^A-z0-9 -]', '', destination)\n", " origin = re.sub(r'[^A-z0-9 -]', '', origin)\n", "\n", " results = get_geo_api_results(origin, destination, mode)\n", "\n", " if results.get('error_message'):\n", " component_logger.info(f\"Got error with {origin}: {results.get('error_message')}\")\n", " return np.nan\n", " else:\n", " if results['rows'][0]['elements'][0]['status'] != 'OK':\n", " # Sometimes we get \"ZERO_RESULTS\" (e.g. from Berlin, Germany to Hollywood, walking distance)\n", " component_logger.info(f\"Got not OK status with {origin}: {results['rows'][0]['elements'][0]['status']}\")\n", " return np.nan\n", " try:\n", " duration_value = results['rows'][0]['elements'][0]['duration']['value']\n", " except Exception as e:\n", " component_logger.error(e)\n", " component_logger.info(results)\n", " duration_value = np.nan\n", " # distance_text = results['rows'][0]['elements'][0]['distance']['text']\n", " # distance_value = results['rows'][0]['elements'][0]['distance']['value']\n", " # duration_text = results['rows'][0]['elements'][0]['duration']['text']\n", "\n", " return duration_value # distance_text, distance_value, duration_text, duration_value\n", "\n", "def enrich_distance(source_df):\n", "\n", " if not {'userEventLocation'}.issubset(source_df.columns):\n", " component_logger.info(\"userEventLocation not in columns list, cannot enrich distance\")\n", " return source_df\n", "\n", " component_logger.info(f\"Enriching driving distance ...\")\n", " mode = 'driving'\n", " try:\n", " source_df[f\"distance_{mode}_time\"] = source_df.apply(\n", " lambda x: get_distance_between_addresses(x, mode), axis=1)\n", " except Exception as e:\n", " component_logger.error(f\"Did not manage to enrich distance: {e}\")\n", "\n", " return source_df" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "df.to_sql(f'fan_address_google_geocoded',\n", " engine,\n", " schema=client_id,\n", " if_exists='append',\n", " index_label='fan_id',\n", " # index=False\n", " )" ] }, { "cell_type": "code", "execution_count": 3, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "All done!\n" ] } ], "source": [ "# Get fan_address table and concatenate the stuff we have, comma separated\n", "# ocean alley\n", "# schema = \"a441ffb172866cb4928c84a73de403ca15da4a54cc535e704413ab620\"\n", "# IHW\n", "#schema = \"bd345f915775993a4d3de1dae65b93b067ad69dde4286a1e1639e5cd2\"\n", "# pixies\n", "# schema='ace3c2bb4abd9fcaf9807975e6ab940131386dc9ecddb2b7e31288ff8'\n", "# Janis joplin\n", "schema='af65a0ef98e7420800afdea2beb81059bde2bc59d797cc0851412ac41'\n", "query2 = f\"\"\"\n", "SELECT fan_id\n", ", concat(fan_country, ', ', fan_city, ', ', fan_state, ', ',fan_area, ',',fan_address) as fan_location\n", "FROM {schema}.fan_address\n", "where concat(fan_country, ', ', fan_city, ', ', fan_state, ', ', fan_address) != ', , , '\n", ";\n", "\"\"\"\n", "query2 = f\"\"\"\n", "select f.fan_id,en.address as fan_location\n", "from {schema}.enriched_input en inner join\n", "{schema}.fan_table f on f.root_email=en.email where en.address is not null\n", ";\n", "\"\"\"\n", "\n", "df = pd.read_sql(query2, engine)\n", "\n", "for index, row in df.iterrows():\n", " address = row['fan_location']\n", "\n", " try:\n", " # Make API call to google geocoding api\n", " geolocation = get_geolocation(address)\n", "\n", " # Parse results to dataframe\n", " df = parse_geolocation_to_df(geolocation)\n", "\n", " df['fan_id'] = row['fan_id']\n", " #print(df)\n", " # Insert one row to database table\n", " df.to_sql(f'fan_address_geocoded_enriched',\n", " engine,\n", " schema=schema,\n", " if_exists='append',\n", " #index_label='fan_id',\n", " index=False\n", " )\n", " except Exception as e:\n", " print(e)\n", "print(\"All done!\")" ] }, { "cell_type": "code", "execution_count": 15, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ " administrative_area_level_1 administrative_area_level_2 country latitude \\\n", "0 Henan Sheng None China 34.508707 \n", "\n", " locality longitude \n", "0 Shangqiu Shi 116.015173 \n" ] } ], "source": [ "# use this cell to geocode aadress if we only have latitute and longitude\n", "# to test function uncomment next 5 lines- this example takes coordinates from San Diego\n", "\n", "# lat='-23.7136000'\n", "# lon='-46.5400000'\n", "# resp = get_geolocation_coord(lat,lon)\n", "# df2 = parse_geolocation_to_df(resp)\n", "# print(df2)" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Janis joplin\n", "schema='af65a0ef98e7420800afdea2beb81059bde2bc59d797cc0851412ac41'\n", "query3 = f\"\"\"\n", "select fag.fan_id, adr.fan_coordinates_lat as fan_latitude,\n", "adr.fan_coordinates_lon as fan_longitude from {schema}.fan_address_geocoded fag\n", "inner join {schema}.fan_address adr on adr.fan_id=fag.fan_id\n", "where fag.locality is null and adr.fan_coordinates_lat is not null and adr.fan_coordinates_lon is not null\n", ";\n", "\"\"\"\n", "\n", "df2 = pd.read_sql(query3, engine)\n", "df2['fan_latitude']=df2['fan_latitude'].str.replace(r\"[\\\"\\',]\", '')\n", "df2['fan_longitude']=df2['fan_longitude'].str.replace(r\"[\\\"\\',]\", '')\n", "print(df2)\n", "for index, row in df2.iterrows():\n", " lat = row['fan_latitude']\n", " lon = row['fan_longitude']\n", " \n", " try:\n", " # Make API call to google geocoding api\n", " geolocation = get_geolocation_coord(lat,lon)\n", "\n", " # Parse results to dataframe\n", " df = parse_geolocation_to_df(geolocation)\n", "\n", " df['fan_id'] = row['fan_id']\n", " #print(df)\n", " # Insert one row to database table\n", " df.to_sql(f'fan_address_geocoded_co',\n", " engine,\n", " schema=schema,\n", " if_exists='append',\n", " #index_label='fan_id',\n", " index=False\n", " )\n", " except Exception as e:\n", " print(e)\n", "print(\"All done!\")" ] } ], "metadata": { "kernelspec": { "display_name": "conda_python3", "language": "python", "name": "conda_python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.6.10" } }, "nbformat": 4, "nbformat_minor": 4 }