from selenium import webdriver from selenium.webdriver.common.by import By from selenium.webdriver.common.keys import Keys from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions from selenium.webdriver.chrome.service import Service from selenium.webdriver.chrome.options import Options from uszipcode import SearchEngine import pandas as pd gasBuddyurl = 'https://www.gasbuddy.com/home' # To make this work as of 08/16/2024 # Make sure to use sqlalchemy_mate = 2.0.0.0 def fetchZipCodes(): engine = SearchEngine() allzips = {} population_threshold = 10000 density_threshold = 1000 for i in range(100000): #Get zipcode info for every possible 5-digit combination zipcode = str(i).zfill(5) try: allzips[zipcode] = engine.by_zipcode(zipcode).to_dict() except: pass #Convert dictionary to DataFrame allzips = pd.DataFrame(allzips).T.reset_index(drop = True) allzips = allzips.convert_dtypes() allzips.dropna(inplace=True) columnsNeeded = ['zipcode', 'zipcode_type', 'major_city','county', 'state', 'population', 'population_density'] zipsRequired = allzips[columnsNeeded] zipsRequired = zipsRequired.query('zipcode_type == "STANDARD"') # Example thresholds (you can adjust these based on your criteria) # Assuming your DataFrame is named `df` and has columns 'population' and 'population_density' filtered_df = zipsRequired[(zipsRequired['population'] > population_threshold) & (zipsRequired['population_density'] > density_threshold)] # Excluding Territories exclude_states = ['PR', 'VI', 'GU'] # Filter the DataFrame zipsRequired = filtered_df[~filtered_df['state'].isin(exclude_states)] return filtered_df def GasBuddy(url, zipCodes, headless = True): chrome_options = Options() if headless: chrome_options.add_argument("--headless") # Run in headless mode chrome_options.add_argument("--no-sandbox") # Sandbox mode disabled (Linux-specific) driverLocal = webdriver.Chrome(service=Service(), options=chrome_options) driverLocal.get(url) price_Heads = {} price_Heads.fromkeys(zipCodes) for z in zipCodes: price_Heads[z] = {} price_Heads[z].setdefault('price', []) price_Heads[z].setdefault('station', []) # Enter the Zip Code and Hit Enter try: enterZipCode = WebDriverWait(driverLocal, 5).until( expected_conditions.presence_of_element_located((By.XPATH, \ ".//div[@class='input__inputGroup___3ogvw'] \ /input[@id='search'] \ "))) enterZipCode.clear() enterZipCode.send_keys(z) enterZipCode.send_keys(Keys.RETURN) # Check for the presence of the error message error_message = None try: error_message = WebDriverWait(driverLocal, 2).until( expected_conditions.presence_of_element_located((By.XPATH, ".//div/em[contains(text(), 'No stations found')]"))) except: pass if error_message: price_Heads[z] = f"No stations found for ZIP code: {z}" continue blocks = WebDriverWait(driverLocal, 100).until( expected_conditions.presence_of_all_elements_located((By.XPATH, \ "\ //div[@class='GenericStationListItem-module__stationListItem___3Jmn4'] \ "))) for b in blocks: t = b.find_element(By.XPATH, ".//h3[contains(@class, 'StationDisplay-module__stationNameHeader___1A2q8')]").text p = b.find_element(By.XPATH, ".//span[@class='text__xl___2MXGo text__left___1iOw3 StationDisplayPrice-module__price___3rARL']").text price_Heads[z].get('station').append(t) price_Heads[z].get('price').append(p) except: pass driverLocal.close() return price_Heads def convertDF(data): # Flatten the dictionary to create a list of records records = [] for zipcode, details in data.items(): if isinstance(details, dict): for price, heading in zip(details['price'], details['station']): # Handle missing price by replacing '- - -' with 0, and remove the '$' sign if price == '- - -': continue else: price_value = float(price.replace('$', '')) records.append({'zipcode': zipcode, 'station': heading.strip(), 'price': price_value}) return pd.DataFrame(records) def loadLottieFile(filePath: str): with open(file=filePath, mode = "r") as f: return json.load(f) import streamlit as st import json from streamlit_lottie import st_lottie_spinner st.set_page_config(layout="wide", page_title="Portfolio Optimization For Commodities", page_icon="Portfolio_Icon.jpg") if 'loadOnce' not in st.session_state: anm = loadLottieFile("Anims/location.json") with st_lottie_spinner(anm , height=400): st.session_state.loadOnce = fetchZipCodes() if 'scrappedData' not in st.session_state: st.session_state.scrappedData = None # Group by major_city and state, and aggregate zip codes into lists grouped = st.session_state.loadOnce.groupby(['major_city', 'state'])['zipcode'].apply(list).reset_index() # Create a dictionary with formatted keys and list of zip codes as values city_state_zip_dict = { f"{row['major_city']}, {row['state']}": row['zipcode'] for _, row in grouped.iterrows() } _,center,_ = st.columns((2,4,2)) with center.form(key = "parameters"): st.subheader("Please select locations and configuration parameters as per your scraping needs") #-------------------------------------------- Zip Codes ---------------------------------------------------------------- st.write("") selected_locations = st.multiselect('Select Locations and Data will be scrapped as per their zip codes', list(city_state_zip_dict.keys()), default=list()) st.write("") head = st.checkbox(label="Please check if you want headless scrapping") submitted = st.form_submit_button("Download") _, _leftDf, cenButtons, _rightDf, _ = st.columns((2,4,2,4,2)) if submitted: # Fetch values for the specified keys and flatten the list zipcodes = [zipcode for key in selected_locations if key in city_state_zip_dict for zipcode in city_state_zip_dict[key]] anm = loadLottieFile("Anims/citations.json") with st_lottie_spinner(anm , height=400): prices = GasBuddy(url=gasBuddyurl, zipCodes=zipcodes, headless=head) finalDf = convertDF(prices) finalDf2 = pd.merge(finalDf, st.session_state.loadOnce[['zipcode', 'major_city', 'state']], on='zipcode', how='left') st.session_state.scrappedData = finalDf2 _leftDf.dataframe(finalDf2, use_container_width=True) cols = st.session_state.scrappedData.columns if st.session_state.scrappedData is not None else [] with cenButtons.form(key='parametersForm2'): # if st.session_state.scrappedData is not None: ind = st.multiselect(label='Column Names for index (new row identifiers) of Pivot Table', options= cols) val = st.multiselect(label='Column Names for values of Pivot Table', options= cols) col = st.multiselect(label='Column Names for columns of Pivot Table', options= cols) submitted2 = st.form_submit_button("Pivot Data") if submitted2: try: pivoted = st.session_state.scrappedData.pivot_table(index=ind, values=val, columns=col) _rightDf.dataframe(pivoted, use_container_width = True) except: cenButtons.error("Your Configurations for Pivot Table are not accepable. Do you mind trying a different combination?")