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.support.wait import WebDriverWait from selenium.webdriver.common.by import By from selenium.webdriver.common.keys import Keys import selenium.common.exceptions import json import time import os import re import sys 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', 'Displayed Price', 'ld+json Price', 'Minimum Quantity', 'Total Price', 'Sales Count', 'Sold Status']) 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 def close_modals(): try: # time.sleep(2) # Clear any modal dialog modal_close_buttons = driver.find_elements(By.XPATH, "//*[contains(concat(' ', normalize-space(@class), ' '), ' icon-icon-close-s ')]") print(f"modal_close = {modal_close}") for button in modal_close_buttons: try: button.click() except Exception as e: print(f"Exception when clicking on close button for element {button}: {e}") # time.sleep(2) except Exception as e: print(f"Exception when trying to close discount modal dialog: {e}") 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") print(f"notice_close = {notice_close}") notice_close.click() # time.sleep(1) except selenium.common.exceptions.NoSuchElementException as nsee: print("Purchase option change dialog close button not found") except Exception as e: print(f"Exception when trying to close purchase option change dialog: {e}") def get_element_text(class_name_prefix): ''' Return the text in the first element with a given class name. Note that XPath doesn't have a function to check if a given word is present in a list of words, so using contains() will match a class name even if the class name is a substring of the class name being matched against. To prevent this, we use spaces as a delimiter at the beginning and end of the class name, and ensure there's only one space before and after all class names in the class name list. Example: consider two class names with cache-busting suffixes: * Header_soldStatusNotice__12345 * Header_soldStatusNoticeBackground__ABCDE Matching against Header_soldStatusNotice using a naive implementation with `contains()` would match both class names, even though only the first one is the correct match. ''' # Construct xpath selector. xpath_str = "//*[contains(" # Step 1: search for a class name by matching the beginning of the class name against... xpath_str += "concat(' ', " # Step 2: a leading space, followed by... xpath_str += "normalize-space(@class), " # Step 3: the list of class names, separated by single spaces... xpath_str += "' '), " # Step 4: a trailing space... xpath_str += f"' {class_name_prefix} ')]" # Step 5: compared against the class name with a single space on either side. try: match = driver.find_element(By.XPATH, xpath_str) return match.get_attribute('textContent') except selenium.common.exceptions.NoSuchElementException as nsee: pass # We didn't find a match, try again by doing a prefix match and assume # the full class name ends with two underscores followed by a cache- # busting string. xpath_str = "//*[contains(" # Step 1: search for a class name by matching the beginning of the class name against... xpath_str += "concat(' ', " # Step 2: a leading space, followed by... xpath_str += "normalize-space(@class)), " # Step 3: the list of class names, separated by single spaces... xpath_str += f"' {class_name_prefix}__')]" # Step 4: compared against the class name prefixed with a single space and ending with two underscores. try: match = driver.find_element(By.XPATH, xpath_str) return match.get_attribute('textContent') except selenium.common.exceptions.NoSuchElementException as nsee: pass # No match, return empty string. return '' def callback(message): ''' Callback function for Pub/Sub messages ''' # Extract zip code from message data zip_code = message.message.data.decode('utf-8') print(f"Received zip code: {zip_code}") while True: try: # Enter zip code in the input field and click the "Check" button WebDriverWait(driver, 2).until(lambda d: d.find_element(By.XPATH, '/html/body/div[2]/div/header/div[3]/div/div/div[2]/div[1]')) zip_button = driver.find_element(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) WebDriverWait(driver, 2).until(lambda d: d.find_element(By.XPATH, '//div/label/following-sibling::input')) zip_input = driver.find_element(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\b\b\b\b\b") for i in range(1,10): zip_input.send_keys(Keys.DELETE) # print("sending zip code") zip_input.send_keys(zip_code) WebDriverWait(driver, 2).until(lambda d: d.find_element(By.XPATH, "//button[text() = 'Confirm']")) check_button = driver.find_element(By.XPATH, "//button[text() = 'Confirm']") print(f"{check_button:}") check_button.click() # time.sleep(2) break except Exception as e: print(f"Exception when changing zip code: {e}, trying to close modal dialogs") close_modals() # 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") # print(f"notice_close = {notice_close}") # notice_close.click() # time.sleep(2) # except: # pass 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 # Record sales count, if any sales_count = get_element_text('Header_count') # Record any sold status notice sold_status = get_element_text('soldStatusNotice') # Check if there's a minimum order quantity min_quantity: float = 0 total_price: float = 0 try: minimum_quantity_str = driver.find_element(By.XPATH, "//div[contains(concat(' ', normalize-space(@class), ' '), ' minQuantity ')]").get_attribute('innerHTML') if minimum_quantity_str: regexp="Minimum purchase: (\d+) \(\$(\d+\.\d{2}?)" match = re.search(regexp, minimum_quantity_str) min_quantity, total_price = match.groups() except selenium.common.exceptions.NoSuchElementException as nsee: print("No minimum purchase string found") except Exception as e: print(f"Exception when finding minimum purchase string: {e}") # Get the price as displayed on the web page displayed_price = driver.find_element(By.XPATH, "//div[contains(concat(' ', normalize-space(@class), ' '), ' price_price ')]").get_attribute('innerText').replace('$', '') print(f"displayed_price = {displayed_price}") # 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) ''' # Get the price from the ld+json element ld_json_element = driver.find_element(By.XPATH, "//script[@type='application/ld+json']") ld_json_text = ld_json_element.get_attribute('innerHTML') ld_json_start_index = ld_json_text.index('{') ld_json_end_index = ld_json_text.rindex('}') + 1 ld_json_json = ld_json_text[ld_json_start_index:ld_json_end_index] ld_json = json.loads(ld_json_json) ld_json_price = ld_json['offers']['price'] # Print the price print(f"ld_json_price = {ld_json_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], 'Displayed Price': [displayed_price], 'ld+json Price': [ld_json_price], 'Minimum Quantity': [min_quantity], 'Total Price': [total_price], 'Sales Count': [sales_count], 'Sold Status': [sold_status]}) 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 {type(e)}: {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}") df.to_hdf('product1.h5', key='df',mode='w') else: print(f"No messages in subscription {subscription_path} at {time.asctime()}, trying again...") continue # Open the HDF5 file in a separate window if sys.platform.startswith("win32"): 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)