Added logging level

[fba.git] / api.py

diff --git a/api.py b/api.py
index 43967e48757c428a749e64d2fb3a8b978a0744e2..8e95e58d7e6e3c58964993d175e47ab61235a734 100644 (file)
--- a/api.py
+++ b/api.py
@@ -7,11 +7,13 @@ from requests import get
 from json import loads
 from re import sub
 from datetime import datetime
+from email import utils
 
 with open("config.json") as f:
     config = loads(f.read())
     base_url = config["base_url"]
     port = config["port"]
+    log_level = config["log_level"]
 app = FastAPI(docs_url=base_url+"/docs", redoc_url=base_url+"/redoc")
 templates = Jinja2Templates(directory=".")
 
@@ -22,7 +24,7 @@ def get_hash(domain: str) -> str:
 def info():
     conn = sqlite3.connect("blocks.db")
     c = conn.cursor()
-    c.execute("select (select count(domain) from instances), (select count(domain) from instances where software in ('pleroma', 'mastodon', 'misskey', 'gotosocial', 'friendica')), (select count(blocker) from blocks)")
+    c.execute("SELECT (SELECT count(domain) FROM  instances), (SELECT count(domain) FROM  instances WHERE software in ('pleroma', 'mastodon', 'misskey', 'gotosocial', 'friendica')), (SELECT count(blocker) FROM  blocks)")
     known, indexed, blocks = c.fetchone()
     c.close()
     return {
@@ -40,11 +42,11 @@ def top(blocked: int = None, blockers: int = None):
     elif blocked != None:
         if blocked > 500:
             raise HTTPException(status_code=400, detail="Too many results")
-        c.execute("select blocked, count(blocked) from blocks where block_level = 'reject' group by blocked order by count(blocked) desc limit ?", (blocked,))
+        c.execute("SELECT blocked, count(blocked) FROM  blocks WHERE block_level = 'reject' group by blocked order by count(blocked) desc limit ?", (blocked,))
     elif blockers != None:
         if blockers > 500:
             raise HTTPException(status_code=400, detail="Too many results")
-        c.execute("select blocker, count(blocker) from blocks where block_level = 'reject' group by blocker order by count(blocker) desc limit ?", (blockers,))
+        c.execute("SELECT blocker, count(blocker) FROM  blocks WHERE block_level = 'reject' group by blocker order by count(blocker) desc limit ?", (blockers,))
     scores = c.fetchall()
     c.close()
 
@@ -68,12 +70,12 @@ def blocked(domain: str = None, reason: str = None, reverse: str = None):
     if domain != None:
         wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
         punycode = domain.encode('idna').decode('utf-8')
-        c.execute("select blocker, blocked, block_level, reason, first_added, last_seen from blocks where blocked = ? or blocked = ? or blocked = ? or blocked = ? or blocked = ? or blocked = ? order by first_added asc",
+        c.execute("SELECT blocker, blocked, block_level, reason, first_added, last_seen FROM  blocks WHERE blocked = ? OR blocked = ? OR blocked = ? OR blocked = ? OR blocked = ? OR blocked = ? order by first_added asc",
                   (domain, "*." + domain, wildchar, get_hash(domain), punycode, "*." + punycode))
     elif reverse != None:
-        c.execute("select blocker, blocked, block_level, reason, first_added, last_seen from blocks where blocker = ? order by first_added asc", (reverse,))
+        c.execute("SELECT blocker, blocked, block_level, reason, first_added, last_seen FROM  blocks WHERE blocker = ? order by first_added asc", (reverse,))
     else:
-        c.execute("select blocker, blocked, block_level, reason, first_added, last_seen from blocks where reason like ? and reason != '' order by first_added asc", ("%"+reason+"%",))
+        c.execute("SELECT blocker, blocked, block_level, reason, first_added, last_seen FROM  blocks WHERE reason like ? AND reason != '' order by first_added asc", ("%"+reason+"%",))
     blocks = c.fetchall()
     c.close()
 
@@ -136,13 +138,14 @@ def mutual(domains: list[str] = Query()):
     c = conn.cursor()
     c.execute(
         "SELECT block_level FROM blocks " \
-        "WHERE ((blocker = :a OR blocker = :b) AND (blocked = :b OR blocked = :a OR blocked = :w)) " \
+        "WHERE ((blocker = :a OR blocker = :b) AND (blocked = :b OR blocked = :a OR blocked = :aw OR blocked = :bw)) " \
         "AND block_level = 'reject' " \
         "LIMIT 1",
         {
             "a": domains[0],
-            "w": "*." + domains[0],
             "b": domains[1],
+            "aw": "*." + domains[0],
+            "bw": "*." + domains[1],
         },
     )
     res = c.fetchone()
@@ -153,6 +156,32 @@ def mutual(domains: list[str] = Query()):
     # No known blocks
     return responses.JSONResponse(status_code=200, content={})
 
-if __name__ == "__main__":
-    uvicorn.run("api:app", host="127.0.0.1", port=port, log_level="info")
+@app.get(base_url+"/rss")
+def rss(request: Request, domain: str = None):
+    conn = sqlite3.connect("blocks.db")
+    c = conn.cursor()
+    if domain != None:
+        wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
+        punycode = domain.encode('idna').decode('utf-8')
+        c.execute("SELECT blocker, blocked, block_level, reason, first_added, last_seen FROM  blocks WHERE blocked = ? OR blocked = ? OR blocked = ? OR blocked = ? OR blocked = ? OR blocked = ? order by first_added desc limit 50",
+                  (domain, "*." + domain, wildchar, get_hash(domain), punycode, "*." + punycode))
+    else:
+        c.execute("SELECT blocker, blocked, block_level, reason, first_added, last_seen FROM  blocks order by first_added desc limit 50")
+    blocks = c.fetchall()
+    c.close()
+
+    result = []
+    for blocker, blocked, block_level, reason, first_added, last_seen in blocks:
+        first_added = utils.format_datetime(datetime.fromtimestamp(first_added))
+        if reason == None or reason == '':
+            reason = "No reason provided."
+        else:
+            reason = "Provided reason: '" + reason + "'"
+        result.append({"blocker": blocker, "blocked": blocked, "block_level": block_level, "reason": reason, "first_added": first_added})
+
+    timestamp = utils.format_datetime(datetime.now())
 
+    return templates.TemplateResponse("rss.xml", {"request": request, "timestamp": timestamp, "domain": domain, "blocks": result}, headers={"Content-Type": "application/rss+xml"})
+
+if __name__ == "__main__":
+    uvicorn.run("api:app", host="127.0.0.1", port=port, log_level=log_level)