48 lines
1.4 KiB
Python
48 lines
1.4 KiB
Python
import httpx
|
|
from nyutils import pp
|
|
from nyutils.sqlutils import db_to_py, sqlite3
|
|
|
|
def fetch_servers(l):
|
|
# Get each server's name / address from a List
|
|
return [Server(0, l.lid, s['name'], s['address']) for s in httpx.get(l.url).json()['servers']] # type: ignore
|
|
|
|
def down_servers(l):
|
|
# Get the servers from a List that are down / return errors
|
|
servers = fetch_servers(l)
|
|
scodes = {s.name: httpx.get(s.url, headers={"User-Agent": "SideStore"}).status_code for s in servers}
|
|
return {s: scodes[s] for s in scodes if scodes[s] != 200}
|
|
|
|
if __name__ == '__main__':
|
|
# Open / Create db
|
|
ani = sqlite3.connect('ani.db')
|
|
|
|
# Run creation schema script
|
|
with open("schema.sql", "r") as f:
|
|
ani.executescript(f.read())
|
|
|
|
# Convert db tables into Python types
|
|
db_to_py(ani, globals())
|
|
# Dynamically generated List + classmethod
|
|
l = List.select(ani) # type: ignore
|
|
|
|
if len(l) == 0:
|
|
# Insert default if none exist
|
|
List(0, "https://servers.sidestore.io/servers.json").insert(ani) # type: ignore
|
|
l = List.select(ani)[0] # type: ignore
|
|
else:
|
|
# Select first list
|
|
l = l[0]
|
|
|
|
# Fetch servers from list and attempt to insert them into db
|
|
for s in fetch_servers(l):
|
|
s.insert(ani)
|
|
|
|
# Commit table creations / insertions
|
|
ani.commit()
|
|
|
|
# print info
|
|
pp(List.select(ani)) # type: ignore
|
|
pp(Server.select(ani)) # type: ignore
|
|
pp(down_servers(l))
|
|
|