lint: tidying up code in block tasks

app/terraform/block_external.py

@@ -1,5 +1,5 @@
 import datetime
-from typing import Tuple
+from typing import Tuple, List, Dict
 
 from bs4 import BeautifulSoup
 import requests
@@ -12,37 +12,51 @@ from app.terraform import BaseAutomation
 
 
 class BlockExternalAutomation(BaseAutomation):
+    """
+    Automation task to import proxy reachability results from external source.
+    """
     short_name = "block_external"
     description = "Import proxy reachability results from external source"
 
-    def automate(self, full: bool = False) -> Tuple[bool, str]:
+    content: bytes
+    results: Dict[str, List[str]]
+
+    def _fetch(self) -> None:
         user_agent = {'User-agent': 'BypassCensorship/1.0'}
         page = requests.get(app.config['EXTERNAL_CHECK_URL'], headers=user_agent)
-        soup = BeautifulSoup(page.content, 'html.parser')
-        h2 = soup.find_all('h2')
+        self.content = page.content
+
+    def _parse(self) -> None:
+        soup = BeautifulSoup(self.content, 'html.parser')
+        h2 = soup.find_all('h2')  # pylint: disable=invalid-name
         div = soup.find_all('div', class_="overflow-auto mb-5")
-
         results = {}
-
         i = 0
         while i < len(h2):
             if not div[i].div:
                 urls = []
-                a = div[i].find_all('a')
+                anchors = div[i].find_all('a')
                 j = 0
-                while j < len(a):
-                    urls.append(a[j].text)
+                while j < len(anchors):
+                    urls.append(anchors[j].text)
                     j += 1
                 results[h2[i].text] = urls
             else:
                 results[h2[i].text] = []
             i += 1
+        self.results = results
+
+    def automate(self, full: bool = False) -> Tuple[bool, str]:
+        # TODO: handle errors in fetching remote content
+        # TODO: handle errors in parsing the remote content
+        self._fetch()
+        self._parse()
         activities = []
         blocked_proxies = []
-        for vp in results:
-            if vp not in app.config['EXTERNAL_VANTAGE_POINTS']:
+        for vantage_point, urls in self.results.items():
+            if vantage_point not in app.config['EXTERNAL_VANTAGE_POINTS']:
                 continue
-            for url in results[vp]:
+            for url in urls:
                 print(f"Found {url} blocked")
                 proxy = Proxy.query.filter(
                     Proxy.provider == "cloudfront",
@@ -54,9 +68,6 @@ class BlockExternalAutomation(BaseAutomation):
                 if not proxy.origin.auto_rotation:
                     print("Proxy auto-rotation forbidden for origin")
                     continue
-                if proxy.deprecated:
-                    print("Proxy already marked blocked")
-                    continue
                 if proxy.added > datetime.datetime.utcnow() - datetime.timedelta(hours=3):
                     activities.append(Activity(
                         activity_type="block_warning",
@@ -78,8 +89,8 @@ class BlockExternalAutomation(BaseAutomation):
                 activity_type="block_warning",
                 text=(
                     "More than 15 proxies were marked blocked according to external source. REFUSING to rotate.")))
-        for a in activities:
-            a.notify()
-            db.session.add(a)
+        for activity in activities:
+            activity.notify()
+            db.session.add(activity)
         db.session.commit()
         return True, ""