1
0

Added basic documentation

This commit is contained in:
Rokas Puzonas 2020-08-09 19:50:03 +03:00
parent 62ef2956cc
commit 27808789ee
5 changed files with 227 additions and 59 deletions

View File

@ -1,3 +1,4 @@
{ {
"autoDocstring.docstringFormat": "sphinx",
"python.pythonPath": "${workspaceFolder}/venv/bin/python3" "python.pythonPath": "${workspaceFolder}/venv/bin/python3"
} }

View File

@ -1,5 +1,3 @@
# MineHost Interface # MineHost Interface
A python library for executing commands or gettings information on MineHost minecraft servers. A python library for executing commands or gettings information about MineHost minecraft servers.
Documentation will be added soon.

View File

@ -1,9 +1,10 @@
from bs4 import BeautifulSoup, Tag from bs4 import BeautifulSoup, Tag
import re import re
from datetime import datetime from datetime import datetime
import math
from .server import MCServer from .server import MCServer
from .session import Session from .session import Session, InvalidSessionException
# The changing of the password and profile info, are intentionally not implemented. # The changing of the password and profile info, are intentionally not implemented.
# It's just too much power # It's just too much power
@ -11,34 +12,70 @@ from .session import Session
datetime_format = "%Y-%m-%d %H:%M:%S" datetime_format = "%Y-%m-%d %H:%M:%S"
class Account: class Account:
def __init__(self, username: str = None, password: str = None, session: Session = None): """Used to get servers, history, balance and other details associated to a specific account.
self.username = username """
if username is not None and password is not None:
self.session = Session(username, password) def __init__(self, email: str = None, password: str = None, session: Session = None):
"""Initializes an account
:param email: email used to login, defaults to None
:type email: str, optional
:param password: password used to login, defaults to None
:type password: str, optional
:param session: an already created session can be provided, defaults to None
:type session: :class:`Session <Session>`, optional
:raises InvalidSessionException: Raised when the given custom session is invalid or when nothing was provided
"""
self.email = email
self.session = None
if email is not None and password is not None:
self.session = Session(email, password)
elif session is not None: elif session is not None:
self.session = session self.session = session
else:
raise Exception("Session or logins must be given") if self.session is None or not self.session.isValid():
raise InvalidSessionException()
def __repr__(self): def __repr__(self):
return f"<Account({self.username or self._cookies.get('PHPSESSID', 'NONE')})>" return f"<Account({self.email})>"
def getServers(self): def getServers(self) -> list:
"""Returns a list of minecraft server objects.
:return: List of MCServer
"""
control_res = self.session.get("/mano-serveriai") control_res = self.session.get("/mano-serveriai")
servers = [] servers = []
for server_id in re.findall("/minecraft-serverio-valdymas/(\d*)/", control_res.text): for server_id in re.findall("/minecraft-serverio-valdymas/(\d*)/", control_res.text):
servers.append(MCServer(server_id, self.session)) servers.append(MCServer(server_id, self.session))
return servers return servers
def getServer(self, i: int = 0): def getServer(self, i: int = 0) -> MCServer:
"""Helper method get a server quickly.
Most people will only want to interact with 1 server.
:param i: Index of server (0 indexed), defaults to 0
:type i: int, optional
:return: A MCServer
"""
return self.getServers()[i] return self.getServers()[i]
def getBalance(self): def getBalance(self) -> float:
"""Current balance in account.
:return: A float representing the current balance.
"""
balance_res = self.session.get("/balanso-pildymas") balance_res = self.session.get("/balanso-pildymas")
balance_match = re.search(r"balanse yra (\d+\.?\d*)", balance_res.text) balance_match = re.search(r"balanse yra (\d+\.?\d*)", balance_res.text)
return float(balance_match.group(1)) return float(balance_match.group(1))
def getProfileInfo(self): def getDetails(self) -> dict:
"""Returns a dictionary containing general details about account.
Available details: email, name, surname, phone, skype.
:return: A dictionary with account details.
"""
profile_res = self.session.get("/profilio-nustatymai") profile_res = self.session.get("/profilio-nustatymai")
soup = BeautifulSoup(profile_res.text, "lxml") soup = BeautifulSoup(profile_res.text, "lxml")
return { return {
@ -49,7 +86,14 @@ class Account:
"skype": soup.find("input", id="v5")["value"] "skype": soup.find("input", id="v5")["value"]
} }
def getLoginHistory(self, limit: int = 10): def getLoginHistory(self, limit: int = math.inf) -> list:
"""Returns a list of entries, where each entry holds the date and ip of who logged in.
Entry keys: date, ip.
:param limit: The maximum number of entries it should try getting, defaults to math.inf
:type limit: int, optional
:return: A list of entries where each entry is a dictionary.
"""
history_res = self.session.get("/istorija/prisijungimu-istorija") history_res = self.session.get("/istorija/prisijungimu-istorija")
soup = BeautifulSoup(history_res.text, "lxml") soup = BeautifulSoup(history_res.text, "lxml")
history = [] history = []
@ -64,11 +108,18 @@ class Account:
"date": datetime.strptime(fields[1].text, datetime_format), "date": datetime.strptime(fields[1].text, datetime_format),
"ip": fields[2].text "ip": fields[2].text
}) })
if len(history) == limit: break if len(history) >= limit: break
return history return history
def getFinanceHistory(self, limit: int = 10): def getFinanceHistory(self, limit: int = math.inf) -> list:
"""Returns a list of entries where each entry describes a transaction.
Entry keys: date, action, balance_change, balance_remainder
:param limit: The maximum number of entries it should try getting, defaults to math.inf
:type limit: int, optional
:return: A list of entries where each entry is a dictionary.
"""
history_res = self.session.get("/balanso-pildymas/ataskaita") history_res = self.session.get("/balanso-pildymas/ataskaita")
soup = BeautifulSoup(history_res.text, "lxml") soup = BeautifulSoup(history_res.text, "lxml")
history = [] history = []
@ -85,11 +136,18 @@ class Account:
"balance_change": float(fields[2].text[:-4]), "balance_change": float(fields[2].text[:-4]),
"balance_remainder": float(fields[3].text[:-3]), "balance_remainder": float(fields[3].text[:-3]),
}) })
if len(history) == limit: break if len(history) >= limit: break
return history return history
def getProfileInfoHistory(self, limit: int = 10): def getProfileDetailsHistory(self, limit: int = math.inf) -> list:
"""Returns a list of entries where each entry describes what was changed and by who.
Entry keys: date, name, surname, phone, skype, ip.
:param limit: The maximum number of entries it should try getting, defaults to math.inf
:type limit: int, optional
:return: A list of entries where each entry is a dictionary.
"""
history_res = self.session.get("/profilio-nustatymai") history_res = self.session.get("/profilio-nustatymai")
soup = BeautifulSoup(history_res.text, "lxml") soup = BeautifulSoup(history_res.text, "lxml")
history = [] history = []
@ -108,7 +166,6 @@ class Account:
"skype": fields[4].text, "skype": fields[4].text,
"ip": fields[5].text "ip": fields[5].text
}) })
if len(history) == limit: break if len(history) >= limit: break
return history
return history

View File

@ -1,37 +1,57 @@
from bs4 import BeautifulSoup, Tag from bs4 import BeautifulSoup, Tag
import re import re
from ftplib import FTP import ftplib
from datetime import datetime import datetime
import paramiko import paramiko
import time import time
datetime_format = "%Y-%m-%d %H:%M:%S" from .session import Session
class InvalidDomainException(Exception): class InvalidDomainException(Exception):
"""Raised when trying to change a server address with an unsupported domain.
"""
pass pass
# TODO: Make the sending of commands not need to sleep inbetween, If It dosen't sleep then it dosen'y always send the commands to the server # TODO: Make the sending of commands not need to sleep inbetween, If It dosen't sleep then it dosen'y always send the commands to the server
class CommandSender: class CommandSender:
def __init__(self, host, username, password, port=22): """Used to directly send commands to a minecraft server.
Rather than loggin into ssh, entering console manually.
"""
def __init__(self, host: str, password: str, port: int = 22):
"""Initializes ssh client and tries connecting to given host
:param host: Host address used while connecting to ssh
:type host: str
:param password: Password used for authentication
:type password: str
:param port: Port used while connecting to ssh, defaults to 22
:type port: int, optional
"""
self.ssh = paramiko.SSHClient() self.ssh = paramiko.SSHClient()
self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
self.ssh.connect(host, port=port, username="console", password=password) self.ssh.connect(host, port=port, username="console", password=password)
self.channel = None self.channel = None
def __enter__(self): def __enter__(self):
if self.channel is None: self.open()
self.open()
return self return self
def __exit__(self, _1, _2, _3): def __exit__(self, _1, _2, _3):
if self.channel is not None: self.close()
self.close()
def __del__(self): def __del__(self):
self.ssh.close() self.ssh.close()
def send(self, *args): def send(self, *args):
"""Send commands to the server. You are able to send to multiple commands by giving multiple arguments.
:param \*args: Commands you would like the server to execute.
:type \*args: str
:raises Exception: Raised if channel is not open
"""
if self.channel is None: if self.channel is None:
raise Exception("Channel is not open") raise Exception("Channel is not open")
@ -39,6 +59,10 @@ class CommandSender:
time.sleep(0.5) time.sleep(0.5)
def open(self): def open(self):
"""Opens a channel used to send commands.
:raises Exception: Raised if channel is already open
"""
if self.channel is not None: if self.channel is not None:
raise Exception("Channel is already open") raise Exception("Channel is already open")
@ -47,14 +71,31 @@ class CommandSender:
time.sleep(0.5) time.sleep(0.5)
def close(self): def close(self):
"""Closes channel used to send commands.
:raises Exception: Raised if channel is already closed
"""
if self.channel is None: if self.channel is None:
raise Exception("Channel is not open") raise Exception("Channel is not open")
self.channel.close() self.channel.close()
self.channel = None self.channel = None
# TODO: A way to edit the server.properties file easier.
class MCServer: class MCServer:
def __init__(self, server_id, session): """Used to control a minehost minecraft server
"""
__attrs__ = [ "start_file" ]
def __init__(self, server_id: str, session: Session):
"""Initializes minecraft server instance. Retrieves some initial data like: address, ip and port.
:param server_id: [description]
:type server_id: str
:param session: [description]
:type session: Session
"""
self.id = server_id self.id = server_id
self.session = session self.session = session
@ -71,22 +112,34 @@ class MCServer:
self.__key = soup.find("input", id="khd")["value"] self.__key = soup.find("input", id="khd")["value"]
def __repr__(self): def __repr__(self):
return f"<MinecraftServer({self.short_address})>" return f"<MCServer({self.ip}:{self.port})>"
@property @property
def __url(self): def __url(self) -> str:
return f"/minecraft-serverio-valdymas/{self.id}" return f"/minecraft-serverio-valdymas/{self.id}"
def getPassword(self): def getPassword(self) -> str:
"""Retrieves the password used to login to SSH, FTP and SFTP.
:return: A string containing the password
"""
res = self.session.get(f"{self.__url}/failai") res = self.session.get(f"{self.__url}/failai")
soup = BeautifulSoup(res.text, "lxml") soup = BeautifulSoup(res.text, "lxml")
return soup.select("table td span:nth-child(2)")[0].text return soup.select("table td span:nth-child(2)")[0].text
def changePassword(self): def changePassword(self) -> bool:
"""Change the current password to a new randomly generated password. Passwords can only be changed every few minutes.
:return: Whether the change was successful
"""
res = self.session.get(f"{self.__url}/failai/change-password") res = self.session.get(f"{self.__url}/failai/change-password")
return res.text.find("class=\"alert alert-info\"") > 0 return res.text.find("class=\"alert alert-info\"") > 0
def getInfo(self): def getStats(self) -> dict:
"""Returns a dictionary containing the current statistcs of the server.
:return: A dictionary containing "state", "version", "ping", "online_players" and "max_players"
"""
res = self.session.get(self.__url) res = self.session.get(self.__url)
soup = BeautifulSoup(res.text, "lxml") soup = BeautifulSoup(res.text, "lxml")
@ -110,10 +163,14 @@ class MCServer:
"max_players": max_players "max_players": max_players
} }
def getExpirationDate(self): def getExpirationDate(self) -> datetime.datetime:
"""Returns the date at which the server will expire.
:return: A datetime object
"""
res = self.session.get(f"{self.__url}/finansai") res = self.session.get(f"{self.__url}/finansai")
date_match = re.search(r"Serveris galioja iki: (\d{4}-\d{2}-\d{2} \d{1,2}:\d{2}:\d{2})", res.text) date_match = re.search(r"Serveris galioja iki: (\d{4}-\d{2}-\d{2} \d{1,2}:\d{2}:\d{2})", res.text)
return datetime.strptime(date_match.group(1), datetime_format) return datetime.datetime.strptime(date_match.group(1), "%Y-%m-%d %H:%M:%S")
@property @property
def start_file(self): def start_file(self):
@ -168,20 +225,32 @@ class MCServer:
if res.content.decode("UTF-8") == "Šis subdomenas jau užimtas. Bandykite kitą": if res.content.decode("UTF-8") == "Šis subdomenas jau užimtas. Bandykite kitą":
raise InvalidDomainException() raise InvalidDomainException()
def FTP(self): def FTP(self) -> ftplib.FTP:
ftp = FTP() """Creates a new FTP connection to the server. The password will be automatically inputted.
:return: An open FTP connection
"""
ftp = ftplib.FTP()
ftp.connect(self.ip, self.ftp_port) ftp.connect(self.ip, self.ftp_port)
ftp.login("serveris", self.getPassword()) ftp.login("serveris", self.getPassword())
return ftp return ftp
def SSH(self): def SSH(self) -> paramiko.SSHClient:
"""Creates a new SSH connection to the server. The password will be automatically inputted.
:return: An open SSH connection
"""
ssh = paramiko.SSHClient() ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(self.ip, port=self.ssh_port, username="console", password=self.getPassword()) ssh.connect(self.ip, port=self.ssh_port, username="console", password=self.getPassword())
return ssh return ssh
def CommandSender(self): def CommandSender(self) -> CommandSender:
return CommandSender(self.ip, "serveris", self.getPassword(), self.ssh_port) """Creates a new CommandSender which allow you to easily send commands to the console.
:return: The newly created CommandSender.
"""
return CommandSender(self.ip, self.getPassword(), self.ssh_port)
def __isControlLocked(self): def __isControlLocked(self):
res = self.session.post("/query/user_loadingct.php", data={ res = self.session.post("/query/user_loadingct.php", data={
@ -189,7 +258,11 @@ class MCServer:
}) })
return res.text != "" return res.text != ""
def start(self): def start(self) -> bool:
"""Tries starting the server.
:return: Whether if it succeded in starting the server.
"""
if self.__isControlLocked(): if self.__isControlLocked():
return False return False
@ -200,7 +273,11 @@ class MCServer:
}) })
return True return True
def stop(self): def stop(self) -> bool:
"""Tries stopping the server.
:return: Whether if it succeded in stopping the server.
"""
if self.__isControlLocked(): if self.__isControlLocked():
return False return False
@ -211,7 +288,11 @@ class MCServer:
}) })
return True return True
def kill(self): def kill(self) -> bool:
"""Tries killing the server. This is the same as stopping but killing dosen't save anything.
:return: Whether if it succeded in killing the server.
"""
if self.__isControlLocked(): if self.__isControlLocked():
return False return False

View File

@ -1,34 +1,61 @@
import requests import requests
# The "Cookies" functions are used, when you already have dictionary of cookies,
# So that the function internally would not need to create one
class IncorrectLoginException(Exception): class IncorrectLoginException(Exception):
"""Raised when the given email and passwords are incorrect.
"""
pass pass
class InvalidSessionException(Exception): class InvalidSessionException(Exception):
"""Raised when the current session is logged out or timed out.
"""
pass pass
class Session(requests.Session): class Session(requests.Session):
def __init__(self, username: str = None, password: str = None): """Wrapper object around the requests.Session.
super().__init__() Checks if it's logged on every request.
if username is not None and password is not None: """
self.login(username, password)
def login(self, username: str, password: str): def __init__(self, email: str = None, password: str = None):
"""Initializes session if email and password are given.
:param email: If provided with password will try to login.
:type email: str, optional
:param password: If provided with email will try to login.
:type password: str, optional
"""
super().__init__()
if email is not None and password is not None:
self.login(email, password)
def login(self, email: str, password: str):
"""Attempts to login using the given credentials.
:param email: Email used to login.
:type email: str
:param password: Password associated to given email.
:type password: str
:raises IncorrectLoginException: Raised on failed login.
"""
requests.Session.request(self, "POST", "https://minehost.lt/prisijungimas-prie-sistemos", data = { requests.Session.request(self, "POST", "https://minehost.lt/prisijungimas-prie-sistemos", data = {
"login": username, "login": email,
"password": password "password": password
}) })
if not self.isValid(): if not self.isValid():
raise IncorrectLoginException() raise IncorrectLoginException()
def logout(self): def logout(self):
"""The session becomes invalid and will raise error if tried to request anything.
"""
self.get("/logout") self.get("/logout")
def request(self, method, url, *args, **kvargs): def request(self, method, url, *args, **kvargs) -> requests.Response:
"""Wrapper function around the default requests.request method.
Instead of giving the whole url like "https://minehost.lt/logout" now you only need to write "/logout".`
:return: requests.Response object
"""
response = requests.Session.request(self, method, "https://minehost.lt"+url, *args, **kvargs) response = requests.Session.request(self, method, "https://minehost.lt"+url, *args, **kvargs)
# Basic check to see if the session is still logged in. Looks for the logout button. # Basic check to see if the session is still logged in. Looks for the logout button.
if response.text.find("href=\"/logout\"") == -1 and response.text.find("src=\"/img/logo.png\"") > 0: if response.text.find("href=\"/logout\"") == -1 and response.text.find("src=\"/img/logo.png\"") > 0:
@ -36,7 +63,11 @@ class Session(requests.Session):
return response return response
def isValid(self): def isValid(self) -> bool:
"""Returns true if current session is logged in, otherwise false.
:return: True if logged in.
"""
res = self.get("/valdymo-pultas") res = self.get("/valdymo-pultas")
return res.text.find("<meta http-equiv=\"refresh\" content=\"0;url=/prisijungimas-prie-sistemos\">") == -1 return res.text.find("<meta http-equiv=\"refresh\" content=\"0;url=/prisijungimas-prie-sistemos\">") == -1