Continued:

[fba.git] / fba / networks / mastodon.py

diff --git a/fba/networks/mastodon.py b/fba/networks/mastodon.py
index cda3cc4786b328cc2e20307de4b9721da831dbd8..be22e94f9eba52f0d413ad2503733fc3e3623310 100644 (file)
--- a/fba/networks/mastodon.py
+++ b/fba/networks/mastodon.py
@@ -19,11 +19,11 @@ import validators
 
 import bs4
 
+from fba.helpers import blacklist
 from fba.helpers import config
 from fba.helpers import domain as domain_helper
 from fba.helpers import tidyup
 
-from fba.http import federation
 from fba.http import network
 
 from fba.models import blocks
@@ -63,10 +63,12 @@ def fetch_blocks_from_about(domain: str) -> dict:
     logger.debug("domain='%s' - CALLED!", domain)
     domain_helper.raise_on(domain)
 
-    if not instances.is_registered(domain):
+    if blacklist.is_blacklisted(domain):
+        raise Exception(f"domain='{domain}' is blacklisted but function is invoked.")
+    elif not instances.is_registered(domain):
         raise Exception(f"domain='{domain}' is not registered but function is invoked.")
 
-    logger.debug("Fetching mastodon blocks from domain='%s'", domain)
+    logger.info("Fetching mastodon blocks from domain='%s'", domain)
     doc = None
     for path in ["/about/more", "/about"]:
         try:
@@ -115,10 +117,24 @@ def fetch_blocks_from_about(domain: str) -> dict:
         if header_text in blocklist or header_text.lower() in blocklist:
             # replaced find_next_siblings with find_all_next to account for instances that e.g. hide lists in dropdown menu
             for line in header.find_all_next("table")[0].find_all("tr")[1:]:
+                domain = line.find("span").text
+                digest = line.find("span")["title"][9:]
+                reason = line.find_all("td")[1].text
+
+                logger.debug("domain='%s',reason='%s' - BEFORE!", domain, reason)
+                domain = tidyup.domain(domain) if domain != "" else None
+                reason = tidyup.reason(reason) if reason != "" else None
+
+                logger.debug("domain='%s',reason='%s' - AFTER!", domain, reason)
+                if domain is None or domain == "":
+                    logger.warning("domain='%s' is empty,line='%s' - SKIPPED!", domain, line)
+                    continue
+
+                logger.debug("Appending domain='%s',digest='%s',reason='%s' to blocklist header_text='%s' ...", domain, digest, reason, blocklist)
                 blocklist[header_text].append({
-                    "domain": tidyup.domain(line.find("span").text),
-                    "hash"  : tidyup.domain(line.find("span")["title"][9:]),
-                    "reason": tidyup.reason(line.find_all("td")[1].text),
+                    "domain": domain,
+                    "digest": digest,
+                    "reason": reason,
                 })
         else:
             logger.warning("header_text='%s' not found in blocklist()=%d", header_text, len(blocklist))
@@ -134,18 +150,15 @@ def fetch_blocks(domain: str) -> list:
     logger.debug("domain='%s' - CALLED!", domain)
     domain_helper.raise_on(domain)
 
-    if not instances.is_registered(domain):
+    if blacklist.is_blacklisted(domain):
+        raise Exception(f"domain='{domain}' is blacklisted but function is invoked.")
+    elif not instances.is_registered(domain):
         raise Exception(f"domain='{domain}' is not registered but function is invoked.")
 
     blocklist = list()
 
-    logger.debug("Invoking federation.fetch_blocks(%s) ...", domain)
-    rows = federation.fetch_blocks(domain)
-
-    logger.debug("rows[%s]()=%d", type(rows), len(rows))
-    if len(rows) == 0:
-        logger.debug("domain='%s' has returned zero rows, trying /about/more page ...", domain)
-        rows = fetch_blocks_from_about(domain)
+    logger.debug("Invoking fetch_blocks_from_about(%s) ...", domain)
+    rows = fetch_blocks_from_about(domain)
 
     logger.debug("rows[%s]()=%d", type(rows), len(rows))
     if len(rows) > 0:
@@ -157,6 +170,7 @@ def fetch_blocks(domain: str) -> list:
                 logger.debug("block[]='%s' is of type 'dict' - SKIPPED!", type(block))
                 continue
             elif "domain" not in block:
+                logger.debug("block='%s'", block)
                 logger.warning("block()=%d does not contain element 'domain' - SKIPPED!", len(block))
                 continue
             elif not domain_helper.is_wanted(block["domain"]):
@@ -178,7 +192,7 @@ def fetch_blocks(domain: str) -> list:
             blocklist.append({
                 "blocker"    : domain,
                 "blocked"    : block["domain"],
-                "hash"       : block["digest"] if "digest" in block else None,
+                "digest"     : block["digest"] if "digest" in block else None,
                 "reason"     : reason,
                 "block_level": blocks.alias_block_level(block["severity"]),
             })