Proxy Rotation

.gitignore

@@ -24,3 +24,4 @@ logs/*.log*
 tags
 stores/store_data/item_cache.p
 Amazon_aio.bat
+myutils

Pipfile

@@ -5,6 +5,7 @@ verify_ssl = true
 
 [dev-packages]
 pyinstaller = "*"
+uvloop = "*"
 
 [packages]
 requests = {extras = ["socks"], version = "*"}
@@ -39,6 +40,10 @@ dnspython = "*"
 fake-useragent = "*"
 pysocks = "*"
 aiohttp-proxy = "*"
+uvloop = "*"
+fake-headers = "*"
+bs4 = "*"
+html5lib = "*"
 
 [requires]
 python_version = "3.8"

amazoncaptcha_aio/__init__.py

@@ -0,0 +1,46 @@
+# -*- coding: utf-8 -*-
+
+"""Solver for Amazon's image captcha.
+
+The motivation behind the creation of this library is taking its start from
+the genuinely simple idea: "I don't want to use pytesseract or some other
+non-amazon-specific OCR services, nor do I want to install some executables to
+just solve a captcha. I desire to get a solution within 1-2 lines of code
+without any heavy add-ons. Using a pure Python."
+
+Examples:
+    Browsing Amazon using selenium and stuck on captcha? The class method
+    below will do all the "dirty" work of extracting an image from the webpage
+    for you. Practically, it takes a screenshot from your webdriver, crops the
+    captcha, and stores it into bytes array, which is then used to create an
+    AmazonCaptcha instance. This also means avoiding any local savings.
+
+        from amazoncaptcha import AmazonCaptcha
+        from selenium import webdriver
+
+        driver = webdriver.Chrome() # This is a simplified example
+        driver.get('https://www.amazon.com/errors/validateCaptcha')
+
+        captcha = AmazonCaptcha.fromdriver(driver)
+        solution = captcha.solve()
+
+
+    If you are not using selenium or the previous method is not just the case for
+    you, it is possible to use a captcha link directly. This class method will
+    request the url, check the content type and store the response content into bytes
+    array to create an instance of AmazonCaptcha.
+
+        from amazoncaptcha import AmazonCaptcha
+
+        link = 'https://images-na.ssl-images-amazon.com/captcha/usvmgloq/Captcha_kwrrnqwkph.jpg'
+
+        captcha = AmazonCaptcha.fromlink(link)
+        solution = captcha.solve()
+
+"""
+
+from .solver import AmazonCaptcha
+from .devtools import AmazonCaptchaCollector
+from .exceptions import ContentTypeError, NotFolderError
+
+#--------------------------------------------------------------------------------------------------------------

amazoncaptcha_aio/__version__.py

@@ -0,0 +1,8 @@
+__title__ = 'amazoncaptcha'
+__description__ = "Pure Python, lightweight, Pillow-based solver for Amazon's text captcha."
+__url__ = 'https://github.com/a-maliarov/amazoncaptcha'
+__version__ = '0.5.0'
+__author__ = 'Anatolii Maliarov'
+__author_email__ = 'tly.mov@gmail.com'
+__license__ = 'MIT'
+__copyright__ = 'Copyright 2020 Anatolii Maliarov'

amazoncaptcha_aio/devtools.py

@@ -0,0 +1,150 @@
+# -*- coding: utf-8 -*-
+
+"""
+amazoncaptcha.devtools
+~~~~~~~~~~~~~~~~~~~~~~
+
+This module contains the set of amazoncaptcha's devtools.
+"""
+
+from .solver import AmazonCaptcha
+from .exceptions import NotFolderError
+from .__version__ import __version__
+
+from io import BytesIO
+import multiprocessing
+import requests
+import os
+
+#--------------------------------------------------------------------------------------------------------------
+
+class AmazonCaptchaCollector(object):
+
+    def __init__(self, output_folder_path, keep_logs=True, accuracy_test=False):
+        """
+        Initializes the AmazonCaptchaCollector instance.
+
+        Args:
+            output_folder (str): Folder where images or logs should be stored.
+            keep_logs (bool, optional): Is set to True, unsolved captcha links
+                will be stored separately.
+            accuracy_test (bool, optional): If set to True, AmazonCaptchaCollector
+                will not download images, but just solve them and log the results.
+
+        """
+
+        self.output_folder = output_folder_path
+        self.keep_logs = keep_logs
+        self.accuracy_test = accuracy_test
+
+        if not os.path.exists(self.output_folder):
+            os.mkdir(self.output_folder)
+
+        elif not os.path.isdir(self.output_folder):
+            raise NotFolderError(self.output_folder)
+
+        self.collector_logs = os.path.join(self.output_folder, f'collector-logs-{__version__.replace(".", "")}.log')
+        self.test_results = os.path.join(self.output_folder, 'test-results.log')
+        self.not_solved_logs = os.path.join(self.output_folder, 'not-solved-captcha.log')
+
+    def _extract_captcha_link(self, captcha_page):
+        """Extracts a captcha link from an html page.
+
+        Args:
+            captcha_page (str): A page's html in string format.
+
+        Returns:
+            str: Captcha link.
+
+        """
+
+        return captcha_page.text.split('<img src="')[1].split('">')[0]
+
+    def _extract_captcha_id(self, captcha_link):
+        """
+        Extracts a captcha id from a captcha link.
+
+        Args:
+            captcha_link (str): A link to the captcha image.
+
+        Returns:
+            str: Captcha ID.
+
+        """
+
+        return ''.join(captcha_link.split('/captcha/')[1].replace('.jpg', '').split('/Captcha_'))
+
+    def get_captcha_image(self):
+        """
+        Requests the page with Amazon's captcha, gets random captcha.
+        Creates AmazonCaptcha instance, stores an original image before solving.
+
+        If it is not an accuracy test, the image will be stored in a specified
+        folder with the solution within its name. Otherwise, only the logs
+        will be stored, mentioning the captcha link being processes and the result.
+
+        """
+
+        captcha_page = requests.get('https://www.amazon.com/errors/validateCaptcha')
+        captcha_link = self._extract_captcha_link(captcha_page)
+
+        response = requests.get(captcha_link)
+        captcha = AmazonCaptcha(BytesIO(response.content))
+        captcha._image_link = captcha_link
+        original_image = captcha.img
+
+        solution = captcha.solve(keep_logs=self.keep_logs, logs_path=self.not_solved_logs)
+        log_message = f'{captcha.image_link}::{solution}'
+
+        if solution != 'Not solved' and not self.accuracy_test:
+            print(log_message)
+            captcha_name = 'dl_' + self._extract_captcha_id(captcha.image_link) + '_' + solution + '.png'
+            original_image.save(os.path.join(self.output_folder, captcha_name))
+
+        else:
+            print(log_message)
+            with open(self.collector_logs, 'a', encoding='utf-8') as f:
+                f.write(log_message + '\n')
+
+    def _distribute_collecting(self, milestone):
+        """Distribution function for multiprocessing."""
+
+        for step in milestone:
+            self.get_captcha_image()
+
+    def start(self, target, processes):
+        """
+        Starts the process of collecting captchas of conducting a test.
+
+        Args:
+            target (int): Number of captchas to be processed.
+            processes (int): Number of simultaneous processes.
+
+        """
+
+        goal = list(range(target))
+        milestones = [goal[x: x + target // processes] for x in range(0, len(goal), target // processes)]
+
+        jobs = list()
+        for j in range(processes):
+            p = multiprocessing.Process(target=self._distribute_collecting, args=(milestones[j], ))
+            jobs.append(p)
+            p.start()
+
+        for proc in jobs:
+            proc.join()
+
+        if self.accuracy_test:
+            with open(self.collector_logs, 'r', encoding='utf-8') as f:
+                output = f.readlines()
+
+            all_captchas = len(output)
+            solved_captchas = len([i for i in output if 'Not solved' not in i])
+            success_percentage = round((solved_captchas / all_captchas) * 100, 5)
+            result = f'::Test::Ver{__version__}::Cap{all_captchas}::Per{success_percentage}::'
+
+            with open(self.test_results, 'w', encoding='utf-8') as f:
+                print(result)
+                f.write(result)
+
+#--------------------------------------------------------------------------------------------------------------

amazoncaptcha_aio/exceptions.py

@@ -0,0 +1,38 @@
+# -*- coding: utf-8 -*-
+
+"""
+amazoncaptcha.exceptions
+~~~~~~~~~~~~~~~~~~~~~~~~
+
+This module contains the set of amazoncaptcha's exceptions.
+"""
+
+#--------------------------------------------------------------------------------------------------------------
+
+class ContentTypeError(Exception):
+    """
+    Requested url, which was supposed to be the url to the captcha image
+    contains unsupported content type within response headers.
+    """
+
+    def __init__(self, content_type, message='is not supported as a Content-Type. Cannot extract the image.'):
+        self.content_type = content_type
+        self.message = message
+
+    def __str__(self):
+        return f'"{self.content_type}" {self.message}'
+
+class NotFolderError(Exception):
+    """
+    Given path, which was supposed to be a path to the folder, where
+    script can store images, is not a folder.
+    """
+
+    def __init__(self, path, message='is not a folder. Cannot store images there.'):
+        self.path = path
+        self.message = message
+
+    def __str__(self):
+        return f'"{self.path}" {self.message}'
+
+#--------------------------------------------------------------------------------------------------------------