HIWIFI-SS-Scheduler

.gitignore

__pycache__
config.py

config.py.example

host = "192.168.199.1"
port = 80
username = "admin"
password = "12345678"
servers = [
    {
        "server": "1.2.3.4",
        "server_port": "1337",
        "password": "12345678",
        "timeout": "300",
        "method": "aes-256-cfb",
        "defaultroute": "0",
        "dnsserver": "8.8.4.4",
        "plugin_opts": "obfs=http;obfs-host=www.bing.com",
        "udp_relay": "0",
        "plugin_enable": "0",
    },
]

main.py

#!/usr/bin/env python

import requests
import json
import config
import random
import functools

def login_required(func):
    @functools.wraps(func)
    def returned_wrapper(hiwifi, *args, **kwargs):
        if hiwifi.stok == None:
            print("Login required!")
            return False
        else:
            return func(hiwifi, *args, **kwargs)
    return returned_wrapper

class HIWIFI:
    def __init__(self, host, port, password, username="admin"):
        self.host = host
        self.port = port
        self.password = password
        self.username = username
        self.session = requests.Session()
        self.stok = None
        self.login()
        self.ss_baseurl = "http://{}:{}/cgi-bin/turbo/;stok={}/api/prometheus/".format(self.host, self.port, self.stok)

    def login(self):
        print("Logining as {}...".format(self.username))
        params = {
            "username": self.username,
            "password": self.password
        }
        url = "http://{}:{}/cgi-bin/turbo/api/login/login_admin".format(self.host, self.port)
        response = self.session.get(url, params=params)
        data = json.loads(response.text)
        if data["stok"] != "":
            self.stok = data["stok"]
            print("Login succeed!")
            return True
        print("Login failed!")
        return False

    @login_required
    def status(self):
        print("Checking status...")
        url = "{}/get_ss_status".format(self.ss_baseurl)
        response = self.session.get(url)
        data = json.loads(response.text)
        result = data["accel"] == "yes"
        print("Status: {}".format(result))
        return result
    
    @login_required
    def stop(self):
        print("Stoping server...")
        params = {"enable": 0}
        url = "{}/set_ss_switch".format(self.ss_baseurl)
        response = self.session.get(url, params=params)
        data = json.loads(response.text)
        return data["enable"] == "0"

    @login_required
    def start(self):
        print("Starting server...")
        params = {"enable": 1}
        url = "{}/set_ss_switch".format(self.ss_baseurl)
        response = self.session.get(url, params=params)
        data = json.loads(response.text)
        return data["enable"] == "1"

    @login_required
    def update(self, server):
        print("Updating server config...")
        params = server
        url = "{}/set_ss_cfg".format(self.ss_baseurl)
        response = self.session.get(url, params=params)
        data = json.loads(response.text)
        print(data)

def main():
    hiwifi = HIWIFI(
        host=config.host,
        port=config.port,
        username=config.username,
        password=config.password,
    )
    status = hiwifi.status()
    while not status:
        print("Need to switch server...")
        hiwifi.stop()
        server = random.choice(config.servers)
        print("Selected server: {}:{}".format(server["server"], server["server_port"]))
        hiwifi.update(server)
        hiwifi.start()
        status = hiwifi.status()

if __name__ == "__main__":
    main()