6
\$\begingroup\$
  • 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.')
\$\endgroup\$
5
  • \$\begingroup\$ Is there anyone interested in reviewing this? :) \$\endgroup\$ – bullseye Oct 16 '19 at 12:55
  • 2
    \$\begingroup\$ There’s a lot of code. Be patient 🙃 \$\endgroup\$ – Grajdeanu Alex Oct 16 '19 at 16:06
  • \$\begingroup\$ @GrajdeanuAlex. yeah I'm sorry for that. \$\endgroup\$ – bullseye Oct 16 '19 at 16:22
  • \$\begingroup\$ I'm interested in reviewing that, I've had a quick look already, but I haven't had time to go further yet ;) \$\endgroup\$ – Cyril D. Oct 18 '19 at 7:05
  • \$\begingroup\$ @Cyril D people around here lose interest quickly over long codes \$\endgroup\$ – bullseye Oct 18 '19 at 10:38
2
\$\begingroup\$

As it's a large chunck of code for this format, I'm sitcking to reviewing _get_amazon_section_product_urls but what I mention here can be applied elsewhere. If you choose to reply with an updated version, I could then look at the remaining.

Code Style

Overall, there is a good job of trying to make the code readable, and I give you bonus points for using type hinting. However, a docstring start on the same line as the brackets, with a short sentence explaining the function, then a blank line, then a paragraph (then I put args and return. I like numpy's style):

# Clear enough, no need for a docstring:
def randint():
  return 4  # Chosen by fair dice roll

# A single line is sufficiently explanatory:
def _cleanup_empty_files(dir_path):
        """Delete empty cached files in a given folder.

        dir_path: a string containing path to target directory.
        """

Moreover, if you use type hinting, I think it's okay to omit the type specification in the docstring:

def _get_amazon_category_names_urls(self, section: str, category_class: str, print_progress=False,
                                    cache_contents=True, delimiter='&&&', cleanup_empty=True):
    """Get  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: specify section to scrape.
        Check `self.amazon_modes` when `amazon` is specified.
    category_class: '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, delete any empty files left once done.
    """

I also find that the class' docstring is not so helpful. I gather it was written when getting started, but should be revisited. It does not help to know what's being scrapped, why, and whether it's scraping search results, or the latest offers, or their css, or...

I'm explicitly not talking about line length, as the official recommendation is 80, I've worked with codebase going to a 100, and here the length seems to go up to 120. Is that what you want? Ok.

The mixing of functionalities

I find worriesome that certain actions are mixed up:

if print_progress:
    try:
        if open(file_name + '.txt').read(1):  # This file is never closed!
            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  # Why is there a pass?

Opening the file is a different action from printing. I'd do:

msg = 'failure'  # Assume the worst in the default state
try:
    with open(f'{file_name}.txt') as fin:
        if fin.read(1):
            msg = 'done'
except FileNotFoundError:
    pass
if print_progress:
    print(f'Category {category_name}: {msg}.')

Perhaps, instead of printing, you could look into logging. You could then set the logging level, avoid these ifs all over the place, and the print_progress argument become irrelevant.

Now that these two operations are decoupled, we can think of integrating the logging with the operation, and remove this test, as what it effectively does it check that the user had the rights to modify this file, where an error would have been thrown earlier. There also a code-path which is not tested: what happens if the files have never been cached, and use_cached_content is False? Then the test returns false, even though function returns an empty list (a I think it should?)

The following block is prone to errors:

if cache_contents:
    if folder_name not in os.listdir(self.path + 'Amazon/'):
        os.mkdir(folder_name)
os.chdir(folder_name)

This is because the path we're really working with is os.path.join(self.path, 'Amazon', folder_name), but then we're changing directory to folder_name only.

path = os.path.join(self.path, 'Amazon', folder_name)
if cache_content:
    if not os.path.exists(path):
        os.mkdirs(path)
os.chdir(path)

I myself am not keen on changing paths all the time, because I find it hard to keep track of where I am. That's why I'd rather build the full path, create directories and work from where I am:

with open(os.path.join(path, filename)):
   ...

Naming

I'm more of a linux guy, and I'm uncomfortable with whitespaces in directory names, as found in _get_amazon_category_names_urls. I think that read_only is not as telling as cached_only or use_cache or use_cached_content. The operation file_name + '.txt' is done is several places. The extention is effectively part of the filename, consider concatenating it once.

Aerate the code

Since it's a relatively large function, which has several operations and code-paths, I would add blank lines to separate the various blocks in to logical units.

With all this, that function looks like that:

        def _get_amazon_section_product_urls(self, section: str, category_class: str, cache_contents=True,
                                         cleanup_empty=True, read_only=False):
        """Get links to all products within all categories available in an Amazon section (as defined in
         self.amazon_modes).

        section: the amazon category to scrape. If previously cached, the files will be read and required
            data will be returned, otherwise, required data will be scraped.
        category_class: 'categories' or 'sub_categories'.
        cache_content: if the data was not previously cached, category names mapped to their urls will 
            be saved to a text file.
        cleanup_empty: if True, delete any empty files left once done.
        use_cached_content: only use previously cached contents. (no scraping 
            attempts in case of missing category/sub-category urls). Returns an empty list if not cache exists.
        """
        all_products = []
        names_urls = self._get_amazon_category_names_urls(section, category_class, print_progress, 
                                                          cache_contents, cleanup_empty=cleanup_empty)

        path = ' '.join([self.amazon_modes[section], category_class.title(), 'Product URLs'])

        if cache_content:
            if not os.path.exists(path):
                os.mkdir(path)

        for category_name, category_url in names_urls:
            logger.info(f'Processing category {category_name} ...')
            msg = 'done'

            filename = '-'.join([self.amazon_modes[section], category_class, category_name])
            filename += '.txt'
            filepath = os.path.join(path, filename)    

            if use_cached_content:
                try:
                    with open(filepath) as fin:
                        urls = [line.rstrip() for line in fin.readlines()]
                        all_products.append((category_name, urls))
                except UnsupportedOperation as e:
                    msg = f'failed: cannot read file ({e})'
            else:
                urls = self._get_amazon_page_product_urls(category_url, print_progress)
                all_products.append((category_name, urls))
                if cache_contents:
                    with open(filepath, 'w') as fout:
                        try:
                            for url in urls:
                                fout.write(url + '\n')
                                logger.debug(f'Saved {url}')
                        except PermissionError as e:
                            msg = f'failed: cannot write file ({e})'

            logger.info(f'Category {category_name}: {msg}.')

        if cleanup_empty:
            self._cleanup_empty_files(path)
        return all_products

Now, here, the errors are silenced, and the function returns an empty list. However, perhaps in the scheme of your software, it would make sense to bubble the errors up the chain and handle the errors more appropriately? Here, an empty list is returned, is that the desired failure mode up the chain?

As a bonus, I'm now toying with Visual Studio Code, and it has this neat functionality of showing me text that I highlight everywhere where it happens. It's a way to notice where you are repeating yourself: Visual Code Studio showing highlighted repeated code

There's a lot more to review, I think it's a good start though.

\$\endgroup\$
3
  • \$\begingroup\$ Thank you for this, I'm almost done with another version of the code that's entirely different and I added a multi-threading functionality, I'll comment with the link here when I'm done. \$\endgroup\$ – bullseye Oct 21 '19 at 22:32
  • \$\begingroup\$ The new version: codereview.stackexchange.com/questions/231126/… \$\endgroup\$ – bullseye Oct 22 '19 at 1:31
  • \$\begingroup\$ I need you to suggest ways to eliminate failures that are randomly occasional on certain sections not including best sellers and most wished for in the new code version above and I'll take take into consideration both feedbacks in my next follow up. \$\endgroup\$ – bullseye Oct 22 '19 at 2:22

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.