api.py

   1 from fastapi import FastAPI, Request, HTTPException
   2 import sqlite3
   3 from hashlib import sha256
   4 from fastapi.templating import Jinja2Templates
   5 from requests import get
   6
   7 base_url = ""
   8 app = FastAPI(docs_url=base_url+"/docs", redoc_url=base_url+"/redoc")
   9 templates = Jinja2Templates(directory=".")
  10
  11 def get_hash(domain: str) -> str:
  12     return sha256(domain.encode("utf-8")).hexdigest()
  13
  14 @app.get(base_url+"/info")
  15 def info():
  16     conn = sqlite3.connect("blocks.db")
  17     c = conn.cursor()
  18     c.execute("select (select count(domain) from instances), (select count(domain) from instances where software in ('pleroma', 'mastodon')), (select count(blocker) from blocks)")
  19     known, indexed, blocks = c.fetchone()
  20     c.close()
  21     return {
  22         "known_instances": known,
  23         "indexed_instances": indexed,
  24         "blocks_recorded": blocks,
  25         "source_code": "https://gitlab.com/EnjuAihara/fedi-block-api",
  26     }
  27
  28 @app.get(base_url+"/api")
  29 def blocked(domain: str = None):
  30     if domain == None:
  31         raise HTTPException(status_code=400, detail="No domain specified")
  32     conn = sqlite3.connect("blocks.db")
  33     c = conn.cursor()
  34     wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
  35     c.execute("select blocker, block_level, reason from blocks where blocked = ? or blocked = ? or blocked = ?", (domain, wildchar, get_hash(domain)))
  36     blocks = c.fetchall()
  37     conn.close()
  38
  39     result = {}
  40     reasons = {}
  41
  42     for domain, block_level, reason in blocks:
  43         if block_level in result:
  44             result[block_level].append(domain)
  45         else:
  46             result[block_level] = [domain]
  47
  48         if reason != "":
  49             if block_level in reasons:
  50                 reasons[block_level][domain] = reason
  51             else:
  52                 reasons[block_level] = {domain: reason}
  53
  54     return {"blocks": result, "reasons": reasons}
  55
  56 @app.get(base_url+"/")
  57 def index(request: Request, domain: str = None):
  58     blocks = get(f"http://127.0.0.1:8069{base_url}/api?domain={domain}")
  59     info = None
  60     if domain == None:
  61         info = get(f"http://127.0.0.1:8069{base_url}/info")
  62         if not info.ok:
  63             raise HTTPException(status_code=info.status_code, detail=info.text)
  64         info = info.json()
  65     if not blocks.ok:
  66         raise HTTPException(status_code=blocks.status_code, detail=blocks.text)
  67     return templates.TemplateResponse("index.html", {"request": request, "domain": domain, "blocks": blocks.json(), "info": info})