from google.cloud import pubsub_v1 from google.api_core.exceptions import DeadlineExceeded from selenium import webdriver from selenium.webdriver.chrome.options import Options from selenium.webdriver.common.by import By import json import time import os from concurrent.futures import TimeoutError import pandas as pd from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC # os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "C:/Users/Karen/AppData/Roaming/gcloud/application_default_credentials.json" # Create an empty dataframe to store the prices df = pd.DataFrame(columns=['Product ID', 'Zip Code', 'Price']) print(f"Created a dataframe: {df}") # Configure the Selenium driver options options = Options() options.add_argument('--disable-gpu') driver = webdriver.Chrome(options=options) # Set up Pub/Sub subscription project_id = "fp-20230425" subscription_name = "fp-subscription-NFL-zipcodes" subscriber = pubsub_v1.SubscriberClient() subscription_path = subscriber.subscription_path(project_id, subscription_name) # Load the product page and extract the __NEXT_DATA__ data url = 'https://www.sayweee.com/en/product/Want-Want-Senbei-Rice-Crackers/17119' product_id = url.split("/")[-1] driver.delete_all_cookies driver.get(url) # time.sleep(5) # time.sleep(2) # Close the cookies setting window try: # cookie_button = driver.find_element(By.XPATH, "/html/body/div[3]/div[2]/div/div/div[3]/button") cookie_button = WebDriverWait(driver, 2).until( EC.presence_of_element_located((By.XPATH, "/html/body/div[3]/div[2]/div/div/div[3]/button")) ) print(f"cookie_button = {cookie_button}") cookie_button.click() except: pass # time.sleep(1) # Clear any modal dialog try: # modal_close = driver.find_element(By.XPATH, "//*[contains(concat(' ', normalize-space(@class), ' '), ' icon-icon-close-s ')]") modal_close = WebDriverWait(driver, 2).until( EC.presence_of_element_located((By.XPATH, "//*[contains(concat(' ', normalize-space(@class), ' '), ' icon-icon-close-s ')]")) ) print(f"modal_close = {modal_close}") modal_close.click() #time.sleep(2) except: pass # Define callback function for Pub/Sub messages def callback(message): # Extract zip code from message data zip_code = message.message.data.decode('utf-8') print(f"Received zip code: {zip_code}") try: try: # time.sleep(2) # Clear any modal dialog # modal_close = driver.find_element(By.XPATH, "//*[contains(concat(' ', normalize-space(@class), ' '), ' icon-icon-close-s ')]") modal_close = WebDriverWait(driver, 2).until( EC.presence_of_element_located((By.XPATH, "//*[contains(concat(' ', normalize-space(@class), ' '), ' icon-icon-close-s ')]")) ) # print(f"modal_close = {modal_close}") modal_close.click() # time.sleep(2) except: pass try: # time.sleep(1) # Clear any notice # notice_close = driver.find_element(By.XPATH, "/html/body/div[3]/div/div[2]/div/div[2]/div/div/div[1]/div[2]/button") notice_close = WebDriverWait(driver, 5).until( EC.presence_of_element_located((By.XPATH, "/html/body/div[3]/div/div[2]/div/div[2]/div/div/div[1]/div[2]/button")) ) # print(f"notice_close = {notice_close}") notice_close.click() # time.sleep(1) except: pass # Enter zip code in the input field and click the "Check" button # zip_button = driver.find_element(By.XPATH, '/html/body/div[2]/div/header/div[3]/div/div/div[2]/div[1]') zip_button = WebDriverWait(driver, 5).until( EC.presence_of_element_located((By.XPATH, '/html/body/div[2]/div/header/div[3]/div/div/div[2]/div[1]')) ) # print(f"{zip_button:}") zip_button.click() # time.sleep(1) # zip_input = driver.find_element(By.XPATH, '//div/label/following-sibling::input') zip_input = WebDriverWait(driver, 2).until( EC.presence_of_element_located((By.XPATH, '//div/label/following-sibling::input')) ) # print(f"{zip_input:}") zip_input.click() # print("clearing zip_input") zip_input.send_keys("\b\b\b\b\b") # print("sending zip code") zip_input.send_keys(zip_code) check_button = driver.find_element(By.XPATH, "//button[text() = 'Confirm']") check_button = WebDriverWait(driver, 2).until(EC.presence_of_element_located((By.XPATH,"//button[text() = 'Confirm']"))) # print(f"{check_button:}") check_button.click() # time.sleep(2) try: # time.sleep(2) # Clear any notice # notice_close = driver.find_element(By.XPATH, "/html/body/div[3]/div/div[2]/div/div[2]/div/div/div[1]/div[2]/button") notice_close = WebDriverWait(driver, 2).until(EC.presence_of_element_located((By.XPATH, "/html/body/div[3]/div/div[2]/div/div[2]/div/div/div[1]/div[2]/button"))) # print(f"notice_close = {notice_close}") notice_close.click() #time.sleep(2) except: pass try: # Clear any modal dialog modal_close = WebDriverWait(driver, 2).until(EC.presence_of_element_located((By.XPATH, "//*[contains(concat(' ', normalize-space(@class), ' '), ' icon-icon-close-s ')]"))) # print(f"modal_close = {modal_close}") modal_close.click() except: pass # Extract the price of rice crackers for this zip code ''' price_element = driver.find_element(By.XPATH, '/html/body/div[2]/div/div[1]/div[1]/div/div/div[2]/div[2]/div[1]/div[4]/div[1]/div') price = price_element.text.strip()[1:] print(price) ''' next_data_element = driver.find_element(By.XPATH, "//script[@type='application/ld+json']") next_data_text = next_data_element.get_attribute('innerHTML') next_data_start_index = next_data_text.index('{') next_data_end_index = next_data_text.rindex('}') + 1 next_data_json = next_data_text[next_data_start_index:next_data_end_index] next_data = json.loads(next_data_json) price = next_data['offers']['price'] # Print the price print(price) subscriber.acknowledge(ack_ids = [message.ack_id], subscription = subscription_path) # acknowledge receipt of message # Append the zip code and price to the dataframe #append_to_df(zip_code, price) new_row = pd.DataFrame({'Product ID': [product_id], 'Zip Code': [zip_code], 'Price': [price]}) return pd.concat([df, new_row], ignore_index=True) #print(f"Recorded the zip code and price: {zip_code},{price}") #print(f"Updated a dataframe: {df}") except Exception as e: print(f"Caught exception: {e}") print(f"Price not found for zip code {zip_code}") price = 'Price not found' return df # Start receiving messages (sync version) while True: pull_request = pubsub_v1.types.PullRequest(max_messages = 1, subscription = subscription_path) messages = [] try: pull_response = subscriber.pull(request = pull_request) messages = pull_response.received_messages except DeadlineExceeded: pass if messages: df = callback(messages[0]) print(f"Updated a dataframe: {df}") else: print(f"no messages, trying again at {time.asctime()}") break df.to_hdf('product1.h5', key='df',mode='w') # Open the HDF5 file in a separate window os.startfile('product1.h5') # Start receiving messages (async version) #flow_control = pubsub_v1.types.FlowControl(max_messages = 1) #streaming_pull_future = subscriber.subscribe(subscription_path, # callback=callback, # flow_control=flow_control) # print(f'Started receiving messages from subscription {subscription_path}...') # Wrap subscriber in a 'with' block to automatically call close() when done. # with subscriber: # try: # When `timeout` is not set, result() will block indefinitely, # unless an exception is encountered first. # streaming_pull_future.result(timeout=timeout) # except TimeoutError: # streaming_pull_future.cancel() # Trigger the shutdown. # streaming_pull_future.result() # Block until the shutdown is complete. # Keep the script running indefinitely to continue receiving messages #while True: # time.sleep(10)