add projects and metrics routes
This commit is contained in:
335
admin/admin.py
335
admin/admin.py
@ -30,8 +30,6 @@ import requests as req
|
||||
from os import getenv
|
||||
from sys import argv
|
||||
|
||||
API_URL_ENV = "API_URL"
|
||||
|
||||
|
||||
# logger used by the script
|
||||
class Log:
|
||||
@ -138,11 +136,32 @@ class AdminAPI:
|
||||
|
||||
self.PUT("/v1/admin/service/add", service)
|
||||
|
||||
def del_service(self, service: str) -> None:
|
||||
if service == "":
|
||||
def del_service(self, name: str) -> None:
|
||||
if name == "":
|
||||
raise Exception("Service name cannot be empty")
|
||||
|
||||
self.DELETE("/v1/admin/service/del?name=%s" % quote_plus(service))
|
||||
self.DELETE("/v1/admin/service/del?name=%s" % quote_plus(name))
|
||||
|
||||
def add_project(self, project: Dict[str, str]):
|
||||
if "name" not in project or project["name"] == "":
|
||||
raise Exception('Project structure is missing required "name" field')
|
||||
|
||||
if "desc" not in project:
|
||||
raise Exception('Project structure is missing required "desc" field')
|
||||
|
||||
if not self._check_multilang_field(project["desc"]):
|
||||
raise Exception(
|
||||
'Project structure field "desc" needs at least '
|
||||
+ "one supported language entry"
|
||||
)
|
||||
|
||||
self.PUT("/v1/admin/project/add", project)
|
||||
|
||||
def del_project(self, name: str) -> None:
|
||||
if name == "":
|
||||
raise Exception("Project name cannot be empty")
|
||||
|
||||
self.DELETE("/v1/admin/project/del?name=%s" % quote_plus(name))
|
||||
|
||||
def check_services(self) -> None:
|
||||
self.GET("/v1/admin/service/check")
|
||||
@ -174,184 +193,212 @@ class AdminAPI:
|
||||
|
||||
self.PUT("/v1/admin/news/add", news)
|
||||
|
||||
def del_news(self, news: str) -> None:
|
||||
if news == "":
|
||||
def del_news(self, id: str) -> None:
|
||||
if id == "":
|
||||
raise Exception("News ID cannot be empty")
|
||||
|
||||
self.DELETE("/v1/admin/news/del?id=%s" % quote_plus(news))
|
||||
self.DELETE("/v1/admin/news/del?id=%s" % quote_plus(id))
|
||||
|
||||
def logs(self) -> List[Dict[str, Any]]:
|
||||
return self.GET("/v1/admin/logs")
|
||||
|
||||
|
||||
# local helper functions used by the script
|
||||
def __format_time(ts: int) -> str:
|
||||
return datetime.fromtimestamp(ts, UTC).strftime("%H:%M:%S %d/%m/%Y")
|
||||
class AdminScript:
|
||||
def __init__(self):
|
||||
self.log: Log = Log()
|
||||
self.api: AdminAPI = None
|
||||
self.commands = {
|
||||
"add_service": self.add_service,
|
||||
"del_service": self.del_service,
|
||||
"add_project": self.add_project,
|
||||
"del_project": self.del_project,
|
||||
"add_news": self.add_news,
|
||||
"del_news": self.del_news,
|
||||
"check_services": self.check_services,
|
||||
"logs": self.get_logs,
|
||||
}
|
||||
self.api_url_env = "API_URL"
|
||||
|
||||
def __format_time(self, ts: int) -> str:
|
||||
return datetime.fromtimestamp(ts, UTC).strftime("%H:%M:%S %d/%m/%Y")
|
||||
|
||||
def __load_json_file(file: str) -> Dict[str, Any]:
|
||||
with open(file, "r") as f:
|
||||
data = loads(f.read())
|
||||
return data
|
||||
def __load_json_file(self, file: str) -> Dict[str, Any]:
|
||||
with open(file, "r") as f:
|
||||
data = loads(f.read())
|
||||
return data
|
||||
|
||||
def __dump_json_file(self, data: Dict[str, Any], file: str) -> None:
|
||||
with open(file, "w") as f:
|
||||
data = dumps(data, indent=2)
|
||||
f.write(data)
|
||||
|
||||
def __dump_json_file(data: Dict[str, Any], file: str) -> None:
|
||||
with open(file, "w") as f:
|
||||
data = dumps(data, indent=2)
|
||||
f.write(data)
|
||||
def run(self) -> bool:
|
||||
if len(argv) < 2 or len(argv) > 3:
|
||||
self.log.error("Usage: %s [command] <file>" % argv[0])
|
||||
self.log.info("Here is a list of available commands:")
|
||||
|
||||
for command in self.commands.keys():
|
||||
print("\t%s" % command)
|
||||
|
||||
# command handlers
|
||||
def __handle_command(log: Log, api: AdminAPI, cmd: str) -> None:
|
||||
match cmd:
|
||||
case "add_service":
|
||||
return False
|
||||
|
||||
url = getenv(self.api_url_env)
|
||||
valid_cmd = False
|
||||
|
||||
if url is None:
|
||||
self.log.error(
|
||||
"Please specify the API URL using %s environment variable"
|
||||
% self.api_url_env
|
||||
)
|
||||
return False
|
||||
|
||||
for cmd in self.commands:
|
||||
if argv[1] == cmd:
|
||||
valid_cmd = True
|
||||
break
|
||||
|
||||
if not valid_cmd:
|
||||
self.log.error(
|
||||
"Invalid command, run the script with no commands to list the available commands"
|
||||
)
|
||||
return False
|
||||
|
||||
try:
|
||||
password = self.log.password("Please enter the admin password")
|
||||
self.api = AdminAPI(url, password)
|
||||
|
||||
if len(argv) == 2:
|
||||
self.handle_command(argv[1])
|
||||
|
||||
elif len(argv) == 3:
|
||||
self.handle_command(argv[1], argv[2])
|
||||
|
||||
except KeyboardInterrupt:
|
||||
self.log.error("Command cancelled")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
self.log.error("Command failed: %s" % e)
|
||||
return False
|
||||
|
||||
# service commands
|
||||
def add_service(self, data: Dict[str, Any] = None) -> None:
|
||||
if data is None:
|
||||
data: Dict[str, str] = {}
|
||||
data["desc"] = {}
|
||||
|
||||
data["name"] = log.input("Serivce name")
|
||||
for lang in api.languages:
|
||||
data["desc"][lang] = log.input("Serivce desc (%s)" % lang)
|
||||
data["check_url"] = log.input("Serivce status check URL")
|
||||
data["clear"] = log.input("Serivce clearnet URL")
|
||||
data["onion"] = log.input("Serivce onion URL")
|
||||
data["i2p"] = log.input("Serivce I2P URL")
|
||||
data["name"] = self.log.input("Serivce name")
|
||||
|
||||
api.add_service(data)
|
||||
log.info("Service has been added")
|
||||
for lang in self.api.languages:
|
||||
data["desc"][lang] = self.log.input("Serivce desc (%s)" % lang)
|
||||
|
||||
case "del_service":
|
||||
api.del_service(log.input("Serivce name"))
|
||||
log.info("Service has been deleted")
|
||||
data["check_url"] = self.log.input("Serivce status check URL")
|
||||
data["clear"] = self.log.input("Serivce clearnet URL")
|
||||
data["onion"] = self.log.input("Serivce onion URL")
|
||||
data["i2p"] = self.log.input("Serivce I2P URL")
|
||||
|
||||
case "check_services":
|
||||
api.check_services()
|
||||
log.info("Requested status check for all the services")
|
||||
self.api.add_service(data)
|
||||
self.log.info("Service has been added")
|
||||
|
||||
case "add_news":
|
||||
def del_service(self, data: Dict[str, Any] = None) -> None:
|
||||
if data is None:
|
||||
data: Dict[str, str] = {}
|
||||
data["name"] = self.log.input("Service name")
|
||||
|
||||
self.api.del_service(data["name"])
|
||||
self.log.info("Service has been deleted")
|
||||
|
||||
# project commands
|
||||
def add_project(self, data: Dict[str, Any] = None) -> None:
|
||||
if data is None:
|
||||
data: Dict[str, str] = {}
|
||||
data["desc"] = {}
|
||||
|
||||
data["name"] = self.log.input("Project name")
|
||||
|
||||
for lang in self.api.languages:
|
||||
data["desc"][lang] = self.log.input("Project desc (%s)" % lang)
|
||||
|
||||
data["url"] = self.log.input("Project URL")
|
||||
data["license"] = self.log.input("Project license")
|
||||
|
||||
self.api.add_project(data)
|
||||
self.log.info("Project has been added")
|
||||
|
||||
def del_project(self, data: Dict[str, Any] = None) -> None:
|
||||
if data is None:
|
||||
data: Dict[str, str] = {}
|
||||
data["name"] = self.log.input("Project name")
|
||||
|
||||
self.api.del_project(data["name"])
|
||||
self.log.info("Project has been deleted")
|
||||
|
||||
# news command
|
||||
def add_news(self, data: Dict[str, Any] = None) -> None:
|
||||
if data is None:
|
||||
news: Dict[str, str] = {}
|
||||
news["title"] = {}
|
||||
news["content"] = {}
|
||||
|
||||
data["id"] = log.input("News ID")
|
||||
for lang in api.languages:
|
||||
data["title"][lang] = log.input("News title (%s)" % lang)
|
||||
data["author"] = log.input("News author")
|
||||
for lang in api.languages:
|
||||
data["content"][lang] = log.input("News content (%s)" % lang)
|
||||
data["id"] = self.log.input("News ID")
|
||||
|
||||
api.add_news(data)
|
||||
log.info("News has been added")
|
||||
for lang in self.api.languages:
|
||||
data["title"][lang] = self.log.input("News title (%s)" % lang)
|
||||
|
||||
case "del_news":
|
||||
api.del_news(log.input("News ID"))
|
||||
log.info("News has been deleted")
|
||||
data["author"] = self.log.input("News author")
|
||||
|
||||
case "logs":
|
||||
logs = api.logs()
|
||||
for lang in self.api.languages:
|
||||
data["content"][lang] = self.log.input("News content (%s)" % lang)
|
||||
|
||||
if logs["result"] is None or len(logs["result"]) == 0:
|
||||
return log.info("No available logs")
|
||||
self.api.add_news(data)
|
||||
self.log.info("News has been added")
|
||||
|
||||
for log in logs["result"]:
|
||||
log.info(
|
||||
"Time: %s | Action: %s"
|
||||
% (__format_time(log["time"]), log["action"])
|
||||
)
|
||||
def del_news(self, data: Dict[str, Any] = None) -> None:
|
||||
if data is None:
|
||||
data: Dict[str, str] = {}
|
||||
data["id"] = self.log.input("News ID")
|
||||
|
||||
self.api.del_project(data["id"])
|
||||
self.log.info("News has been deleted")
|
||||
|
||||
def __handle_command_with_file(log: Log, api: AdminAPI, cmd: str, file: str) -> None:
|
||||
match cmd:
|
||||
case "add_service":
|
||||
data = __load_json_file(file)
|
||||
api.add_service(data)
|
||||
log.info("Service has been added")
|
||||
def check_services(self, data: Dict[str, Any] = None) -> None:
|
||||
self.api.check_services()
|
||||
self.log.info("Requested status check for all the services")
|
||||
|
||||
case "del_service":
|
||||
data = __load_json_file(file)
|
||||
api.del_service(data["name"])
|
||||
log.info("Service has been deleted")
|
||||
def get_logs(self, data: Dict[str, Any] = None) -> None:
|
||||
logs = self.api.logs()
|
||||
|
||||
case "check_services":
|
||||
api.check_services()
|
||||
log.info("Requested status check for all the services")
|
||||
if logs["result"] is None or len(logs["result"]) == 0:
|
||||
return self.log.info("No available logs")
|
||||
|
||||
case "add_news":
|
||||
data = __load_json_file(file)
|
||||
api.add_news(data)
|
||||
log.info("News has been added")
|
||||
for log in logs["result"]:
|
||||
self.log.info(
|
||||
"Time: %s | Action: %s"
|
||||
% (self.__format_time(log["time"]), log["action"])
|
||||
)
|
||||
|
||||
case "del_news":
|
||||
data = __load_json_file(file)
|
||||
api.del_news(data["id"])
|
||||
log.info("News has been deleted")
|
||||
def handle_command(self, cmd: str, file: str = None) -> bool:
|
||||
for command in self.commands.keys():
|
||||
if command != cmd:
|
||||
continue
|
||||
|
||||
case "logs":
|
||||
logs = api.logs()
|
||||
data = None
|
||||
|
||||
if logs["result"] is None or len(logs["result"]) == 0:
|
||||
return log.info("No available logs")
|
||||
try:
|
||||
if file != "" and file is not None:
|
||||
data = self.__load_json_file(file)
|
||||
|
||||
__dump_json_file(logs["result"], file)
|
||||
log.info("Logs has been saved")
|
||||
self.commands[cmd](data)
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
self.log.error("Command failed: %s" % e)
|
||||
return False
|
||||
|
||||
self.log.error("Invalid command: %s", cmd)
|
||||
return False
|
||||
|
||||
commands = [
|
||||
"add_service",
|
||||
"del_service",
|
||||
"check_services",
|
||||
"add_news",
|
||||
"del_news",
|
||||
"logs",
|
||||
]
|
||||
|
||||
if __name__ == "__main__":
|
||||
log = Log()
|
||||
|
||||
if len(argv) < 2 or len(argv) > 3:
|
||||
log.error("Usage: %s [command] <file>" % argv[0])
|
||||
log.info("Here is a list of available commands:")
|
||||
print("\tadd_service")
|
||||
print("\tdel_service")
|
||||
print("\tcheck_services")
|
||||
print("\tadd_news")
|
||||
print("\tdel_news")
|
||||
print("\tlogs")
|
||||
exit(1)
|
||||
|
||||
url = getenv(API_URL_ENV)
|
||||
valid_cmd = False
|
||||
|
||||
for cmd in commands:
|
||||
if argv[1] == cmd:
|
||||
valid_cmd = True
|
||||
break
|
||||
|
||||
if not valid_cmd:
|
||||
log.error(
|
||||
"Invalid command, run the script with no commands to list the available commands"
|
||||
)
|
||||
exit(1)
|
||||
|
||||
if url is None:
|
||||
log.error(
|
||||
"Please specify the API URL using %s environment variable" % API_URL_ENV
|
||||
)
|
||||
exit(1)
|
||||
|
||||
try:
|
||||
password = log.password("Please enter the admin password")
|
||||
api = AdminAPI(url, password)
|
||||
|
||||
if len(argv) == 2:
|
||||
__handle_command(log, api, argv[1])
|
||||
elif len(argv) == 3:
|
||||
__handle_command_with_file(log, api, argv[1], argv[2])
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print()
|
||||
log.error("Command cancelled")
|
||||
exit(1)
|
||||
|
||||
except Exception as e:
|
||||
log.error("Command failed: %s" % e)
|
||||
exit(1)
|
||||
script = AdminScript()
|
||||
exit(script.run() if 1 else 0)
|
||||
|
||||
Reference in New Issue
Block a user