]> git.mxchange.org Git - fba.git/blobdiff - fba/fba.py
projects / fba.git / blobdiff
? search:
summary | shortlog | log | commit | commitdiff | tree
raw | inline | side by side
Continued:
[fba.git] / fba / fba.py
diff --git a/fba/fba.py b/fba/fba.py
index 14d500b1a55429350c535bcea153382b0d14ebce..982cc628b3b3bb0eec1247e8aca425fb0616623b 100644 (file)
--- a/fba/fba.py
+++ b/fba/fba.py
@@ -25,11 +25,9 @@ import time
 import validators
 
 from fba import cache
+from fba import config
 from fba import instances
 
-with open("config.json") as f:
-    config = json.loads(f.read())
-
 # Don't check these, known trolls/flooders/testing/developing
 blacklist = [
     # Floods network with fake nodes as "research" project
@@ -66,12 +64,12 @@ nodeinfo_identifier = [
 
 # HTTP headers for non-API requests
 headers = {
-    "User-Agent": config["useragent"],
+    "User-Agent": config.get("useragent"),
 }
 
 # HTTP headers for API requests
 api_headers = {
-    "User-Agent": config["useragent"],
+    "User-Agent": config.get("useragent"),
     "Content-Type": "application/json",
 }
 
@@ -400,8 +398,8 @@ def log_error(domain: str, res: any):
             ])
 
         # Cleanup old entries
-        # DEBUG: print(f"DEBUG: Purging old records (distance: {config['error_log_cleanup']})")
-        cursor.execute("DELETE FROM error_log WHERE created < ?", [time.time() - config["error_log_cleanup"]])
+        # DEBUG: print(f"DEBUG: Purging old records (distance: {config.get('error_log_cleanup')})")
+        cursor.execute("DELETE FROM error_log WHERE created < ?", [time.time() - config.get("error_log_cleanup")])
     except BaseException as e:
         print(f"ERROR: failed SQL query: domain='{domain}',exception[{type(e)}]:'{str(e)}'")
         sys.exit(255)
@@ -485,7 +483,7 @@ def get_peers(domain: str, software: str) -> list:
     if software == "misskey":
         # DEBUG: print(f"DEBUG: domain='{domain}' is misskey, sending API POST request ...")
         offset = 0
-        step = config["misskey_offset"]
+        step = config.get("misskey_offset")
 
         # iterating through all "suspended" (follow-only in its terminology)
         # instances page-by-page, since that troonware doesn't support
@@ -510,9 +508,9 @@ def get_peers(domain: str, software: str) -> list:
             if len(fetched) == 0:
                 # DEBUG: print("DEBUG: Returned zero bytes, exiting loop:", domain)
                 break
-            elif len(fetched) != config["misskey_offset"]:
-                # DEBUG: print(f"DEBUG: Fetched '{len(fetched)}' row(s) but expected: '{config['misskey_offset']}'")
-                offset = offset + (config["misskey_offset"] - len(fetched))
+            elif len(fetched) != config.get("misskey_offset"):
+                # DEBUG: print(f"DEBUG: Fetched '{len(fetched)}' row(s) but expected: '{config.get('misskey_offset')}'")
+                offset = offset + (config.get("misskey_offset") - len(fetched))
             else:
                 # DEBUG: print("DEBUG: Raising offset by step:", step)
                 offset = offset + step
@@ -550,7 +548,7 @@ def get_peers(domain: str, software: str) -> list:
     elif software == "lemmy":
         # DEBUG: print(f"DEBUG: domain='{domain}' is Lemmy, fetching JSON ...")
         try:
-            res = reqto.get(f"https://{domain}/api/v3/site", headers=api_headers, timeout=(config["connection_timeout"], config["read_timeout"]))
+            res = reqto.get(f"https://{domain}/api/v3/site", headers=api_headers, timeout=(config.get("connection_timeout"), config.get("read_timeout")))
 
             data = res.json()
             # DEBUG: print(f"DEBUG: res.ok={res.ok},res.status_code='{res.status_code}',data[]='{type(data)}'")
@@ -587,7 +585,7 @@ def get_peers(domain: str, software: str) -> list:
             # DEBUG: print(f"DEBUG: domain='{domain}',mode='{mode}'")
             while True:
                 try:
-                    res = reqto.get(f"https://{domain}/api/v1/server/{mode}?start={start}&count=100", headers=headers, timeout=(config["connection_timeout"], config["read_timeout"]))
+                    res = reqto.get(f"https://{domain}/api/v1/server/{mode}?start={start}&count=100", headers=headers, timeout=(config.get("connection_timeout"), config.get("read_timeout")))
 
                     data = res.json()
                     # DEBUG: print(f"DEBUG: res.ok={res.ok},res.status_code='{res.status_code}',data[]='{type(data)}'")
@@ -624,13 +622,13 @@ def get_peers(domain: str, software: str) -> list:
 
     # DEBUG: print(f"DEBUG: Fetching get_peers_url='{get_peers_url}' from '{domain}' ...")
     try:
-        res = reqto.get(f"https://{domain}{get_peers_url}", headers=api_headers, timeout=(config["connection_timeout"], config["read_timeout"]))
+        res = reqto.get(f"https://{domain}{get_peers_url}", headers=api_headers, timeout=(config.get("connection_timeout"), config.get("read_timeout")))
 
         data = res.json()
         # DEBUG: print(f"DEBUG: res.ok={res.ok},res.status_code={res.status_code},data[]='{type(data)}'")
         if not res.ok or res.status_code >= 400:
             # DEBUG: print(f"DEBUG: Was not able to fetch '{get_peers_url}', trying alternative ...")
-            res = reqto.get(f"https://{domain}/api/v3/site", headers=api_headers, timeout=(config["connection_timeout"], config["read_timeout"]))
+            res = reqto.get(f"https://{domain}/api/v3/site", headers=api_headers, timeout=(config.get("connection_timeout"), config.get("read_timeout")))
 
             data = res.json()
             # DEBUG: print(f"DEBUG: res.ok={res.ok},res.status_code={res.status_code},data[]='{type(data)}'")
@@ -679,7 +677,7 @@ def post_json_api(domain: str, path: str, parameter: str, extra_headers: dict =
     # DEBUG: print("DEBUG: Sending POST to domain,path,parameter:", domain, path, parameter, extra_headers)
     data = {}
     try:
-        res = reqto.post(f"https://{domain}{path}", data=parameter, headers={**api_headers, **extra_headers}, timeout=(config["connection_timeout"], config["read_timeout"]))
+        res = reqto.post(f"https://{domain}{path}", data=parameter, headers={**api_headers, **extra_headers}, timeout=(config.get("connection_timeout"), config.get("read_timeout")))
 
         data = res.json()
         # DEBUG: print(f"DEBUG: res.ok={res.ok},res.status_code={res.status_code},data[]='{type(data)}'")
@@ -727,7 +725,7 @@ def fetch_nodeinfo(domain: str, path: str = None) -> list:
 
         try:
             # DEBUG: print("DEBUG: Fetching request:", request)
-            res = reqto.get(request, headers=api_headers, timeout=(config["connection_timeout"], config["read_timeout"]))
+            res = reqto.get(request, headers=api_headers, timeout=(config.get("connection_timeout"), config.get("read_timeout")))
 
             data = res.json()
             # DEBUG: print(f"DEBUG: res.ok={res.ok},res.status_code={res.status_code},data[]='{type(data)}'")
@@ -763,7 +761,7 @@ def fetch_wellknown_nodeinfo(domain: str) -> list:
     data = {}
 
     try:
-        res = reqto.get(f"https://{domain}/.well-known/nodeinfo", headers=api_headers, timeout=(config["connection_timeout"], config["read_timeout"]))
+        res = reqto.get(f"https://{domain}/.well-known/nodeinfo", headers=api_headers, timeout=(config.get("connection_timeout"), config.get("read_timeout")))
 
         data = res.json()
         # DEBUG: print("DEBUG: domain,res.ok,data[]:", domain, res.ok, type(data))
@@ -814,7 +812,7 @@ def fetch_generator_from_path(domain: str, path: str = "/") -> str:
 
     try:
         # DEBUG: print(f"DEBUG: Fetching path='{path}' from '{domain}' ...")
-        res = reqto.get(f"https://{domain}{path}", headers=headers, timeout=(config["connection_timeout"], config["read_timeout"]))
+        res = reqto.get(f"https://{domain}{path}", headers=headers, timeout=(config.get("connection_timeout"), config.get("read_timeout")))
 
         # DEBUG: print("DEBUG: domain,res.ok,res.status_code,res.text[]:", domain, res.ok, res.status_code, type(res.text))
         if res.ok and res.status_code < 300 and len(res.text) > 0:
@@ -1186,13 +1184,13 @@ def send_bot_post(instance: str, blocks: dict):
     if truncated:
         message = message + "(the list has been truncated to the first 20 entries)"
 
-    botheaders = {**api_headers, **{"Authorization": "Bearer " + config["bot_token"]}}
+    botheaders = {**api_headers, **{"Authorization": "Bearer " + config.get("bot_token")}}
 
     req = reqto.post(
-        f"{config['bot_instance']}/api/v1/statuses",
+        f"{config.get('bot_instance')}/api/v1/statuses",
         data={
             "status"      : message,
-            "visibility"  : config['bot_visibility'],
+            "visibility"  : config.get('bot_visibility'),
             "content_type": "text/plain"
         },
         headers=botheaders,
@@ -1218,7 +1216,7 @@ def get_mastodon_blocks(domain: str) -> dict:
 
     try:
         doc = bs4.BeautifulSoup(
-            reqto.get(f"https://{domain}/about", headers=headers, timeout=(config["connection_timeout"], config["read_timeout"])).text,
+            reqto.get(f"https://{domain}/about", headers=headers, timeout=(config.get("connection_timeout"), config.get("read_timeout"))).text,
             "html.parser",
         )
     except BaseException as e:
@@ -1263,7 +1261,7 @@ def get_friendica_blocks(domain: str) -> dict:
 
     try:
         doc = bs4.BeautifulSoup(
-            reqto.get(f"https://{domain}/friendica", headers=headers, timeout=(config["connection_timeout"], config["read_timeout"])).text,
+            reqto.get(f"https://{domain}/friendica", headers=headers, timeout=(config.get("connection_timeout"), config.get("read_timeout"))).text,
             "html.parser",
         )
     except BaseException as e:
@@ -1304,7 +1302,7 @@ def get_misskey_blocks(domain: str) -> dict:
     }
 
     offset = 0
-    step = config["misskey_offset"]
+    step = config.get("misskey_offset")
     while True:
         # iterating through all "suspended" (follow-only in its terminology)
         # instances page-by-page, since that troonware doesn't support
@@ -1333,9 +1331,9 @@ def get_misskey_blocks(domain: str) -> dict:
             if len(fetched) == 0:
                 # DEBUG: print("DEBUG: Returned zero bytes, exiting loop:", domain)
                 break
-            elif len(fetched) != config["misskey_offset"]:
-                # DEBUG: print(f"DEBUG: Fetched '{len(fetched)}' row(s) but expected: '{config['misskey_offset']}'")
-                offset = offset + (config["misskey_offset"] - len(fetched))
+            elif len(fetched) != config.get("misskey_offset"):
+                # DEBUG: print(f"DEBUG: Fetched '{len(fetched)}' row(s) but expected: '{config.get('misskey_offset')}'")
+                offset = offset + (config.get("misskey_offset") - len(fetched))
             else:
                 # DEBUG: print("DEBUG: Raising offset by step:", step)
                 offset = offset + step
@@ -1382,9 +1380,9 @@ def get_misskey_blocks(domain: str) -> dict:
             if len(fetched) == 0:
                 # DEBUG: print("DEBUG: Returned zero bytes, exiting loop:", domain)
                 break
-            elif len(fetched) != config["misskey_offset"]:
-                # DEBUG: print(f"DEBUG: Fetched '{len(fetched)}' row(s) but expected: '{config['misskey_offset']}'")
-                offset = offset + (config["misskey_offset"] - len(fetched))
+            elif len(fetched) != config.get("misskey_offset"):
+                # DEBUG: print(f"DEBUG: Fetched '{len(fetched)}' row(s) but expected: '{config.get('misskey_offset')}'")
+                offset = offset + (config.get("misskey_offset") - len(fetched))
             else:
                 # DEBUG: print("DEBUG: Raising offset by step:", step)
                 offset = offset + step
@@ -1422,7 +1420,7 @@ def tidyup_domain(domain: str) -> str:
     # No port number
     domain = re.sub("\:\d+$", "", domain)
 
-    # No protocol, sometimes with the slashes
+    # No protocol, sometimes without the slashes
     domain = re.sub("^https?\:(\/*)", "", domain)
 
     # No trailing slash
Personal clone of "fedi-block-api", if you like to use the command fetch_instances_social you need to get an API key from https://instances.social .