fetch_instances.py

   1 from reqto import get
   2 from hashlib import sha256
   3 import sqlite3
   4 import sys
   5 import json
   6
   7 with open("config.json") as f:
   8     config = json.loads(f.read())
   9
  10 domain = sys.argv[1]
  11
  12 blacklist = [
  13     "activitypub-troll.cf",
  14     "gab.best",
  15     "4chan.icu",
  16     "social.shrimpcam.pw",
  17     "mastotroll.netz.org"
  18 ]
  19
  20 headers = {
  21     "user-agent": config["useragent"]
  22 }
  23
  24
  25 def get_hash(domain: str) -> str:
  26     return sha256(domain.encode("utf-8")).hexdigest()
  27
  28
  29 def get_peers(domain: str) -> str:
  30     try:
  31         res = get(f"https://{domain}/api/v1/instance/peers", headers=headers, timeout=5)
  32         return res.json()
  33     except:
  34         return None
  35
  36 peerlist = get_peers(domain)
  37
  38 def get_type(instdomain: str) -> str:
  39     try:
  40         res = get(f"https://{instdomain}/nodeinfo/2.1.json", headers=headers, timeout=5)
  41         if res.status_code == 404:
  42             res = get(f"https://{instdomain}/nodeinfo/2.0", headers=headers, timeout=5)
  43         if res.status_code == 404:
  44             res = get(f"https://{instdomain}/nodeinfo/2.0.json", headers=headers, timeout=5)
  45         if res.ok and "text/html" in res.headers["content-type"]:
  46             res = get(f"https://{instdomain}/nodeinfo/2.1", headers=headers, timeout=5)
  47         if res.ok:
  48             if res.json()["software"]["name"] in ["akkoma", "rebased"]:
  49                 return "pleroma"
  50             elif res.json()["software"]["name"] in ["hometown", "ecko"]:
  51                 return "mastodon"
  52             elif res.json()["software"]["name"] in ["calckey", "groundpolis", "foundkey", "cherrypick"]:
  53                 return "misskey"
  54             else:
  55                 return res.json()["software"]["name"]
  56         elif res.status_code == 404:
  57             res = get(f"https://{instdomain}/api/v1/instance", headers=headers, timeout=5)
  58         if res.ok:
  59             return "mastodon"
  60     except:
  61         return None
  62
  63
  64 conn = sqlite3.connect("blocks.db")
  65 c = conn.cursor()
  66
  67 c.execute(
  68     "SELECT domain FROM  instances WHERE 1"
  69 )
  70
  71 for instance in peerlist:
  72     instance = instance.lower()
  73
  74     blacklisted = False
  75     for domain in blacklist:
  76         if domain in instance:
  77             blacklisted = True
  78
  79     if blacklisted:
  80         continue
  81
  82     print(instance)
  83     try:
  84         c.execute(
  85             "SELECT domain FROM  instances WHERE domain = ?", (instance,)
  86         )
  87         if c.fetchone() == None:
  88             c.execute(
  89                 "INSERT INTO instances SELECT ?, ?, ?",
  90                 (instance, get_hash(instance), get_type(instance)),
  91             )
  92         conn.commit()
  93     except Exception as e:
  94         print("error:", e, instance)
  95 conn.close()