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 ]
  18
  19 headers = {
  20     "user-agent": config["useragent"]
  21 }
  22
  23
  24 def get_hash(domain: str) -> str:
  25     return sha256(domain.encode("utf-8")).hexdigest()
  26
  27
  28 def get_peers(domain: str) -> str:
  29     try:
  30         res = get(f"https://{domain}/api/v1/instance/peers", headers=headers, timeout=5)
  31         return res.json()
  32     except:
  33         return None
  34
  35 peerlist = get_peers(domain)
  36
  37 def get_type(instdomain: str) -> str:
  38     try:
  39         res = get(f"https://{instdomain}/nodeinfo/2.1.json", headers=headers, timeout=5)
  40         if res.status_code == 404:
  41             res = get(f"https://{instdomain}/nodeinfo/2.0", headers=headers, timeout=5)
  42         if res.status_code == 404:
  43             res = get(f"https://{instdomain}/nodeinfo/2.0.json", headers=headers, timeout=5)
  44         if res.ok and "text/html" in res.headers["content-type"]:
  45             res = get(f"https://{instdomain}/nodeinfo/2.1", headers=headers, timeout=5)
  46         if res.ok:
  47             if res.json()["software"]["name"] in ["akkoma", "rebased"]:
  48                 return "pleroma"
  49             elif res.json()["software"]["name"] in ["hometown", "ecko"]:
  50                 return "mastodon"
  51             elif res.json()["software"]["name"] in ["calckey", "groundpolis", "foundkey", "cherrypick"]:
  52                 return "misskey"
  53             else:
  54                 return res.json()["software"]["name"]
  55         elif res.status_code == 404:
  56             res = get(f"https://{instdomain}/api/v1/instance", headers=headers, timeout=5)
  57         if res.ok:
  58             return "mastodon"
  59     except:
  60         return None
  61
  62
  63 conn = sqlite3.connect("blocks.db")
  64 c = conn.cursor()
  65
  66 c.execute(
  67     "select domain from instances where 1"
  68 )
  69
  70 for instance in peerlist:
  71     instance = instance.lower()
  72
  73     blacklisted = False
  74     for domain in blacklist:
  75         if domain in instance:
  76             blacklisted = True
  77
  78     if blacklisted:
  79         continue
  80
  81     print(instance)
  82     try:
  83         c.execute(
  84             "select domain from instances where domain = ?", (instance,)
  85         )
  86         if c.fetchone() == None:
  87             c.execute(
  88                 "insert into instances select ?, ?, ?",
  89                 (instance, get_hash(instance), get_type(instance)),
  90             )
  91         conn.commit()
  92     except Exception as e:
  93         print("error:", e, instance)
  94 conn.close()