This repository has been archived on 2022-10-07. You can view files and clone it, but cannot push or open issues or pull requests.
skynet-webportal/setup-scripts/bot_utils.py

195 lines
5.9 KiB
Python

#!/usr/bin/env python3
from urllib.request import urlopen, Request
from dotenv import load_dotenv
from pathlib import Path
from datetime import datetime
import urllib, json, os, traceback, discord, sys, re, subprocess, requests, io
# sc_precision is the number of hastings per siacoin
sc_precision = 10 ** 24
# Environment variable globals
api_endpoint, port, portal_name, bot_token, password = None, None, None, None, None
discord_client = None
setup_done = False
# Get the container name as an argument or use "sia" as default.
CONTAINER_NAME = "sia"
if len(sys.argv) > 2:
CONTAINER_NAME = sys.argv[2]
# find out local siad ip by inspecting its docker container
def get_docker_container_ip(container_name):
ip_regex = re.compile(r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$")
docker_cmd = (
"docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' "
+ container_name
)
output = subprocess.check_output(docker_cmd, shell=True).decode("utf-8")
return ip_regex.findall(output)[0]
# find siad api password by getting it out of the docker container
def get_api_password():
api_password_regex = re.compile(r"^\w+$")
docker_cmd = "docker exec {} cat /sia-data/apipassword".format(CONTAINER_NAME)
output = subprocess.check_output(docker_cmd, shell=True).decode("utf-8")
return api_password_regex.findall(output)[0]
def setup():
# Load dotenv file if possible.
# TODO: change all scripts to use named flags/params
if len(sys.argv) > 1:
env_path = Path(sys.argv[1])
load_dotenv(dotenv_path=env_path, override=True)
global bot_token
bot_token = os.getenv("DISCORD_BOT_TOKEN")
global bot_channel
bot_channel = os.getenv("DISCORD_BOT_CHANNEL", "skynet-server-health")
global bot_notify_role
bot_notify_role = os.getenv("DISCORD_BOT_NOTIFY_ROLE", "skynet-prod")
global portal_name
portal_name = os.getenv("SKYNET_SERVER_API")
global port
port = os.getenv("API_PORT", "9980")
global api_endpoint
api_endpoint = "http://{}:{}".format(get_docker_container_ip(CONTAINER_NAME), port)
siad.initialize()
global setup_done
setup_done = True
return bot_token
# send_msg sends the msg to the specified discord channel. If force_notify is set to true it adds "@here".
async def send_msg(client, msg, force_notify=False, file=None):
await client.wait_until_ready()
guild = client.guilds[0]
chan = None
for c in guild.channels:
if c.name == bot_channel:
chan = c
break
if chan is None:
print("Can't find channel {}".format(bot_channel))
# Get the prod team role
role = None
for r in guild.roles:
if r.name == bot_notify_role:
role = r
break
# Add the portal name.
msg = "**{}**: {}".format(portal_name, msg)
if file and isinstance(file, str):
is_json = is_json_string(file)
content_type = "application/json" if is_json else "text/plain"
ext = "json" if is_json else "txt"
filename = "{}-{}.{}".format(
CONTAINER_NAME, datetime.utcnow().strftime("%Y-%m-%d-%H:%M:%S"), ext
)
skylink = upload_to_skynet(file, filename, content_type=content_type)
if skylink:
msg = "{} {}".format(msg, skylink) # append skylink to message
file = None # clean file reference, we're using a skylink
else:
file = discord.File(
io.BytesIO(file.encode()), filename=filename
) # wrap text into discord file wrapper
if force_notify and role:
msg = "{} /cc {}".format(msg, role.mention)
await chan.send(msg, file=file)
def upload_to_skynet(contents, filename="file.txt", content_type="text/plain"):
files = {"file": (filename, contents, content_type)}
res = requests.post("https://siasky.net/skynet/skyfile", files=files)
if res.status_code == requests.codes["ok"]:
res_json = res.json()
return "https://siasky.net/" + res_json["skylink"]
return None
def is_json_string(str):
try:
json.loads(str)
return True
except ValueError:
return False
# siad class provides wrappers for the necessary siad commands.
class siad:
# initializes values for using the API (password and
# user-agent) so that all calls to urllib.request.urlopen have these set.
@staticmethod
def initialize():
# Setup a handler with the API password
username = ""
password_mgr = urllib.request.HTTPPasswordMgrWithDefaultRealm()
password_mgr.add_password(None, api_endpoint, username, get_api_password())
handler = urllib.request.HTTPBasicAuthHandler(password_mgr)
# Setup an opener with the correct user agent
opener = urllib.request.build_opener(handler)
opener.addheaders = [("User-agent", "Sia-Agent")]
# Install the opener.
# Now all calls to urllib.request.urlopen use our opener.
urllib.request.install_opener(opener)
# load_json reads the http response and decodes the JSON value
@staticmethod
def load_json(resp):
return json.loads(resp.decode("utf-8"))
@staticmethod
def get(endpoint):
if not setup_done:
setup()
resp = urllib.request.urlopen(api_endpoint + endpoint).read()
return siad.load_json(resp)
@staticmethod
def get_wallet():
if not setup_done:
setup()
resp = urllib.request.urlopen(api_endpoint + "/wallet").read()
return siad.load_json(resp)
@staticmethod
def get_renter():
if not setup_done:
setup()
resp = urllib.request.urlopen(api_endpoint + "/renter").read()
return siad.load_json(resp)
@staticmethod
def get_renter_contracts():
if not setup_done:
setup()
resp = urllib.request.urlopen(api_endpoint + "/renter/contracts").read()
return siad.load_json(resp)