]> git.mxchange.org Git - fba.git/blobdiff - fba/utils.py
Continued:
[fba.git] / fba / utils.py
index 56a004364ca86177e568d7376566aaeb0d4dfb58..6652c19261e7747d268f5213d3d220e207f1af14 100644 (file)
@@ -20,9 +20,8 @@ from urllib.parse import urlparse
 
 import bs4
 import requests
-import validators
 
-from fba.helpers import blacklist
+from fba.helpers import config
 from fba.helpers import domain as domain_helper
 from fba.helpers import tidyup
 
@@ -47,6 +46,7 @@ def get_hash(domain: str) -> str:
 
 def fetch_url(url: str, headers: dict, timeout: tuple) -> requests.models.Response:
     logger.debug("url='%s',headers()=%d,timeout(%d)='%s' - CALLED!", url, len(headers), len(timeout), timeout)
+
     if not isinstance(url, str):
         raise ValueError(f"Parameter url[]='{type(url)}' is not of type 'str'")
     elif url == "":
@@ -62,15 +62,16 @@ def fetch_url(url: str, headers: dict, timeout: tuple) -> requests.models.Respon
     # Invoke other function, avoid trailing ?
     logger.debug("components[%s]='%s'", type(components), components)
     if components.query != "":
-        response = network.fetch_response(components.netloc, f"{components.path}?{components.query}", headers, timeout)
+        response = network.fetch_response(components.netloc.split(":")[0], f"{components.path}?{components.query}", headers, timeout)
     else:
-        response = network.fetch_response(components.netloc, components.path if isinstance(components.path, str) and components.path != '' else '/', headers, timeout)
+        response = network.fetch_response(components.netloc.split(":")[0], components.path if isinstance(components.path, str) and components.path != '' else '/', headers, timeout)
 
     logger.debug("response[]='%s' - EXIT!", type(response))
     return response
 
 def find_domains(tags: bs4.element.ResultSet, search: str) -> list:
     logger.debug("tags[%s]()=%d,search='%s' - CALLED!", type(tags), len(tags), search)
+
     if not isinstance(tags, bs4.element.ResultSet):
         raise ValueError(f"Parameter tags[]='{type(tags)}' is not of type 'ResultSet'")
     elif not isinstance(search, str):
@@ -98,7 +99,7 @@ def find_domains(tags: bs4.element.ResultSet, search: str) -> list:
         domain = domain.encode("idna").decode("utf-8")
         logger.debug("domain='%s' - AFTER!", domain)
 
-        if not is_domain_wanted(domain):
+        if not domain_helper.is_wanted(domain):
             logger.debug("domain='%s' is not wanted - SKIPPED!", domain)
             continue
 
@@ -108,41 +109,6 @@ def find_domains(tags: bs4.element.ResultSet, search: str) -> list:
     logger.debug("domains()=%d - EXIT!", len(domains))
     return domains
 
-def is_domain_wanted(domain: str) -> bool:
-    logger.debug("domain='%s' - CALLED!", domain)
-
-    wanted = True
-    if not isinstance(domain, str):
-        raise ValueError(f"Parameter domain[]='{type(domain)}' is not of type 'str'")
-    elif domain == "":
-        raise ValueError("Parameter 'domain' is empty")
-    elif domain.lower() != domain:
-        wanted = False
-    elif not validators.domain(domain.split("/")[0]):
-        logger.debug("domain='%s' is not a valid domain name - settings False ...", domain)
-        wanted = False
-    elif domain.endswith(".arpa"):
-        logger.debug("domain='%s' is a domain for reversed IP addresses - settings False ...", domain)
-        wanted = False
-    elif domain.endswith(".onion"):
-        logger.debug("domain='%s' is a TOR .onion domain - settings False ...", domain)
-        wanted = False
-    elif domain.endswith(".tld"):
-        logger.debug("domain='%s' is a fake domain - settings False ...", domain)
-        wanted = False
-    elif blacklist.is_blacklisted(domain):
-        logger.debug("domain='%s' is blacklisted - settings False ...", domain)
-        wanted = False
-    elif domain.find("/profile/") > 0 or domain.find("/users/") > 0 or (instances.is_registered(domain.split("/")[0]) and domain.find("/c/") > 0):
-        logger.debug("domain='%s' is a single user", domain)
-        wanted = False
-    elif domain.find("/tag/") > 0:
-        logger.debug("domain='%s' is a tag", domain)
-        wanted = False
-
-    logger.debug("wanted='%s' - EXIT!", wanted)
-    return wanted
-
 def deobfuscate(domain: str, blocker: str, domain_hash: str = None) -> str:
     logger.debug("domain='%s',blocker='%s',domain_hash='%s' - CALLED!", domain, blocker, domain_hash)
     domain_helper.raise_on(blocker)
@@ -154,8 +120,12 @@ def deobfuscate(domain: str, blocker: str, domain_hash: str = None) -> str:
     elif not isinstance(domain_hash, str) and domain_hash is not None:
         raise ValueError(f"Parameter domain_hash[]='{type(domain_hash)}' is not of type 'str'")
 
+    logger.debug("Setting has_obfuscation=False for blocker='%s' ...", blocker)
+    instances.set_has_obfuscation(blocker, False)
+
     if domain.find("*") >= 0:
         logger.debug("blocker='%s' uses obfuscated domains", blocker)
+        instances.set_has_obfuscation(blocker, True)
 
         # Obscured domain name with no hash
         row = instances.deobfuscate("*", domain, domain_hash)
@@ -166,9 +136,9 @@ def deobfuscate(domain: str, blocker: str, domain_hash: str = None) -> str:
             domain = row["domain"]
         else:
             logger.warning("blocker='%s' has domain='%s' that cannot be deobfuscated.", blocker, domain)
-            instances.set_has_obfuscation(blocker, True)
     elif domain.find("?") >= 0:
         logger.debug("blocker='%s' uses obfuscated domains", blocker)
+        instances.set_has_obfuscation(blocker, True)
 
         # Obscured domain name with no hash
         row = instances.deobfuscate("?", domain, domain_hash)
@@ -179,32 +149,11 @@ def deobfuscate(domain: str, blocker: str, domain_hash: str = None) -> str:
             domain = row["domain"]
         else:
             logger.warning("blocker='%s' has domain='%s' that cannot be deobfuscated.", blocker, domain)
-            instances.set_has_obfuscation(blocker, True)
     else:
         logger.debug("domain='%s' is not obfuscated", domain)
 
     logger.debug("domain='%s' - EXIT!", domain)
     return domain
 
-def alias_block_level(block_level: str) -> str:
-    logger.debug("block_level='%s' - CALLED!", block_level)
-    if not isinstance(block_level, str):
-        raise ValueError(f"Parameter block_level[]='%s' is not of type 'str'", type(block_level))
-    elif block_level == "":
-        raise ValueError("Parameter 'block_level' is empty")
-
-    if block_level == "silence":
-        logger.debug("Block level 'silence' has been changed to 'silenced'")
-        block_level = "silenced"
-    elif block_level == "suspend":
-        logger.debug("Block level 'suspend' has been changed to 'suspended'")
-        block_level = "suspended"
-    elif block_level == "nsfw":
-        logger.debug("Block level 'nsfw' has been changed to 'media_nsfw'")
-        block_level = "media_nsfw"
-    elif block_level == "quarantined_instances":
-        logger.debug("Block level 'quarantined_instances' has been changed to 'quarantined'")
-        block_level = "quarantined"
-
-    logger.debug("block_level='%s' - EXIT!", block_level)
-    return block_level
+def base_url() -> str:
+    return f"{config.get('scheme')}://{config.get('hostname')}{config.get('base_url')}"