Spaces:
Runtime error
Runtime error
| 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?") | |