Intro
This is a follow-up question to my earlier post. Although @alecxe had very nice remarks in hist last answer, I decided to get all the products a bit different.
Instead of searching for a space (i.e. ASCII url-encoded %20) (which wasn't the best way to go as there were products without a space in their name / description), I decided to take every category from the index page and recursively go through all of them until I reach the product page.
Moreover, regarding the requests.Session(), I decided not to use it because I've created a txt file containing some user-agents (in my head, it didn't make sense to have multiple user-agents within the same session - maybe I'm wrong).
Other changes that I made:
- sorted the header fields so that the
product_idgoes first and the images at the end - improved some of the xpaths
- added a
retry()function which recursively tries to get a specific link
Concerns
Unfortunately, I didn't have the necessary time to go through Scrapy and wrap my head around it, so I didn't use it (even tough it might improve it considerably). As it is, the below code runs very slowly (for example, since morning it only went through 30k products). How can I speed it up more?
Code
"""
Scraper for https://www.richelieu.com/
Python version: 3.6.X
"""
from collections import ChainMap
from csv import DictWriter, QUOTE_MINIMAL
from lxml import html
from random import choice, randint
from string import ascii_uppercase
from time import sleep
import os
import requests
base_dir = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
utils = os.path.join(base_dir, 'utils')
user_agents = os.path.join(base_dir, 'utils')
result = os.path.join(base_dir, 'result')
INDEX_PAGE = 'https://www.richelieu.com/us/en/index'
CATEGORY_LINKS_FILE = os.path.join(utils, 'big_categories.txt')
PRODUCT_LINKS_FILE = os.path.join(utils, 'products.txt')
RESULT_FILE = os.path.join(result, 'products2.csv')
USER_AGENTS_FILE = os.path.join(user_agents, 'user_agents.txt')
def random_user_agent():
return choice(open(USER_AGENTS_FILE).readlines()).strip()
def check_if_more_products(tree):
more_prods = [
all_prod.attrib['href']
for all_prod in tree.xpath("//div[@id='pm2_prodTableForm']//tbody/tr/td[1]//a")
]
if not more_prods:
return False
return more_prods
def get_product_number(tree):
try:
prod_number = tree.xpath("//h2[contains(., 'Product number')]/following-sibling::span")[0].text
except IndexError:
prod_number = ""
return {"Product number": prod_number}
def get_product_name(tree):
try:
prod_name = tree.xpath("//section[@id='pm2_topInfo']//h1/span")[0].text
except IndexError:
prod_name = ""
return {"Name": prod_name}
def get_product_category(tree):
try:
breadcrumb = tree.xpath("/html/body/div[2]//li//span")
category = " / ".join([x.text for x in breadcrumb][1:-1])
except IndexError:
category = ""
return {"Category": category}
def get_product_description(tree):
try:
description = tree.xpath("//*[@id='pm2_topInfo']//p")[0].text
except IndexError:
description = ''
return {"Description": description}
def get_extra_data(tree):
keys = [
a.text
for a in tree.xpath("//table//tr/th[position()>1]/a")
]
vals = [
td.text
for td in tree.xpath("//table//tr[1]/td[position()>1]")
]
return dict(zip(keys, vals))
def create_technical_tables(tree):
keys = [
el.text
for el in tree.xpath("//table[@class='table']//span")
]
vals = [
el.text.strip()
for el in tree.xpath("//table[@class='table']//td")
]
info_table = get_extra_data(tree)
data, extra_data = {}, {}
table_1, table_2 = '', ''
for key, val in zip(keys, vals):
if key is not None:
key = key.lower()
if val.startswith("From") and key == "suggested price":
continue
if key in ["product number", "our divisions"]:
continue
if "material" in key:
extra_data[key] = val
continue
if "color" in key:
extra_data[key] = val
continue
if "finish" in key:
extra_data[key] = val
continue
data[key] = val
for key, val in info_table.items():
if key is not None:
key = key.lower()
if 'material' in key:
extra_data[key] = val
if 'color' in key:
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):
try:
ts = create_technical_tables(tree)[0]
except Exception:
ts = ""
return {"Technical specifications": ts}
def second_table(tree):
try:
info = create_technical_tables(tree)[1]
except Exception:
info = ""
return {"Info": info}
def get_catalog_link(tree):
try:
catalog_link = tree.xpath(
"//h2[contains(., 'RELATED DOCUMENTS')]/following-sibling::div[contains(@id, 'carouselSegment')]//li[1]//a"
)[0].attrib['href']
if not catalog_link.startswith('https://'):
catalog_link = 'https://www.richelieu.com{}'.format(catalog_link)
except IndexError:
catalog_link = ''
return {"Catalog link": catalog_link}
def get_line_art_link(tree):
try:
line_art = tree.xpath(
"//h2[contains(., 'RELATED DOCUMENTS')]/following-sibling::"
"div[contains(@id, 'carouselSegment')]//li[last()]//a"
)[0].attrib['href']
if not line_art.startswith('https://'):
line_art = 'https://www.richelieu.com{}'.format(line_art)
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)
data = [
product_number,
product_name,
product_category,
product_description,
first_table_product,
second_table_product,
catalog_link,
line_art_link,
right_part_info,
product_images
]
return dict(ChainMap(*data))
def retry(link):
wait = randint(0, 10)
try:
return requests.get(link, headers={"User-Agent": random_user_agent()}).text
except Exception as e:
print('Retrying product page in {} seconds because: {}'.format(wait, e))
sleep(wait)
return retry(link)
def rearrange(header):
return sorted(header, key=lambda x: (x.startswith('image_') + (x != 'Product number'), x))
def get_categories():
categories = set()
html_ = retry(INDEX_PAGE)
page = html.fromstring(html_)
for letter in ascii_uppercase:
for link in page.xpath("//div[@id='index-{}']//li/a".format(letter)):
categories.add(link.attrib['href'])
for link in page.xpath("//div[@id='index-0-9']//li/a"):
categories.add(link.attrib['href'])
with open(CATEGORY_LINKS_FILE, 'w', encoding='utf-8') as f:
for category in categories:
f.write('{}?imgMode=m&sort=&nbPerPage=200'.format(category) + '\n')
def dig_up_products(url):
html_ = retry(url)
page = html.fromstring(html_)
for link in page.xpath('//h2[contains(., "CATEGORIES")]/following-sibling::*[@id="carouselSegment2b"]//li//a'):
yield from dig_up_products(link.attrib["href"])
for link in page.xpath('//ul[@id="prodResult"]/li//div[@class="imgWrapper"]/a'):
yield link.attrib["href"]
def main():
print('[# INFO #] START getting all product categories...')
get_categories()
print('[# INFO #] END getting all product categories...\n')
print('[# INFO #] START getting all product links...')
all_product_links = set()
with open(CATEGORY_LINKS_FILE) as in_file:
for start in in_file:
start = start.strip()
for link in dig_up_products(start):
all_product_links.add(link)
with open(PRODUCT_LINKS_FILE, 'a+') as out_file:
for pl in all_product_links:
if not any(line.strip() == pl for line in out_file):
out_file.write(pl + '\n')
print('[# INFO #] END getting all product links...\n')
print('[# INFO #] START getting all products info...')
fieldnames, data = set(), []
with open(PRODUCT_LINKS_FILE) as links:
i = 0
for link in links:
link = link.strip()
product_page = retry(link)
if product_page:
product_tree = html.fromstring(product_page)
more_products = check_if_more_products(product_tree)
if not more_products:
i += 1
data.append(prepare_product(product_tree))
print('Appended one more (if not more products): {}'.format(i))
else:
for product_link in more_products:
new_page = retry(product_link)
if new_page:
i += 1
new_product_tree = html.fromstring(new_page)
data.append(prepare_product(new_product_tree))
print('Appended one more (if new_page): {}'.format(i))
for dict_ in data:
fieldnames.update(dict_.keys())
print('[# INFO #] END getting all products info...\n')
print('[# INFO #] START writing products info to file...')
with open(RESULT_FILE, 'w', newline='', encoding='utf-8') as f:
writer = DictWriter(f, fieldnames=rearrange(fieldnames), delimiter=';', quoting=QUOTE_MINIMAL)
writer.writeheader()
for dict_ in data:
writer.writerow(dict_)
print('[# INFO #] END writing products info to file...\n\n')
if __name__ == '__main__':
main()
PS: for those of you who are going test it, here is the list of user-agents.
