- Description: This is a simple script for scraping Amazon and eBay category, sub-category and product URLs and saving contents to files. In case of previously saved files, the files will be read and no attempts to re-scrape contents will be executed.
- Documentation: you will find the most of the information needed in the docstrings.
- Private methods: methods defined are private
_method_name
because this is an initial part of the program that will feed future methods that I will add which would be getting product details and polishing data for applying some data analysis. I will be posting follow ups to this code when I'm done. _get_amazon_category_names_urls()
returns a list mapping, not a dictionary because sometimes there are duplicate titles with different links.- For reviewers: As this is my first scraping attempt, it might not be the best way of doing it so a whole general review is most welcome of course.
- Focus points:
- How to improve the program structure.
- How to decrease if not eliminate scraping failures(sometimes some random urls are not scraped and result in empty .txt files which will be handled by a defined method, however I want to eliminate this.)
- Is the defined
self.headers
doing the job (which I think is preventing the target website from blocking the connection after many requests) or is there a better to do it?
Code
Formatted version:
#!/usr/bin/env python3
from bs4 import BeautifulSoup
from time import perf_counter
import os
import requests
class WebScraper:
"""
A tool for scraping websites including:
- Amazon
- eBay
"""
def __init__(
self, website: str, target_url=None, path=None
):
"""
website: A string indication of a website:
- 'ebay'
- 'Amazon'
target_url: A string containing a single url to scrape.
"""
self.supported_websites = ["Amazon", "ebay"]
if website not in self.supported_websites:
raise ValueError(
f"Website {website} not supported."
)
self.website = website
self.target_url = target_url
if not path:
self.path = (
"/Users/user_name/Desktop/code/web scraper/"
)
if path:
self.path = path
self.headers = {
"User-Agent": "Safari/5.0 (Windows NT 6.3; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/54.0.2840.71 Safari/537.36"
}
self.amazon_modes = {
"bs": "Best Sellers",
"nr": "New Releases",
"gi": "Gift Ideas",
"ms": "Movers and Shakers",
"mw": "Most Wished For",
}
def _cache_category_urls(
self,
text_file_names: dict,
section: str,
category_class: str,
website: str,
content_path: str,
categories: list,
print_progress=False,
cleanup_empty=True,
):
"""
Write scraped category/sub_category urls to files.
text_file_names: a dictionary containing .txt file names to save data under.
section: a string indicating section to scrape.
For Amazon:
check self.amazon_modes.
category_class: a string 'categories' or 'sub_categories'.
website: a string 'ebay' or 'Amazon'.
content_path: a string containing path to folder for saving URLs.
categories: a list containing category or sub_category urls to be saved.
print_progress: if True, progress will be displayed.
cleanup_empty: if True, after writing the .txt files, empty files(failures) will be deleted.
"""
os.chdir(content_path + website + "/")
with open(
text_file_names[section][category_class], "w"
) as cats:
for category in categories:
cats.write(category + "\n")
if print_progress:
if open(
text_file_names[section][
category_class
],
"r",
).read(1):
print(
f"Saving {category} ... done."
)
else:
print(
f"Saving {category} ... failure."
)
if cleanup_empty:
self._cleanup_empty_files(
self.path + website + "/"
)
def _read_urls(
self,
text_file_names: dict,
section: str,
category_class: str,
content_path: str,
website: str,
cleanup_empty=True,
):
"""
Read saved urls from a file and return a sorted list containing the urls.
text_file_names: a dictionary containing .txt file names to save data under.
section: a string indicating section to scrape.
For Amazon:
check self.amazon_modes.
category_class: a string 'categories' or 'sub_categories'.
website: a string 'ebay' or 'Amazon'.
content_path: a string containing path to folder for saving URLs.
print_progress: if True, progress will be displayed.
cleanup_empty: if True, if any empty files found during the execution, the files will be deleted.
"""
os.chdir(content_path + website + "/")
if text_file_names[section][
category_class
] in os.listdir(content_path + "Amazon/"):
with open(
text_file_names[section][category_class]
) as cats:
if cleanup_empty:
self._cleanup_empty_files(
self.path + website + "/"
)
return [
link.rstrip()
for link in cats.readlines()
]
def _scrape_urls(
self,
starting_target_urls: dict,
section: str,
category_class: str,
prev_categories=None,
print_progress=False,
):
"""
Scrape urls of a category class and return a list of URLs.
starting_target_urls: a dictionary containing the initial websites that will start the web crawling.
section: a string indicating section to scrape.
For Amazon:
check self.amazon_modes.
category_class: a string 'categories' or 'sub_categories'.
prev_categories: if sub_categories are scraped, prev_categories a list of category urls.
print_progress: if True, progress will be displayed.
"""
target_url = starting_target_urls[section][1]
if category_class == "categories":
starting_url = requests.get(
starting_target_urls[section][0],
headers=self.headers,
)
html_content = BeautifulSoup(
starting_url.text, features="lxml"
)
target_url_part = starting_target_urls[section][
1
]
if not print_progress:
return sorted(
{
str(link.get("href"))
for link in html_content.findAll(
"a"
)
if target_url_part in str(link)
}
)
if print_progress:
categories = set()
for link in html_content.findAll("a"):
if target_url_part in str(link):
link_to_add = str(link.get("href"))
categories.add(link_to_add)
print(
f"Fetched {section}-{category_class}: {link_to_add}"
)
return categories
if category_class == "sub_categories":
if not print_progress:
responses = [
requests.get(
category, headers=self.headers
)
for category in prev_categories
]
category_soups = [
BeautifulSoup(
response.text, features="lxml"
)
for response in responses
]
pre_sub_category_links = [
str(link.get("href"))
for category in category_soups
for link in category.findAll("a")
if target_url in str(link)
]
return sorted(
{
link
for link in pre_sub_category_links
if link not in prev_categories
}
)
if print_progress:
responses, pre, sub_categories = (
[],
[],
set(),
)
for category in prev_categories:
response = requests.get(
category, headers=self.headers
)
responses.append(response)
print(
f"Got response {response} for {section}-{category}"
)
category_soups = [
BeautifulSoup(
response.text, features="lxml"
)
for response in responses
]
for soup in category_soups:
for link in soup.findAll("a"):
if target_url in str(link):
fetched_link = str(
link.get("href")
)
pre.append(fetched_link)
print(
f"Fetched {section}-{fetched_link}"
)
return sorted(
{
link
for link in pre
if link not in prev_categories
}
)
def _get_amazon_category_urls(
self,
section: str,
subs=True,
cache_urls=True,
print_progress=False,
cleanup_empty=True,
):
"""
Return a list containing Amazon category and sub-category(optional) urls, if previously cached, the files
will be read and required data will be returned, otherwise, required data will be scraped.
section: a string indicating section to scrape.
For Amazon:
check self.amazon_modes.
subs: if subs, category and sub-category urls will be returned.
cache_urls: if cache_urls and content not previously cached, the content will be saved to .txt files.
print_progress: if True, progress will be displayed.
cleanup_empty: if True, if any empty files are left after the execution, will be deleted..
"""
starting_target_urls = {
"bs": (
"https://www.amazon.com/gp/bestsellers/",
"https://www.amazon.com/Best-Sellers",
),
"nr": (
"https://www.amazon.com/gp/new-releases/",
"https://www.amazon.com/gp/new-releases/",
),
"ms": (
"https://www.amazon.com/gp/movers-and-shakers/",
"https://www.amazon.com/gp/movers-and-shakers/",
),
"gi": (
"https://www.amazon.com/gp/most-gifted/",
"https://www.amazon.com/gp/most-gifted",
),
"mw": (
"https://www.amazon.com/gp/most-wished-for/",
"https://www.amazon.com/gp/most-wished-for/",
),
}
text_file_names = {
"bs": {
"categories": "bs_categories.txt",
"sub_categories": "bs_sub_categories.txt",
},
"nr": {
"categories": "nr_categories.txt",
"sub_categories": "nr_sub_categories.txt",
},
"ms": {
"categories": "ms_categories.txt",
"sub_categories": "ms_sub_categories.txt",
},
"gi": {
"categories": "gi_categories.txt",
"sub_categories": "gi_sub_categories.txt",
},
"mw": {
"categories": "mw_categories.txt",
"sub_categories": "mw_sub_categories.txt",
},
}
if self.website != "Amazon":
raise ValueError(
f"Cannot fetch Amazon data from {self.website}"
)
if section not in text_file_names:
raise ValueError(f"Invalid section {section}")
os.chdir(self.path)
if "Amazon" not in os.listdir(self.path):
os.mkdir("Amazon")
os.chdir("Amazon")
if "Amazon" in os.listdir(self.path):
categories = self._read_urls(
text_file_names,
section,
"categories",
self.path,
"Amazon",
cleanup_empty=cleanup_empty,
)
if not subs:
if cleanup_empty:
self._cleanup_empty_files(
self.path + "Amazon/"
)
return sorted(categories)
sub_categories = self._read_urls(
text_file_names,
section,
"sub_categories",
self.path,
"Amazon",
cleanup_empty=cleanup_empty,
)
try:
if categories and sub_categories:
if cleanup_empty:
self._cleanup_empty_files(
self.path + "Amazon/"
)
return (
sorted(categories),
sorted(sub_categories),
)
except UnboundLocalError:
pass
if not subs:
categories = self._scrape_urls(
starting_target_urls,
section,
"categories",
print_progress=print_progress,
)
if cache_urls:
self._cache_category_urls(
text_file_names,
section,
"categories",
"Amazon",
self.path,
categories,
print_progress=print_progress,
cleanup_empty=cleanup_empty,
)
if cleanup_empty:
self._cleanup_empty_files(
self.path + "Amazon/"
)
return sorted(categories)
if subs:
categories = self._scrape_urls(
starting_target_urls,
section,
"categories",
print_progress=print_progress,
)
if cache_urls:
self._cache_category_urls(
text_file_names,
section,
"categories",
"Amazon",
self.path,
categories,
print_progress=print_progress,
)
sub_categories = self._scrape_urls(
starting_target_urls,
section,
"sub_categories",
categories,
print_progress=print_progress,
)
if cache_urls:
self._cache_category_urls(
text_file_names,
section,
"sub_categories",
"Amazon",
self.path,
sub_categories,
print_progress=print_progress,
cleanup_empty=cleanup_empty,
)
if cleanup_empty:
self._cleanup_empty_files(
self.path + "Amazon/"
)
return (
sorted(categories),
sorted(sub_categories),
)
def _get_ebay_urls(
self, cache_urls=True, cleanup_empty=True
):
"""
Return a sorted list containing ebay category and sub-category URLs if previously cached, the files
will be read and required data will be returned, otherwise, required data will be scraped.
cache_urls: if cache_urls and content not previously cached, the content will be saved to .txt files.
cleanup_empty: if True, if any empty files are left after the execution, will be deleted.
"""
if self.website != "ebay":
raise ValueError(
f"Cannot fetch ebay data from {self.website}"
)
target_url = "https://www.ebay.com/b/"
if "ebay" not in os.listdir(self.path):
os.mkdir("ebay")
os.chdir("ebay")
if "ebay" in os.listdir(self.path):
os.chdir(self.path + "ebay/")
if "categories.txt" in os.listdir(
self.path + "ebay/"
):
with open("categories.txt") as cats:
categories = [
link.rstrip()
for link in cats.readlines()
]
if cleanup_empty:
self._cleanup_empty_files(
self.path + "ebay/"
)
return categories
initial_html = requests.get(
"https://www.ebay.com/n/all-categories",
self.headers,
)
initial_soup = BeautifulSoup(
initial_html.text, features="lxml"
)
categories = sorted(
{
str(link.get("href"))
for link in initial_soup.findAll("a")
if target_url in str(link)
}
)
if cache_urls:
with open("categories.txt", "w") as cats:
for category in categories:
cats.write(category + "\n")
if cleanup_empty:
self._cleanup_empty_files(self.path + "ebay/")
return categories
def _get_amazon_page_product_urls(
self, page_url: str, print_progress=False
):
"""
Return a sorted list of links to all products found on a single Amazon page.
page_url: a string containing target Amazon_url.
"""
prefix = "https://www.amazon.com"
page_response = requests.get(
page_url, headers=self.headers
)
page_soup = BeautifulSoup(
page_response.text, features="lxml"
)
if not print_progress:
return sorted(
{
prefix + str(link.get("href"))
for link in page_soup.findAll("a")
if "psc=" in str(link)
}
)
if print_progress:
links = set()
for link in page_soup.findAll("a"):
if "psc=" in str(link):
link_to_get = prefix + str(
link.get("href")
)
links.add(link_to_get)
print(f"Got link {link_to_get}")
return sorted(links)
@staticmethod
def _cleanup_empty_files(dir_path):
"""
Delete empty cached files in a given folder.
dir_path: a string containing path to target directory.
"""
try:
for file_name in os.listdir(dir_path):
if not os.path.isdir(dir_path + file_name):
if not open(file_name).read(1):
os.remove(file_name)
except UnicodeDecodeError:
pass
def _get_amazon_category_names_urls(
self,
section: str,
category_class: str,
print_progress=False,
cache_contents=True,
delimiter="&&&",
cleanup_empty=True,
):
"""
Return a list of pairs [category name, url] if previously cached, the files
will be read and required data will be returned, otherwise, required data will be scraped.
section: a string indicating section to scrape.
For Amazon:
check self.amazon_modes.
category_class: a string 'categories' or 'sub_categories'.
print_progress: if True, progress will be displayed.
cache_contents: if data is not previously cached, category names mapped to their urls will be saved to .txt.
delimiter: delimits category name and the respective url in the .txt cached file.
cleanup_empty: if True, if any empty files are left after the execution, will be deleted.
"""
file_names = {
"categories": section + "_category_names.txt",
"sub_categories": section
+ "_sub_category_names.txt",
}
names_urls = []
os.chdir(self.path)
if "Amazon" in os.listdir(self.path):
os.chdir("Amazon")
file_name = file_names[category_class]
if file_name in os.listdir(
self.path + "Amazon"
):
with open(file_name) as names:
if cleanup_empty:
self._cleanup_empty_files(
self.path + "Amazon/"
)
return [
line.rstrip().split(delimiter)
for line in names.readlines()
]
if "Amazon" not in os.listdir(self.path):
os.mkdir("Amazon")
os.chdir("Amazon")
categories, sub_categories = self._get_amazon_category_urls(
section,
cache_urls=cache_contents,
print_progress=print_progress,
cleanup_empty=cleanup_empty,
)
for category in eval("eval(category_class)"):
category_response = requests.get(
category, headers=self.headers
)
category_html = BeautifulSoup(
category_response.text, features="lxml"
)
try:
category_name = category_html.h1.span.text
names_urls.append((category_name, category))
if cache_contents:
with open(
file_names[category_class], "w"
) as names:
names.write(
category_name
+ delimiter
+ category
+ "\n"
)
if print_progress:
if open(
file_names[category_class], "r"
).read(1):
print(
f"{section}-{category_class[:-3]}y: {category_name} ... done."
)
else:
print(
f"{section}-{category_class[:-3]}y: {category_name} ... failure."
)
except AttributeError:
pass
if cleanup_empty:
self._cleanup_empty_files(self.path + "Amazon/")
return names_urls
def _get_amazon_section_product_urls(
self,
section: str,
category_class: str,
print_progress=False,
cache_contents=True,
cleanup_empty=True,
read_only=False,
):
"""
Return links to all products within all categories available in an Amazon section(check self.amazon_modes).
section: a string indicating section to scrape. If previously cached, the files
will be read and required data will be returned, otherwise, required data will be scraped..
For Amazon:
check self.amazon_modes.
category_class: a string 'categories' or 'sub_categories'.
print_progress: if True, progress will be displayed.
cache_contents: if data is not previously cached, category names mapped to their urls will be saved to .txt.
cleanup_empty: if True, if any empty files are left after the execution, will be deleted.
read_only: if files are previously cached and only cached contents are required(no scraping attempts in case of
missing category/sub-category urls).
"""
all_products = []
names_urls = self._get_amazon_category_names_urls(
section,
category_class,
print_progress,
cache_contents,
cleanup_empty=cleanup_empty,
)
folder_name = " ".join(
[
self.amazon_modes[section],
category_class.title(),
"Product URLs",
]
)
if cache_contents:
if folder_name not in os.listdir(
self.path + "Amazon/"
):
os.mkdir(folder_name)
os.chdir(folder_name)
for category_name, category_url in names_urls:
if print_progress:
print(
f"Processing category {category_name} ..."
)
file_name = "-".join(
[
self.amazon_modes[section],
category_class,
category_name,
]
)
if file_name + ".txt" in os.listdir(
self.path + "Amazon/" + folder_name + "/"
):
with open(
file_name + ".txt"
) as product_urls:
urls = [
line.rstrip()
for line in product_urls.readlines()
]
all_products.append(
(category_name, urls)
)
else:
if not read_only:
urls = self._get_amazon_page_product_urls(
category_url, print_progress
)
all_products.append(
(category_name, urls)
)
if cache_contents:
with open(
file_name + ".txt", "w"
) as current:
for url in urls:
current.write(url + "\n")
if print_progress:
print(f"Saving {url}")
if print_progress:
try:
if open(file_name + ".txt").read(1):
print(
f"Category {category_name} ... done."
)
else:
print(
f"Category {category_name} ... failure."
)
except FileNotFoundError:
if print_progress:
print(
f"Category {category_name} ... failure."
)
pass
if cleanup_empty:
self._cleanup_empty_files(
self.path + "Amazon/" + folder_name
)
return all_products
if __name__ == "__main__":
start_time = perf_counter()
path_to_folder = input(
"Enter path for content saving: "
).rstrip()
abc = WebScraper("Amazon", path=path_to_folder)
print(
abc._get_amazon_section_product_urls(
"bs", "categories", print_progress=True
)
)
end_time = perf_counter()
print(f"Time: {end_time - start_time} seconds.")
If you prefer the non-formatted version:
#!/usr/bin/env python3
from bs4 import BeautifulSoup
from time import perf_counter
import os
import requests
class WebScraper:
"""
A tool for scraping websites including:
- Amazon
- eBay
"""
def __init__(self, website: str, target_url=None, path=None):
"""
website: A string indication of a website:
- 'ebay'
- 'Amazon'
target_url: A string containing a single url to scrape.
"""
self.supported_websites = ['Amazon', 'ebay']
if website not in self.supported_websites:
raise ValueError(f'Website {website} not supported.')
self.website = website
self.target_url = target_url
if not path:
self.path = '/Users/user_name/Desktop/web scraper/'
if path:
self.path = path
self.headers = {'User-Agent': 'Safari/5.0 (Windows NT 6.3; Win64; x64) AppleWebKit/537.36 '
'(KHTML, like Gecko) Chrome/54.0.2840.71 Safari/537.36'}
self.amazon_modes = {'bs': 'Best Sellers', 'nr': 'New Releases', 'gi': 'Gift Ideas',
'ms': 'Movers and Shakers', 'mw': 'Most Wished For'}
def _cache_category_urls(self, text_file_names: dict, section: str, category_class: str, website: str,
content_path: str, categories: list, print_progress=False, cleanup_empty=True):
"""
Write scraped category/sub_category urls to file.
text_file_names: a dictionary containing .txt file names to save data under.
section: a string indicating section to scrape.
For Amazon:
check self.amazon_modes.
category_class: a string 'categories' or 'sub_categories'.
website: a string 'ebay' or 'Amazon'.
content_path: a string containing path to folder for saving URLs.
categories: a list containing category or sub_category urls to be saved.
print_progress: if True, progress will be displayed.
cleanup_empty: if True, after writing the .txt files, empty files(failures) will be deleted.
"""
os.chdir(content_path + website + '/')
with open(text_file_names[section][category_class], 'w') as cats:
for category in categories:
cats.write(category + '\n')
if print_progress:
if open(text_file_names[section][category_class], 'r').read(1):
print(f'Saving {category} ... done.')
else:
print(f'Saving {category} ... failure.')
if cleanup_empty:
self._cleanup_empty_files(self.path + website + '/')
def _read_urls(self, text_file_names: dict, section: str, category_class: str, content_path: str, website: str,
cleanup_empty=True):
"""
Read saved urls from a file and return a sorted list containing the urls.
text_file_names: a dictionary containing .txt file names to save data under.
section: a string indicating section to scrape.
For Amazon:
check self.amazon_modes.
category_class: a string 'categories' or 'sub_categories'.
website: a string 'ebay' or 'Amazon'.
content_path: a string containing path to folder for saving URLs.
print_progress: if True, progress will be displayed.
cleanup_empty: if True, if any empty files found during the execution, the files will be deleted.
"""
os.chdir(content_path + website + '/')
if text_file_names[section][category_class] in os.listdir(content_path + 'Amazon/'):
with open(text_file_names[section][category_class]) as cats:
if cleanup_empty:
self._cleanup_empty_files(self.path + website + '/')
return [link.rstrip() for link in cats.readlines()]
def _scrape_urls(self, starting_target_urls: dict, section: str, category_class: str, prev_categories=None,
print_progress=False):
"""
Scrape urls of a category class and return a list of URLs.
starting_target_urls: a dictionary containing the initial websites that will start the web crawling.
section: a string indicating section to scrape.
For Amazon:
check self.amazon_modes.
category_class: a string 'categories' or 'sub_categories'.
prev_categories: if sub_categories are scraped, prev_categories a list of category urls.
print_progress: if True, progress will be displayed.
"""
target_url = starting_target_urls[section][1]
if category_class == 'categories':
starting_url = requests.get(starting_target_urls[section][0], headers=self.headers)
html_content = BeautifulSoup(starting_url.text, features='lxml')
target_url_part = starting_target_urls[section][1]
if not print_progress:
return sorted({str(link.get('href')) for link in html_content.findAll('a')
if target_url_part in str(link)})
if print_progress:
categories = set()
for link in html_content.findAll('a'):
if target_url_part in str(link):
link_to_add = str(link.get('href'))
categories.add(link_to_add)
print(f'Fetched {section}-{category_class}: {link_to_add}')
return categories
if category_class == 'sub_categories':
if not print_progress:
responses = [requests.get(category, headers=self.headers) for category in prev_categories]
category_soups = [BeautifulSoup(response.text, features='lxml') for response in responses]
pre_sub_category_links = [str(link.get('href')) for category in category_soups
for link in category.findAll('a') if target_url in str(link)]
return sorted({link for link in pre_sub_category_links if link not in prev_categories})
if print_progress:
responses, pre, sub_categories = [], [], set()
for category in prev_categories:
response = requests.get(category, headers=self.headers)
responses.append(response)
print(f'Got response {response} for {section}-{category}')
category_soups = [BeautifulSoup(response.text, features='lxml') for response in responses]
for soup in category_soups:
for link in soup.findAll('a'):
if target_url in str(link):
fetched_link = str(link.get('href'))
pre.append(fetched_link)
print(f'Fetched {section}-{fetched_link}')
return sorted({link for link in pre if link not in prev_categories})
def _get_amazon_category_urls(self, section: str, subs=True, cache_urls=True, print_progress=False,
cleanup_empty=True):
"""
Return a list containing Amazon category and sub-category(optional) urls, if previously cached, the files
will be read and required data will be returned, otherwise, required data will be scraped.
section: a string indicating section to scrape.
For Amazon:
check self.amazon_modes.
subs: if subs, category and sub-category urls will be returned.
cache_urls: if cache_urls and content not previously cached, the content will be saved to .txt files.
print_progress: if True, progress will be displayed.
cleanup_empty: if True, if any empty files are left after the execution, will be deleted..
"""
starting_target_urls = {'bs': ('https://www.amazon.com/gp/bestsellers/',
'https://www.amazon.com/Best-Sellers'),
'nr': ('https://www.amazon.com/gp/new-releases/',
'https://www.amazon.com/gp/new-releases/'),
'ms': ('https://www.amazon.com/gp/movers-and-shakers/',
'https://www.amazon.com/gp/movers-and-shakers/'),
'gi': ('https://www.amazon.com/gp/most-gifted/',
'https://www.amazon.com/gp/most-gifted'),
'mw': ('https://www.amazon.com/gp/most-wished-for/',
'https://www.amazon.com/gp/most-wished-for/')}
text_file_names = {'bs': {'categories': 'bs_categories.txt', 'sub_categories': 'bs_sub_categories.txt'},
'nr': {'categories': 'nr_categories.txt', 'sub_categories': 'nr_sub_categories.txt'},
'ms': {'categories': 'ms_categories.txt', 'sub_categories': 'ms_sub_categories.txt'},
'gi': {'categories': 'gi_categories.txt', 'sub_categories': 'gi_sub_categories.txt'},
'mw': {'categories': 'mw_categories.txt', 'sub_categories': 'mw_sub_categories.txt'}}
if self.website != 'Amazon':
raise ValueError(f'Cannot fetch Amazon data from {self.website}')
if section not in text_file_names:
raise ValueError(f'Invalid section {section}')
os.chdir(self.path)
if 'Amazon' not in os.listdir(self.path):
os.mkdir('Amazon')
os.chdir('Amazon')
if 'Amazon' in os.listdir(self.path):
categories = self._read_urls(text_file_names, section, 'categories', self.path, 'Amazon',
cleanup_empty=cleanup_empty)
if not subs:
if cleanup_empty:
self._cleanup_empty_files(self.path + 'Amazon/')
return sorted(categories)
sub_categories = self._read_urls(text_file_names, section, 'sub_categories', self.path, 'Amazon',
cleanup_empty=cleanup_empty)
try:
if categories and sub_categories:
if cleanup_empty:
self._cleanup_empty_files(self.path + 'Amazon/')
return sorted(categories), sorted(sub_categories)
except UnboundLocalError:
pass
if not subs:
categories = self._scrape_urls(starting_target_urls, section, 'categories', print_progress=print_progress)
if cache_urls:
self._cache_category_urls(text_file_names, section, 'categories', 'Amazon', self.path, categories,
print_progress=print_progress, cleanup_empty=cleanup_empty)
if cleanup_empty:
self._cleanup_empty_files(self.path + 'Amazon/')
return sorted(categories)
if subs:
categories = self._scrape_urls(starting_target_urls, section, 'categories', print_progress=print_progress)
if cache_urls:
self._cache_category_urls(text_file_names, section, 'categories', 'Amazon', self.path, categories,
print_progress=print_progress)
sub_categories = self._scrape_urls(starting_target_urls, section, 'sub_categories', categories,
print_progress=print_progress)
if cache_urls:
self._cache_category_urls(text_file_names, section, 'sub_categories', 'Amazon', self.path,
sub_categories, print_progress=print_progress, cleanup_empty=cleanup_empty)
if cleanup_empty:
self._cleanup_empty_files(self.path + 'Amazon/')
return sorted(categories), sorted(sub_categories)
def _get_ebay_urls(self, cache_urls=True, cleanup_empty=True):
"""
Return a sorted list containing ebay category and sub-category URLs if previously cached, the files
will be read and required data will be returned, otherwise, required data will be scraped.
cache_urls: if cache_urls and content not previously cached, the content will be saved to .txt files.
cleanup_empty: if True, if any empty files are left after the execution, will be deleted.
"""
if self.website != 'ebay':
raise ValueError(f'Cannot fetch ebay data from {self.website}')
target_url = 'https://www.ebay.com/b/'
if 'ebay' not in os.listdir(self.path):
os.mkdir('ebay')
os.chdir('ebay')
if 'ebay' in os.listdir(self.path):
os.chdir(self.path + 'ebay/')
if 'categories.txt' in os.listdir(self.path + 'ebay/'):
with open('categories.txt') as cats:
categories = [link.rstrip() for link in cats.readlines()]
if cleanup_empty:
self._cleanup_empty_files(self.path + 'ebay/')
return categories
initial_html = requests.get('https://www.ebay.com/n/all-categories', self.headers)
initial_soup = BeautifulSoup(initial_html.text, features='lxml')
categories = sorted({str(link.get('href')) for link in initial_soup.findAll('a') if target_url in str(link)})
if cache_urls:
with open('categories.txt', 'w') as cats:
for category in categories:
cats.write(category + '\n')
if cleanup_empty:
self._cleanup_empty_files(self.path + 'ebay/')
return categories
def _get_amazon_page_product_urls(self, page_url: str, print_progress=False):
"""
Return a sorted list of links to all products found on a single Amazon page.
page_url: a string containing target Amazon_url.
"""
prefix = 'https://www.amazon.com'
page_response = requests.get(page_url, headers=self.headers)
page_soup = BeautifulSoup(page_response.text, features='lxml')
if not print_progress:
return sorted({prefix + str(link.get('href')) for link in page_soup.findAll('a') if 'psc=' in str(link)})
if print_progress:
links = set()
for link in page_soup.findAll('a'):
if 'psc=' in str(link):
link_to_get = prefix + str(link.get('href'))
links.add(link_to_get)
print(f'Got link {link_to_get}')
return sorted(links)
@staticmethod
def _cleanup_empty_files(dir_path):
"""
Delete empty cached files in a given folder.
dir_path: a string containing path to target directory.
"""
try:
for file_name in os.listdir(dir_path):
if not os.path.isdir(dir_path + file_name):
if not open(file_name).read(1):
os.remove(file_name)
except UnicodeDecodeError:
pass
def _get_amazon_category_names_urls(self, section: str, category_class: str, print_progress=False,
cache_contents=True, delimiter='&&&', cleanup_empty=True):
"""
Return a list of pairs [category name, url] if previously cached, the files
will be read and required data will be returned, otherwise, required data will be scraped.
section: a string indicating section to scrape.
For Amazon:
check self.amazon_modes.
category_class: a string 'categories' or 'sub_categories'.
print_progress: if True, progress will be displayed.
cache_contents: if data is not previously cached, category names mapped to their urls will be saved to .txt.
delimiter: delimits category name and the respective url in the .txt cached file.
cleanup_empty: if True, if any empty files are left after the execution, will be deleted.
"""
file_names = {'categories': section + '_category_names.txt',
'sub_categories': section + '_sub_category_names.txt'}
names_urls = []
os.chdir(self.path)
if 'Amazon' in os.listdir(self.path):
os.chdir('Amazon')
file_name = file_names[category_class]
if file_name in os.listdir(self.path + 'Amazon'):
with open(file_name) as names:
if cleanup_empty:
self._cleanup_empty_files(self.path + 'Amazon/')
return [line.rstrip().split(delimiter) for line in names.readlines()]
if 'Amazon' not in os.listdir(self.path):
os.mkdir('Amazon')
os.chdir('Amazon')
categories, sub_categories = self._get_amazon_category_urls(section, cache_urls=cache_contents,
print_progress=print_progress,
cleanup_empty=cleanup_empty)
for category in eval('eval(category_class)'):
category_response = requests.get(category, headers=self.headers)
category_html = BeautifulSoup(category_response.text, features='lxml')
try:
category_name = category_html.h1.span.text
names_urls.append((category_name, category))
if cache_contents:
with open(file_names[category_class], 'w') as names:
names.write(category_name + delimiter + category + '\n')
if print_progress:
if open(file_names[category_class], 'r').read(1):
print(f'{section}-{category_class[:-3]}y: {category_name} ... done.')
else:
print(f'{section}-{category_class[:-3]}y: {category_name} ... failure.')
except AttributeError:
pass
if cleanup_empty:
self._cleanup_empty_files(self.path + 'Amazon/')
return names_urls
def _get_amazon_section_product_urls(self, section: str, category_class: str, print_progress=False,
cache_contents=True, cleanup_empty=True, read_only=False):
"""
Return links to all products within all categories available in an Amazon section(check self.amazon_modes).
section: a string indicating section to scrape. If previously cached, the files
will be read and required data will be returned, otherwise, required data will be scraped..
For Amazon:
check self.amazon_modes.
category_class: a string 'categories' or 'sub_categories'.
print_progress: if True, progress will be displayed.
cache_contents: if data is not previously cached, category names mapped to their urls will be saved to .txt.
cleanup_empty: if True, if any empty files are left after the execution, will be deleted.
read_only: if files are previously cached and only cached contents are required(no scraping attempts in case of
missing category/sub-category urls).
"""
all_products = []
names_urls = self._get_amazon_category_names_urls(section, category_class, print_progress, cache_contents,
cleanup_empty=cleanup_empty)
folder_name = ' '.join([self.amazon_modes[section], category_class.title(),
'Product URLs'])
if cache_contents:
if folder_name not in os.listdir(self.path + 'Amazon/'):
os.mkdir(folder_name)
os.chdir(folder_name)
for category_name, category_url in names_urls:
if print_progress:
print(f'Processing category {category_name} ...')
file_name = '-'.join([self.amazon_modes[section], category_class, category_name])
if file_name + '.txt' in os.listdir(self.path + 'Amazon/' + folder_name + '/'):
with open(file_name + '.txt') as product_urls:
urls = [line.rstrip() for line in product_urls.readlines()]
all_products.append((category_name, urls))
else:
if not read_only:
urls = self._get_amazon_page_product_urls(category_url, print_progress)
all_products.append((category_name, urls))
if cache_contents:
with open(file_name + '.txt', 'w') as current:
for url in urls:
current.write(url + '\n')
if print_progress:
print(f'Saving {url}')
if print_progress:
try:
if open(file_name + '.txt').read(1):
print(f'Category {category_name} ... done.')
else:
print(f'Category {category_name} ... failure.')
except FileNotFoundError:
if print_progress:
print(f'Category {category_name} ... failure.')
pass
if cleanup_empty:
self._cleanup_empty_files(self.path + 'Amazon/' + folder_name)
return all_products
if __name__ == '__main__':
start_time = perf_counter()
path_to_folder = input('Enter path for content saving: ').rstrip()
abc = WebScraper('Amazon', path=path_to_folder)
print(abc._get_amazon_section_product_urls('bs', 'categories', print_progress=True))
end_time = perf_counter()
print(f'Time: {end_time - start_time} seconds.')