some refactoring

Author: DevAlone

collectors/collector.py

@@ -5,12 +5,13 @@
 import models
 
 
+# TODO: refactor saving state
 class AbstractCollector:
     """Base class for all types of collectors"""
 
     async def collect(self):
         """
-        this method should return proxies in any of the following formats:
+        This method should return proxies in any of the following formats:
 
         ::
 
@@ -19,22 +20,54 @@ async def collect(self):
             protocol://ip:port
             protocol://domain:port
 
+
+        ip can be both ipv4 and ipv6
+
+        will support yield in the future, now just return list
         """
 
         return []
 
     async def _collect(self):
-        """do not use!"""
-        return (await self.collect())[:settings.COLLECTOR_MAXIMUM_NUMBER_OF_PROXIES_PER_REQUEST]
+        """Do not use! It is called on collector's processing automatically"""
+
+        # i = 0
+        # async for proxy in self.collect():
+        #     if i > settings.COLLECTOR_MAXIMUM_NUMBER_OF_PROXIES_PER_REQUEST:
+        #         break
+
+        #     yield proxy
+        #     i += 1
+        proxies = list(await self.collect())
+        proxies = proxies[:settings.COLLECTOR_MAXIMUM_NUMBER_OF_PROXIES_PER_REQUEST]
+        self.last_processing_proxies_count = len(proxies)
+        return proxies
 
     async def load_state(self, state: models.CollectorState):
+        """
+        Function for loading collector's state from database model.
+        It's called automatically, don't worry. All you can do is
+        to override without forgetting to call parent's method like this:
+
+        ::
+
+            async def load_state(self, state):
+                super(MyCollector, self).load_state(state)
+                # do something here
+        """
         self.last_processing_time = state.last_processing_time
         self.processing_period = state.processing_period
+        self.last_processing_proxies_count = state.last_processing_proxies_count
         self.data = json.loads(state.data) if state.data is not None and state.data else {}
 
-    async def set_state(self, state: models.CollectorState):
+    async def save_state(self, state: models.CollectorState):
+        """
+        Function for saving collector's state to database model.
+        It's called automatically, don't worry.
+        """
         state.last_processing_time = self.last_processing_time
         state.processing_period = self.processing_period
+        state.last_processing_proxies_count = self.last_processing_proxies_count
         state.data = json.dumps(self.data)
 
     last_processing_time = 0
@@ -53,16 +86,25 @@ async def set_state(self, state: models.CollectorState):
     override_maximum_processing_period = None
     """
     ignore settings' maximum processing period and set
-    it to value of this variable
+    it to the value of this variable
     """
 
     override_minimum_processing_period = None
+    """
+    ignore settings' minimum processing period and set
+    it to the value of this variable, for example
+    when some collector has requests time limit
+    """
 
     data = {}
+    # TODO: consider namespacing
     """
     here you can store some information,
     it will be written into and read from database
     by magic, don't worry about it :)
-    Just don't use names starting with underscore
+    If you're curious, see process_collector_of_state() function
+    from processor.py file
+
+    Don't use names starting with the underscore
     like this one: _last_page
     """

collectors/pages_collector.py

@@ -5,16 +5,31 @@
 # TODO: save pages to collector state
 class PagesCollector(AbstractCollector):
     """
-    Collector for paginated APIs.
-    Here you should override ``process_page(page_index)`` method
+    Collector for paginated APIs. Pages are started from 0.
+    Here you should override ``process_page(page_index)`` method.
+    This collector will care about pages, increment it on each processing
+    and will reset it to 0 if there is no proxies on the page or if proxies
+    are the same as those on the previous one. If you don't want such smart
+    behavior, just set dynamic_pages_count to false and set pages_count manually.
     """
 
-    async def collect(self):
+    async def load_state(self, state):
+        super(PagesCollector, self).load_state(state)
         if "_current_page" in self.data:
             self.current_page = self.data["_current_page"]
 
-        proxies = (await self.process_page(self.current_page)
-                   )[:settings.COLLECTOR_MAXIMUM_NUMBER_OF_PROXIES_PER_REQUEST]
+        if "_pages_count" in self.data:
+            self.pages_count = self.data["_pages_count"]
+
+    async def save_state(self, state):
+        super(PagesCollector, self).save_state(state)
+        self.data["_current_page"] = self.current_page
+        self.data["_pages_count"] = self.pages_count
+
+    async def collect(self):
+        proxies = list(
+            await self.process_page(self.current_page)
+        )[:settings.COLLECTOR_MAXIMUM_NUMBER_OF_PROXIES_PER_REQUEST]
 
         if self.dynamic_pages_count:
             if proxies:
@@ -37,7 +52,6 @@ async def collect(self):
         if self.current_page >= self.pages_count:
             self.current_page = 0
 
-        self.data["_current_page"] = self.current_page
 
         return proxies
 

collectors_list.py

@@ -8,8 +8,11 @@
 
 collectors = {}
 
-_collector_dirs = settings.COLLECTORS_DIRS if type(settings.COLLECTORS_DIRS) is list else [settings.COLLECTORS_DIRS]
-for collectors_dir in _collector_dirs:
+_collectors_dirs = settings.COLLECTORS_DIRS
+if type(_collectors_dirs) is not list:
+    _collectors_dirs = [_collectors_dirs]
+
+for collectors_dir in _collectors_dirs:
     for root, dirs, files in os.walk(collectors_dir):
         for file in files:
             if file.endswith(".py"):

processor.py

@@ -257,7 +257,7 @@ async def process_collector_of_state(self, collector_state):
             self.logger.debug(
                 "start processing collector of type \"{}\"".format(type(collector))
             )
-            proxies = set(await collector._collect())
+            proxies = await collector._collect()
 
             if not proxies:
                 self.collectors_logger.warning(
@@ -278,7 +278,6 @@ async def process_collector_of_state(self, collector_state):
         finally:
             collector.last_processing_time = int(time.time())
             await collector.set_state(collector_state)
-            collector_state.last_processing_proxies_count = len(proxies)
             # TODO: save new proxies count
             await db.update(collector_state)