SVIG/pricer
7
0
Fork 0
Code Issues 5 Pull Requests Releases Wiki Activity
Files
104ae2e903f96e052f4a6fe9dd878612cf9e7914
pricer/fp2_14.py
T

281 lines
12 KiB
Python
Raw Blame History

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
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
import selenium.common.exceptions
import re
from datetime import datetime
import h5py
import sys
import ntplib
import arrow
# This fp2_14.py is used to load the new price and the timestamp, and append such data in an existing key in the HDF5 file "product1.h5"
def get_product_prices(project_id, subscription_name, url):
# Create an empty dataframe to store the prices
df_new = pd.DataFrame(columns=['Product ID', 'Zip Code', 'Shown Price', 'Minimum Quantity', 'Total Price', 'Unit Price','Sales Count','Sold Status'])
print(f"Created a dataframe: {df_new}")
# 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/House-Foods-Firm-Tofu-/85175?trace_id=c93f3d1a-80a4-4e1a-8d41-a402ae43186f'
product_id = url.split("/")[-1][:5]
# product_id = url.split("/")[-1]
driver.delete_all_cookies
driver.get(url)
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 modal_close():
try:
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
def notice_close():
try:
# Clear any notice
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()
except:
pass
def cookie_close():
try:
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
cookie_close()
modal_close()
# 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:
modal_close()
notice_close()
# Enter zip code in the input field and click the "Check" button
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()
# 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()
notice_close()
modal_close()
# Record sales count, if any
sales_count_element = get_element_text('Header_count')
count_start_index = sales_count_element.index('sold ') + len('sold ')
count_end_index = sales_count_element.index('+')
sales_count = sales_count_element[count_start_index:count_end_index]
# Record any sold status notice
sold_status = get_element_text('UnavailableCartbtn_container')
if sold_status == 'Notify me':
sold_status = 'Sold Out'
# Extract the price of rice crackers for this zip code
# Check if there is a minimum order quantity
min_quantity: float = 0
total_price: float = 0
unit_price: float = 0
try:
minimum_quantity_str = WebDriverWait(driver, 5).until(EC.presence_of_element_located(((By.XPATH, "//div[contains(@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 purhcase string found")
except Exception as e:
print(f"Exception when finding minimum purchase string: {e}")
ld_json_element = WebDriverWait(driver, 5).until(EC.presence_of_element_located((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(ld_json_price)
unit_price_element = WebDriverWait(driver, 5).until(EC.presence_of_element_located(((By.XPATH,"//div[contains(@class, 'price_price')]").get_attribute('innerText'))))
unit_price = unit_price_element.replace("$", "")
print(f"unit_price = {unit_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], 'Shown Price': [ld_json_price], 'Minimum Quantity': [min_quantity], 'Total Price': [total_price], 'Unit Price': [unit_price], 'Sales Count': [sales_count], 'Sold Status': [sold_status]})
df_new = pd.concat([df_new, new_row], ignore_index=True)
return df_new
#print(f"Recorded the zip code and price: {zip_code},{price}")
#print(f"Updated a dataframe: {df_new}")
except Exception as e:
print(f"Caught exception: {e}")
print(f"Price not found for zip code {zip_code}")
return df_new
# 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_new = callback(messages[0])
print(f"Updated a dataframe: {df_new}")
else:
print(f"no messages, trying again at {time.asctime()}")
break
return df_new
'''
If using sys to get the function, please remove the following part of code.
'''
if __name__ == '__main__':
project_id = input("Enter the project ID: ")
subscription_name = input("Enter the subscription name: ")
url = input("Enter the URL: ")
get_product_prices(project_id, subscription_name, url)
df_new = get_product_prices("fp-20230425", "fp-subscription-NFL-zipcodes", "https://www.sayweee.com/en/product/Want-Want-Senbei-Rice-Crackers/17119")
# Now we have the new dataframe "df" recording the new price listed in weee's page
# Load such data in the existing HDF5 file
file_path = 'product1.h5'
with h5py.File(file_path, 'a') as file:
# Load the existing dataframe 'df' from the HDF5 file
# REPLACE KEY TO GET THE EXISTING DATAFRAME FOR EACH DIFFERENT PRODUCT
df = pd.read_hdf(file, key='df')
# Add timestamp to the new dataframe "df_new"
# timestamp = datetime.now()
'''
To deal with the issue of inconsistent time when running a script on different systems, we can use 'ntplib' that
provides time synchronization with a reliable time server.
Also, to ensure that the local time is interpreted in the system's local time zone, we use 'arrow' which offers a
sensible and human-friendly approach to format and convert the timestamps.
'''
# Fetch the time from an NTP server
ntp_client = ntplib.NTPClient()
response = ntp_client.request('pool.ntp.org')
ntp_time = datetime.fromtimestamp(response.tx_time)
timestamp = arrow.get(ntp_time).to('local')
df_new['Timestamp'] = timestamp
# Append 'df_new' to the existing df stored in the key
df_update = df.append(df_new, ignore_index=True)
# Save 'df2' to the HDF5 file with the key 'df'
# REPLACE KEY FOR DIFFERENT PRODUCT
df_update.to_hdf(file, key='df', mode='a')
print("Data appended and saved successfully.")
Reference in New Issue View Git Blame Copy Permalink