I wanted to see how would I deal with a large amount of data being scraped and written into a CSV file, so I decided to get the info out of a random website.
First off, I found a way to search for all the products on the website and I figured out that any product will have a space in its name or description, so I searched for that.
Second, I had to know how many pages there are so I can go through all of them and take some information about every product.
from csv import DictWriter, QUOTE_MINIMAL
from lxml import html
import requests
SEARCH_PAGE_URL = "https://www.richelieu.com/us/en/search?s=%20"
DYNAMIC_PRODUCTS_PAGE = "https://www.richelieu.com/us/en/search/?s=%20&imgMode=m&sort=&nbPerPage=48&page={}#results"
HEADERS = {
"User-Agent": "User-Agent:Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_6) AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/61.0.3163.100 Safari/537.36"
}
def get_total_products():
"""Return the total number of products"""
search_page = requests.get(SEARCH_PAGE_URL, headers=HEADERS).text
search_page_tree = html.fromstring(search_page)
return int(search_page_tree.xpath("//div[@class='ts_resultOptions']//p[1]")[0].text.split()[2])
def get_product_url(total_products):
"""Yield the links for products pages"""
for page_number in range(1, total_products + 1):
products_url = DYNAMIC_PRODUCTS_PAGE.format(page_number)
products_page = requests.get(products_url, headers=HEADERS).text
products_page_tree = html.fromstring(products_page)
products_links = [a_tag.attrib["href"]
for a_tag in products_page_tree.xpath("//*[@id='prodResult']/li//div[@class='imgWrapper']/a")]
for link in products_links:
yield link
break
def check_if_more_products(tree):
"""Check to see if there are more SKU's for the same product"""
more_prods = [all_prod.attrib['href'] for all_prod in tree.xpath("//td[@class='sku ']/div/a")]
if not more_prods:
return False
return more_prods
def get_product_number(tree):
return {
"Product number": tree.xpath("//div[@id='pm2_questions']/div[@class='noProdReadOnly']/span")[0].text
}
def get_product_name(tree):
return {
"Name": tree.xpath("//*[@id='pm2_topInfo']/div/h1/span")[0].text
}
def get_product_category(tree):
breadcrumb = tree.xpath("/html/body/div[2]//li//span")
return {
"Category": " / ".join([x.text for x in breadcrumb][1:-1])
}
def get_product_description(tree):
try:
description = tree.xpath("//*[@id='pm2_topInfo']/div/p")[0].text
except IndexError:
description = ''
return {
"Description": description
}
def get_extra_data(tree):
data = {}
keys = [a.text for a in tree.xpath("//div[@class='prodTableContainer']/table/thead/tr/th[position()>1]/a")]
vals = [td.text for td in tree.xpath("//div[@class='prodTableContainer']/table/tbody/tr[1]/td[position()>1]")]
for key, val in zip(keys, vals):
data[key] = val
return data
def create_technical_tables(tree):
keys = [el.text for el in tree.xpath("//table[@class='table']/tbody/tr/th/span")]
vals = [el.text.strip() for el in tree.xpath("//table[@class='table']/tbody/tr/td")]
info_table = get_extra_data(tree)
data, extra_data = {}, {}
table_1, table_2 = '', ''
for key, val in zip(keys, vals):
if val.startswith("From") and key.lower() == "suggested price":
continue
if key.lower() == "product number":
continue
if key.lower() == "our divisions":
continue
if key.lower() == "material":
extra_data[key] = val
continue
if "color" in key.lower():
extra_data[key] = val
continue
if "finish" in key.lower():
extra_data[key] = val
continue
data[key] = val
for key, val in info_table.items():
if 'material' in key.lower():
extra_data[key] = val
if 'color' in key.lower():
extra_data[key] = val
for key, val in data.items():
table_1 += "<p><strong>{}:</strong>{}</p>".format(key, val)
for key, val in extra_data.items():
table_2 += "<p><strong>{}</strong>{}</p>".format(key, val)
return table_1, table_2
def first_table(tree):
return {
"Technical specifications": create_technical_tables(tree)[0]
}
def second_table(tree):
return {
"Info": create_technical_tables(tree)[1]
}
def get_catalog_link(tree):
try:
catalog_link = tree.xpath("//*[@id='carouselSegment1b']/ul/li[1]/div/a")[0].attrib['href']
except IndexError:
catalog_link = ''
return {
"Catalog link": catalog_link
}
def get_line_art_link(tree):
try:
line_art = "https://www.richelieu.com{}".format(
tree.xpath("//*[@id='carouselSegment1b']/ul/li[2]/div/a")[0].attrib['href']
)
except IndexError:
line_art = ''
return {
"Line art": line_art
}
def get_right_part_info(tree):
data = {}
extras = [a.text.strip()
for a in tree.xpath("//div[@id='pm2_blocDroitFixe']/div[@class='feAncetres clearfix']/ul/li/a")]
for item in extras:
name, value = item.split(': ')
data[name] = value
return data
def get_product_images(tree):
data = {}
links = ["https://www.richelieu.com{}".format(a.attrib["href"])
for a in tree.xpath("//div[@id='rcMediaPlayerCarousel']//li/a")]
for i, link in enumerate(links, start=1):
key = 'image_{}'.format(i)
data[key] = link
return data
def prepare_product(tree):
product_number = get_product_number(tree)
product_name = get_product_name(tree)
product_category = get_product_category(tree)
product_description = get_product_description(tree)
first_table_product = first_table(tree)
second_table_product = second_table(tree)
catalog_link = get_catalog_link(tree)
line_art_link = get_line_art_link(tree)
right_part_info = get_right_part_info(tree)
product_images = get_product_images(tree)
row = [
product_number,
product_name,
product_category,
product_description,
first_table_product,
second_table_product,
catalog_link,
line_art_link,
right_part_info,
product_images
]
super_row = [], {}
for d in row:
for k, v in d.items():
super_row[k] = v
return super_row
def main():
total_products = get_total_products()
all_fieldnames = set()
for link in get_product_url(total_products):
product_page = requests.get(link, headers=HEADERS).text
product_tree = html.fromstring(product_page)
more_products = check_if_more_products(product_tree)
if not more_products:
row = prepare_product(product_tree)
fieldnames = row.keys()
all_fieldnames.update(fieldnames)
else:
for product_link in more_products:
new_page = requests.get(product_link, headers=HEADERS).text
new_product_tree = html.fromstring(new_page)
row = prepare_product(new_product_tree)
fieldnames = row.keys()
all_fieldnames.update(fieldnames)
with open('products.csv', 'a', newline='') as f:
writer = DictWriter(f, fieldnames=list(all_fieldnames), delimiter=';', quoting=QUOTE_MINIMAL)
writer.writeheader()
for link in get_product_url(total_products):
product_page = requests.get(link, headers=HEADERS).text
product_tree = html.fromstring(product_page)
more_products = check_if_more_products(product_tree)
if not more_products:
row = prepare_product(product_tree)
writer.writerow(row)
else:
for product_link in more_products:
new_page = requests.get(product_link, headers=HEADERS).text
new_product_tree = html.fromstring(new_page)
row = prepare_product(new_product_tree)
writer.writerow(row)
if __name__ == '__main__':
main()
Performance
Because the data for products from different categories are different - or even for the products in the same category (some of them, not all), I had to first go through all the products and create the header for my CSV file and just after that, get the actual info. Because of this and because there is a huge number of products (40k+), the code is really slow.
For anybody who'll want to test it, I'd recommend throwing a break after the loop in the get_product_url() method.
What I'm looking for:
- Suggestions regarding ways to improve the code performance
- Suggestions about improving my
xpaths (or maybebs4would help out more) - I wouldn't like to use Scrapy, although I know it might help (I won't mind a quick intro into how this might be used in my case)
