120 lines
3.4 KiB
Python
120 lines
3.4 KiB
Python
import json
|
|
import logging
|
|
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
import typing
|
|
|
|
import requests
|
|
|
|
log: logging.Logger = logging.getLogger(__name__)
|
|
|
|
|
|
class Client:
|
|
host: str = ""
|
|
cookies: typing.Dict[str, str] = {}
|
|
headers: typing.Dict[str, str] = {}
|
|
|
|
def url(self, path):
|
|
if self.host:
|
|
return self.host + path
|
|
return path
|
|
|
|
def get(self, url, **kwargs) -> requests.models.Response:
|
|
return requests.get(self.url(url), cookies=self.cookies, headers=self.headers, **kwargs)
|
|
|
|
def post(self, url, data, **kwargs) -> requests.models.Response:
|
|
return requests.post(self.url(url), data, cookies=self.cookies, headers=self.headers, **kwargs)
|
|
|
|
def download_file(self, url, target, **kwargs) -> bool:
|
|
with open(target, "wb") as out:
|
|
try:
|
|
download = self.get(url, stream=True, **kwargs)
|
|
shutil.copyfileobj(download.raw, out)
|
|
except Exception as e:
|
|
log.exception(e)
|
|
os.remove(target)
|
|
return False
|
|
return True
|
|
|
|
def download_files(self, urls, **kwargs) -> tempfile.TemporaryDirectory:
|
|
target = tempfile.TemporaryDirectory()
|
|
for path in urls:
|
|
filename = os.path.join(target.name, path.split("/")[-1])
|
|
self.download_file(path, filename, **kwargs)
|
|
return target
|
|
|
|
def login(self):
|
|
pass
|
|
|
|
def list(self):
|
|
pass
|
|
|
|
|
|
class BiogamesClient(Client):
|
|
config_fields: typing.Dict[str, typing.List[str]] = {
|
|
'login': ('username', 'password', 'host'),
|
|
'session': ('sessionid', 'csrftoken', 'host'),
|
|
}
|
|
login_url: str = "/game2/auth/json-login"
|
|
list_url: str = "/game2/instance/log/list/"
|
|
headers: typing.Dict[str, str] = {'Accept': 'application/json'}
|
|
|
|
def __init__(self, **kwargs):
|
|
match = {j: all([i in kwargs for i in self.config_fields[j]]) for j in self.config_fields}
|
|
valid = filter(lambda x: match[x], match)
|
|
if not valid:
|
|
raise ValueError("missing parameter (" + str(self.config_fields) + ")")
|
|
self.config = kwargs
|
|
self.cookies = {}
|
|
self.host = self.config['host']
|
|
if 'session' in valid:
|
|
self.cookies = kwargs
|
|
|
|
def login(self) -> bool:
|
|
csrf_request = self.get(self.list_url)
|
|
if not csrf_request.ok:
|
|
log.exception(ConnectionError("Unable to obtain CSRF token (" + str(csrf_request) + ")"))
|
|
return False
|
|
if not 'csrftoken' in self.cookies:
|
|
self.cookies['csrftoken'] = csrf_request.cookies['csrftoken']
|
|
login_payload = {
|
|
'username': self.config['username'],
|
|
'password': self.config['password'],
|
|
'next': '',
|
|
'csrfmiddlewaretoken': 'csrftoken',
|
|
}
|
|
login = self.post(self.login_url, json.dumps(login_payload))
|
|
if not login.ok:
|
|
log.exception(ConnectionError("Unable to authenticate", login, login.text))
|
|
return False
|
|
self.cookies['sessionid'] = login.cookies['sessionid']
|
|
print(self.cookies)
|
|
return True
|
|
|
|
def list(self) -> dict:
|
|
print(self.cookies)
|
|
logs = self.get(self.list_url)
|
|
if not logs.ok:
|
|
raise ConnectionError("HTTP fail", logs, logs.text)
|
|
return logs.json()
|
|
|
|
def load_all_logs(self) -> tempfile.TemporaryDirectory:
|
|
return self.download_files([i["file_url"] for i in self.list()])
|
|
|
|
|
|
CLIENTS: typing.Dict[str, typing.Type[Client]] = {
|
|
"Biogames": BiogamesClient,
|
|
}
|
|
|
|
if __name__ == '__main__':
|
|
# c = BiogamesClient(host="http://biodiv", username="ba", password="853451")
|
|
# print(c.login())
|
|
# print(json.dumps(c.list(), indent=1))
|
|
# print(type(c.load_all_logs()))
|
|
# print(type(c.get("/")))
|
|
c = BiogamesClient(host="http://biodiv", **{'csrftoken': 'IgbwP83iEibW6RE7IADIFELYdbx0dvqQ',
|
|
'sessionid': 'zntsj09d92tjos1b6ruqjthlzv60xdin'})
|
|
print(json.dumps(c.list(), indent=1))
|