black

setup-scripts/blocklist-airtable.py

@@ -6,75 +6,95 @@ from bot_utils import setup, send_msg
 bot_token = setup()
 client = discord.Client()
 
-AIRTABLE_API_KEY = os.getenv('AIRTABLE_API_KEY')
-AIRTABLE_BASE = os.getenv('AIRTABLE_BASE', 'app89plJvA9EqTJEc')
-AIRTABLE_TABLE = os.getenv('AIRTABLE_TABLE', 'Table%201')
-AIRTABLE_FIELD = os.getenv('AIRTABLE_FIELD', 'Link')
+AIRTABLE_API_KEY = os.getenv("AIRTABLE_API_KEY")
+AIRTABLE_BASE = os.getenv("AIRTABLE_BASE", "app89plJvA9EqTJEc")
+AIRTABLE_TABLE = os.getenv("AIRTABLE_TABLE", "Table%201")
+AIRTABLE_FIELD = os.getenv("AIRTABLE_FIELD", "Link")
+
+
+def exec(command):
+    return os.popen(command).read().strip()
+
 
 async def block_skylinks_from_airtable():
-    print("Pulling blocked skylinks from airtable via api integration")
+    print("Pulling blocked skylinks from Airtable via api integration")
     headers = {"Authorization": "Bearer " + AIRTABLE_API_KEY}
     skylinks = []
     offset = None
     while len(skylinks) == 0 or offset:
-        print("Requesting a batch of records from airtable with " + (offset if offset else "empty") + " offset")
-        query = '&'.join(['fields%5B%5D=' + AIRTABLE_FIELD, ('offset=' + offset) if offset else ''])
-        airtable = requests.get(
-            "https://api.airtable.com/v0/" + AIRTABLE_BASE + "/" + AIRTABLE_TABLE + "?" + query, headers=headers
+        print("Requesting a batch of records from Airtable with " + (offset if offset else "empty") + " offset")
+        query = "&".join(["fields%5B%5D=" + AIRTABLE_FIELD, ("offset=" + offset) if offset else ""])
+        response = requests.get(
+            "https://api.airtable.com/v0/" + AIRTABLE_BASE + "/" + AIRTABLE_TABLE + "?" + query,
+            headers=headers,
         )
 
-        if airtable.status_code != 200:
-            message = "Airtable blocklist integration responded with code " + str(airtable.status_code) + ": " + (airtable.text or "empty response")
+        if response.status_code != 200:
+            status_code = str(response.status_code)
+            response_text = response.text or "empty response"
+            message = "Airtable blocklist integration responded with code " + status_code + ": " + response_text
             return print(message) or await send_msg(client, message, force_notify=False)
 
-        airtable_data = airtable.json()
-        skylinks = skylinks + [entry['fields'][AIRTABLE_FIELD] for entry in airtable_data['records']]
+        data = response.json()
+        skylinks = skylinks + [entry["fields"][AIRTABLE_FIELD] for entry in data["records"]]
 
         if len(skylinks) == 0:
             return print("Airtable returned 0 skylinks - make sure your configuration is correct")
 
-        offset = airtable_data.get('offset')
+        offset = data.get("offset")
 
     print("Airtable returned total " + str(len(skylinks)) + " skylinks to block")
 
-    apipassword = os.popen('docker exec sia cat /sia-data/apipassword').read().strip()
-    ipaddress = os.popen('docker inspect -f \'{{range.NetworkSettings.Networks}}{{.IPAddress}}{{end}}\' sia').read().strip()
+    apipassword = exec("docker exec sia cat /sia-data/apipassword")
+    ipaddress = exec("docker inspect -f '{{range.NetworkSettings.Networks}}{{.IPAddress}}{{end}}' sia")
 
     print("Sending blocklist request to siad")
-    headers = { 'user-agent': 'Sia-Agent' }
-    auth = ('', apipassword)
-    data = json.dumps({ 'add': skylinks })
-    response = requests.post('http://' + ipaddress + ':9980/skynet/blocklist', auth = auth, headers = headers, data = data)
+    response = requests.post(
+        "http://" + ipaddress + ":9980/skynet/blocklist",
+        auth=("", apipassword),
+        headers={"user-agent": "Sia-Agent"},
+        data=json.dumps({"add": skylinks}),
+    )
 
     if response.status_code == 204:
         print("Siad blocklist successfully updated with provided skylink")
     else:
-        message = "Siad blocklist endpoint responded with code " + str(response.status_code) + ": " + (response.text or "empty response")
+        status_code = str(response.status_code)
+        response_text = response.text or "empty response"
+        message = "Siad blocklist endpoint responded with code " + status_code + ": " + response_text
         return print(message) or await send_msg(client, message, force_notify=False)
 
     print("Searching nginx cache for blocked files")
-    cached_files_command = '/usr/bin/find /data/nginx/cache/ -type f | /usr/bin/xargs --no-run-if-empty -n1000 /bin/grep -Els \'^KEY: .*(' + '|'.join(skylinks) + ')\''
-    cached_files_count = int(os.popen('docker exec -it nginx bash -c "' + cached_files_command + ' | wc -l"').read().strip())
+    cached_files_command = (
+        "/usr/bin/find /data/nginx/cache/ -type f | /usr/bin/xargs --no-run-if-empty -n1000 /bin/grep -Els '^KEY: .*("
+        + "|".join(skylinks)
+        + ")'"
+    )
+    cached_files_count = int(exec('docker exec -it nginx bash -c "' + cached_files_command + ' | wc -l"'))
 
     if cached_files_count == 0:
         return print("No nginx cached files matching blocked skylinks were found")
 
-    os.popen('docker exec -it nginx bash -c "' + cached_files_command + ' | xargs rm"')
-    message = 'Purged ' + str(cached_files_count) + ' blocklisted files from nginx cache'
+    exec('docker exec -it nginx bash -c "' + cached_files_command + ' | xargs rm"')
+    message = "Purged " + str(cached_files_count) + " blocklisted files from nginx cache"
     return print(message) or await send_msg(client, message)
 
+
 async def exit_after(delay):
     await asyncio.sleep(delay)
     os._exit(0)
 
+
 @client.event
 async def on_ready():
     try:
         await block_skylinks_from_airtable()
     except:  # catch all exceptions
-        await send_msg(client, "```\n{}\n```".format(traceback.format_exc()), force_notify=False)
+        message = "```\n{}\n```".format(traceback.format_exc())
+        await send_msg(client, message, force_notify=False)
     asyncio.create_task(exit_after(3))
 
+
 client.run(bot_token)
 
 # asyncio.run(on_ready())