SVIG/python-bootcamp
7
0
Fork 1
Code Issues Pull Requests 1 Releases Wiki Activity

Defined get_price function & modified timestamp

Browse Source
This commit is contained in:
Karenzz
2023-05-23 17:59:16 -04:00
parent 93d386733a
commit 1aac3a15b1
3 changed files with 276 additions and 220 deletions
Download Patch File Download Diff File Expand all files Collapse all files
__pycache__/fp2_14.cpython-311.pyc
BIN
View File
Binary file not shown.
fp2_14.py
+65 -72
View File
@@ -11,11 +11,15 @@ from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import selenium.common.exceptions
import re
import datetime
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}")
@@ -26,18 +30,20 @@ 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'
# 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)
# time.sleep(5)
def get_element_text(class_name_prefix):
'''
@@ -87,10 +93,25 @@ def get_element_text(class_name_prefix):
return ''
# time.sleep(2)
# Close the cookies setting window
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 = 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"))
)
@@ -99,19 +120,8 @@ try:
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
cookie_close()
modal_close()
# Define callback function for Pub/Sub messages
def callback(message):
# Extract zip code from message data
@@ -120,41 +130,16 @@ def callback(message):
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
modal_close()
notice_close()
# 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(
@@ -171,26 +156,9 @@ def callback(message):
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
notice_close()
modal_close()
# Record sales count, if any
sales_count_element = get_element_text('Header_count')
@@ -208,7 +176,7 @@ def callback(message):
total_price: float = 0
unit_price: float = 0
try:
minimum_quantity_str = driver.find_element(By.XPATH, "//div[contains(@class, 'minQuantity')]").get_attribute('innerHTML')
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)
@@ -220,7 +188,7 @@ def callback(message):
ld_json_element = driver.find_element(By.XPATH, "//script[@type='application/ld+json']")
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
@@ -230,7 +198,7 @@ def callback(message):
# Print the price
print(ld_json_price)
unit_price_element = driver.find_element(By.XPATH,"//div[contains(@class, 'price_price')]").get_attribute('innerText')
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}" )
@@ -238,14 +206,14 @@ def callback(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]})
return pd.concat([df_new, new_row], ignore_index=True)
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}")
price = 'Price not found'
return df_new
# Start receiving messages (sync version)
@@ -260,10 +228,24 @@ while True:
if messages:
df_new = callback(messages[0])
print(f"Updated a dataframe: {df}")
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
@@ -274,7 +256,18 @@ with h5py.File(file_path, 'a') as file:
df = pd.read_hdf(file, key='df')
# Add timestamp to the new dataframe "df_new"
timestamp = datetime.datetime.now()
# 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
test_fp2_14.py
+63
View File
@@ -0,0 +1,63 @@
from fp2_14 import get_product_prices
import h5py
import pandas as pd
import datetime
import sys
from datetime import datetime
import ntplib
import arrow
project_id = "fp-20230425"
subscription_name = "fp-subscription-NFL-zipcodes"
url = "https://www.sayweee.com/en/product/House-Foods-Firm-Tofu-/85175?trace_id=c93f3d1a-80a4-4e1a-8d41-a402ae43186f"
df_new = get_product_prices(project_id, subscription_name, url)
'''
# When using sys to call the 'get_product_prices' function
if __name__ == '__main__':
# Check if the required command-line arguments are provided
if len(sys.argv) < 4:
print("Usage: python test_fp2_14.py project_id subscription_name url")
sys.exit(1)
project_id = sys.argv[1]
subscription_name = sys.argv[2]
url = sys.argv[3:]
get_product_prices(project_id, subscription_name, url)
'''
# 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.")