5
\$\begingroup\$

This is a follow up to the code here: Web scraper that extracts urls from Amazon and eBay A multi-threaded modification to the previous version that is Amazon focused and most of the necessary documentation is in the docstrings.

You'll find a copy of the source code as well as necessary files here including (proxies.txt, amazon_log.txt, user_agents.txt) to be enclosed within the same folder as the code's.

Features:

  • Multi-threaded scraping of contents.
  • Save urls to .txt files
  • Scrape Amazon sections including: best sellers, new releases, most wished for ...
  • Save names to .txt files.
  • Map names to urls.
  • Caching of contents for further re-use.
  • Extraction of product features including(name, title, url, features, technical details ...

I'll be implementing another class that manages this one with public methods organizing files into csv/json files and perform some data analysis as well as optimizations to this one. I'll be posting follow ups when I'm done.

For reviewers:

  • Modifications: I made a lot of modifications in this version and it's completely different than the previous one. It's Amazon only focused and lots of unnecessary former method parameters print_progress, cleanup_empty are now class attributes. Sequential extraction is now optional as well as multi-threaded extraction which is 500 x faster. Docstrings are up to date and completely changed in terms of style and content. The code is much more organized in this version and much more readable.
  • Shorter code suggestions: I want to shorten the code and eliminate repetition(if any), most of the code is repetition free, but tasks are repetitive in usually different forms.
  • Proxies and user agents: Concerning the responses gathered using the _get_response() method, are proxies and and headers parameters doing the necessary job? are proxies working this way? are there any improvements that could be done?
  • Random occasional failures: There are occasional and random occurrences of failures in feature extraction in sections that do not include best sellers or most wished for. Why these failures sometimes happen and sometimes they do not? and how to control this and get the least failure percentage possible?
  • Private methods: Methods defined here are private _private() because this class will be used by another class that manages the extraction and will contain public methods mostly.
  • Suggestions: General suggestions to improve the code are most welcome and feel free to ask questions if you need to clarify things.

Note: For people downvoting this, unless you work at Google or Nasa maybe or even Alan Turing at some other dimension at least give me the honor of letting me know why this might not have passed your super godly standards.

Code

#!/usr/bin/env python3
from requests.exceptions import HTTPError, ConnectionError, ConnectTimeout
from concurrent.futures import ThreadPoolExecutor, as_completed
from bs4 import BeautifulSoup
from time import perf_counter
from random import choice
import requests
import bs4
import os


class AmazonScraper:
    """
    A tool to scrape Amazon different sections.

    Sections:
    Best Sellers - New Releases - Gift Ideas - Movers and Shakers - Most Wished For.

    Features:
    Category/Subcategory Urls and names.
    Product Urls and details(title, features, technical details, price, review count)
    """

    def __init__(
            self, path=None, print_progress=False, cache_contents=True, cleanup_empty=True, threads=1, log=None):
        """
        Args:
            path: Folder path to save scraped and cached contents.
            print_progress: If True then the progress will be displayed.
            cache_contents: If True then the scraped contents will be cached for further re-use.
            cleanup_empty: If True, empty .txt files that might result will be deleted.
            threads: If number of threads(1 by default) is increased, multiple threads will be used.
            log: If print_progress is True, content will be saved to the log (a file name + .txt).
        """
        if not path:
            self.path = '/Users/user_name/Desktop/Amazon Scraper/'
        if path:
            self.path = path
        self.headers = [{'User-Agent': item.rstrip()} for item in open('user_agents.txt').readlines()]
        self.print_progress = print_progress
        self.cache_contents = cache_contents
        self.cleanup_empty = cleanup_empty
        self.session = requests.session()
        self.threads = threads
        if log:
            if log in os.listdir(self.path):
                os.remove(log)
            self.log = open(log, 'w')
        self.proxies = [{'https:': 'https://' + item.rstrip(), 'http':
                        'http://' + item.rstrip()} for item in open('proxies.txt').readlines()]
        self.modes = {'bs': 'Best Sellers', 'nr': 'New Releases', 'gi': 'Gift Ideas',
                      'ms': 'Movers and Shakers', 'mw': 'Most Wished For'}
        self.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/')}

    def _cache_main_category_urls(self, text_file_names: dict, section: str, category_class: str,
                                  content_path: str, categories: list):
        """
        Cache the main category/subcategory URLs to .txt files.
        Args:
            text_file_names: Section string indications mapped to their corresponding .txt filenames.
            section: Keyword indication of target section.
                'bs': Best Sellers
                'nr': New Releases
                'ms': Movers & Shakers
                'gi': Gift Ideas
                'mw': Most Wished For
            category_class: Category level indication 'categories' or 'subcategories'.
            content_path: Path to folder to save cached files.
            categories: The list of category/subcategory urls to be saved.
        Return:
             None
        """
        os.chdir(content_path + 'Amazon/')
        with open(text_file_names[section][category_class], 'w') as cats:
            for category in categories:
                cats.write(category + '\n')
                if self.print_progress:
                    if not open(text_file_names[section][category_class]).read().isspace():
                        print(f'Saving {category} ... done.')
                        if self.log:
                            print(f'Saving {category} ... done.', file=self.log, end='\n')
                    if open(text_file_names[section][category_class]).read().isspace():
                        print(f'Saving {category} ... failure.')
                        if self.log:
                            print(f'Saving {category} ... failure.', file=self.log, end='\n')
        if self.cleanup_empty:
            self._cleanup_empty_files(self.path)

    def _read_main_category_urls(self, text_file_names: dict, section: str, category_class: str, content_path: str):
        """
        Read the main category/subcategory cached urls from their respective .txt files.
        Args:
            text_file_names: Section string indications mapped to their corresponding .txt filenames.
            section: Keyword indication of target section.
                'bs': Best Sellers
                'nr': New Releases
                'ms': Movers & Shakers
                'gi': Gift Ideas
                'mw': Most Wished For
            category_class: Category level indication 'categories' or 'subcategories'.
            content_path: Path to folder to save cached files.
        Return:
             A list of the main category/subcategory urls specified.
        """
        os.chdir(content_path + 'Amazon')
        if text_file_names[section][category_class] in os.listdir(content_path + 'Amazon/'):
            with open(text_file_names[section][category_class]) as cats:
                if self.cleanup_empty:
                    self._cleanup_empty_files(self.path)
                return [link.rstrip() for link in cats.readlines()]

    def _get_response(self, url):
        """
        Send a get request to target url.
        Args:
            url: Target Url.
        Return:
             Response object.
        """
        return self.session.get(url, headers=choice(self.headers), proxies=choice(self.proxies))

    def _scrape_main_category_urls(self, section: str, category_class: str, prev_categories=None):
        """
        Scrape links of all main category/subcategory Urls of the specified section.
        Args:
            section: Keyword indication of target section.
                'bs': Best Sellers
                'nr': New Releases
                'ms': Movers & Shakers
                'gi': Gift Ideas
                'mw': Most Wished For
            category_class: Category level indication 'categories' or 'subcategories'.
            prev_categories: A list containing parent category Urls.
        Return:
             A sorted list of scraped category/subcategory Urls.
        """
        target_url = self.starting_target_urls[section][1]
        if category_class == 'categories':
            starting_url = self._get_response(self.starting_target_urls[section][0])
            html_content = BeautifulSoup(starting_url.text, features='lxml')
            target_url_part = self.starting_target_urls[section][1]
            if not self.print_progress:
                return sorted({str(link.get('href')) for link in html_content.findAll('a')
                               if target_url_part in str(link)})
            if self.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 {self.modes[section]}-{category_class[:-3]}y: {link_to_add}')
                        if self.log:
                            print(f'Fetched {self.modes[section]}-{category_class[:-3]}y: '
                                  f'{link_to_add}', file=self.log, end='\n')
                return categories
        if category_class == 'subcategories':
            if not self.print_progress:
                if self.threads == 1:
                    responses = [self._get_response(category)
                                 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 self.threads > 1:
                    with ThreadPoolExecutor(max_workers=self.threads) as executor:
                        future_html = {
                            executor.submit(self._get_response, category): category for category in prev_categories}
                        responses = [future.result() for future in as_completed(future_html)]
                        category_soups = [BeautifulSoup(response.text) 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 self.print_progress:
                if self.threads == 1:
                    responses, pre, subcategories = [], [], set()
                    for category in prev_categories:
                        response = self._get_response(category)
                        responses.append(response)
                        print(f'Got response {response} for {self.modes[section]}-{category}')
                        if self.log:
                            print(f'Got response {response} for {self.modes[section]}-{category}',
                                  file=self.log, end='\n')

                    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 {self.modes[section]}-{fetched_link}')
                                if self.log:
                                    print(f'Fetched {self.modes[section]}-{fetched_link}', file=self.log,
                                          end='\n')
                    return sorted({link for link in pre if link not in prev_categories})
                if self.threads > 1:
                    with ThreadPoolExecutor(max_workers=self.threads) as executor:
                        category_soups = []
                        future_responses = {
                            executor.submit(self._get_response, category): category for category in prev_categories}
                        for future in as_completed(future_responses):
                            url = future_responses[future]
                            try:
                                response = future.result()
                                print(f'Got response {response} for {self.modes[section]}-{url}')
                                if self.log:
                                    print(f'Got response {response} for {self.modes[section]}-{url}',
                                          file=self.log, end='\n')
                            except(HTTPError, ConnectTimeout, ConnectionError):
                                print(f'Failed to get response from {url}')
                                if self.log:
                                    print(f'Failed to get response from {url}', file=self.log, end='\n')
                            else:
                                category_soups.append(BeautifulSoup(response.text, features='lxml'))
                        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})

    def _get_main_category_urls(self, section: str, subs=True):
        """
        Manage the scrape/read from previous session cache operations and return section Urls.
        If the program found previously cached files, will read and return existing data, else
        new content will be scraped and returned.
        Args:
            section: Keyword indication of target section.
                'bs': Best Sellers
                'nr': New Releases
                'ms': Movers & Shakers
                'gi': Gift Ideas
                'mw': Most Wished For
            subs: If False, only categories will be returned.
        Return:
            2 sorted lists: categories and subcategories.
        """
        text_file_names = \
            {section_short: {'categories': self.modes[section_short] + ' Category Urls.txt',
                             'subcategories': self.modes[section_short] + ' Subcategory Urls.txt'}
             for section_short in self.modes}
        if 'Amazon' not in os.listdir(self.path):
            os.mkdir('Amazon')
            os.chdir(self.path + 'Amazon')
        if 'Amazon' in os.listdir(self.path):
            categories = self._read_main_category_urls(text_file_names, section, 'categories', self.path)
            if not subs:
                if self.cleanup_empty:
                    self._cleanup_empty_files(self.path)
                return sorted(categories)
            subcategories = self._read_main_category_urls(text_file_names, section, 'subcategories', self.path)
            try:
                if categories and subcategories:
                    if self.cleanup_empty:
                        self._cleanup_empty_files(self.path)
                    return sorted(categories), sorted(subcategories)
            except UnboundLocalError:
                pass
        if not subs:
            categories = self._scrape_main_category_urls(section, 'categories')
            if self.cache_contents:
                self._cache_main_category_urls(text_file_names, section, 'categories', self.path, categories)
            if self.cleanup_empty:
                self._cleanup_empty_files(self.path)
            return sorted(categories)
        if subs:
            categories = self._scrape_main_category_urls(section, 'categories')
            if self.cache_contents:
                self._cache_main_category_urls(text_file_names, section, 'categories', self.path, categories)
            subcategories = self._scrape_main_category_urls(section, 'subcategories', categories)
            if self.cache_contents:
                self._cache_main_category_urls(text_file_names, section, 'subcategories', self.path, subcategories)
            if self.cleanup_empty:
                self._cleanup_empty_files(self.path)
            return sorted(categories), sorted(subcategories)

    def _extract_page_product_urls(self, page_url: str):
        """
        Extract product Urls from an Amazon page and the page title.
        Args:
            page_url: Target page.
        Return:
             The page category title(string) and a sorted list of product Urls.
        """
        prefix = 'https://www.amazon.com'
        response = self._get_response(page_url)
        soup = BeautifulSoup(response.text, features='lxml')
        try:
            title = soup.h1.text.strip()
        except AttributeError:
            title = 'N/A'
        product_links = {prefix + link.get('href') for link in soup.findAll('a') if 'psc=' in str(link)}
        return title, sorted(product_links)

    @staticmethod
    def _cleanup_empty_files(dir_path: str):
        """
        Cleanup a given folder from empty .txt files.
        Args:
            dir_path: Path to the target folder to be cleaned up.
        Return:
             None
        """
        for file_name in [file for file in os.listdir(dir_path)]:
            if not os.path.isdir(file_name):
                try:
                    contents = open(file_name).read().strip()
                    if not contents:
                        os.remove(file_name)
                except(UnicodeDecodeError, FileNotFoundError):
                    pass

    def _category_page_title_to_url(self, section: str, category_class: str, delimiter='&&&'):
        """
        Map category/subcategory names to their respective Urls.
        Args:
        section:
            'bs': Best Sellers
            'nr': New Releases
            'ms': Movers & Shakers
            'gi': Gift Ideas
            'mw': Most Wished For
        category_class: Category level indication 'categories' or 'subcategories'.
        delimiter: Delimits category/subcategory names and their respective Urls in the .txt files.
        Return:
             A list of lists(pairs): [[category/subcategory name, Url], ...]
        """
        file_names = {'categories': self.modes[section] + ' Category Names.txt',
                      'subcategories': self.modes[section] + ' Subcategory 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 self.cleanup_empty:
                        self._cleanup_empty_files(self.path)
                    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, subcategories = self._get_main_category_urls(section)
        if not self.print_progress:
            if self.threads == 1:
                responses_urls = [(self._get_response(url), url)
                                  for url in eval('eval(category_class)')]
                soups_urls = [(BeautifulSoup(item[0].text, features='lxml'), item[1]) for item in responses_urls]
                for soup, url in soups_urls:
                    try:
                        title = soup.h1.text.strip()
                        names_urls.append([title, url])
                    except AttributeError:
                        pass
            if self.threads > 1:
                with ThreadPoolExecutor(max_workers=self.threads) as executor:
                    future_responses = {
                        executor.submit(self._get_response, category): category
                        for category in eval('eval(category_class)')}
                    responses = [future.result() for future in as_completed(future_responses)]
                    responses_urls = [
                        (response, url) for response, url in zip(responses, eval('eval(category_class)'))]
                    soups_urls = [
                        (BeautifulSoup(item[0].text, features='lxml'), item[1]) for item in responses_urls]
                    for soup, url in soups_urls:
                        try:
                            title = soup.h1.text.strip()
                            names_urls.append([title, url])
                        except AttributeError:
                            pass
        if self.print_progress:
            if self.threads == 1:
                for url in eval('eval(category_class)'):
                    response = self._get_response(url)
                    print(f'Got response {response} for {url}')
                    print(f'Fetching name of {url} ...')
                    if self.log:
                        print(f'Got response {response} for {url}', file=self.log, end='\n')
                        print(f'Fetching name of {url} ...', file=self.log, end='\n')

                    soup = BeautifulSoup(response.text, features='lxml')
                    try:
                        title = soup.h1.text.strip()
                        names_urls.append([title, url])
                        print(f'Fetching name {title} ... done')
                        if self.log:
                            print(f'Fetching name {title} ... done', file=self.log, end='\n')
                    except AttributeError:
                        print(f'Fetching name failure for {url}')
                        if self.log:
                            print(f'Fetching name failure for {url}', file=self.log, end='\n')
            if self.threads > 1:
                with ThreadPoolExecutor(max_workers=self.threads) as executor:
                    future_responses = {
                        executor.submit(self._get_response, category): category
                        for category in eval('eval(category_class)')}
                    for future_response in as_completed(future_responses):
                        response = future_response.result()
                        url = future_responses[future_response]
                        print(f'Got response {response} for {url}')
                        if self.log:
                            print(f'Got response {response} for {url}', file=self.log, end='\n')
                        soup = BeautifulSoup(response.text, features='lxml')
                        try:
                            title = soup.h1.text.strip()
                            names_urls.append([title, url])
                            print(f'Fetching name {title} ... done')
                            if self.log:
                                print(f'Fetching name {title} ... done', file=self.log, end='\n')
                        except AttributeError:
                            print(f'Fetching name failure for {url}')
                            if self.log:
                                print(f'Fetching name failure for {url}', file=self.log, end='\n')

            if self.cache_contents:
                with open(file_names[category_class], 'w') as names:
                    for name, url in names_urls:
                        names.write(name + delimiter + url + '\n')
            if self.cleanup_empty:
                self._cleanup_empty_files(self.path + 'Amazon')
        return names_urls

    def _extract_section_products(self, section: str, category_class: str):
        """
        For every category/subcategory successfully scraped from the given section, product urls will be extracted.
        Args:
            section:
                'bs': Best Sellers
                'nr': New Releases
                'ms': Movers & Shakers
                'gi': Gift Ideas
                'mw': Most Wished For
            category_class: Category level indication 'categories' or 'subcategories'.
        Return:
             List of tuples(category name, product urls) containing product Urls for each scraped category/subcategory.
        """
        products = []
        names_urls = self._category_page_title_to_url(section, category_class)
        urls = [item[1] for item in names_urls]
        folder_name = ' '.join([self.modes[section], category_class[:-3].title() + 'y', 'Product Urls'])
        if not self.print_progress:
            if self.threads == 1:
                products = [
                    (category_name, [product_url for product_url in self._extract_page_product_urls(category_url)[1]])
                    for category_name, category_url in names_urls]
                products = [item for item in products if item[1]]
            if self.threads > 1:
                with ThreadPoolExecutor(max_workers=self.threads) as executor:
                    future_products = {executor.submit(self._extract_page_product_urls, category_url): category_url
                                       for category_url in urls}
                    products = [future.result() for future in as_completed(future_products)]
                    products = [item for item in products if item[1]]
        if self.print_progress:
            products = []
            if self.threads == 1:
                for category_name, category_url in names_urls:
                    product_urls = self._extract_page_product_urls(category_url)
                    if product_urls[1]:
                        print(f'Extraction of {category_name} products ... done')
                        if self.log:
                            print(f'Extraction of {category_name} products ... done', file=self.log, end='\n')
                        products.append(product_urls)
                    else:
                        print(f'Extraction of {category_name} products ... failure')
                        if self.log:
                            print(f'Extraction of {category_name} products ... failure', file=self.log, end='\n')
            if self.threads > 1:
                with ThreadPoolExecutor(max_workers=self.threads) as executor:
                    future_products = {executor.submit(self._extract_page_product_urls, category_url): category_url
                                       for category_url in urls}
                    for future in as_completed(future_products):
                        category_name, category_urls = future.result()
                        if category_urls:
                            print(f'Extraction of {category_name} products ... done')
                            if self.log:
                                print(f'Extraction of {category_name} products ... done', file=self.log, end='\n')
                            products.append((category_name, category_urls))
                        else:
                            print(f'Extraction of {category_name} products ... failure')
                            if self.log:
                                print(f'Extraction of {category_name} products ... failure', file=self.log, end='\n')
        if self.cache_contents:
            if folder_name not in os.listdir(self.path + 'Amazon'):
                os.mkdir(folder_name)
            os.chdir(folder_name)
            for category_name, category_product_urls in products:
                with open(category_name + '.txt', 'w') as links:
                    for url in category_product_urls:
                        links.write(url + '\n')
        if self.cleanup_empty:
            self._cleanup_empty_files(self.path + 'Amazon/' + folder_name)
        return products

    def _get_amazon_product_details(self, product_url: str):
        """
        Extract product details including:
            [Price, Title, URL, Rating, Number of reviews, Sold by, Features, Technical table]
        Args:
            product_url: Target product.
        Return:
            A dictionary with the scraped details.
        """
        product_html_details, text_details = {}, {}
        response = self._get_response(product_url).text
        html_content = BeautifulSoup(response, features='lxml')
        product_html_details['Price'] = html_content.find('span', {'id': 'price_inside_buybox'})
        product_html_details['Url'] = product_url
        product_html_details['Title'] = html_content.title
        product_html_details['Rating'] = html_content.find('span',
                                                           {'class': 'reviewCountTextLinkedHistogram noUnderline'})
        product_html_details['Number of reviews'] = html_content.find('span', {'id': 'acrCustomerReviewText'})
        product_html_details['Sold by'] = html_content.find('a', {'id': 'bylineInfo'})
        product_html_details['Features'] = html_content.find('div', {'id': 'feature-bullets'})
        if product_html_details['Features']:
            product_html_details['Features'] = product_html_details['Features'].findAll('li')
        technical_table = html_content.find('table', {'class': 'a-keyvalue prodDetTable'})
        if technical_table:
            product_html_details['Technical details'] = list(
                zip([item.text.strip() for item in technical_table.findAll('th')],
                    [item.text.strip() for item in technical_table.findAll('td')]))
        for item in product_html_details:
            if isinstance(product_html_details[item], bs4.element.Tag):
                text_details[item] = product_html_details[item].text.strip()
            if isinstance(product_html_details[item], bs4.element.ResultSet):
                text_details[item] = ' • '.join([tag.text.strip() for tag in product_html_details[item]])
            if isinstance(product_html_details[item], str):
                text_details[item] = product_html_details[item]
            if item == 'Technical details':
                text_details[item] = ' • '.join([' : '.join(pair) for pair in product_html_details[item]])
        return text_details


if __name__ == '__main__':
    start_time = perf_counter()
    path = input('Enter path to save files: ')
    session = AmazonScraper(print_progress=True, threads=20, log='amazon_log.txt', path=path)
    print(session._extract_section_products('bs', 'categories'))
    print(session._extract_section_products('bs', 'subcategories'))
    end_time = perf_counter()
    print(f'Time: {end_time - start_time} seconds.')
\$\endgroup\$
  • \$\begingroup\$ Dare I ask why you aren't using docs.aws.amazon.com/AWSECommerceService/latest/DG/Welcome.html ? \$\endgroup\$ – Reinderien Oct 22 '19 at 17:36
  • \$\begingroup\$ @Renderien yeah ask whatever you want. I'm not using AWS because I heard that product details are not available and that it might be limited in terms of access and so. I'm not sure if this is True, I read it somewhere and if i'm wrong, I would definitely use it. \$\endgroup\$ – bullseye Oct 22 '19 at 19:28
  • \$\begingroup\$ @Renderien may i ask for a review? \$\endgroup\$ – bullseye Oct 22 '19 at 19:53
  • \$\begingroup\$ Yep - after work ;-) \$\endgroup\$ – Reinderien Oct 22 '19 at 20:10
2
\$\begingroup\$

Default arguments

This default:

path=None

isn't effectively None, but instead '/Users/user_name/Desktop/Amazon Scraper/'. That's an immutable value, so it's safe to put into the default directly.

An obvious issue with that path is that it's absolute and not per-user. Consider using os.path.expanduser with ~ instead.

Dict formatting

Writing this:

self.proxies = [{'https:': 'https://' + item.rstrip(), 'http':
                'http://' + item.rstrip()} for item in open('proxies.txt').readlines()]

should have one dict item per line or it'll get confusing. In other words,

self.proxies = [{'https:': 'https://' + item.rstrip(),
                 'http': 'http://' + item.rstrip()}
                for item in open('proxies.txt').readlines()]

Avoid backslash continuation

    self.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/')}

can be

    self.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/')
    }

Avoid manual path concatenation

This:

    os.chdir(content_path + 'Amazon/')

should use pathlib and the / operator instead.

Use a log library

This:

                    if self.log:
                        print(f'Saving {category} ... failure.', file=self.log, end='\n')

shouldn't be writing to files directly. Instead, you should be setting up the stock Python logging with a file handler that goes to that file. It's more flexible and maintainable.

Implicit line iteration

For lines like this:

            return [link.rstrip() for link in cats.readlines()]

You don't need to call readlines. Iterating over a file object iterates over its lines.

HTTP error checking

_get_response should include a call to raise_for_status. It's a quick and easy way to get better validation on your HTTP calls.

| improve this answer | |
\$\endgroup\$
  • \$\begingroup\$ Thanks for this, I will be modifying all of the points you mentioned in my next follow up, do you have any idea about how to ensure proxies and user agents are working effectively? Because i'm getting some occasional robot checks, how to bypass the robot checks? I thought proxies and user agents should deal with such problems but it looks like not. \$\endgroup\$ – bullseye Oct 23 '19 at 1:12
  • \$\begingroup\$ What exactly do you mean by robot check? Can you show an example of a failure produced by one? \$\endgroup\$ – Reinderien Oct 23 '19 at 1:18
  • \$\begingroup\$ okay, to see examples of when this happens, try running under main session = AmazonScraper(print_progress=True, threads=10) and choose your path then for item in ['ms', 'nr', 'gi']: print(session._extract_section_products(item, 'subcategories')) and change every line similar to title = soup.h1.text to title = soup.title.text and you'll encounter occasional pages named 'robot check' as title, then you can examine the html content if you want. \$\endgroup\$ – bullseye Oct 23 '19 at 1:40
  • \$\begingroup\$ I think that question is better-suited to StackOverflow, but my guess is that the server is doing what it's supposed to - your client is literally a robot. \$\endgroup\$ – Reinderien Oct 23 '19 at 1:42
  • \$\begingroup\$ And this i think this shouldn't be happening with proxies and user agents if I understand how this works correctly \$\endgroup\$ – bullseye Oct 23 '19 at 1:43

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.