17 Commits
Author SHA1 Message Date
leonhuang da2c779bcd adfafafafsafafasfasfasfafafafsagagafafaagagagagagag 2023-05-18 19:27:06 -05:00
leonhuang 7259509d9e adadjalkfjlaskfjlaksfjlaskfjlkasfjlakfjakfjlasfjlgjslga 2023-05-18 19:27:06 -05:00
scott.liu 59e84426af A brain teaser 2023-05-18 16:59:41 -07:00
scott.liu 97f93ebd8e A brain teaser w/ answer 2023-05-18 16:58:54 -07:00
scott.liu 35c8889e77 A brain teaser 2023-05-18 16:56:56 -07:00
cheezedoodle2 dca7e9d599 foo 2023-05-18 12:47:27 -07:00
cheezedoodle2 ce856e5626 foo 2023-05-18 12:44:01 -07:00
cheezedoodle2 072032e80e foo 2023-05-18 12:39:46 -07:00
cheezedoodle2 235e7b5268 foo 2023-05-18 12:38:03 -07:00
cheezedoodle2 22fa096271 Removed foo 2023-05-18 03:08:11 -07:00
scott.liu 13eb12bf69 An update on the test file 2023-05-18 02:49:43 -07:00
scott.liu efa189d0b6 Commit 2 2023-05-18 02:33:11 -07:00
scott.liu 111f551a39 This is a test commit 2023-05-18 02:33:11 -07:00
cheezedoodle2 e0fd81dc1d foo 2023-05-18 02:21:48 -07:00
cheezedoodle2 420194ef54 Removed test file 2023-05-17 17:31:15 -07:00
cheezedoodle2 632448c249 added test file 2023-05-17 17:30:22 -07:00
Sloetjes c3405d56e5 Homework for web server 2023-04-19 20:45:08 -04:00
70 changed files with 223 additions and 4048 deletions
+100
View File
@@ -0,0 +1,100 @@
# A quick brain teaser
print('What has a head and a tail but no body?')
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
# A coin
-276
View File
@@ -1,276 +0,0 @@
import multiprocessing
import subprocess
import concurrent.futures
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
import selenium.common.exceptions
import re
from fp2_14 import get_product_prices
# Define the subset of urls
url_list = [
'https://www.sayweee.com/en/product/Want-Want-Senbei-Rice-Crackers/17119',
'https://www.sayweee.com/en/product/Lays-Potato-Chips-Cucumber-Flavor/51261',
'https://www.sayweee.com/en/product/Glico-Pocky-Cookies-and-Cream-Family-Pack/79467',
'https://www.sayweee.com/en/product/Xiang-Na-Lan-Thai-Jasmine-Rice/19566?category=dried01',
'https://www.sayweee.com/en/product/Sekka-Premium-Japanese-Medium-Grain-Rice/81224',
'https://www.sayweee.com/en/product/Big-Green-Spring-Farm-Organic-Millet/90991?category=dried02',
'https://www.sayweee.com/en/product/Liuzhou-Guangxi-Specialty-LuoSiFen-Hot-Spicy-Noodles/95442?category=instant01',
'https://www.sayweee.com/en/product/House-Foods-Firm-Tofu-/85175?trace_id=c93f3d1a-80a4-4e1a-8d41-a402ae43186f',
'https://www.sayweee.com/en/product/Lactaid-Lactose-Free-2--Reduced-Fat-Milk/95353',
'https://www.sayweee.com/en/product/Washington-Fuji-Apples/94817?category=fruits03',
'https://www.sayweee.com/en/product/Formosa-Pork-Sung-/23528?trace_id=3c143dd4-4a35-4e8d-93ee-d91f209717e7'
]
# Create a pool of worker processes
# pool = multiprocessing.Pool(cpu_count)
# Define the function to run the scraper
def run_scraper(url):
# scraping logic
df = 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}")
product_id = url.split("/")[-1][:5]
options = Options()
options.add_argument('--disable-gpu')
driver = webdriver.Chrome(options=options)
project_id = "fp-20230425"
subscription_name = "fp-subscription-NFL-zipcodes"
subscriber = pubsub_v1.SubscriberClient()
subscription_path = subscriber.subscription_path(project_id, subscription_name)
# product_id = url.split("/")[-1]
driver.delete_all_cookies
driver.get(url)
def get_element_text(class_name_prefix):
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
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
return ''
# Close the cookies setting window
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
# Clear any modal dialog
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
# 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:
# 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
try:
# Clear any notice
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()
except:
pass
# 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 = 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:
# 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
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_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 = driver.find_element(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 = 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(ld_json_price)
unit_price_element = driver.find_element(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]})
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
# subprocess.call(['python', 'fp2_13.py', url])
return df
if __name__ == '__main__':
# Determine the number of available CPUs
# cpu_count = multiprocessing.cpu_count()
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
future_to_url = {executor.submit(run_scraper, url): url for url in url_list}
results = []
for future in concurrent.futures.as_completed(future_to_url):
url = future_to_url[future]
try:
result = future.result()
print(f"Scraped data for URL: {url}")
results.append(result)
except Exception as e:
print(f"Failed to scrape data for URL: {url}\nError: {e}")
'''
# Run the scraper in parallel for the subset of items
results = pool.map(run_scraper, url_list)
# Close the pool and wait for the processes to finish
pool.close()
pool.join()
'''
with pd.HDFStore('product_data.h5') as store:
for url, result in zip(url_list, results):
# Extract the product ID from the URL
product_id = url.split("/")[-1][:5]
# Save the dataframe to the HDF5 file with a unique key
key = f'df_{product_id}'
store[key] = result
# result.to_hdf('product_data.h5', key=key, mode='a')
-33
View File
@@ -1,33 +0,0 @@
ID,Team Name,Arena Name,Abbreviation,Conference,Division,Arena Location,ZIP Code
1,Arizona Cardinals,State Farm Stadium,ARI,NFC,West,"Glendale, Arizona",85305
2,Atlanta Falcons,Mercedes-Benz Stadium,ATL,NFC,South,"Atlanta, Georgia",30313
3,Baltimore Ravens,M&T Bank Stadium,BAL,AFC,North,"Baltimore, Maryland",21230
4,Buffalo Bills,Bills Stadium,BUF,AFC,East,"Orchard Park, New York",14127
5,Carolina Panthers,Bank of America Stadium,CAR,NFC,South,"Charlotte, North Carolina",28202
6,Chicago Bears,Soldier Field,CHI,NFC,North,"Chicago, Illinois",60605
7,Cincinnati Bengals,Paul Brown Stadium,CIN,AFC,North,"Cincinnati, Ohio",45202
8,Cleveland Browns,First Energy Stadium,CLE,AFC,North,"Cleveland, Ohio",44114
9,Dallas Cowboys,At&T Stadium,DAL,NFC,East,"Arlington, Texas",76011
10,Denver Broncos,Empower Field at Mile High,DEN,AFC,West,"Denver, Colorado",80204
11,Detroit Lions,Ford Field,DET,NFC,North,"Detroit, Michigan",48226
12,Green Bay Packers,Lambeau Field,GB,NFC,North,"Green Bay, Wisconsin",54304
13,Houston Texans,NRG Stadium,HOU,AFC,South,"Houston, Texas",77054
14,Indianapolis Colts,Lucas Oil Stadium,IND,AFC,South,"Indianapolis, Indiana",46225
15,Jacksonville Jaguars,TIAA Bank Field,JAX,AFC,South,"Jacksonville, Florida",32202
16,Kansas City Chiefs,Arrowhead Stadium,KC,AFC,West,"Kansas City, Missouri",64129
17,Las Vegas Raiders,Allegiant Stadium,LV,AFC,West,"Paradise, Nevada",89118
18,Los Angeles Chargers,SoFi Stadium,LAC,AFC,West,"Inglewood, California",90301
19,Los Angeles Rams,SoFi Stadium,LAR,NFC,West,"Inglewood, California",90301
20,Miami Dolphins,Hard Rock Stadium,MIA,AFC,East,"Miami Gardens, Florida",33056
21,Minnesota Vikings,U.S. Bank Stadium,MIN,NFC,North,"Minneapolis, Minnesota",55415
22,New England Patriots,Gillette Stadium,NE,AFC,East,"Foxborough, Massachusetts",02035
23,New Orleans Saints,Mercedes-Benz Superdome,NO,NFC,South,"New Orleans, Louisiana",70112
24,New York Giants,MetLife Stadium,NYG,NFC,East,"East Rutherford, New Jersey",07073
25,New York Jets,MetLife Stadium,NYJ,AFC,East,"East Rutherford, New Jersey",07073
26,Philadelphia Eagles,Lincoln Financial Field,PHI,NFC,East,"Philadelphia, Pennsylvania",19148
27,Pittsburgh Steelers,Heinz Field,PIT,AFC,North,"Pittsburgh, Pennsylvania",15212
28,San Francisco 49ers,Levi's Stadium,SF,NFC,West,"Santa Clara, California",95054
29,Seattle Seahawks,CenturyLink Field,SEA,NFC,West,"Seattle, Washington",98134
30,Tampa Bay Buccaneers,Raymond James Stadium,TB,NFC,South,"Tampa, Florida",33607
31,Tennessee Titans,Nissan Stadium,TEN,AFC,South,"Nashville, Tennessee",37213
32,Washington Redskins,FedExField,WAS,NFC,East,"Landover, Maryland",20785
1 ID Team Name Arena Name Abbreviation Conference Division Arena Location ZIP Code
2 1 Arizona Cardinals State Farm Stadium ARI NFC West Glendale, Arizona 85305
3 2 Atlanta Falcons Mercedes-Benz Stadium ATL NFC South Atlanta, Georgia 30313
4 3 Baltimore Ravens M&T Bank Stadium BAL AFC North Baltimore, Maryland 21230
5 4 Buffalo Bills Bills Stadium BUF AFC East Orchard Park, New York 14127
6 5 Carolina Panthers Bank of America Stadium CAR NFC South Charlotte, North Carolina 28202
7 6 Chicago Bears Soldier Field CHI NFC North Chicago, Illinois 60605
8 7 Cincinnati Bengals Paul Brown Stadium CIN AFC North Cincinnati, Ohio 45202
9 8 Cleveland Browns First Energy Stadium CLE AFC North Cleveland, Ohio 44114
10 9 Dallas Cowboys At&T Stadium DAL NFC East Arlington, Texas 76011
11 10 Denver Broncos Empower Field at Mile High DEN AFC West Denver, Colorado 80204
12 11 Detroit Lions Ford Field DET NFC North Detroit, Michigan 48226
13 12 Green Bay Packers Lambeau Field GB NFC North Green Bay, Wisconsin 54304
14 13 Houston Texans NRG Stadium HOU AFC South Houston, Texas 77054
15 14 Indianapolis Colts Lucas Oil Stadium IND AFC South Indianapolis, Indiana 46225
16 15 Jacksonville Jaguars TIAA Bank Field JAX AFC South Jacksonville, Florida 32202
17 16 Kansas City Chiefs Arrowhead Stadium KC AFC West Kansas City, Missouri 64129
18 17 Las Vegas Raiders Allegiant Stadium LV AFC West Paradise, Nevada 89118
19 18 Los Angeles Chargers SoFi Stadium LAC AFC West Inglewood, California 90301
20 19 Los Angeles Rams SoFi Stadium LAR NFC West Inglewood, California 90301
21 20 Miami Dolphins Hard Rock Stadium MIA AFC East Miami Gardens, Florida 33056
22 21 Minnesota Vikings U.S. Bank Stadium MIN NFC North Minneapolis, Minnesota 55415
23 22 New England Patriots Gillette Stadium NE AFC East Foxborough, Massachusetts 02035
24 23 New Orleans Saints Mercedes-Benz Superdome NO NFC South New Orleans, Louisiana 70112
25 24 New York Giants MetLife Stadium NYG NFC East East Rutherford, New Jersey 07073
26 25 New York Jets MetLife Stadium NYJ AFC East East Rutherford, New Jersey 07073
27 26 Philadelphia Eagles Lincoln Financial Field PHI NFC East Philadelphia, Pennsylvania 19148
28 27 Pittsburgh Steelers Heinz Field PIT AFC North Pittsburgh, Pennsylvania 15212
29 28 San Francisco 49ers Levi's Stadium SF NFC West Santa Clara, California 95054
30 29 Seattle Seahawks CenturyLink Field SEA NFC West Seattle, Washington 98134
31 30 Tampa Bay Buccaneers Raymond James Stadium TB NFC South Tampa, Florida 33607
32 31 Tennessee Titans Nissan Stadium TEN AFC South Nashville, Tennessee 37213
33 32 Washington Redskins FedExField WAS NFC East Landover, Maryland 20785
BIN
View File
Binary file not shown.

Before

Width:  |  Height:  |  Size: 4.8 KiB

-23
View File
@@ -17,27 +17,4 @@ the focus is on *working results*, not rote memorization. Homework questions
are simple enough that they can be reviewed interactively in 1:1 or up to 1:4 are simple enough that they can be reviewed interactively in 1:1 or up to 1:4
sessions with study groups. sessions with study groups.
This branch contains a prototype for a retail price scraper for getweee.com.
To run it, install:
* Python 3.9+,
* chrome,
* chromedriver,
* and the python depedencies:
```
pip install -r /path/to/requirements.txt
```
Once those are installed, install the GCP SDK:
https://cloud.google.com/sdk/docs/install
You will also have to log into GCP *twice*, once for the gcloud CLI tools,
and again for the GCP SDK libraries:
https://googleapis.dev/python/google-api-core/latest/auth.html
-42
View File
@@ -1,42 +0,0 @@
from abc import ABC, abstractmethod
class Simulation(ABC):
@abstractmethod
def start(self):
"Start the simulated object."
pass
@abstractmethod
def tick(self) -> bool:
"""
Simulate one clock tick.
Returns:
bool: True if the simulation could advance one clock tick, or False if the object is expired.
"""
pass
@abstractmethod
def get_uses(self):
"""
Return the uses of the simulated object. These are verb strings which are passed to the `use()` method.
Returns:
tuple: A dict of key-value pairs, where the keys are strings of
verbs to be passed to `use`, and the values are strings explaining
what the verb does in English.
"""
@abstractmethod
def use(self, *args, **kwargs):
"""
Use the simulated object.
Args:
args (tuple): A tuple of verbs to be used on the object.
kwargs (dict): A dictionary of key-value pairs of arguments to the verb(s).
Returns:
True if the action succeeded, False if the action did not succeed and the object needs to be expired.
"""
pass
@abstractmethod
def stop(self):
"Stop the simulated object."
pass
-42
View File
@@ -1,42 +0,0 @@
from importlib import import_module
import tkinter as tk
import os
import sys
from random import choice
from time import sleep
from tkinter import filedialog
from abc import ABC, abstractmethod
class SimulationRunner:
def __init__(self, simulation) -> None:
self.simulation = simulation()
self.simulation.start()
uses = list(self.simulation.get_uses().keys())
while self.simulation.tick() and self.simulation.use(choice(uses)):
pass
self.simulation.stop()
if __name__ == '__main__':
if len(sys.argv) == 1:
# Create a hidden root window
root = tk.Tk()
root.withdraw()
# Open the file chooser dialog
file_path = filedialog.askopenfilename()
print(f"Selected file: {file_path}")
else:
file_path = sys.argv[1]
# Get the python file's directory and filename
module_dir, module_file = os.path.split(file_path)
# Get the python file's name without any extension
module_name, _ = os.path.splitext(module_file)
# Add the module directory to sys.path so we can import it unqualified
sys.path.append(module_dir)
simulation = import_module(module_name)
# Load the simulation module's simulation class, which should be the same
# name as the module name.
simulationrunner = SimulationRunner(getattr(simulation, module_name))
-402
View File
@@ -1,402 +0,0 @@
{
"cells": [
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"## Guide\n",
"1. Please write your solutions at the appropriate location **# YOUR CODE HERE**\n",
"2. Do not modify other cells"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"import pandas as pd\n",
"\n",
"# Numerical or binary attributes\n",
"num_or_bin= ['Income', 'Kidhome','Recency', 'MntWines', 'MntFruits', 'MntMeatProducts', 'MntFishProducts', 'MntSweetProducts','MntGoldProds', 'NumDealsPurchases', 'NumWebPurchases',\n",
" 'NumCatalogPurchases', 'NumStorePurchases', 'NumWebVisitsMonth','AcceptedCmp3', 'AcceptedCmp4', 'AcceptedCmp5', 'AcceptedCmp1','AcceptedCmp2']\n",
"\n",
"# Categorical attributes\n",
"cat = ['Education','Marital_Status']\n",
"\n",
"# Load the data source\n",
"pd.set_option('display.max_columns', None)\n",
"df = pd.read_csv(\"https://raw.githubusercontent.com/rpi-techfundamentals/website_fall_2021/master/site/public/marketing_campaign.csv\", delimiter='\\t')\n",
"df.head()"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"## Q1. Load and preprocess the data.\n",
"Build and transform your dataframe `df` using the following steps before modeling. Create a new dataframe `X` that has the following properties:\n",
"\n",
"1. It should include only the columns from lists `num_or_bin` and `cat` (in the code above).\n",
"2. Variables in cat should be transformed to `n-1` dummy variables (where `n` is the number of categories for each variable.)\n",
"3. Missing variables in the `Income` table should be filled with the median of the `income` table.\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"raise NotImplementedError()\n",
"# YOUR CODE HERE"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Answer\n",
"'''\n",
"all_cols=num_or_bin+cat\n",
"X=pd.get_dummies(df[all_cols],columns=cat,drop_first=True)\n",
"X.loc[X['Income'].isnull(),'Income']=X['Income'].median()\n",
"'''"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"#Please do not modify/delete this cell\n",
"# 1 Test Cell \n",
"#this tests the columns. \n",
"ecol = set(['Income', 'Kidhome', 'Recency', 'MntWines', 'MntFruits',\n",
" 'MntMeatProducts', 'MntFishProducts', 'MntSweetProducts',\n",
" 'MntGoldProds', 'NumDealsPurchases', 'NumWebPurchases',\n",
" 'NumCatalogPurchases', 'NumStorePurchases', 'NumWebVisitsMonth',\n",
" 'AcceptedCmp3', 'AcceptedCmp4', 'AcceptedCmp5', 'AcceptedCmp1',\n",
" 'AcceptedCmp2', 'Education_Basic', 'Education_Graduation',\n",
" 'Education_Master', 'Education_PhD', 'Marital_Status_Alone',\n",
" 'Marital_Status_Divorced', 'Marital_Status_Married',\n",
" 'Marital_Status_Single', 'Marital_Status_Together',\n",
" 'Marital_Status_Widow', 'Marital_Status_YOLO'])\n",
"assert len(ecol-set(X.columns))==0\n",
"assert len(set(X.columns))-len(ecol)==0"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"#Please do not modify/delete this cell\n",
"# 2 Test Cell \n",
"#this tests the columns. \n",
"assert X.shape==(2240, 30)\n",
"q2=X.copy()\n",
"q2['Response']=df['Response']\n",
"q2['Year_Birth']=df['Year_Birth']\n",
"q2['Dt_Customer']=df['Dt_Customer']\n",
"q2.to_csv('q2.csv',index=False)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"#Please do not modify/delete this cell\n",
"# 3 Test Cell \n",
"assert len(X.loc[X['Income']==51381.5,'Income'])==24"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"#Please do not modify/delete this cell\n",
"# 4 Test Cell\n",
"assert X.isnull().sum().sum()==0"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Reload data\n",
"import pandas as pd\n",
"df_cl=pd.read_csv('q2.csv')"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"## Q2. Feature Creation.\n",
"1. In the `df_cl` dataframe, create an integer feature `age` which is `2021` minus `Year_Birth`.\n",
"2. In the `df_cl` dataframe, create an integer feature `holiday` which is 1 if the month of `Dt_Customer` is equal to 12 and 0 otherwise. You can get the month by accessing `mm` from the date format `dd-mm-yyyy`."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"raise NotImplementedError()\n",
"# YOUR CODE HERE"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Answer\n",
"'''\n",
"df_cl['age']=(2021-df_cl['Year_Birth']).astype(int)\n",
"df_cl['Dt_Customer_tmp']=df_cl['Dt_Customer'].apply(lambda x:x[3:5])\n",
"df_cl.loc[df_cl['Dt_Customer_tmp']=='12','holiday']=1\n",
"df_cl.loc[df_cl['Dt_Customer_tmp']!='12','holiday']=0\n",
"df_cl['holiday']=df_cl['holiday'].astype(int)\n",
"'''"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"#Please do not modify/delete this cell\n",
"# 5 Test Cell\n",
"assert df_cl['age'].sum()==116915"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"#Please do not modify/delete this cell\n",
"# 6 Test Cell\n",
"import numpy as np\n",
"assert isinstance(df_cl['age'][0], (np.int64,np.int32))"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"#Please do not modify/delete this cell\n",
"# 7 Test Cell\n",
"assert df_cl['holiday'].sum()==175"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"#Please do not modify/delete this cell\n",
"# 8 Test Cell\n",
"import numpy as np\n",
"assert isinstance(df_cl['holiday'][0], (np.int64,np.int32))"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"import pandas as pd\n",
"df_cl2=pd.read_csv('dfcl2.csv')\n",
"y_reload=df_cl2['Response']\n",
"X_reload=df_cl2.drop(['Response'], axis=1)"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"## Q3. Train Test Split\n",
"Now split the data `X_reload`, `y_reload` intro training (50%) and testing data (50%) by using these variable names while making sure there are Make sure that there are equal number of positive cases (1's) in the train and the test set.\n",
"\n",
"`X_train`: Training feature columns\n",
"`X_test`: Testing feature columns\n",
"`y_train`: Training DV\n",
"`y_test`: Testing DV\n",
"\n",
"When doing so set the `random_state` parameter to 100 so you get the same split that is in the solution."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"raise NotImplementedError()\n",
"# YOUR CODE HERE"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Answer\n",
"'''\n",
"y_reload=df_cl2['Response']\n",
"X_reload=df_cl2.drop(['Response'], axis=1)\n",
"from sklearn.model_selection import train_test_split\n",
"X_train,X_test,y_train,y_test=train_test_split(X_reload,y_reload,\n",
" train_size=0.5,\n",
" test_size=0.5,\n",
" random_state=100,\n",
" stratify=y_reload)\n",
"'''"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"#Please do not modify/delete this cell\n",
"# 9 Test Cell\n",
"assert X_train.shape==(1120, 31)\n",
"assert X_test.shape==(1120, 31)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"#Please do not modify/delete this cell\n",
"# 10 Test Cell\n",
"assert y_train.sum()==y_test.sum()"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"import pandas as pd\n",
"df_cl2=pd.read_csv('dfcl2.csv')\n",
"y_reload=df_cl2['Response']\n",
"X_reload=df_cl2.drop(['Response'], axis=1)\n",
"\n",
"X_train2, X_test2, y_train2, y_test2 = train_test_split(X_reload, y_reload, test_size=0.3, random_state = 100)"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"## Q4. Classification\n",
"Use `X_train2`, `X_test2`, `y_train2`, `y_test2` from above. Use a Gradient Boosting classifier (as instantiated below).\n",
"\n",
"Please calculate:\n",
"1. `test_acc` The accuracy of the classifier on the test set.\n",
"2. `test_roc` The area under the ROC curve on on the test set."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from sklearn import metrics\n",
"from sklearn.ensemble import GradientBoostingClassifier\n",
"#use this classifier\n",
"classifier = GradientBoostingClassifier(n_estimators=100, learning_rate=1.0, max_depth=1, random_state=0)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"raise NotImplementedError()\n",
"# YOUR CODE HERE"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Answer\n",
"'''\n",
"classifier.fit(X_train2,y_train2)\n",
"test_y_pred=classifier.predict(X_test2)\n",
"test_y_pred_prob=classifier.predict_proba(X_test2)[:,1]\n",
"test_acc=metrics.accuracy_score(y_test2,test_y_pred)\n",
"test_roc=metrics.roc_auc_score(y_test2,test_y_pred_prob)\n",
"'''"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"#Please do not modify/delete this cell\n",
"# 11 Test Cell\n",
"assert round(test_acc,3)==0.860"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"#Please do not modify/delete this cell\n",
"# 12 Test Cell\n",
"assert round(test_roc,3)==0.846"
]
}
],
"metadata": {
"language_info": {
"name": "python"
},
"orig_nbformat": 4
},
"nbformat": 4,
"nbformat_minor": 2
}
-74
View File
@@ -1,74 +0,0 @@
import h5py
import pandas as pd
import numpy as np
import plotly.graph_objects as go
f = h5py.File('product1.h5','r')
list(f.keys())
for key in f.keys():
print(f[key])
df = pd.read_hdf('product1.h5', key='df')
df = df._append(pd.read_hdf('product1.h5', key))
# print(f[key].shape)
# print(f[key].value)
# df = pd.read_hdf('product1.h5', key='df7')
print(df)
'''
# Open the HDF5 file
with h5py.File('product1.h5', 'r') as file:
dataframes=pd.DataFrame(columns=['Product ID', 'Zip Code', 'Shown Price', 'Minimum Quantity', 'Total Price', 'Unit Price','Sales Count','Sold Status'])
for key in file.keys():
# Read the dataframe from the HDF5 file
dataframe = pd.read_hdf('product1.h5', key)
# Append the dataframe to the list
dataframes._append(dataframe)
# Print the number of dataframes read from the HDF5 file
print(f"Read {len(dataframes)} dataframes from the HDF5 file")
item_std_devs = {}
for dataframe in dataframes:
item_name = dataframe['Product ID'].iloc[0]
std_dev = np.std(dataframe['Unit Price'])
item_std_devs[item_name] = std_dev
# Print the item names and their corresponding standard deviations
for item_name, std_dev in item_std_devs.items():
print(f"Item: {item_name}, Std Dev: {std_dev}")
# Prepare the data for the graph
x = list(item_std_devs.keys())
y = list(item_std_devs.values())
# Create a bar graph
fig = go.Figure(data=[go.Bar(x=x, y=y)])
# Customize the graph
fig.update_layout(
title="Standard Deviation of Pricing for Each Item",
xaxis_title="Item",
yaxis_title="Standard Deviation",
)
# Save the graph as an HTML file
report_file = 'test2.html'
fig.write_html(report_file)
'''
'''
# Calculate the standard deviation using pandas' function
std_dev_unit = df.groupby('Product ID')['Unit Price'].std()
std_dev_displayed = df.groupby('Zip Code')['Shown Price'].std()
# Generate the graph using Plotly
fig = go.Figure()
fig.add_trace(go.Bar(x=std_dev_unit.index, y=std_dev_unit.values))
fig.update_layout(title='Standard Deviation of the Unit Pricing by Item', xaxis_title='Product ID', yaxis_title='Standard Deviation')
report_file = 'test2.html'
fig.write_html(report_file)
'''
Binary file not shown.
Binary file not shown.

Before

Width:  |  Height:  |  Size: 16 KiB

BIN
View File
Binary file not shown.

Before

Width:  |  Height:  |  Size: 26 KiB

-81
View File
@@ -1,81 +0,0 @@
'''
class Vault:
def __init__(self):
self._treasure = "gold"
self._attributes = dict(self.__dict__)
def __setattr__(self, name, value):
if name != "_attributes" and name not in self._attributes:
print("attribute deleted!")
self._attributes[name] = value
super().__setattr__("_attributes", dict(self.__dict__))
@property
def treasure(self):
if self.__dict__ != self._attributes:
print("attribute deleted!")
return self._treasure
v=Vault()
print(v.treasure)
'''
'''
class Vault:
def __init__(self):
self._treasure = "gold"
self._original_dict = dict(self.__dict__)
def __setattr__(self, name, value):
if name != "_treasure" and name in self.__dict__ and self._original_dict.get(name) != value:
print("attribute deleted!")
self.__dict__[name] = value
@property
def treasure(self):
if self.__dict__ != self._original_dict:
print("attribute deleted!")
return self._treasure
v = Vault()
print(v.treasure)
'''
'''
class Vault:
def __init__(self):
self._treasure = "gold"
self._attributes = dict(self.__dict__)
def __setattr__(self, name, value):
if name != "_attributes" and name not in self._attributes:
print("attribute deleted!")
self._attributes[name] = value
super().__setattr__("_attributes", dict(self.__dict__))
@property
def treasure(self):
if self.__dict__ != self._attributes:
print("attribute deleted!")
return self._treasure
v = Vault()
print(v.treasure)
'''
class Vault:
def __init__(self):
self._treasure = "gold"
def __setattr__(self, name, value):
if name != "_attributes" and name not in self.__dict__:
print("attribute deleted!")
self.__dict__[name] = value
self._attributes = dict(self.__dict__)
@property
def treasure(self):
if self.__dict__ != self._attributes:
print("attribute deleted!")
return self._treasure
v = Vault()
print(v.treasure)
-47
View File
@@ -1,47 +0,0 @@
import importlib.util
import os
def get_module_size(module_name):
"""Get the total number of bytes in the cached .pyc file or .py file of a module or package.
Args:
module_name (str): The name of the module or package.
Returns:
int: The total number of bytes in the cached .pyc file or .py file.
"""
spec = importlib.util.find_spec(module_name)
if spec is not None and spec.has_location:
# If the module is found and has a location, check for .pyc file first
pyc_file_path = spec.origin + 'c'
if os.path.exists(pyc_file_path):
return os.stat(pyc_file_path).st_size
else:
# If .pyc file doesn't exist, check for .py file
py_file_path = spec.origin
if os.path.exists(py_file_path):
return os.stat(py_file_path).st_size
return 0 # Return 0 if no .pyc or .py file found
import sys
def get_total_module_size():
"""Get the total number of bytes for all modules listed in sys.modules.
Returns:
int: The total number of bytes for all modules.
"""
total_size = 0
for module_name in sys.modules:
module_size = get_module_size(module_name)
total_size += module_size
return total_size
if __name__ == '__main__' and __package__ is None:
__spec__ = importlib.util.spec_from_file_location('__main__', __file__)
total_size = get_total_module_size()
print("Total size of all modules in sys.modules: {} bytes".format(total_size))
BIN
View File
Binary file not shown.

Before

Width:  |  Height:  |  Size: 19 KiB

BIN
View File
Binary file not shown.

Before

Width:  |  Height:  |  Size: 37 KiB

BIN
View File
Binary file not shown.

Before

Width:  |  Height:  |  Size: 37 KiB

-65
View File
@@ -1,65 +0,0 @@
from google.cloud import pubsub_v1
from google.oauth2 import service_account
import os
import asyncio
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "C:/Users/Karen/AppData/Roaming/gcloud/application_default_credentials.json"
project_id = "fp-20230425"
'''
topic_name = "fp-topic-Karen"
subscription_name = "fp-subscription-Karen"
subscriber = pubsub_v1.SubscriberClient()
subscription_path = f"projects/{project_id}/subscriptions/{subscription_name}"
subscriber.delete_subscription(request={"subscription": subscription_path})
print("Subscription deleted: {}".format(subscription_path))
# Delete the topic
publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(project_id, topic_name)
publisher.delete_topic(request={"topic": topic_path})
print(f"Topic '{topic_name}' deleted.")
subscription_name = "fp-subscription-Karen"
subscription_path = f"projects/{project_id}/subscriptions/{subscription_name}"
'''
# Create a pubsub topic
publisher = pubsub_v1.PublisherClient()
topic_name = "fp-topic-Karen"
topic_path = publisher.topic_path(project_id, topic_name)
topic = publisher.create_topic(request={"name": topic_path})
# Create a Pub/Sub subscription
subscriber = pubsub_v1.SubscriberClient()
subscription_name = "fp-subscription-Karen"
subscription_path = subscriber.subscription_path(project_id, subscription_name)
subscription = subscriber.create_subscription(
name=subscription_path, topic=topic_path
)
# Publish a message to the topic
message = b"Hello!"
publisher.publish(topic_path, data=message)
print("Message published.")
async def receive_messages():
response = subscriber.pull(subscription=subscription_path, max_messages=1)
for received_message in response.received_messages:
print("Received message: {}".format(received_message.message.data))
asyncio.run(receive_messages())
'''
# Pull the message from the subscription
response = subscriber.pull(subscription=subscription_path, max_messages=1)
for received_message in response.received_messages:
print("Received message: {}".format(received_message.message.data))
# Acknowledge the message to mark it as processed
subscriber.acknowledge(subscription_path, [received_message.ack_id])
'''
-10
View File
@@ -1,10 +0,0 @@
import http.client
# The product ID for the Pete-and-Gerry's organic large brown egges is 65902
conn = http.client.HTTPSConnection("www.sayweee.com")
payload = ''
headers = {}
conn.request("GET", "/en/product/Pete-and-Gerry-s-Organic-Large-Brown-Eggs/65902?category=dairy02", payload, headers)
res = conn.getresponse()
data = res.read()
print(data.decode("utf-8"))
-67
View File
@@ -1,67 +0,0 @@
from google.cloud import pubsub_v1
from google.api_core.exceptions import DeadlineExceeded
import csv
import os
import asyncio
# os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "C:/Users/Karen/AppData/Roaming/gcloud/application_default_credentials.json"
project_id = "fp-20230425"
# project_id = "svig-384801"
'''
topic_name = "fp-topic-NFL-zipcodes"
subscription_name = "fp-subscription-NFL-zipcodes"
subscriber = pubsub_v1.SubscriberClient()
subscription_path = f"projects/{project_id}/subscriptions/{subscription_name}"
subscriber.delete_subscription(request={"subscription": subscription_path})
print("Subscription deleted: {}".format(subscription_path))
# Delete the topic
publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(project_id, topic_name)
publisher.delete_topic(request={"topic": topic_path})
print(f"Topic '{topic_name}' deleted.")
'''
# pubsub topic
publisher = pubsub_v1.PublisherClient()
topic_name = "fp-topic-NFL-zipcodes"
topic_path = publisher.topic_path(project_id, topic_name)
# Create a Pub/Sub subscription
subscriber = pubsub_v1.SubscriberClient()
'''
subscription_name = "fp-subscription-zipcodes"
subscription_path = subscriber.subscription_path(project_id, subscription_name)
subscription = subscriber.create_subscription(
name=subscription_path, topic=topic_path
)
'''
# Publish a message to the topic
# Read the data from the csv file
# file_path = "C:/Users/Karen/Desktop/zipcodes.csv"
file_path = "zipcodes.csv"
with open(file_path, "r") as csv_file:
reader = csv.reader(csv_file)
next(reader) # skip header row
for row in reader:
# Format the data as a string and publish it to the topic
data = ",".join(row).encode("utf-8")
try:
future = publisher.publish(topic_path, data)
print(f"Published message ID {future.result()}")
except DeadlineExceeded as de:
print(f"Caught DeadlineExceeded exception: {de}")
'''
async def receive_messages():
response = subscriber.pull(subscription=subscription_path, max_messages=1)
for received_message in response.received_messages:
print("Received message: {}".format(received_message.message.data))
asyncio.run(receive_messages())
'''
-372
View File
@@ -1,372 +0,0 @@
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)
-199
View File
@@ -1,199 +0,0 @@
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
# Create an empty dataframe to store the prices
df = pd.DataFrame(columns=['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'
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
# Extract the price of rice crackers for this zip code
# next_data_element = driver.find_element(By.XPATH, "//script[@type='application/ld+json']")
next_data_element = WebDriverWait(driver, 2).until(EC.presence_of_element_located((By.XPATH, '//*[@id="__next"]/div/div[1]/div[1]/div/div/div[2]/div[2]/div[1]/div[4]/div[1]/div')))
next_data_text = next_data_element.get_attribute('nodeValue')
price = next_data_text.text
'''
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['price']
'''
# Print the price
print(price)
'''
try:
min_quantity_element = WebDriverWait(driver, 2).until(EC.presence_of_element_located((By.XPATH, '//script[@id="__NEXT_DATA__"]')))
min_quantity_text = min_quantity_element.get_attribute('innerHTML')
min_quantity_start_index = min_quantity_text.index('{')
min_quantity_end_index = min_quantity_text.rindex('}') + 1
min_quantity_json = min_quantity_text[min_quantity_start_index:min_quantity_end_index]
min_quantity = json.loads(min_quantity_json)
price = min_quantity['product']['price']
except:
next_data_element = WebDriverWait(driver, 2).until(EC.presence_of_element_located((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']
finally:
# 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({'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()}")
-271
View File
@@ -1,271 +0,0 @@
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
import selenium.common.exceptions
import re
# 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', 'Shown Price', 'Minimum Quantity', 'Total Price', 'Unit 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/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):
'''
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 ''
# 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
# 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 = driver.find_element(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 = 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(ld_json_price)
unit_price_element = driver.find_element(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]})
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')
# df.to_hdf('product1.h5', key='df',mode='a')
# Open the HDF5 file in a separate window
# os.startfile('product1.h5')
-280
View File
@@ -1,280 +0,0 @@
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.")
-32
View File
@@ -1,32 +0,0 @@
import http.client
import xml.etree.ElementTree as ET
conn = http.client.HTTPSConnection("www.sayweee.com")
payload = ''
headers = {}
conn.request("GET", "/en/product/Pete-and-Gerry-s-Organic-Large-Brown-Eggs/65902?category=dairy02", payload, headers)
res = conn.getresponse()
data = res.read()
# Parse the HTML page content using ElementTree
root = ET.fromstring(data.decode('utf-8'))
if root is not None:
# Find the script element with __NEXT_DATA__ data
next_data_element = None
for script_element in root.iter('script'):
if script_element.text is not None and '__NEXT_DATA__' in script_element.text:
next_data_element = script_element
break
# Extract the __NEXT_DATA__ data
if next_data_element is not None:
next_data_text = next_data_element.txt
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]
print(next_data_json)
else:
print('__NEXT_DATA__ element not found')
else:
print('Failed to parse HTML content')
-32
View File
@@ -1,32 +0,0 @@
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import json
# Configure the Selenium driver options
options = Options()
options.add_argument('--headless')
options.add_argument('--disable-gpu')
driver = webdriver.Chrome(options=options)
# Load the page and extract the __NEXT_DATA__ data
url = 'https://www.sayweee.com/en/product/Pete-and-Gerry-s-Organic-Large-Brown-Eggs/65902?category=dairy02'
driver.get(url)
wait = WebDriverWait(driver, 10)
next_data_element = wait.until(EC.presence_of_element_located((By.XPATH, '//script[contains(text(), "__NEXT_DATA__")]')))
next_data_text = next_data_element.text
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)
# Extract the price from the __NEXT_DATA__ data
price = next_data['props']['pageProps']['product']['price']
# Close the Selenium driver
driver.quit()
# Print the price
print(price)
-43
View File
@@ -1,43 +0,0 @@
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
import json
import time
# Configure the Selenium driver options
options = Options()
#options.add_argument('--headless')
options.add_argument('--disable-gpu')
driver = webdriver.Chrome(options=options)
# Load the page and extract the __NEXT_DATA__ data
url = 'https://www.sayweee.com/en/product/Want-Want-Senbei-Rice-Crackers/17119'
driver.get(url)
time.sleep(5)
next_data_element = None
#try:
next_data_element = driver.find_element(By.XPATH, "//script[@type='application/ld+json']")
#except:
# print('__NEXT_DATA__ element not found')
#time.sleep(1000)
if next_data_element is not None:
next_data_text = next_data_element.get_attribute('innerHTML')
print(next_data_text)
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)
# Extract the price from the __NEXT_DATA__ data
# price = next_data['props']['pageProps']['product']['price']
print(next_data)
price = next_data['offers']['price']
# Print the price
print(price)
else:
print('__NEXT_DATA__ element not found')
time.sleep(1000)
# Close the Selenium driver
#driver.quit()
-33
View File
@@ -1,33 +0,0 @@
import re
import json
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
# Configure the Selenium driver options
options = Options()
options.add_argument('--headless')
options.add_argument('--disable-gpu')
driver = webdriver.Chrome(options=options)
# Load the page and extract the __NEXT_DATA__ data
url = 'https://www.sayweee.com/en/product/Want-Want-Senbei-Rice-Crackers/17119'
driver.get(url)
next_data_element = None
for script_element in driver.find_elements(By.XPATH, '//script'):
script_text = script_element.get_attribute('innerHTML')
match = re.search(r'__NEXT_DATA__ = (\{.*\});', script_text)
if match:
next_data_json = match.group(1)
next_data = json.loads(next_data_json)
# Extract the price from the __NEXT_DATA__ data
price = next_data['props']['pageProps']['product']['price']
# Print the price
print(price)
break
if next_data_element is None:
print('__NEXT_DATA__ element not found')
# Close the Selenium driver
driver.quit()
-126
View File
@@ -1,126 +0,0 @@
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
import json
import time
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
# Configure the Selenium driver options
options = Options()
options.add_argument('--disable-gpu')
driver = webdriver.Chrome(options=options)
# Define the zip codes for which to extract the price
zip_codes = ['85305', '30313', '21230', '14127','28202','60605','45202','44114','76011','80204','48226','54304','77054','46225','32202','64129','89118','90301','33056','55415','02035','70112','07073','19148','15212','95054','98134','33607','37213','20785']
# Load the page and extract the __NEXT_DATA__ data
#url = 'https://www.sayweee.com/en/product/Want-Want-Senbei-Rice-Crackers/17119'
url = 'https://www.sayweee.com/en/product/Lays-Potato-Chips-Cucumber-Flavor/51261'
#url = 'https://www.sayweee.com/en/product/Glico-Pocky-Cookies-and-Cream-Family-Pack/79467'
#url = 'https://www.sayweee.com/en/product/Xiang-Na-Lan-Thai-Jasmine-Rice/19566?category=dried01'
#url = 'https://www.sayweee.com/en/product/Sekka-Premium-Japanese-Medium-Grain-Rice/81224'
#url = 'https://www.sayweee.com/en/product/Big-Green-Spring-Farm-Organic-Millet/90991?category=dried02'
#url = 'https://www.sayweee.com/en/product/Liuzhou-Guangxi-Specialty-LuoSiFen-Hot-Spicy-Noodles/95442?category=instant01'
#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/Lactaid-Lactose-Free-2--Reduced-Fat-Milk/95353?trace_id=7847dd6e-c511-48e2-a76e-a283399967b1'
#url = 'https://www.sayweee.com/en/product/Washington-Fuji-Apples/94817?category=fruits03'
driver.delete_all_cookies
driver.get(url)
# Close the cookies setting window
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
# Clear any modal dialog
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 callback(zip_code):
try:
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
try:
# Clear any notice
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()
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
# 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 = WebDriverWait(driver, 5).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)
try:
check_button = WebDriverWait(driver, 5).until(EC.presence_of_element_located((By.XPATH, "//button[text() = 'Confirm']")))
# print(f"{check_button:}")
except:
check_button = WebDriverWait(driver, 5).until(EC.presence_of_element_located((By.XPATH, "/html/body/div[5]/div/div[2]/div/div[2]/div[2]/div/div[1]/form/div/div/div[1]/div/div/div[3]/div/button")))
finally:
check_button.click()
try:
# Clear any notice
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()
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
try:
sold_out_notice = driver.find_element(By.XPATH, '//*[@id="soldStatusNotice"]')
print(f"No such product in this area = {zip_code, sold_out_notice}")
except:
print(f"The product is available in this area {zip_code}")
except Exception as e:
print(f"Caught exception: {e}")
print(f"Price not found for zip code {zip_code}")
for zip_code in zip_codes:
callback(zip_code)
driver.quit()
-64
View File
@@ -1,64 +0,0 @@
from google.cloud import pubsub_v1
import csv
import os
import asyncio
# os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "C:/Users/Karen/AppData/Roaming/gcloud/application_default_credentials.json"
# project_id = "fp-20230425"
project_id = "svig-384801"
'''
topic_name = "fp-topic-NFL-zipcodes"
subscription_name = "fp-subscription-NFL-zipcodes"
subscriber = pubsub_v1.SubscriberClient()
subscription_path = f"projects/{project_id}/subscriptions/{subscription_name}"
subscriber.delete_subscription(request={"subscription": subscription_path})
print("Subscription deleted: {}".format(subscription_path))
# Delete the topic
publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(project_id, topic_name)
publisher.delete_topic(request={"topic": topic_path})
print(f"Topic '{topic_name}' deleted.")
'''
# Create a pubsub topic
publisher = pubsub_v1.PublisherClient()
topic_name = "fp-topic-NFL-zipcodes"
topic_path = publisher.topic_path(project_id, topic_name)
topic = publisher.create_topic(request={"name": topic_path}, timeout=30)
# Create a Pub/Sub subscription
subscriber = pubsub_v1.SubscriberClient()
subscription_name = "fp-subscription-NFL-zipcodes"
subscription_path = subscriber.subscription_path(project_id, subscription_name)
subscription = subscriber.create_subscription(
name=subscription_path, topic=topic_path
)
'''
# Publish a message to the topic
# Read the data from the csv file
# file_path = "C:/Users/Karen/Desktop/NFL_Cities.csv"
file_path = zipcodes.csv
with open(file_path, "r") as csv_file:
reader = csv.reader(csv_file)
next(reader) # skip header row
for row in reader:
# Format the data as a string and publish it to the topic
data = ",".join(row).encode("utf-8")
future = publisher.publish(topic_path, data)
print(f"Published message ID {future.result()}")
async def receive_messages():
response = subscriber.pull(subscription=subscription_path, max_messages=1)
for received_message in response.received_messages:
print("Received message: {}".format(received_message.message.data))
asyncio.run(receive_messages())
'''
-55
View File
@@ -1,55 +0,0 @@
import json
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
import time
from google.cloud import pubsub_v1
import os
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "C:/Users/Karen/AppData/Roaming/gcloud/application_default_credentials.json"
options = Options()
options.add_argument('--headless')
options.add_argument('--disable-gpu')
driver = webdriver.Chrome(options=options)
# Set up Pub/Sub subscription
project_id = "fp-20230425"
subscription_name = "fp-subscription-zipcodes"
subscriber = pubsub_v1.SubscriberClient()
subscription_path = subscriber.subscription_path(project_id, subscription_name)
def callback(message):
try:
# Get the zip code from the message
zip_code = message.data.decode('utf-8')
print(f"Received zip code: {zip_code}")
# Load the webpage with the zip code
url = 'https://www.sayweee.com/en/product/Want-Want-Senbei-Rice-Crackers/17119'
driver.get(url)
driver.find_element(By.XPATH, '/html/body/div[4]/div/div[2]/div/div[2]/div[2]/div/div[1]/form/div/div/div[1]/div/div/div[2]/input').send_keys(zip_code)
driver.find_element(By.XPATH, '/html/body/div[4]/div/div[2]/div/div[2]/div[2]/div/div[1]/form/div/div/div[1]/div/div/div[3]/div').click()
# Extract the price from the webpage
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 result
print(f"Price of rice crackers in zip code {zip_code}: {price}")
except Exception as e:
print(f"Error processing message: {str(e)}")
finally:
message.ack()
subscriber.subscribe(subscription_path, callback=callback)
while True:
time.sleep(1)
-106
View File
@@ -1,106 +0,0 @@
from google.cloud import pubsub_v1
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
# os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "C:/Users/Karen/AppData/Roaming/gcloud/application_default_credentials.json"
# 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"
project_id = "svig-384801"
subscription_name = "fp-subscription-zipcodes"
subscriber = pubsub_v1.SubscriberClient()
subscription_path = subscriber.subscription_path(project_id, subscription_name)
timeout = 5
# Load the product page and extract the __NEXT_DATA__ data
url = 'https://www.sayweee.com/en/product/Want-Want-Senbei-Rice-Crackers/17119'
driver.get(url)
# time.sleep(5)
'''
next_data_element = driver.find_element(By.XPATH, "//script[@type='application/ld+json']")
if next_data_element is not None:
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)
else:
print('__NEXT_DATA__ element not found')
'''
# Define callback function for Pub/Sub messages
def callback(message):
# Extract zip code from message data
zip_code = message.data.decode('utf-8')
print(f"Received zip code: {zip_code}")
try:
time.sleep(5)
# Clear any modal dialog
modal_close = driver.find_element(By.XPATH, "//*[contains(concat(' ', normalize-space(@class), ' '), ' icon-icon-close-s ')]")
print(f"modal_close = {modal_close}")
modal_close.click()
time.sleep(2)
# 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]/div/div/div[1]')
print(zip_button)
zip_button.click()
time.sleep(1)
zip_input = driver.find_element(By.XPATH, '//div/label/following-sibling::input')
zip_input.clear()
zip_input.send_keys(zip_code)
check_button = driver.find_element(By.XPATH, "//button[text() = 'Confirm']")
check_button.click()
time.sleep(2)
# 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)
message.ack() # acknowledge receipt of message
except:
print(f"Price not found for zip code {zip_code}")
price = 'Price not found'
# Start receiving messages
streaming_pull_future = subscriber.subscribe(subscription_path, callback=callback)
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)
-166
View File
@@ -1,166 +0,0 @@
import sqlalchemy
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, Float, DateTime
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import relationship, sessionmaker
import sys
from flask import Flask, render_template
# Create the Flask application
app = Flask(__name__)
# Create the database engine
engine = create_engine('sqlite:///scraper_design.db')
# Create a session factory
Session = sessionmaker(bind=engine)
# Create a base class
Base = declarative_base()
# Define the product class
class Product(Base):
__tablename__ = 'product'
id = Column(Integer, primary_key=True)
name = Column(String)
SKU_PLU_UPC_number = Column(String)
price_histories = relationship("PriceHistory", back_populates="product")
# Define the retailer class
class Retailer(Base):
__tablename__ = 'retailer'
id = Column(Integer, primary_key=True)
name = Column(String)
products = relationship("Product", secondary="retailer_product")
# Define the price class
class Price(Base):
__tablename__ = 'price'
id = Column(Integer, primary_key=True)
MOQ = Column(Integer)
price = Column(Float)
# Define the scraperjob class
class ScraperJob(Base):
__tablename__ = 'scraper_job'
id = Column(Integer, primary_key=True)
start_time = Column(DateTime)
end_time = Column(DateTime)
parameters = Column(String)
statistics = Column(String)
# Define the price history class
class PriceHistory(Base):
__tablename__ = 'price_history'
id = Column(Integer, primary_key=True)
location = Column(String)
zip_code = Column(Integer)
city_name = Column(String)
price_at_time = Column(DateTime)
price = Column(Float)
scraper_job_id = Column(Integer, ForeignKey('scraper_job.id'))
product_id = Column(Integer, ForeignKey('product.id'))
product = relationship("Product", back_populates="price_histories")
scraper_job = relationship("ScraperJob")
# Define the user class
class User(Base):
__tablename__ = 'user'
id = Column(Integer, primary_key=True)
username = Column(String)
credentials = Column(String)
creation_date = Column(DateTime)
modification_date = Column(DateTime)
def __str__(self):
return f"{self.id=} {self.username=} {self.creation_date=} {self.modification_date=}"
class RetailerProduct(Base):
__tablename__ = 'retailer_product'
retailer_id = Column(Integer, ForeignKey('retailer.id'), primary_key=True)
product_id = Column(Integer, ForeignKey('product.id'), primary_key=True)
# Create the database tables
Base.metadata.create_all(engine)
@app.route('/')
def homepage():
# Create a session
session = Session()
# Get all the classes from SQLAlchemy
classes = Base.__subclasses__()
statement = sqlalchemy.select(User)
users = session.scalars(statement).all()
# Render the template with the classes
return render_template('test3.html', classes=classes, users=users)
# Create command-line arguments for the user class
def main():
if len(sys.argv) < 2:
app.run()
command = sys.argv[1]
if command == "create":
if len(sys.argv) != 3:
print("Usage: python user_tool.py create [username]")
return
username = sys.argv[2]
# Create a new user object
user = User(username=username)
# Add the user to the database
session = Session()
session.add(user)
session.commit()
print("User created successfully.")
elif command == "edit":
if len(sys.argv) != 4:
print("Usage: python user_tool.py edit [user_id] [new_username]")
return
user_id = int(sys.argv[2])
new_username = sys.argv[3]
# Retrieve the user from the database
session = Session()
user = session.query(User).get(user_id)
if user:
# Update the user object
user.username = new_username
session.commit()
print("User updated successfully.")
else:
print("User not found.")
elif command == "delete":
if len(sys.argv) != 3:
print("Usage: python user_tool.py delete [user_id]")
return
user_id = int(sys.argv[2])
# Retrieve the user from the database
session = Session()
user = session.query(User).get(user_id)
if user:
# Delete the user from the database
session.delete(user)
session.commit()
print("User deleted successfully.")
else:
print("User not found.")
else:
print("Usage: python user_tool.py [create|edit|delete]")
return
if __name__ == "__main__":
main()
-100
View File
@@ -1,100 +0,0 @@
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
import concurrent.futures
import pandas as pd
# Define the subset of urls
url_list = [
'https://www.sayweee.com/en/product/Want-Want-Senbei-Rice-Crackers/17119',
'https://www.sayweee.com/en/product/Lays-Potato-Chips-Cucumber-Flavor/51261',
'https://www.sayweee.com/en/product/Glico-Pocky-Cookies-and-Cream-Family-Pack/79467',
'https://www.sayweee.com/en/product/Xiang-Na-Lan-Thai-Jasmine-Rice/19566?category=dried01',
'https://www.sayweee.com/en/product/Sekka-Premium-Japanese-Medium-Grain-Rice/81224',
'https://www.sayweee.com/en/product/Big-Green-Spring-Farm-Organic-Millet/90991?category=dried02',
'https://www.sayweee.com/en/product/Liuzhou-Guangxi-Specialty-LuoSiFen-Hot-Spicy-Noodles/95442?category=instant01',
'https://www.sayweee.com/en/product/House-Foods-Firm-Tofu-/85175?trace_id=c93f3d1a-80a4-4e1a-8d41-a402ae43186f',
'https://www.sayweee.com/en/product/Lactaid-Lactose-Free-2--Reduced-Fat-Milk/95353',
'https://www.sayweee.com/en/product/Washington-Fuji-Apples/94817?category=fruits03',
'https://www.sayweee.com/en/product/Formosa-Pork-Sung-/23528?trace_id=3c143dd4-4a35-4e8d-93ee-d91f209717e7'
]
# Define the function to run the scraper
def run_scraper(url):
project_id = "fp-20230425"
subscription_name = "fp-subscription-NFL-zipcodes"
df_new = get_product_prices(project_id, subscription_name, url)
# Add timestamp to the new dataframe "df_new"
'''
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_new.append(df_new, ignore_index=True)
return df_update
'''
# 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)
'''
if __name__ == '__main__':
# Determine the number of available CPUs
# cpu_count = multiprocessing.cpu_count()
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
future_to_url = {executor.submit(run_scraper, url): url for url in url_list}
results = []
for future in concurrent.futures.as_completed(future_to_url):
url = future_to_url[future]
try:
result = future.result()
print(f"Scraped data for URL: {url}")
results.append(result)
except Exception as e:
print(f"Failed to scrape data for URL: {url}\nError: {e}")
# Now we have the new dataframe "df_update" recording the new price listed in weee's page
# Load such data in the existing HDF5 file
file_path = 'product1.h5'
with pd.HDFStore(file_path) as store:
for url, result in zip(url_list, results):
# Extract the product ID from the URL
product_id = url.split("/")[-1][:5]
# Save the dataframe to the HDF5 file with a unique key
key = f'df_{product_id}'
store[key] = result
result.to_hdf('product_data.h5', key=key, mode='a')
print("Data appended and saved successfully.")
-1
View File
@@ -1 +0,0 @@
print("hello")
+78
View File
@@ -0,0 +1,78 @@
{
"cells": [
{
"cell_type": "code",
"execution_count": null,
"id": "0593e529",
"metadata": {},
"outputs": [],
"source": [
"import os\n",
"from urllib.parse import urlparse\n",
"\n",
"class JPEGRequestHandler:\n",
" def __init__(self, app):\n",
" self.app = app\n",
"\n",
" def __call__(self, environ, start_response):\n",
" url = urlparse(environ['REQUEST_URI'])\n",
" path = url.path\n",
"\n",
" if path.endswith('.jpg'):\n",
" file_path = path[1:]\n",
"\n",
" if os.path.isfile(file_path):\n",
" with open(file_path, 'rb') as file:\n",
" response_body = file.read()\n",
" status = '200 OK'\n",
" headers = [('Content-type', 'image/jpeg'),\n",
" ('Content-Length', str(len(response_body)))]\n",
"\n",
" start_response(status, headers)\n",
" return [response_body]\n",
" else:\n",
" status = '404 Not Found'\n",
" headers = [('Content-type', 'text/plain'),\n",
" ('Content-Length', '0')]\n",
"\n",
" start_response(status, headers)\n",
" return []\n",
" else:\n",
" status = '403 Forbidden'\n",
" headers = [('Content-type', 'text/plain'),\n",
" ('Content-Length', '0')]\n",
"\n",
" start_response(status, headers)\n",
" return []\n",
"\n",
"if __name__ == '__main__':\n",
" from werkzeug.serving import run_simple\n",
" from werkzeug.middleware.dispatcher import DispatcherMiddleware\n",
"\n",
" app = DispatcherMiddleware(JPEGRequestHandler(app), {'/': app})\n",
" run_simple('0.0.0.0', 8000, app, use_reloader=True)"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3 (ipykernel)",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.9.13"
}
},
"nbformat": 4,
"nbformat_minor": 5
}
+44
View File
@@ -0,0 +1,44 @@
import os
from urllib.parse import urlparse
class JPEGRequestHandler:
def __init__(self, app):
self.app = app
def __call__(self, environ, start_response):
url = urlparse(environ['REQUEST_URI'])
path = url.path
if path.endswith('.jpg'):
file_path = path[1:]
if os.path.isfile(file_path):
with open(file_path, 'rb') as file:
response_body = file.read()
status = '200 OK'
headers = [('Content-type', 'image/jpeg'),
('Content-Length', str(len(response_body)))]
start_response(status, headers)
return [response_body]
else:
status = '404 Not Found'
headers = [('Content-type', 'text/plain'),
('Content-Length', '0')]
start_response(status, headers)
return []
else:
status = '403 Forbidden'
headers = [('Content-type', 'text/plain'),
('Content-Length', '0')]
start_response(status, headers)
return []
if __name__ == '__main__':
from werkzeug.serving import run_simple
from werkzeug.middleware.dispatcher import DispatcherMiddleware
app = DispatcherMiddleware(JPEGRequestHandler(app), {'/': app})
run_simple('0.0.0.0', 8000, app, use_reloader=True)
-54
View File
@@ -1,54 +0,0 @@
'''
import os
from flask import Flask, Response
from urllib.parse import urlparse
app = Flask(__name__)
@app.route('/')
def index():
return 'Hello, welcome to the cat picture server!'
@app.route('/<path:path>')
def serve_jpeg(path):
if path.endswith('.jpg'):
file_path = path[1:]
if os.path.isfile(file_path):
with open(file_path, 'rb') as file:
return Response(file.read(), content_type='image/jpeg')
else:
return 'File not found', 404
else:
return 'Only .jpg files are allowed', 403
if __name__ == '__main__':
app.run(host='0.0.0.0', port=8000)
'''
'''
import os
from fastapi import FastAPI
from fastapi.responses import FileResponse
from urllib.parse import urlparse
app = FastAPI()
@app.get('/')
def index():
return 'Hello, welcome to the cat picture server!'
@app.get('/{path:path}')
async def serve_jpeg(path: str):
if path.endswith('.jpg'):
file_path = path[1:]
if os.path.isfile(file_path):
return FileResponse(file_path, media_type='image/jpeg')
else:
return {'detail': 'File not found'}, 404
else:
return {'detail': 'Only .jpg files are allowed'}, 403
if __name__ == '__main__':
import uvicorn
uvicorn.run(app, host='0.0.0.0', port=8000)
'''
BIN
View File
Binary file not shown.

Before

Width:  |  Height:  |  Size: 30 KiB

-14
View File
@@ -1,14 +0,0 @@
import sys
# Iterate through the modules in sys.modules
for module_name, module in sys.modules.items():
# Check if the module has a loader attribute
if hasattr(module, '__loader__'):
loader = module.__loader__
# Determine the type of loader
if type(loader).__name__ == 'SourceFileLoader':
print(f"Module '{module_name}' has a source loader.")
elif type(loader).__name__ == 'BuiltinImporter':
print(f"Module '{module_name}' has a built-in loader.")
elif type(loader).__name__ == 'ExtensionFileLoader':
print(f"Module '{module_name}' has an extension loader.")
-12
View File
@@ -1,12 +0,0 @@
from cx_Freeze import setup, Executable
executables = [Executable('module-inspector.py')]
setup(
name='wow',
version='1.0',
description='Module-inspector Application',
executables=executables
)
-7
View File
@@ -1,7 +0,0 @@
i = 5
def f(arg=i):
print(arg)
i = 6
f()
-47
View File
@@ -1,47 +0,0 @@
import importlib.util
import os
def get_module_size(module_name):
"""Get the total number of bytes in the cached .pyc file or .py file of a module or package.
Args:
module_name (str): The name of the module or package.
Returns:
int: The total number of bytes in the cached .pyc file or .py file.
"""
spec = importlib.util.find_spec(module_name)
if spec is not None and spec.has_location:
# If the module is found and has a location, check for .pyc file first
pyc_file_path = spec.origin + 'c'
if os.path.exists(pyc_file_path):
return os.stat(pyc_file_path).st_size
else:
# If .pyc file doesn't exist, check for .py file
py_file_path = spec.origin
if os.path.exists(py_file_path):
return os.stat(py_file_path).st_size
return 0 # Return 0 if no .pyc or .py file found
import sys
def get_total_module_size():
"""Get the total number of bytes for all modules listed in sys.modules.
Returns:
int: The total number of bytes for all modules.
"""
total_size = 0
for module_name in sys.modules:
module_size = get_module_size(module_name)
total_size += module_size
return total_size
if __name__ == '__main__' and __package__ is None:
__spec__ = importlib.util.spec_from_file_location('__main__', __file__)
total_size = get_total_module_size()
print("Total size of all modules in sys.modules: {} bytes".format(total_size))
-74
View File
@@ -1,74 +0,0 @@
import os
import sys
import importlib.util
import types
from collections import defaultdict
from sys import getsizeof
def get_bytecode_size(module_name):
"""Get the size of the cached .pyc bytecode file for a module."""
spec = importlib.util.find_spec(module_name)
if spec and spec.origin and spec.origin.endswith('.py'):
pyc_file = importlib.util.cache_from_source(spec.origin)
if os.path.exists(pyc_file):
return os.path.getsize(pyc_file)
return 0
def get_dso_size(module_name):
"""Get the size of the DSO (Dynamic Shared Object) module file for a module."""
spec = importlib.util.find_spec(module_name)
if spec and spec.origin and spec.origin.endswith(('.so', '.dll')):
return os.path.getsize(spec.origin)
return 0
def generate_module_report():
"""Generate a report summarizing the module environment."""
total_modules = len(sys.modules)
bytecode_size = sum(get_bytecode_size(module_name) for module_name in sys.modules)
dso_size = sum(get_dso_size(module_name) for module_name in sys.modules)
python_size = sum(
getsizeof(sys.modules[module_name]) - getsizeof(module_name)
for module_name in sys.modules
if isinstance(sys.modules[module_name], types.ModuleType)
)
total_size = bytecode_size + dso_size + python_size
# Calculate percentage of each type of module
bytecode_percent = bytecode_size / total_size * 100
dso_percent = dso_size / total_size * 100
python_percent = python_size / total_size * 100
# Generate the report
report = f"Total modules: {total_modules}\n"
report += f"Total size: {total_size} bytes\n"
report += f"Bytecode modules: {bytecode_percent:.2f}% ({bytecode_size} bytes)\n"
report += f"DSO modules: {dso_percent:.2f}% ({dso_size} bytes)\n"
report += f"Python code modules: {python_percent:.2f}% ({python_size} bytes)\n"
# Generate histogram of module directory names
dir_histogram = defaultdict(int)
for module_name in sys.modules:
module_file = getattr(sys.modules[module_name], '__file__', None)
if module_file:
module_dir = os.path.dirname(module_file)
dir_histogram[module_dir] += 1
report += "\nModule Directory Histogram:\n"
for module_dir, count in dir_histogram.items():
report += f"{module_dir}: {'*' * count}\n"
return report
if __name__ == "__main__":
__spec__ = importlib.util.spec_from_file_location('__main__', __file__)
report = generate_module_report()
print(report)
Binary file not shown.

Before

Width:  |  Height:  |  Size: 16 KiB

BIN
View File
Binary file not shown.

Before

Width:  |  Height:  |  Size: 10 KiB

+1
View File
@@ -0,0 +1 @@
print('Hi')
BIN
View File
Binary file not shown.

Before

Width:  |  Height:  |  Size: 4.2 KiB

-5
View File
@@ -2,7 +2,6 @@ import sys
import tkinter as tk import tkinter as tk
from tkinter import ttk from tkinter import ttk
def on_treeview_heading_click(column, tree): def on_treeview_heading_click(column, tree):
data = [(tree.set(child, column), child) for child in tree.get_children("")] data = [(tree.set(child, column), child) for child in tree.get_children("")]
reverse = (column == tree.heading(column)["text"]) and not column.startswith("_") reverse = (column == tree.heading(column)["text"]) and not column.startswith("_")
@@ -39,10 +38,6 @@ tree.column('#0', width=20, stretch=False)
tree.tag_configure("selected", background="#ccc") tree.tag_configure("selected", background="#ccc")
# Call the on_treeview_heading_click function when a column heading is clicked
for col in tree["columns"]:
tree.heading(col, text=col, anchor=tk.W, command=lambda col=col: on_treeview_heading_click(col, tree))
root.mainloop()
# Call the on_treeview_heading_click function when a column heading is clicked # Call the on_treeview_heading_click function when a column heading is clicked
for col in tree["columns"]: for col in tree["columns"]:
tree.heading(col, text=col, anchor=tk.W, command=lambda col=col: on_treeview_heading_click(col, tree)) tree.heading(col, text=col, anchor=tk.W, command=lambda col=col: on_treeview_heading_click(col, tree))
BIN
View File
Binary file not shown.

Before

Width:  |  Height:  |  Size: 40 KiB

BIN
View File
Binary file not shown.
-9
View File
@@ -1,9 +0,0 @@
---- BEGIN SSH2 PUBLIC KEY ----
Comment: "rsa-key-20230531"
AAAAB3NzaC1yc2EAAAADAQABAAABAQDEZW0iQpBPPRGM7QCdc3GqefkqUqq5i727
MEYvf7gy0h7Zvz72cvnDtnVF6MknCGVZ1S3uY+HeKUtujE3+z9tJ5zIxO+TxthJe
8gbH9MUkElKX8qJN/nF1H4dLiW+77ARsT70rDoHayWccGGnwC/Gs9xXVF+Ni0JD5
TKcGoye3Ps2Ni/KJAzkn78JyfrAb+Ail4DG4snCf1q/jXTuB39qYTTg94FJEqwe1
dhPIzRiOjFT3eSmHAXdSuAXFjpXAq+bV8R2uxKsgNfvxyV4vlKwFhYTq+kHwDWX7
RtLQUUs8mSTfbnL2LuE+T2d8sBtj9wi+gODFl5dDAVZkmuKrRgLF
---- END SSH2 PUBLIC KEY ----
-8
View File
@@ -1,8 +0,0 @@
google.cloud.pubsub
plotly
pandas
numpy
h5py
selenium
sqlalchemy
flask
-21
View File
@@ -1,21 +0,0 @@
<!DOCTYPE html>
<html>
<head>
<title>Class Dump</title>
</head>
<body>
<h1>Class Dump</h1>
<ul>
{% for class in classes %}
<li>{{ class.__name__ }}</li>
{% endfor %}
</ul>
<h1>User Dump</h1>
<ul>
{% for user in users %}
<li>{{ user }}</li>
{% endfor %}
</ul>
</body>
</html>
-31
View File
@@ -1,31 +0,0 @@
class Count:
def __init__(self,mymin,mymax):
self.mymin=mymin
self.mymax=mymax
def __getattr__(self, item):
self.__dict__[item]=0
return 0
obj1 = Count(1,10)
print(obj1.mymin)
print(obj1.mymax)
print(obj1.mycurrent1)
class Count:
def __init__(self,mymin,mymax):
self.mymin=mymin
self.mymax=mymax
self.current=None
def __getattribute__(self, item):
if item.startswith('cur'):
raise AttributeError
return object.__getattribute__(self,item)
# or you can use ---return super().__getattribute__(item)
obj1 = Count(1,10)
print(obj1.mymin)
print(obj1.mymax)
print(obj1.current)
-14
View File
@@ -1,14 +0,0 @@
import sys
# Iterate through the modules in sys.modules
for module_name, module in sys.modules.items():
# Check if the module has a loader attribute
if hasattr(module, '__loader__'):
loader = module.__loader__
# Determine the type of loader
if type(loader).__name__ == 'SourceFileLoader':
print(f"Module '{module_name}' has a source loader.")
elif type(loader).__name__ == 'BuiltinImporter':
print(f"Module '{module_name}' has a built-in loader.")
elif type(loader).__name__ == 'ExtensionFileLoader':
print(f"Module '{module_name}' has an extension loader.")
-12
View File
@@ -1,12 +0,0 @@
from cx_Freeze import setup, Executable
executables = [Executable('module-inspector.py')]
setup(
name='wow',
version='1.0',
description='Module-inspector Application',
executables=executables
)
-58
View File
@@ -1,58 +0,0 @@
import time
'''import matplotlib.pyplot as plt'''
# Durations to measure in seconds
durations = [0.01, 0.1, 1.0]
# Number of repetitions
num_repetitions = 100
# Lists to store jitter for each duration
jitter_10ms = []
jitter_100ms = []
jitter_1s = []
# Function to measure jitter
def measure_jitter(duration):
start_time = time.perf_counter_ns()
time.sleep(duration)
end_time = time.perf_counter_ns()
return end_time - start_time
# Measure jitter for each duration and repeat for num_repetitions times
for duration in durations:
for _ in range(num_repetitions):
jitter = measure_jitter(duration)
if duration == 0.01:
jitter_10ms.append(jitter)
elif duration == 0.1:
jitter_100ms.append(jitter)
elif duration == 1.0:
jitter_1s.append(jitter)
'''
# Create histograms for each duration's jitter
plt.hist(jitter_10ms, bins=10, color='r', alpha=0.5, label='10ms')
plt.hist(jitter_100ms, bins=10, color='g', alpha=0.5, label='100ms')
plt.hist(jitter_1s, bins=10, color='b', alpha=0.5, label='1s')
plt.xlabel('Jitter (ns)')
plt.ylabel('Frequency')
plt.title('Jitter of time.sleep() for Different Durations')
plt.legend()
plt.show()
'''
import random
import secrets
# Create a dictionary with 1 million integer keys
dict = {}
for i in range(1000000):
# Generate a random 64-byte value as bytes
random_value = secrets.token_bytes(64)
# Convert the bytes value to hexadecimal representation
hex_value = random_value.hex()
# Add the key-value pair to the dictionary
dict[i] = hex_value
print("Dictionary created with 1 million integer keys.")
-51
View File
@@ -1,51 +0,0 @@
class Vault:
def __init__(self):
self._treasure = "gold"
self._known_good_copy = self.__dict__.copy()
def __getattribute__(self, name):
if name == "treasure":
if self.__dict__ != self._known_good_copy:
print("Attribute deleted!")
return self._treasure
else:
return object.__getattribute__(self, name)
def __setattr__(self, name, value):
if name != "_treasure" and name != "_known_good_copy":
print("Attribute deletion detected!")
self._known_good_copy = self.__dict__.copy()
object.__setattr__(self, name, value)
v = Vault()
Vault.treasure = " haha monkeypatched"
print(v.treasure)
'''
class Vault:
def __init__(self):
self._treasure = "gold"
self._known_good_copy = self.__dict__.copy()
def __get__(self, instance, owner):
if instance is None:
return self
else:
if instance.__dict__ != self._known_good_copy:
print("Attribute deleted!")
return self._treasure
def __set__(self, instance, value):
print("Attribute deletion detected!")
self._known_good_copy = instance.__dict__.copy()
def __delete__(self, instance):
print("Attribute deletion detected!")
self._known_good_copy = instance.__dict__.copy()
def __set_name__(self, owner, name):
self.name = name
v = Vault()
Vault._treasure = " haha monkeypatched"
print(v._treasure)
'''
-17
View File
@@ -1,17 +0,0 @@
import asyncio
# Define a coroutine function
async def greet(name):
print(f"Hello, {name}!")
await asyncio.sleep(1)
print(f"Goodbye, {name}!")
# Call the coroutine function
coroutine_obj = greet("Karen")
# coroutine_obj is a coroutine object, not the actual execution of the coroutine
# Use an event loop to run the coroutine
loop = asyncio.get_event_loop()
loop.run_until_complete(coroutine_obj)
loop.close()
-8
View File
@@ -1,8 +0,0 @@
def think():
x=1
def think2():
nonlocal x
x=2
think2()
return x
print(think())
-33
View File
@@ -1,33 +0,0 @@
'''
import sys
import importlib.util
# Load only the sys module
importlib.import_module('sys')
# Check the type of the value associated with the 'os' key
os_module = sys.modules['os']
os_module_type = type(os_module)
# Print the type of the value associated with the 'os' key
print("Type of the value associated with the 'os' key in sys.modules: {}".format(os_module_type))
'''
'''
import os
print(type(os))
'''
import builtins
# Access the os module from globals()
os = globals().get('os', None)
# Check if os module is available
if os is not None:
# Call os.getlogin()
username = os.getlogin()
print("Username:", username)
else:
print("os module not available")
-54
View File
@@ -1,54 +0,0 @@
import os
import sys
import importlib.util
import types
def get_bytecode_size(module_name):
"""Get the size of the cached .pyc bytecode file for a module."""
spec = importlib.util.find_spec(module_name)
if spec and spec.origin.endswith('.py'):
pyc_file = importlib.util.cache_from_source(spec.origin)
if os.path.exists(pyc_file):
return os.path.getsize(pyc_file)
return 0
def get_dso_size(module_name):
"""Get the size of the DSO (Dynamic Shared Object) module file for a module."""
spec = importlib.util.find_spec(module_name)
if spec and spec.origin.endswith(('.so', '.dll')):
return os.path.getsize(spec.origin)
return 0
def generate_module_report():
"""Generate a report summarizing the module environment."""
total_modules = len(sys.modules)
bytecode_size = sum(get_bytecode_size(module_name) for module_name in sys.modules)
dso_size = sum(get_dso_size(module_name) for module_name in sys.modules)
python_size = sum(
sys.modules[module_name].__sizeof__()
for module_name in sys.modules
if isinstance(sys.modules[module_name], types.ModuleType)
)
total_size = bytecode_size + dso_size + python_size
# Calculate percentage of each type of module
bytecode_percent = bytecode_size / total_size * 100
dso_percent = dso_size / total_size * 100
python_percent = python_size / total_size * 100
# Generate the report
report = f"Total modules: {total_modules}\n"
report += f"Total size: {total_size} bytes\n"
report += f"Bytecode modules: {bytecode_percent:.2f}% ({bytecode_size} bytes)\n"
report += f"DSO modules: {dso_percent:.2f}% ({dso_size} bytes)\n"
report += f"Python code modules: {python_percent:.2f}% ({python_size} bytes)\n"
return report
if __name__ == '__main__' and __package__ is None:
__spec__ = importlib.util.spec_from_file_location('__main__', __file__)
report = generate_module_report()
print(report)
-259
View File
@@ -1,259 +0,0 @@
'''
import os
import sys
import importlib.util
import types
import inspect
from collections import defaultdict
def get_bytecode_size(module_name):
"""Get the size of the cached .pyc bytecode file for a module."""
spec = importlib.util.find_spec(module_name)
if spec and spec.origin.endswith('.py'):
pyc_file = importlib.util.cache_from_source(spec.origin)
if os.path.exists(pyc_file):
return os.path.getsize(pyc_file)
return 0
def get_dso_size(module_name):
"""Get the size of the DSO (Dynamic Shared Object) module file for a module."""
spec = importlib.util.find_spec(module_name)
if spec and spec.origin.endswith(('.so', '.dll')):
return os.path.getsize(spec.origin)
return 0
def generate_module_report():
"""Generate a report summarizing the module environment."""
total_modules = len(sys.modules)
bytecode_size = sum(get_bytecode_size(module_name) for module_name in sys.modules)
dso_size = sum(get_dso_size(module_name) for module_name in sys.modules)
python_size = sum(
sys.modules[module_name].__sizeof__()
for module_name in sys.modules
if isinstance(sys.modules[module_name], types.ModuleType)
)
total_size = bytecode_size + dso_size + python_size
# Calculate percentage of each type of module
bytecode_percent = bytecode_size / total_size * 100
dso_percent = dso_size / total_size * 100
python_percent = python_size / total_size * 100
# Generate the report
report = f"Total modules: {total_modules}\n"
report += f"Total size: {total_size} bytes\n"
report += f"Bytecode modules: {bytecode_percent:.2f}% ({bytecode_size} bytes)\n"
report += f"DSO modules: {dso_percent:.2f}% ({dso_size} bytes)\n"
report += f"Python code modules: {python_percent:.2f}% ({python_size} bytes)\n"
# Generate histogram of module directory names
dir_histogram = defaultdict(int)
for module_name in sys.modules:
module_file = sys.modules[module_name].__file__
if module_file:
module_dir = os.path.dirname(module_file)
dir_histogram[module_dir] += 1
report += "\nModule Directory Histogram:\n"
for module_dir, count in dir_histogram.items():
report += f"{module_dir}: {'*' * count}\n"
return report
if __name__ == '__main__' and __package__ is None:
script_path = inspect.getframeinfo(inspect.currentframe()).filename
__spec__ = importlib.util.spec_from_loader('__main__', None, origin=script_path)
report = generate_module_report()
print(report)
'''
'''
import os
import sys
from importlib.util import find_spec
def get_module_size(module):
spec = find_spec(module)
if spec is None:
return None
if spec.cached:
size = os.stat(spec.cached).st_size
else:
size = os.stat(spec.origin).st_size
return size
def summarize_modules():
total_size = 0
byte_count = 0
dso_count = 0
py_count = 0
directory_histogram = {}
for module_name, module in sys.modules.items():
if module is None:
continue
size = get_module_size(module_name)
if size is None:
continue
total_size += size
if module_name.endswith('_bootstrap') or module_name.endswith('_bootstrap_external'):
continue
if module_name.startswith('_frozen_importlib'):
dso_count += 1
continue
if module_name.startswith('_'):
continue
if module.__file__.endswith('.pyc'):
byte_count += size
elif module.__file__.endswith(('.so', '.dll')):
dso_count += 1
else:
py_count += size
module_directory = os.path.dirname(os.path.abspath(module.__file__))
directory_histogram[module_directory] = directory_histogram.get(module_directory, 0) + 1
print(f"Total module size: {total_size} bytes")
print(f"Bytecode modules: {(byte_count/total_size)*100:.2f}% ({byte_count} bytes)")
print(f"DSO modules: {(dso_count/total_size)*100:.2f}% ({dso_count} modules)")
print(f"Python modules: {(py_count/total_size)*100:.2f}% ({py_count} bytes)")
print("Module directory histogram:")
for directory, count in directory_histogram.items():
print(f"{directory}: {'*' * count}")
if __name__ == '__main__':
summarize_modules()
'''
'''
import os
import sys
from importlib.util import find_spec
def get_module_size(module):
spec = find_spec(module)
if spec is None:
return None
if spec.cached:
size = os.stat(spec.cached).st_size
else:
size = os.stat(spec.origin).st_size
return size
def summarize_modules():
total_size = 0
byte_count = 0
dso_count = 0
py_count = 0
directory_histogram = {}
for module_name, module in sys.modules.items():
if module is None:
continue
size = get_module_size(module_name)
if size is None:
continue
total_size += size
if module_name.endswith('_bootstrap') or module_name.endswith('_bootstrap_external'):
continue
if module_name.startswith('_frozen_importlib'):
dso_count += 1
continue
if module_name.startswith('_'):
continue
if module.__file__.endswith('.pyc'):
byte_count += size
elif module.__file__.endswith(('.so', '.dll')):
dso_count += 1
else:
py_count += size
module_directory = os.path.dirname(os.path.abspath(module.__file__))
directory_histogram[module_directory] = directory_histogram.get(module_directory, 0) + 1
print(f"Total module size: {total_size} bytes")
print(f"Bytecode modules: {(byte_count/total_size)*100:.2f}% ({byte_count} bytes)")
print(f"DSO modules: {(dso_count/total_size)*100:.2f}% ({dso_count} modules)")
print(f"Python modules: {(py_count/total_size)*100:.2f}% ({py_count} bytes)")
print("Module directory histogram:")
for directory, count in directory_histogram.items():
print(f"{directory}: {'*' * count}")
if __name__ == '__main__':
summarize_modules()
'''
import os
import sys
import importlib.util
import types
from collections import defaultdict
def get_bytecode_size(module_name):
"""Get the size of the cached .pyc bytecode file for a module."""
spec = importlib.util.find_spec(module_name)
if spec and spec.origin.endswith('.py'):
pyc_file = importlib.util.cache_from_source(spec.origin)
if os.path.exists(pyc_file):
return os.path.getsize(pyc_file)
return 0
def get_dso_size(module_name):
"""Get the size of the DSO (Dynamic Shared Object) module file for a module."""
spec = importlib.util.find_spec(module_name)
if spec and spec.origin.endswith(('.so', '.dll')):
return os.path.getsize(spec.origin)
return 0
def generate_module_report():
"""Generate a report summarizing the module environment."""
total_modules = len(sys.modules)
bytecode_size = sum(get_bytecode_size(module_name) for module_name in sys.modules)
dso_size = sum(get_dso_size(module_name) for module_name in sys.modules)
python_size = sum(
sys.modules[module_name].__sizeof__()
for module_name in sys.modules
if isinstance(sys.modules[module_name], types.ModuleType)
)
total_size = bytecode_size + dso_size + python_size
# Calculate percentage of each type of module
bytecode_percent = bytecode_size / total_size * 100
dso_percent = dso_size / total_size * 100
python_percent = python_size / total_size * 100
# Generate the report
report = f"Total modules: {total_modules}\n"
report += f"Total size: {total_size} bytes\n"
report += f"Bytecode modules: {bytecode_percent:.2f}% ({bytecode_size} bytes)\n"
report += f"DSO modules: {dso_percent:.2f}% ({dso_size} bytes)\n"
report += f"Python code modules: {python_percent:.2f}% ({python_size} bytes)\n"
# Generate histogram of module directory names
dir_histogram = defaultdict(int)
for module_name in sys.modules:
module_file = sys.modules[module_name].__file__
if module_file:
module_dir = os.path.dirname(module_file)
dir_histogram[module_dir] += 1
report += "\nModule Directory Histogram:\n"
for module_dir, count in dir_histogram.items():
report += f"{module_dir}: {'*' * count}\n"
return report
if __name__ == "__main__":
report = generate_module_report()
print(report)
-77
View File
@@ -1,77 +0,0 @@
import os
import sys
import importlib.util
import types
from collections import defaultdict
from sys import getsizeof
def get_bytecode_size(module_name):
"""Get the size of the cached .pyc bytecode file for a module."""
spec = importlib.util.find_spec(module_name)
if spec and spec.origin and spec.origin.endswith('.py'):
pyc_file = importlib.util.cache_from_source(spec.origin)
if os.path.exists(pyc_file):
return os.path.getsize(pyc_file)
return 0
def get_dso_size(module_name):
"""Get the size of the DSO (Dynamic Shared Object) module file for a module."""
spec = importlib.util.find_spec(module_name)
if spec and spec.origin and spec.origin.endswith(('.so', '.dll')):
return os.path.getsize(spec.origin)
return 0
def generate_module_report():
"""Generate a report summarizing the module environment."""
total_modules = len(sys.modules)
bytecode_size = sum(get_bytecode_size(module_name) for module_name in sys.modules)
dso_size = sum(get_dso_size(module_name) for module_name in sys.modules)
python_size = sum(
getsizeof(sys.modules[module_name]) - getsizeof(module_name)
for module_name in sys.modules
if isinstance(sys.modules[module_name], types.ModuleType)
)
total_size = bytecode_size + dso_size + python_size
# Calculate percentage of each type of module
bytecode_percent = bytecode_size / total_size * 100
dso_percent = dso_size / total_size * 100
python_percent = python_size / total_size * 100
# Generate the report
report = f"Total modules: {total_modules}\n"
report += f"Total size: {total_size} bytes\n"
report += f"Bytecode modules: {bytecode_percent:.2f}% ({bytecode_size} bytes)\n"
report += f"DSO modules: {dso_percent:.2f}% ({dso_size} bytes)\n"
report += f"Python code modules: {python_percent:.2f}% ({python_size} bytes)\n"
# Generate histogram of module directory names
dir_histogram = defaultdict(int)
for module_name in sys.modules:
module_file = getattr(sys.modules[module_name], '__file__', None)
if module_file:
module_dir = os.path.dirname(module_file)
dir_histogram[module_dir] += 1
report += "\nModule Directory Histogram:\n"
for module_dir, count in dir_histogram.items():
report += f"{module_dir}: {'*' * count}\n"
return report
if __name__ == "__main__":
__spec__ = importlib.util.spec_from_file_location('__main__', __file__)
report = generate_module_report()
print(report)
report = generate_module_report()
print(report)
-63
View File
@@ -1,63 +0,0 @@
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.")
-45
View File
@@ -1,45 +0,0 @@
# Selected
# Want-Want Rice Crackers
url = 'https://www.sayweee.com/en/product/Want-Want-Senbei-Rice-Crackers/17119'
# Lay's Potato Chips
url = 'https://www.sayweee.com/en/product/Lays-Potato-Chips-Cucumber-Flavor/51261'
# Pocky Cookies
url = 'https://www.sayweee.com/en/product/Glico-Pocky-Cookies-and-Cream-Family-Pack/79467'
# Thai Jasmine Rice
url = 'https://www.sayweee.com/en/product/Xiang-Na-Lan-Thai-Jasmine-Rice/19566?category=dried01'
# Japanese Medium Grain Rice
url = 'https://www.sayweee.com/en/product/Sekka-Premium-Japanese-Medium-Grain-Rice/81224'
# Millet
url = 'https://www.sayweee.com/en/product/Big-Green-Spring-Farm-Organic-Millet/90991?category=dried02'
# Nature's Soy Firm Tofu: https://www.sayweee.com/zh/product/Nature-s-Soy-Firm-Tofu-/19231?trace_id=ee63a29e-a1e4-4065-b213-96aa14a15521
# Free 2% Reduced Fat Milk: https://www.sayweee.com/zh/product/Lactaid-Lactose-Free-2--Reduced-Fat-Milk/95353?trace_id=3030fe4c-04fc-4f8f-9257-7f6395b9e513
# Washington Fuji Apples: https://www.sayweee.com/zh/product/Washington-Fuji-Apples/94817?trace_id=dc4955fc-bba8-48cf-b1f9-9db3cf077eed
# Greenlandic Sweet Prawns: https://www.sayweee.com/zh/product/Royal-Greenland-Cooked-Wild-Caught-Greenlandic-Sweet-Prawns-Frozen-1/44656?category=seafood01
# Optional
# LuoSiFen
url = 'https://www.sayweee.com/en/product/Liuzhou-Guangxi-Specialty-LuoSiFen-Hot-Spicy-Noodles/95442?category=instant01'
# Rice Noodles
url = 'https://www.sayweee.com/en/product/Sanyangyishi-Fresh-Instant-Rice-Noodles/2156?category=instant01'
# Cup Noodles
url = 'https://www.sayweee.com/en/product/Nissin-Beef-Flavor-Cup-Noodles/102519?category=instant01'
# Pork Sung
url = 'https://www.sayweee.com/en/product/Formosa-Pork-Sung-/23528?category=canned04'
# Nature's Soy Firm Tofu: https://www.sayweee.com/zh/product/Nature-s-Soy-Firm-Tofu-/19231?trace_id=ee63a29e-a1e4-4065-b213-96aa14a15521
# [Not available or sold out in some area] Beef Ribeye Thin Slices: https://www.sayweee.com/zh/product/Beef-Ribeye-Thin-Slices--Frozen-1/4295?category=meat02
# Shiitake Mushrooms: https://www.sayweee.com/zh/product/Shiitake-Mushrooms/45782?trace_id=ba388941-f631-4eaa-8570-5b7d4aba64e7
-33
View File
@@ -1,33 +0,0 @@
ZIP Code
85305
30313
21230
14127
28202
60605
45202
44114
76011
80204
48226
54304
77054
46225
32202
64129
89118
90301
90301
33056
55415
02035
70112
07073
07073
19148
15212
95054
98134
33607
37213
20785
1 ZIP Code
2 85305
3 30313
4 21230
5 14127
6 28202
7 60605
8 45202
9 44114
10 76011
11 80204
12 48226
13 54304
14 77054
15 46225
16 32202
17 64129
18 89118
19 90301
20 90301
21 33056
22 55415
23 02035
24 70112
25 07073
26 07073
27 19148
28 15212
29 95054
30 98134
31 33607
32 37213
33 20785