Mirrored_Repos/pluGET
1
0
Fork 0
mirror of https://github.com/Neocky/pluGET.git synced 2024-04-29 16:12:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #50 from Neocky/dev/1.7.0

Browse Source 
Dev/1.7.0 should work. Tests were all successfull
This commit is contained in:
Neocky 2022-07-02 19:29:14 +02:00 committed by GitHub
commit badf6230bb
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
28 changed files with 2104 additions and 1553 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@ -133,3 +133,6 @@ dmypy.json
# PyCharm Settings
.idea
# pluGET Files
pluGET_config.yaml

13
install_requirements.py Normal file
View File

@ -0,0 +1,13 @@
"""
Install the needed python packages through this script
"""
import os
try:
print("Installing Python packages and dependencies from requirements.txt...\n")
os.system('py -m pip install -r requirements.txt' if os.name=='nt' else 'pip install -r requirements.txt')
print("\nStart pluGET by launching the 'pluGET.py' file!")
except:
print("Requirements couldn't be installed. Check if file 'requirements.txt' is in the same folder and that Python3\
and pip is installed!")

34
installer.bat
View File

@ -1,34 +0,0 @@
@ECHO OFF
cd /d %~dp0
Echo ___ ___ ___ ___ ___ ___
Echo /\ \ /\__\ /\__\ /\ \ /\ \ /\ \
Echo /::\ \ /:/ / /:/ / /::\ \ /::\ \ \:\ \
Echo /:/\:\ \ /:/ / /:/ / /:/\:\ \ /:/\:\ \ \:\ \
Echo /::\~\:\ \ /:/ / /:/ / ___ /:/ \:\ \ /::\~\:\ \ /::\ \
Echo /:/\:\ \:\__\ /:/__/ /:/__/ /\__\ /:/__/_\:\__\ /:/\:\ \:\__\ /:/\:\__\
Echo \/__\:\/:/ / \:\ \ \:\ \ /:/ / \:\ /\ \/__/ \:\~\:\ \/__/ /:/ \/__/
Echo \::/ / \:\ \ \:\ /:/ / \:\ \:\__\ \:\ \:\__\ /:/ /
Echo \/__/ \:\ \ \:\/:/ / \:\/:/ / \:\ \/__/ \/__/
Echo \:\__\ \::/ / \::/ / \:\__\
Echo \/__/ \/__/ \/__/ \/__/
Echo `
Echo `
Echo ------------------------------------
Echo [By Neocky]
Echo https://github.com/Neocky/pluGET
Echo pluGET-Installer
Echo ------------------------------------
Echo `
Echo ----------------------------------------------------------------------------------
Echo Installing Python packages and dependencies...
py -m pip install -r requirements.txt
Echo `
Echo Launching pluGET...
launcher.bat
exit

4
launcher.bat
View File

@ -1,4 +0,0 @@
@ECHO OFF
cd "%~dp0"
cd src
py "__main__.py"

3
launcher.sh
View File

@ -1,3 +0,0 @@
#!/bin/sh
cd src/
python3 "__main__.py"

45
pluget.py Normal file
View File

@ -0,0 +1,45 @@
"""
Handles the main function and the argument passing for the whole pluGET program
"""
import sys
import argparse
# check if folder 'src' is accessible with all modules needed and if not exit
try:
from src.handlers.handle_config import check_config, validate_config
from src.utils.console_output import rename_console_title, clear_console, print_logo, print_console_logo
from src.utils.utilities import check_requirements, api_test_spiget, check_for_pluGET_update
from src.handlers.handle_input import handle_input
except TypeError:
print("Folder 'src' not found in the directory or missing files or broken functions detected! \
\nPlease redownload the files from here: https://www.github.com/Neocky/pluGET")
sys.exit()
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Arguments for pluGET",
formatter_class=argparse.ArgumentDefaultsHelpFormatter)
parser.add_argument("mode", help="Mode (install/update/etc.)", nargs='?', default=None)
parser.add_argument("object", help="Object/Plugin Name", nargs='?', default=None)
parser.add_argument("version", help="Version", nargs='?', default=None)
parser.add_argument("--no-confirmation", action="store_true", help="Skip confirmation messages")
args = vars(parser.parse_args())
rename_console_title()
check_config()
validate_config()
api_test_spiget()
check_requirements()
if args["mode"] is not None and args["object"] is not None:
# arguments were used so call the handle_input function to get the right function call
print_console_logo()
check_for_pluGET_update()
handle_input(args["mode"], args["object"], args["version"], args["no_confirmation"], arguments_from_console=True)
else:
# no arguments were used so start pluGET console
clear_console()
print_logo()
check_for_pluGET_update()
handle_input()

1
requirements.txt
View File

@ -12,3 +12,4 @@ rich >= 9.13.0
commonmark >= 0.9.1
Pygments >= 2.8.1
typing_extensions >= 3.7.4.3
ruamel.yaml >= 0.17.21

16
src/__main__.py
View File

@ -1,16 +0,0 @@
from utils.consoleoutput import consoleTitle, clearConsole, printMainMenu
from utils.utilities import check_requirements
from handlers.handle_input import createInputLists, getInput
from handlers.handle_config import checkConfig
def mainFunction():
consoleTitle()
clearConsole()
checkConfig()
check_requirements()
createInputLists()
printMainMenu()
getInput()
mainFunction()

176
src/handlers/handle_config.py
View File

@ -1,91 +1,109 @@
""""
Handles the logic for the config validation, reading and creating
"""
import os
import sys
import configparser
import ruamel.yaml
from pathlib import Path
from utils.consoleoutput import oColors
from rich.console import Console
class configurationValues:
class config_value():
"""
Class which holds all the available configuration values from the config file and which will be used later in
the process of updating plugins
If bool in config can't be read it will default to 'False'
"""
def __init__(self):
config = configparser.ConfigParser()
config.sections()
config.read("config.ini")
localPluginFolder = config['General']['UseLocalPluginFolder']
self.pathToPluginFolder = Path(config['Local - This Machine']['PathToPluginFolder'])
seperateDownloadPath = config['Local - This Machine']['SeperateDownloadPath']
self.pathToSeperateDownloadPath = Path(config['Local - This Machine']['PathToSeperateDownloadPath'])
self.sftp_server = config['SFTP - Remote Server']['Server']
self.sftp_user = config['SFTP - Remote Server']['Username']
self.sftp_password = config['SFTP - Remote Server']['Password']
sftp_port = config['SFTP - Remote Server']['SFTPPort']
ftp_port = config['SFTP - Remote Server']['FTPPort']
self.sftp_folderPath = config['SFTP - Remote Server']['PluginFolderOnServer']
sftp_useSftp = config['SFTP - Remote Server']['USE_SFTP']
sftp_seperateDownloadPath = config['SFTP - Remote Server']['SeperateDownloadPath']
self.sftp_pathToSeperateDownloadPath = config['SFTP - Remote Server']['PathToSeperateDownloadPath']
self.sftp_port = int(sftp_port)
self.ftp_port = int(ftp_port)
if localPluginFolder == 'True':
self.localPluginFolder = True
else:
self.localPluginFolder = False
if seperateDownloadPath == 'True':
self.seperateDownloadPath = True
else:
self.seperateDownloadPath = False
if sftp_seperateDownloadPath == 'True':
self.sftp_seperateDownloadPath = True
else:
self.sftp_seperateDownloadPath = False
if sftp_useSftp == 'True':
self.sftp_useSftp = True
else:
self.sftp_useSftp = False
yaml = ruamel.yaml.YAML()
with open("pluGET_config.yaml", "r") as config_file:
data = yaml.load(config_file)
self.connection = str(data["Connection"]).lower()
self.path_to_plugin_folder = Path(data["Local"]["PathToPluginFolder"])
self.local_seperate_download_path = True if data["Local"]["SeperateDownloadPath"] == True else False
self.local_path_to_seperate_download_path = Path(data["Local"]["PathToSeperateDownloadPath"])
self.server = data["Remote"]["Server"]
self.username = data["Remote"]["Username"]
self.password = data["Remote"]["Password"]
self.sftp_port = int(data["Remote"]["SFTP_Port"])
self.ftp_port = int(data["Remote"]["FTP_Port"])
self.remote_seperate_download_path = True if data["Remote"]["SeperateDownloadPath"] == True else False
self.remote_path_to_seperate_download_path = data["Remote"]["PathToSeperateDownloadPath"]
self.remote_plugin_folder_on_server = data["Remote"]["PluginFolderOnServer"]
def checkConfig():
configAvailable = os.path.isfile("config.ini")
if not configAvailable:
createConfig()
print(oColors.brightRed + "Config created. Edit config before executing again!" + oColors.standardWhite)
def check_config() -> None:
"""
Check if there is a pluGET_config.yml file in the same folder as pluget.py and if not create a new config
and exit the programm
"""
if not os.path.isfile("pluGET_config.yaml"):
create_config()
return None
def create_config() -> None:
"""
Creates the yaml config in the current directory with the filename pluGET_config.yml
"""
# this is the whole yaml code because of weird formating indention is not possible
configuration = """\
#
# Configuration File for pluGET
# https://www.github.com/Neocky/pluGET
#
# What should be used for the connection (local, sftp, ftp)
Connection: local
Local:
PathToPluginFolder: C:/Users/USER/Desktop/plugins
# If a different folder should be used to store the updated plugins change to (True/False) and the path below
SeperateDownloadPath : False
PathToSeperateDownloadPath: C:/Users/USER/Desktop/plugins
Remote:
Server: 0.0.0.0
Username: user
Password: password
# If a different Port for SFTP/FTP will be used
SFTP_Port: 22
FTP_Port: 21
# If a different folder should be used to store the updated plugins change to (True/False) and the path below
SeperateDownloadPath : False
PathToSeperateDownloadPath: /plugins/updated
# Change the path below if the plugin folder path is different on the SFTP/FTP server (Change only if you know what you are doing)
PluginFolderOnServer: /plugins
"""
# load ruamel.yaml to get the # commands right in the yaml code
yaml = ruamel.yaml.YAML()
code = yaml.load(configuration)
with open("pluGET_config.yaml", "w") as config_file:
yaml.dump(code, config_file)
config_file_path = os.path.abspath("pluGET_config.yaml")
print(f"Path of config file: {config_file_path}")
print("Config created. Edit config before executing again!")
input("Press any key + enter to exit...")
sys.exit()
def createConfig():
config = configparser.ConfigParser(allow_no_value=True)
config['General'] = {}
config['General'][';'] = 'If a local plugin folder exists (True/False) (If False SFTP/FTP will be used):'
config['General']['UseLocalPluginFolder'] = 'True'
config['Local - This Machine'] = {}
config['Local - This Machine']['PathToPluginFolder'] = 'C:/Users/USER/Desktop/plugins'
config['Local - This Machine'][';'] = 'For a different folder to store the updated plugins change to (True/False) and the path below'
config['Local - This Machine']['SeperateDownloadPath'] = 'False'
config['Local - This Machine']['PathToSeperateDownloadPath'] = 'C:/Users/USER/Desktop/plugins'
config['SFTP - Remote Server'] = {}
config['SFTP - Remote Server']['Server'] = '0.0.0.0'
config['SFTP - Remote Server']['Username'] = 'user'
config['SFTP - Remote Server']['Password'] = 'password'
config['SFTP - Remote Server'][';'] = 'If a different Port for SFTP needs to be used (Default: 22)'
config['SFTP - Remote Server']['SFTPPort'] = '22'
config['SFTP - Remote Server'][';_'] = 'If a different Port for FTP needs to be used (Default: 21)'
config['SFTP - Remote Server']['FTPPort'] = '21'
config['SFTP - Remote Server'][';__'] = 'Change the path below if the plugin folder path is different on the SFTP/FTP server (Change only if you know what you are doing)'
config['SFTP - Remote Server']['PluginFolderOnServer'] = '/plugins'
config['SFTP - Remote Server'][';___'] = 'If you want to use FTP instead of SFTP change to (False) else use (True)'
config['SFTP - Remote Server']['USE_SFTP'] = 'True'
config['SFTP - Remote Server'][';____'] = 'For a different folder to store the updated plugins (Only with the update command!) change to (True/False) and the path below'
config['SFTP - Remote Server']['SeperateDownloadPath'] = 'False'
config['SFTP - Remote Server']['PathToSeperateDownloadPath'] = '/plugins'
with open('config.ini', 'w') as configfile:
config.write(configfile)
def validate_config() -> None:
"""
Validates the config variables after config class is loaded and exit if error is detected and print error
"""
accepted_values = [
("local", "sftp", "ftp")
]
# exit afterwards if there is an error in config
exit_afterwards = False
config = config_value()
# rich console for nice colors
console = Console()
if config.connection not in accepted_values[0]:
console.print(f"Error in Config! Accepted values for key 'Connection' are {accepted_values[0]}",
style="bright_red")
exit_afterwards = True
if exit_afterwards:
sys.exit()

180
src/handlers/handle_ftp.py
View File

@ -1,117 +1,187 @@
import os
import sys
import ftplib
import stat
import re
from utils.consoleoutput import oColors
from handlers.handle_config import configurationValues
from src.utils.console_output import rich_print_error
from src.handlers.handle_config import config_value
def createFTPConnection():
configValues = configurationValues()
def ftp_create_connection():
"""
Creates a connection to the ftp server with the given values in the config
:returns: ftp connection type
"""
config_values = config_value()
try:
ftp = ftplib.FTP()
ftp.connect(configValues.sftp_server, configValues.ftp_port)
ftp.login(configValues.sftp_user, configValues.sftp_password)
ftp.connect(config_values.server, config_values.ftp_port)
ftp.login(config_values.username, config_values.password)
return ftp
except UnboundLocalError:
print(oColors.brightRed + "[FTP]: Check your config.ini!" + oColors.standardWhite)
print(oColors.brightRed + "Exiting program..." + oColors.standardWhite)
rich_print_error("Error: [SFTP]: Check your config file!")
rich_print_error("Exiting program...")
sys.exit()
def ftp_showPlugins(ftp):
configValues = configurationValues()
ftp.cwd(configValues.sftp_folderPath)
def ftp_show_plugins(ftp) -> None:
"""
Prints all plugins in the plugin folder
:param ftp: ftp connection
:returns: None
"""
config_values = config_value()
ftp.cwd(config_values.remote_plugin_folder_on_server)
for attr in ftp.dir():
print(attr.filename, attr)
return None
def ftp_upload_file(ftp, itemPath):
configValues = configurationValues()
if configValues.sftp_seperateDownloadPath is True:
uploadFolderPath = configValues.sftp_pathToSeperateDownloadPath
def ftp_upload_file(ftp, path_item) -> None:
"""
Uploads a file to the ftp server
:param ftp: ftp connection
:param path_item: Name of the item which should be uploaded
:returns: None
"""
config_values = config_value()
if config_values.remote_seperate_download_path is True:
path_upload_folder = config_values.remote_path_to_seperate_download_path
else:
uploadFolderPath = configValues.sftp_folderPath
path_upload_folder = config_values.remote_plugin_folder_on_server
try:
ftp.cwd(uploadFolderPath)
itemPath = os.path.relpath(itemPath, 'TempSFTPFolder/')
itemPath = str(itemPath)
currentDirectory = os.getcwd()
ftp.cwd(path_upload_folder)
path_item = os.path.relpath(path_item, 'TempSFTPFolder/')
path_item = str(path_item)
current_directory = os.getcwd()
os.chdir('TempSFTPFolder')
with open (itemPath, 'rb') as plugin_file:
ftp.storbinary('STOR '+ str(itemPath), plugin_file)
with open (path_item, 'rb') as plugin_file:
ftp.storbinary('STOR '+ str(path_item), plugin_file)
except FileNotFoundError:
print(oColors.brightRed + "[FTP]: The 'plugins' folder couldn*t be found on the remote host!" + oColors.standardWhite)
print(oColors.brightRed + "[FTP]: Aborting uploading." + oColors.standardWhite)
os.chdir(currentDirectory)
rich_print_error("Error: [FTP]: The 'plugins' folder couldn't be found on the remote host!")
rich_print_error("Error: [FTP]: Aborting uploading.")
os.chdir(current_directory)
ftp.close()
return None
def ftp_upload_server_jar(ftp, itemPath):
def ftp_upload_server_jar(ftp, path_item) -> None:
"""
Uploads a serverjar to the root folder of the ftp host
:param ftp: ftp connection
:param path_item: Name of the file which should be uploaded
:returns: None
"""
try:
ftp.cwd('.')
itemPath = os.path.relpath(itemPath, 'TempSFTPFolder/')
itemPath = str(itemPath)
currentDirectory = os.getcwd()
path_item = os.path.relpath(path_item, 'TempSFTPFolder/')
path_item = str(path_item)
current_directory = os.getcwd()
os.chdir('TempSFTPFolder')
with open (itemPath, 'rb') as server_jar:
ftp.storbinary('STOR '+ str(itemPath), server_jar)
with open (path_item, 'rb') as server_jar:
ftp.storbinary('STOR '+ str(path_item), server_jar)
except FileNotFoundError:
print(oColors.brightRed + "[FTP]: The 'root' folder couldn*t be found on the remote host!" + oColors.standardWhite)
print(oColors.brightRed + "[FTP]: Aborting uploading." + oColors.standardWhite)
os.chdir(currentDirectory)
rich_print_error("Error: [FTP]: The 'root' folder couldn't be found on the remote host!")
rich_print_error("Error: [FTP]: Aborting uploading.")
os.chdir(current_directory)
ftp.close()
return None
def ftp_listAll(ftp):
configValues = configurationValues()
def ftp_list_all(ftp):
"""
Returns a list with all installed plugins in the plugin folder of the ftp host
:param ftp: ftp connection
:returns: List of all plugins in plugin folder
"""
config_values = config_value()
try:
ftp.cwd(configValues.sftp_folderPath)
installedPlugins = ftp.nlst()
ftp.cwd(config_values.remote_plugin_folder_on_server)
installed_plugins = ftp.nlst()
except FileNotFoundError:
print(oColors.brightRed + "[FTP]: The 'plugins' folder couldn*t be found on the remote host!" + oColors.standardWhite)
rich_print_error("Error: [FTP]: The 'plugins' folder couldn't be found on the remote host!")
try:
return installedPlugins
return installed_plugins
except UnboundLocalError:
print(oColors.brightRed + "[FTP]: No plugins were found." + oColors.standardWhite)
rich_print_error("Error: [FTP]: No plugins were found.")
def ftp_listFilesInServerRoot(ftp):
def ftp_list_files_in_server_root(ftp):
"""
Returns a list with all files in the root folder of the ftp host
:param ftp: ftp connection
:returns: List of all files in root folder
"""
try:
ftp.cwd('.')
filesInServerRoot = ftp.nlst()
except FileNotFoundError:
print(oColors.brightRed + "[FTP]: The 'root' folder couldn*t be found on the remote host!" + oColors.standardWhite)
rich_print_error("Error: [FTP]: The 'root' folder couldn't be found on the remote host!")
try:
return filesInServerRoot
except UnboundLocalError:
print(oColors.brightRed + "[FTP]: No Serverjar was found." + oColors.standardWhite)
rich_print_error("Error: [FTP]: No Serverjar was found.")
def ftp_downloadFile(ftp, downloadPath, fileToDownload):
configValues = configurationValues()
ftp.cwd(configValues.sftp_folderPath)
filedata = open(downloadPath,'wb')
ftp.retrbinary('RETR '+fileToDownload, filedata.write)
def ftp_download_file(ftp, path_download, file) -> None:
"""
Download a file of the ftp server
:param ftp: ftp connection
:param path_download: Path to save downloaded file to
:param file: File to download
:returns None
"""
config_values = config_value()
ftp.cwd(config_values.remote_plugin_folder_on_server)
filedata = open(path_download,'wb')
ftp.retrbinary('RETR '+file, filedata.write)
filedata.close()
ftp.quit()
return None
def ftp_is_file(ftp, pluginPath):
if ftp.nlst(pluginPath) == [pluginPath]:
def ftp_is_file(ftp, plugin_path) -> bool:
"""
Check if file on ftp host is a file and not a directory
:param ftp: ftp connection
:param plugin_path
:returns: True if file is a file and not a directory
"""
if ftp.nlst(plugin_path) == [plugin_path]:
return True
else:
return False
def ftp_validateFileAttributes(ftp, pluginPath):
if ftp_is_file(ftp, pluginPath) is False:
def ftp_validate_file_attributes(ftp, plugin_path) -> bool:
"""
Check if a file is a legitimate plugin file
:param ftp: ftp connection
:param plugin_path: Path of file to check
:returns: If file is a plugin file or not
"""
if ftp_is_file(ftp, plugin_path) is False:
return False
if re.search(r'.jar$', pluginPath):
if re.search(r'.jar$', plugin_path):
return True
else:
return False

196
src/handlers/handle_input.py
View File

@ -1,105 +1,121 @@
import sys
""""
Handles the input through the pluGET command line
"""
from utils.consoleoutput import oColors
from utils.utilities import getHelp, getCommandHelp
from handlers.handle_config import configurationValues
from plugin.plugin_downloader import searchPackage, getSpecificPackage
from plugin.plugin_updatechecker import updateInstalledPackage, checkInstalledPackage
from plugin.plugin_remover import removePlugin
from serverjar.serverjar_checker import checkInstalledServerjar, updateServerjar
from serverjar.serverjar_paper import papermc_downloader
from src.utils.console_output import rich_print_error
from src.plugin.plugin_downloader import get_specific_plugin_spiget, search_specific_plugin_spiget
from src.plugin.plugin_updatechecker import check_installed_plugins, update_installed_plugins
from src.serverjar.serverjar_updatechecker import \
check_update_available_installed_server_jar, update_installed_server_jar
from src.serverjar.serverjar_paper_velocity_waterfall import serverjar_papermc_update
def createInputLists():
global COMMANDLIST
COMMANDLIST = [
'get',
'update',
'check',
'search',
'exit',
'help',
'remove',
'get-paper'
]
global INPUTSELECTEDOBJECT
INPUTSELECTEDOBJECT = [
'all',
'*'
]
# check
# update
# get
# get-paper
# get-waterfall
# get-velocity
# get-purpur ???
# exit
# remove
# search
def handleInput(inputCommand, inputSelectedObject, inputParams):
configValues = configurationValues()
def handle_input(
input_command : str=None,
input_selected_object : str=None,
input_parameter : str=None,
no_confirmation : bool=False,
arguments_from_console : bool=False
) -> None:
"""
Manages the correct function calling from the given input
:param input_command: Command of main function
:param input_selected_object: Name of plugin/serverjar
:param: input_parameter: Optional parameters
:param no_confirmation: If plugins should be updated without no confirmation message
:param arguments_from_console: If arguments were given on script call
:returns None:
"""
while True:
if inputCommand == 'get':
if inputSelectedObject.isdigit():
if not configValues.localPluginFolder:
if configValues.sftp_seperateDownloadPath is True:
pluginPath = configValues.sftp_pathToSeperateDownloadPath
else:
pluginPath = configValues.sftp_folderPath
getSpecificPackage(inputSelectedObject, pluginPath, inputParams)
break
else:
if configValues.seperateDownloadPath is True:
pluginPath = configValues.pathToSeperateDownloadPath
else:
pluginPath = configValues.pathToPluginFolder
getSpecificPackage(inputSelectedObject, pluginPath, inputParams)
break
else:
searchPackage(inputSelectedObject)
break
if inputCommand == 'update':
if inputSelectedObject == 'serverjar':
updateServerjar(inputParams)
else:
updateInstalledPackage(inputSelectedObject)
break
if inputCommand == 'check':
if inputSelectedObject == 'serverjar':
checkInstalledServerjar()
else:
checkInstalledPackage(inputSelectedObject, inputParams)
break
if inputCommand == 'search':
searchPackage(inputSelectedObject)
break
if inputCommand == 'exit':
sys.exit()
if inputCommand == 'help':
if inputSelectedObject == 'command' or inputSelectedObject == 'commands':
getCommandHelp(inputParams)
else:
getHelp()
break
if inputCommand == 'remove':
removePlugin(inputSelectedObject)
break
if inputCommand == 'get-paper':
papermc_downloader(inputSelectedObject, inputParams)
break
else:
print(oColors.brightRed + "Error: Command not found. Please try again. :(" + oColors.standardWhite)
print(oColors.brightRed + "Use: '" + oColors.standardWhite +"help command" + oColors.brightRed +"' to get all available commands" + oColors.standardWhite)
getInput()
getInput()
# when arguemnts were not passed from console ask for input
if arguments_from_console is False:
try:
input_command, input_selected_object, input_parameter = get_input()
except TypeError:
# KeyboardInterrupt was triggered and None was returned so exit
return
match input_command:
case "get":
match input_selected_object.isdigit():
case True:
get_specific_plugin_spiget(input_selected_object, input_parameter)
case _:
search_specific_plugin_spiget(input_selected_object)
case "update":
match input_selected_object:
case "serverjar":
update_installed_server_jar(input_parameter)
case _:
update_installed_plugins(input_selected_object, no_confirmation)
case "check":
match input_selected_object:
case "serverjar":
check_update_available_installed_server_jar()
case _:
check_installed_plugins(input_selected_object, input_parameter)
case "search":
search_specific_plugin_spiget(input_selected_object)
# TODO add remover
#case "remove":
# print("remove package")
# #removePlugin(inputSelectedObject)
case "get-paper":
serverjar_papermc_update(input_selected_object, input_parameter, None, "paper")
case "get-velocity":
serverjar_papermc_update(input_selected_object, input_parameter, None, "velocity")
case "get-waterfall":
serverjar_papermc_update(input_selected_object, input_parameter, None, "waterfall")
case "exit":
return
case _:
rich_print_error("Error: Command not found. Please try again. :(")
rich_print_error("Use: 'help command' to get all available commands")
# return to break out of while loop if pluGET was started with arguments from console
if arguments_from_console:
return None
def getInput():
inputCommand = None
def get_input() -> str:
"""
Gets command line input and calls the handle input function
:returns: Main command to execute
:returns: Selected Object to work with
:returns: Optional parameter
"""
input_command = None
print("\n'STRG + C' to exit")
while True:
try:
inputCommand, inputSelectedObject, *inputParams = input("pluGET >> ").split()
input_command, input_selected_object, *input_parameter = input("pluGET >> ").split()
break
except ValueError:
if inputCommand == None:
if input_command == None:
# request input again if no input was given or not enough
continue
else:
print(oColors.brightRed + "Wrong input! Use: > 'command' 'selectedObject' [optionalParams]" + oColors.standardWhite)
print(oColors.brightRed + "Use: '" + oColors.standardWhite +"help command" + oColors.brightRed +"' to get all available commands" + oColors.standardWhite)
rich_print_error("Wrong input! Use: > 'command' 'selectedObject' [optionalParams]")
rich_print_error("Use: 'help command' to get all available commands")
except KeyboardInterrupt:
sys.exit()
inputParams = inputParams[0] if inputParams else None
handleInput(inputCommand, inputSelectedObject, inputParams)
return
input_parameter = input_parameter[0] if input_parameter else None
return input_command, input_selected_object, input_parameter

171
src/handlers/handle_sftp.py
View File

@ -5,103 +5,164 @@ import paramiko
import stat
import re
from utils.consoleoutput import oColors
from handlers.handle_config import configurationValues
from src.utils.console_output import rich_print_error
from src.handlers.handle_config import config_value
def createSFTPConnection():
configValues = configurationValues()
def sftp_create_connection():
"""
Creates a sftp connection with the given values in the config file
:returns: SFTP connection type
"""
config_values = config_value()
cnopts = pysftp.CnOpts()
cnopts.hostkeys = None # TODO fix this
try:
sftp = pysftp.Connection(configValues.sftp_server, username=configValues.sftp_user, \
password=configValues.sftp_password, port=configValues.sftp_port, cnopts=cnopts)
sftp = pysftp.Connection(config_values.server, username=config_values.username, \
password=config_values.password, port=config_values.sftp_port, cnopts=cnopts)
except paramiko.ssh_exception.AuthenticationException:
print(oColors.brightRed + "[SFTP]: Wrong Username/Password" + oColors.standardWhite)
rich_print_error("Error: [SFTP]: Wrong Username/Password")
except paramiko.ssh_exception.SSHException:
print(oColors.brightRed + "[SFTP]: The SFTP server isn't available." + oColors.standardWhite)
rich_print_error("Error: [SFTP]: The SFTP server isn't available.")
try:
return sftp
except UnboundLocalError:
print(oColors.brightRed + "[SFTP]: Check your config.ini!" + oColors.standardWhite)
print(oColors.brightRed + "Exiting program..." + oColors.standardWhite)
rich_print_error("Error: [SFTP]: Check your config file!")
rich_print_error("Exiting program...")
sys.exit()
def sftp_showPlugins(sftp):
configValues = configurationValues()
sftp.cd(configValues.sftp_folderPath)
def sftp_show_plugins(sftp) -> None:
"""
Prints all plugins in the sftp folder
:param sftp: sftp connection
:returns: None
"""
config_values = config_value()
sftp.cd(config_values.remote_plugin_folder_on_server)
for attr in sftp.listdir_attr():
print(attr.filename, attr)
def sftp_upload_file(sftp, itemPath):
configValues = configurationValues()
if configValues.sftp_seperateDownloadPath is True:
uploadFolderPath = configValues.sftp_pathToSeperateDownloadPath
else:
uploadFolderPath = configValues.sftp_folderPath
try:
sftp.chdir(uploadFolderPath)
sftp.put(itemPath)
except FileNotFoundError:
print(oColors.brightRed + "[SFTP]: The 'plugins' folder couldn*t be found on the remote host!" + oColors.standardWhite)
print(oColors.brightRed + "[SFTP]: Aborting uploading." + oColors.standardWhite)
sftp.close()
return None
def sftp_upload_server_jar(sftp, itemPath):
def sftp_upload_file(sftp, path_item) -> None:
"""
Uploads a file to the set folder from the config file
:param sftp: sftp connection
:param path_item: The upload path with the item name
:returns: None
"""
config_values = config_value()
if config_values.remote_seperate_download_path is True:
path_upload_folder = config_values.remote_path_to_seperate_download_path
else:
path_upload_folder = config_values.remote_plugin_folder_on_server
try:
sftp.chdir(path_upload_folder)
sftp.put(path_item)
except FileNotFoundError:
rich_print_error("Error: [SFTP]: The 'plugins' folder couldn't be found on the remote host!")
rich_print_error("Error: [SFTP]: Aborting uploading.")
sftp.close()
return None
def sftp_upload_server_jar(sftp, path_item) -> None:
"""
Uploads the server jar to the root folder
:param sftp: sftp connection
:param path_item: The upload path with the item name
:returns: None
"""
try:
sftp.chdir('.')
sftp.put(itemPath)
sftp.put(path_item)
except FileNotFoundError:
print(oColors.brightRed + "[SFTP]: The 'root' folder couldn*t be found on the remote host!" + oColors.standardWhite)
print(oColors.brightRed + "[SFTP]: Aborting uploading." + oColors.standardWhite)
rich_print_error("Error: [SFTP]: The 'root' folder couldn't be found on the remote host!")
rich_print_error("Error: [SFTP]: Aborting uploading.")
sftp.close()
return None
def sftp_listAll(sftp):
configValues = configurationValues()
def sftp_list_all(sftp):
"""
List all plugins in the 'plugins' folder on the sftp host
:param sftp: sftp connection
:return: List of plugins in plugin folder
"""
config_values = config_value()
try:
sftp.chdir(configValues.sftp_folderPath)
installedPlugins = sftp.listdir()
sftp.chdir(config_values.remote_plugin_folder_on_server)
installed_plugins = sftp.listdir()
except FileNotFoundError:
print(oColors.brightRed + "[SFTP]: The 'plugins' folder couldn*t be found on the remote host!" + oColors.standardWhite)
rich_print_error("Error: [SFTP]: The 'plugins' folder couldn't be found on the remote host!")
try:
return installedPlugins
return installed_plugins
except UnboundLocalError:
print(oColors.brightRed + "[SFTP]: No plugins were found." + oColors.standardWhite)
rich_print_error("Error: [SFTP]: No plugins were found.")
def sftp_listFilesInServerRoot(sftp):
def sftp_list_files_in_server_root(sftp):
"""
List all files in the root folder on the sftp host
:param sftp: sftp connection
:returns: List of files in root folder
"""
try:
filesInServerRoot = sftp.listdir()
files_in_server_root = sftp.listdir()
except FileNotFoundError:
print(oColors.brightRed + "[SFTP]: The 'root' folder couldn*t be found on the remote host!" + oColors.standardWhite)
rich_print_error("Error: [SFTP]: The 'root' folder couldn't be found on the remote host!")
try:
return filesInServerRoot
return files_in_server_root
except UnboundLocalError:
print(oColors.brightRed + "[SFTP]: No Serverjar was found." + oColors.standardWhite)
rich_print_error("Error: [SFTP]: No Serverjar was found.")
def sftp_downloadFile(sftp, downloadPath, fileToDownload):
configValues = configurationValues()
sftp.cwd(configValues.sftp_folderPath)
currentDirectory = os.getcwd()
def sftp_download_file(sftp, file) -> None:
"""
Downloads a plugin file from the sftp host to a temporary folder
:param sftp: sftp connection
:param file: Filename of plugin
:returns: None
"""
config_values = config_value()
sftp.cwd(config_values.remote_plugin_folder_on_server)
current_directory = os.getcwd()
os.chdir('TempSFTPFolder')
sftp.get(fileToDownload)
sftp.get(file)
sftp.close()
os.chdir(currentDirectory)
os.chdir(current_directory)
return None
def sftp_validateFileAttributes(sftp, pluginPath):
pluginSFTPAttribute = sftp.lstat(pluginPath)
if stat.S_ISDIR(pluginSFTPAttribute.st_mode):
def sftp_validate_file_attributes(sftp, plugin_path) -> bool:
"""
Check if the file is a legitimate plugin file
:param sftp: sftp connection
param plugin_path: Path of the single plugin file
:returns: If file is a plugin file or not
"""
plugin_sftp_attribute = sftp.lstat(plugin_path)
if stat.S_ISDIR(plugin_sftp_attribute.st_mode):
return False
elif re.search(r'.jar$', pluginPath):
elif re.search(r'.jar$', plugin_path):
return True
else:
return False

376
src/plugin/plugin_downloader.py
View File

@ -1,155 +1,245 @@
"""
File and functions which handle the download of the specific plugins
"""
import re
import urllib.request
from urllib.error import HTTPError
from pathlib import Path
import requests
from utils.consoleoutput import oColors
from utils.web_request import doAPIRequest
from utils.utilities import createTempPluginFolder, deleteTempPluginFolder, calculateFileSizeKb, calculateFileSizeMb
from handlers.handle_config import configurationValues
from handlers.handle_sftp import sftp_upload_file, createSFTPConnection
from handlers.handle_ftp import ftp_upload_file, createFTPConnection
from rich.table import Table
from rich.console import Console
from rich.progress import Progress
from src.utils.utilities import convert_file_size_down, remove_temp_plugin_folder, create_temp_plugin_folder
from src.utils.utilities import api_do_request
from src.utils.console_output import rich_print_error
from src.handlers.handle_config import config_value
from src.handlers.handle_sftp import sftp_create_connection, sftp_upload_file
from src.handlers.handle_ftp import ftp_create_connection, ftp_upload_file
def handleRegexPackageName(packageNameFull):
packageNameFull2 = packageNameFull
def handle_regex_plugin_name(full_plugin_name) -> str:
"""
Return the plugin name after trimming clutter from name with regex operations
"""
# trims the part of the package that has for example "[1.1 Off]" in it
unwantedpackageName = re.search(r'(^\[+[a-zA-Z0-9\s\W*\.*\-*\+*\%*\,]*\]+)', packageNameFull)
unwantedpackageNamematch = bool(unwantedpackageName)
if unwantedpackageNamematch:
unwantedpackageNameString = unwantedpackageName.group()
packageNameFull2 = packageNameFull.replace(unwantedpackageNameString, '')
# gets the real packagename "word1 & word2" is not supported only gets word 1
packageName = re.search(r'([a-zA-Z]\d*)+(\s?\-*\_*[a-zA-Z]\d*\+*\-*\'*)+', packageNameFull2)
unwanted_plugin_name = re.search(r'(^\[+[a-zA-Z0-9\s\W*\.*\-*\+*\%*\,]*\]+)', full_plugin_name)
if bool(unwanted_plugin_name):
unwanted_plugin_name_string = unwanted_plugin_name.group()
full_plugin_name = full_plugin_name.replace(unwanted_plugin_name_string, '')
# gets the real plugin_name "word1 & word2" is not supported only gets word1
plugin_name = re.search(r'([a-zA-Z]\d*)+(\s?\-*\_*[a-zA-Z]\d*\+*\-*\'*)+', full_plugin_name)
try:
packageNameFullString = packageName.group()
packageNameOnly = packageNameFullString.replace(' ', '')
plugin_name_full_string = plugin_name.group()
found_plugin_name = plugin_name_full_string.replace(' ', '')
except AttributeError:
packageNameOnly = unwantedpackageNameString
return packageNameOnly
found_plugin_name = unwanted_plugin_name_string
return found_plugin_name
def getVersionID(packageId, packageVersion):
if packageVersion == None or packageVersion == 'latest':
url = f"https://api.spiget.org/v2/resources/{packageId}/versions/latest"
response = doAPIRequest(url)
versionId = response["id"]
return versionId
url = f"https://api.spiget.org/v2/resources/{packageId}/versions?size=100&sort=-name"
versionList = doAPIRequest(url)
for packages in versionList:
packageUpdate = packages["name"]
versionId = packages["id"]
if packageUpdate == packageVersion:
return versionId
return versionList[0]["id"]
def getVersionName(packageId, versionId):
url = f"https://api.spiget.org/v2/resources/{packageId}/versions/{versionId}"
response = doAPIRequest(url)
versionName = response["name"]
return versionName
def searchPackage(resourceName):
configValues = configurationValues()
url = f"https://api.spiget.org/v2/search/resources/{resourceName}?field=name&sort=-downloads"
packageName = doAPIRequest(url)
i = 1
print(oColors.brightBlack + f"Searching: {resourceName}" + oColors.standardWhite)
print("┌─────┬─────────────────────────────┬───────────┬──────────────────────────────────────────────────────────────────────┐")
print("│ No. │ Name │ Downloads │ Description │")
print("└─────┴─────────────────────────────┴───────────┴──────────────────────────────────────────────────────────────────────┘")
for resource in packageName:
pName = resource["name"]
newName = handleRegexPackageName(pName)
pTag = resource["tag"]
pDownloads = resource["downloads"]
print(f" [{i}]".rjust(6), end='')
print(" ", end='')
print(f"{newName}".ljust(30), end='')
print(f"{pDownloads}".rjust(9), end='')
print(" ", end='')
print(f"{pTag}".ljust(120))
i = i + 1
resourceSelected = int(input("Select your wanted resource (No.)(0 to exit): "))
if resourceSelected != 0:
resourceSelected = resourceSelected - 1
resourceId = packageName[resourceSelected]["id"]
if not configValues.localPluginFolder:
if configValues.sftp_seperateDownloadPath is True:
pluginDownloadPath = configValues.sftp_pathToSeperateDownloadPath
else:
pluginDownloadPath = configValues.sftp_folderPath
else:
if configValues.seperateDownloadPath is True:
pluginDownloadPath = configValues.pathToSeperateDownloadPath
else:
pluginDownloadPath = configValues.pathToPluginFolder
try:
getSpecificPackage(resourceId, pluginDownloadPath)
except HTTPError as err:
print(oColors.brightRed + f"Error: {err.code} - {err.reason}" + oColors.standardWhite)
def downloadSpecificVersion(resourceId, downloadPath, versionID='latest'):
configValues = configurationValues()
if versionID != 'latest':
#url = f"https://spigotmc.org/resources/{resourceId}/download?version={versionID}"
print(oColors.brightRed + "Sorry but specific version downloads aren't supported because of cloudflare protection. :(" + oColors.standardWhite)
print(oColors.brightRed + "Reverting to latest version." + oColors.standardWhite)
url = f"https://api.spiget.org/v2/resources/{resourceId}/download"
#url = f"https://api.spiget.org/v2/resources/{resourceId}/versions/latest/download" #throws 403 forbidden error...cloudflare :(
urrlib_opener = urllib.request.build_opener()
urrlib_opener.addheaders = [('User-agent', 'pluGET/1.0')]
urllib.request.install_opener(urrlib_opener)
remotefile = urllib.request.urlopen(url)
filesize = remotefile.info()['Content-Length']
urllib.request.urlretrieve(url, downloadPath)
filesize = int(filesize)
print(" ", end='')
if filesize >= 1000000:
filesizeData = calculateFileSizeMb(filesize)
print("Downloaded " + (str(filesizeData)).rjust(9) + f" MB here {downloadPath}")
else:
filesizeData = calculateFileSizeKb(filesize)
print("Downloaded " + (str(filesizeData)).rjust(9) + f" KB here {downloadPath}")
if not configValues.localPluginFolder:
if configValues.sftp_useSftp:
sftpSession = createSFTPConnection()
sftp_upload_file(sftpSession, downloadPath)
else:
ftpSession = createFTPConnection()
ftp_upload_file(ftpSession, downloadPath)
def getSpecificPackage(resourceId, downloadPath, inputPackageVersion='latest'):
configValues = configurationValues()
if configValues.localPluginFolder == False:
downloadPath = createTempPluginFolder()
url = f"https://api.spiget.org/v2/resources/{resourceId}"
packageDetails = doAPIRequest(url)
try:
packageName = packageDetails["name"]
except KeyError:
print(oColors.brightRed + "Error: Plugin ID couldn't be found" + oColors.standardWhite)
def get_version_id_spiget(plugin_id, plugin_version) -> str:
"""
Returns the version id of the plugin
"""
if plugin_version == None or plugin_version == 'latest':
url = f"https://api.spiget.org/v2/resources/{plugin_id}/versions/latest"
response = api_do_request(url)
if response == None:
return None
packageNameNew = handleRegexPackageName(packageName)
versionId = getVersionID(resourceId, inputPackageVersion)
packageVersion = getVersionName(resourceId, versionId)
packageDownloadName = f"{packageNameNew}-{packageVersion}.jar"
downloadPackagePath = Path(f"{downloadPath}/{packageDownloadName}")
if inputPackageVersion is None or inputPackageVersion == 'latest':
downloadSpecificVersion(resourceId=resourceId, downloadPath=downloadPackagePath)
else:
downloadSpecificVersion(resourceId, downloadPackagePath, versionId)
version_id = response["id"]
return version_id
if not configValues.localPluginFolder:
deleteTempPluginFolder(downloadPath)
url = f"https://api.spiget.org/v2/resources/{plugin_id}/versions?size=100&sort=-name"
version_list = api_do_request(url)
if version_list == None:
return None
for plugins in version_list:
plugin_update = plugins["name"]
version_id = plugins["id"]
if plugin_update == plugin_version:
return version_id
return version_list[0]["id"]
def get_version_name_spiget(plugin_id, plugin_version_id) -> str:
"""
Returns the name of a specific version
"""
url = f"https://api.spiget.org/v2/resources/{plugin_id}/versions/{plugin_version_id}"
response = api_do_request(url)
if response == None:
return None
version_name = response["name"]
return version_name
def get_download_path(config_values) -> str:
"""
Reads the config and gets the path of the plugin folder
"""
match (config_values.connection):
case "local":
match (config_values.local_seperate_download_path):
case True:
return config_values.local_path_to_seperate_download_path
case _:
return config_values.path_to_plugin_folder
case _:
match (config_values.remote_seperate_download_path):
case True:
return config_values.remote_path_to_seperate_download_path
case _:
return config_values.remote_plugin_folder_on_server
def download_specific_plugin_version_spiget(plugin_id, download_path, version_id="latest") -> None:
"""
Download a specific plugin
"""
config_values = config_value()
if version_id != "latest" and version_id != None:
#url = f"https://spigotmc.org/resources/{plugin_id}/download?version={versionID}"
rich_print_error("Sorry but specific version downloads aren't supported because of cloudflare protection. :(")
rich_print_error("Reverting to latest version.")
#throws 403 forbidden error...cloudflare :(
#url = f"https://api.spiget.org/v2/resources/{plugin_id}/versions/latest/download"
url = f"https://api.spiget.org/v2/resources/{plugin_id}/download"
# use rich Progress() to create progress bar
with Progress(transient=True) as progress:
header = {'user-agent': 'pluGET/1.0'}
r = requests.get(url, headers=header, stream=True)
try:
file_size = int(r.headers.get('content-length'))
# create progress bar
download_task = progress.add_task(" [cyan]Downloading...", total=file_size)
except TypeError:
# Content-lenght returned nothing
file_size = 0
with open(download_path, 'wb') as f:
# split downloaded data in chunks of 32768
for data in r.iter_content(chunk_size=32768):
f.write(data)
# don't show progress bar if no content-length was returned
if file_size == 0:
continue
progress.update(download_task, advance=len(data))
#f.flush()
# use rich console for nice colors
console = Console()
if file_size == 0:
console.print(
f" [not bold][bright_green]Downloaded[bright_magenta] file [cyan]→ [white]{download_path}"
)
elif file_size >= 1000000:
file_size_data = convert_file_size_down(convert_file_size_down(file_size))
console.print(" [not bold][bright_green]Downloaded[bright_magenta] " + (str(file_size_data)).rjust(9) + \
f" MB [cyan]→ [white]{download_path}")
else:
file_size_data = convert_file_size_down(file_size)
console.print(" [not bold][bright_green]Downloaded[bright_magenta] " + (str(file_size_data)).rjust(9) + \
f" KB [cyan]→ [white]{download_path}")
if config_values.connection == "sftp":
sftp_session = sftp_create_connection()
sftp_upload_file(sftp_session, download_path)
elif config_values.connection == "ftp":
ftp_session = ftp_create_connection()
ftp_upload_file(ftp_session, download_path)
return None
def get_specific_plugin_spiget(plugin_id, plugin_version="latest") -> None:
"""
Gets the specific plugin and calls the download function
"""
config_values = config_value()
# use a temporary folder to store plugins until they are uploaded
if config_values.connection != "local":
download_path = create_temp_plugin_folder()
else:
download_path = get_download_path(config_values)
url = f"https://api.spiget.org/v2/resources/{plugin_id}"
plugin_details = api_do_request(url)
if plugin_details == None:
return None
try:
plugin_name = plugin_details["name"]
except KeyError:
# exit if plugin id couldn't be found
rich_print_error("Error: Plugin ID couldn't be found")
return None
plugin_name = handle_regex_plugin_name(plugin_name)
plugin_version_id = get_version_id_spiget(plugin_id, plugin_version)
plugin_version_name = get_version_name_spiget(plugin_id, plugin_version_id)
plugin_download_name = f"{plugin_name}-{plugin_version_name}.jar"
download_plugin_path = Path(f"{download_path}/{plugin_download_name}")
# if api requests weren't successfull stop function
if plugin_version_id == None or plugin_version_name == None:
rich_print_error("Error: Webrequest timed out")
return None
# set the plugin_version_id to None if a specific version wasn't given as parameter
if plugin_version == "latest" or plugin_version is None:
plugin_version_id = None
download_specific_plugin_version_spiget(plugin_id, download_plugin_path, plugin_version_id)
if config_values.connection != "local":
remove_temp_plugin_folder()
return None
def search_specific_plugin_spiget(plugin_name) -> None:
"""
Search for a name and return the top 10 results sorted for their download count
Then ask for input and download that plugin
"""
url= f"https://api.spiget.org/v2/search/resources/{plugin_name}?field=name&sort=-downloads"
plugin_search_results = api_do_request(url)
if plugin_search_results == None:
rich_print_error("Error: Webrequest wasn't successfull!")
return None
print(f"Searching for {plugin_name}...")
print(f"Found plugins:")
# create table with rich
rich_table = Table(box=None)
rich_table.add_column("No.", justify="right", style="cyan", no_wrap=True)
rich_table.add_column("Name", style="bright_magenta")
rich_table.add_column("Downloads", justify="right", style="bright_green")
rich_table.add_column("Description", justify="left", style="white")
# start counting at 1 for all my non-programming friends :)
i = 1
for found_plugin in plugin_search_results:
plugin_name = handle_regex_plugin_name(found_plugin["name"])
plugin_downloads = found_plugin["downloads"]
plugin_description = found_plugin["tag"]
rich_table.add_row(str(i), plugin_name, str(plugin_downloads), plugin_description)
i += 1
# print table from rich
rich_console = Console()
rich_console.print(rich_table)
try:
plugin_selected = input("Select your wanted resource (No.)(0 to exit): ")
except KeyboardInterrupt:
return None
if plugin_selected == "0":
return None
try:
plugin_selected = int(plugin_selected) - 1
plugin_selected_id = plugin_search_results[plugin_selected]["id"]
except ValueError:
rich_print_error("Error: Input wasn't a number! Please try again!")
return None
except IndexError:
rich_print_error("Error: Number was out of range! Please try again!")
return None
selected_plugin_name = handle_regex_plugin_name(plugin_search_results[plugin_selected]["name"])
rich_console.print(f"\n [not bold][bright_white]● [bright_magenta]{selected_plugin_name} [bright_green]latest")
get_specific_plugin_spiget(plugin_selected_id)

59
src/plugin/plugin_remover.py
View File

@ -1,59 +0,0 @@
import os
import re
from pathlib import Path
from utils.consoleoutput import oColors
from handlers.handle_config import configurationValues
from handlers.handle_sftp import createSFTPConnection, sftp_listAll
from handlers.handle_ftp import createFTPConnection, ftp_listAll
from plugin.plugin_updatechecker import getFileName, getFileVersion, getInstalledPlugin, createPluginList
def removePlugin(pluginToRemove):
configValues = configurationValues()
createPluginList()
if not configValues.localPluginFolder:
if not configValues.sftp_useSftp:
ftp = createFTPConnection()
pluginList = ftp_listAll(ftp)
else:
sftp = createSFTPConnection()
pluginList = sftp_listAll(sftp)
else:
pluginList = os.listdir(configValues.pathToPluginFolder)
i = 0
try:
for plugin in pluginList:
try:
fileName = getFileName(plugin)
fileVersion = getFileVersion(plugin)
pluginId = getInstalledPlugin(fileName, fileVersion)
except TypeError:
continue
pluginIdStr = str(pluginId)
if pluginToRemove == pluginIdStr or re.search(pluginToRemove, fileName, re.IGNORECASE):
print(f"Removing: {fileName}")
if not configValues.localPluginFolder:
pluginPath = configValues.sftp_folderPath
pluginPath = f"{pluginPath}/{plugin}"
if not configValues.sftp_useSftp:
ftp = createFTPConnection()
ftp.delete(pluginPath)
else:
sftp = createSFTPConnection()
sftp.remove(pluginPath)
print(f"Removed: {fileName}")
i += 1
break
else:
pluginPath = configValues.pathToPluginFolder
pluginPath = Path(f"{pluginPath}/{plugin}")
os.remove(pluginPath)
print(f"Removed: {fileName}")
i += 1
break
except TypeError:
print(oColors.brightRed + f"Aborted removing of: {pluginToRemove}." + oColors.standardWhite)
if i == 0:
print(oColors.brightRed + f"Couldn't remove plugin: {pluginToRemove}" + oColors.standardWhite)

884
src/plugin/plugin_updatechecker.py
View File

@ -1,482 +1,568 @@
"""
Handles the plugin checking and updating
"""
import os
import re
import io
import base64
from zipfile import ZipFile
from urllib.error import HTTPError
from pathlib import Path
import zipfile
from rich.progress import track
from rich.table import Table
from rich.console import Console
from urllib.error import HTTPError
from zipfile import ZipFile
from utils.consoleoutput import oColors
from utils.web_request import doAPIRequest
from handlers.handle_config import configurationValues
from handlers.handle_sftp import createSFTPConnection, sftp_listAll, sftp_downloadFile, sftp_validateFileAttributes
from handlers.handle_ftp import createFTPConnection, ftp_listAll, ftp_downloadFile, ftp_validateFileAttributes
from plugin.plugin_downloader import getSpecificPackage
from utils.utilities import createTempPluginFolder, deleteTempPluginFolder
from src.handlers.handle_config import config_value
from src.handlers.handle_sftp import sftp_create_connection, sftp_download_file, sftp_validate_file_attributes, sftp_list_all
from src.handlers.handle_ftp import ftp_create_connection, ftp_download_file, ftp_validate_file_attributes, ftp_list_all
from src.plugin.plugin_downloader import get_specific_plugin_spiget, get_download_path
from src.utils.console_output import rich_print_error
from src.utils.utilities import api_do_request, create_temp_plugin_folder, remove_temp_plugin_folder
def createPluginList():
class Plugin():
"""
Create plugin class to store installed plugins inside it
"""
def __init__(
self,
plugin_file_name : str,
plugin_name : str,
plugin_file_version : str,
plugin_latest_version : str,
plugin_is_outdated : bool,
plugin_repository : str,
plugin_repository_data : list
) -> None:
self.plugin_file_name = plugin_file_name
self.plugin_name = plugin_name
self.plugin_file_version = plugin_file_version
self.plugin_latest_version = plugin_latest_version
self.plugin_is_outdated = plugin_is_outdated
self.plugin_repository = plugin_repository
self.plugin_repository_data = plugin_repository_data
@staticmethod
def create_plugin_list() -> list:
"""
Creates a global array list to store plugins
"""
global INSTALLEDPLUGINLIST
INSTALLEDPLUGINLIST = []
return INSTALLEDPLUGINLIST
def addToPluginList(localFileName, pluginId, versionId, plugin_latest_version, plugin_is_outdated):
INSTALLEDPLUGINLIST.append([localFileName, pluginId, versionId, plugin_latest_version, plugin_is_outdated])
@staticmethod
def add_to_plugin_list(
plugin_file_name: str,
plugin_name : str,
plugin_file_version : str,
plugin_latest_version : str,
plugin_is_outdated : bool,
plugin_repository : str,
plugin_repository_data : list
) -> None:
"""
Adds a plugin to global installed plugin lists
"""
INSTALLEDPLUGINLIST.append(Plugin(
plugin_file_name,
plugin_name,
plugin_file_version,
plugin_latest_version,
plugin_is_outdated,
plugin_repository,
plugin_repository_data
))
return None
def getFileName(pluginName):
pluginNameFull = pluginName
pluginVersion = re.search(r'([\d.]+[.jar]+)', pluginNameFull)
def get_plugin_file_name(plugin_full_name: str) -> str:
"""
Finds the full plugin name of the given string
Example LuckPerms-5.4.30.jar -> Luckperms
:param plugin_full_name: Full filename of plugin
:returns: Full plugin name
"""
plugin_full_name2 = plugin_full_name
# find number.jar
plugin_file_version = re.search(r'([\d.]+[.jar]+)', plugin_full_name2)
try:
pluginVersionFull = pluginVersion.group()
plugin_file_version_full = plugin_file_version.group()
except AttributeError:
pluginVersionFull = pluginVersion
pluginNameOnlyy = pluginNameFull.replace(pluginVersionFull, '')
pluginNameOnly = re.sub(r'(\-$)', '', pluginNameOnlyy)
pluginNameOnlyy = re.sub(r'(\-v$)', '', pluginNameOnly)
return pluginNameOnlyy
plugin_file_version_full = plugin_file_version
# remove number from plugin name
plugin_name_only = plugin_full_name2.replace(plugin_file_version_full, '')
# remove - from plugin name
plugin_name_only = re.sub(r'(\-$)', '', plugin_name_only)
# remove -v from plugin name
plugin_name_only = re.sub(r'(\-v$)', '', plugin_name_only)
return plugin_name_only
def getFileVersion(pluginName):
pluginNameFull = pluginName
pluginVersion = re.search(r'([\d.]+[.jar]+)', pluginNameFull)
pluginVersionFull = pluginVersion.group()
pluginVersionString = pluginVersionFull.replace('.jar', '')
if pluginVersionString.endswith('.'):
pluginVersionString = ''
if pluginVersionString == '':
pluginVersionString = eggCrackingJar(pluginNameFull, 'version')
return pluginVersionString
def get_plugin_file_version(plugin_full_name: str) -> str:
"""
Gets the version of the plugin
:param plugin_full_name: Full filename of plugin
:returns: Version of plugin as string
"""
plugin_file_version = re.search(r'([\d.]+[.jar]+)', plugin_full_name)
plugin_file_version = plugin_file_version.group()
plugin_file_version = plugin_file_version.replace('.jar', '')
if plugin_file_version.endswith('.'):
print("get_plugin_file_version endswith .")
plugin_file_name, plugin_file_version = egg_cracking_jar(plugin_full_name)
return plugin_file_version
def getLatestPluginVersion(pluginId):
url = f"https://api.spiget.org/v2/resources/{pluginId}/versions/latest"
latestUpdateSearch = doAPIRequest(url)
versionLatestUpdate = latestUpdateSearch["name"]
return versionLatestUpdate
def get_latest_plugin_version_spiget(plugin_id : str) -> str:
"""
Gets the latest spigot plugin version
:param plugin_id: Plugin Spigot ID
:returns: Name of the latest update
"""
url = f"https://api.spiget.org/v2/resources/{plugin_id}/versions/latest"
latest_update_search = api_do_request(url)
return str(latest_update_search["name"])
def getUpdateDescription(pluginId):
url = f"https://api.spiget.org/v2/resources/{pluginId}/updates?size=1&sort=-date"
latestDescriptionSearch = doAPIRequest(url)
versionLatestDescription = latestDescriptionSearch[0]["description"]
versionLatestDescription = base64.b64decode(versionLatestDescription)
versionLatestDescriptionText =versionLatestDescription.decode('utf-8')
htmlRegex = re.compile('<.*?>')
versionLatestDescriptionText = re.sub(htmlRegex, '', versionLatestDescriptionText)
linesChangelogDescription = versionLatestDescriptionText.split("\n")
nonEmptyLines = [line for line in linesChangelogDescription if line.strip() != ""]
stringnonEmptyLines = ""
for line in nonEmptyLines:
stringnonEmptyLines += line + "\n"
stringnonEmptyLines = stringnonEmptyLines[:-1]
return stringnonEmptyLines
def create_plugin_version_tuple(plugin_version_string : str) -> tuple:
"""
Create a tuple of all version numbers
:param plugin_version_string: Plugin Version
:returns: Tuple of all version numbers
"""
return tuple(map(int, (plugin_version_string.split("."))))
def versionTuple(versionString):
return tuple(map(int, (versionString.split("."))))
def get_plugin_version_without_letters(plugin_version_string : str) -> str:
"""
Returns the version without letters from the plugin version
:param plugin_version_string: Plugin Version
:returns: Plugin version without letters
"""
return re.sub(r'([A-Za-z]*)', '', plugin_version_string)
def getVersionWithoutLetters(versionString):
return re.sub(r'([A-Za-z]*)', '', versionString)
def compare_plugin_version(plugin_latest_version : str, plugin_file_version : str) -> bool:
"""
Check if plugin version is outdated
:param plugin_latest_version: Latest available plugin version
:param plugin_file_version: Installed plugin version
def compareVersions(plugin_latest_version, pluginVersion):
:returns: bool if plugin version is outdated
"""
try:
pluginVersionTuple = versionTuple(getVersionWithoutLetters(pluginVersion))
plugin_latest_versionTuple = versionTuple(getVersionWithoutLetters(plugin_latest_version))
plugin_version_tuple = create_plugin_version_tuple(
get_plugin_version_without_letters(plugin_file_version))
plugin_latest_version_tuple = create_plugin_version_tuple(
get_plugin_version_without_letters(plugin_latest_version))
except ValueError:
return False
if pluginVersionTuple < plugin_latest_versionTuple:
if plugin_version_tuple < plugin_latest_version_tuple:
return True
else:
return False
def eggCrackingJar(localJarFileName, searchMode):
configValues = configurationValues()
if not configValues.localPluginFolder:
tempPluginFolderPath = createTempPluginFolder()
if configValues.sftp_useSftp:
sftp = createSFTPConnection()
pathToPluginJar = Path(f"{tempPluginFolderPath}/{localJarFileName}")
sftp_downloadFile(sftp, pathToPluginJar, localJarFileName)
else:
ftp = createFTPConnection()
pathToPluginJar = Path(f"{tempPluginFolderPath}/{localJarFileName}")
ftp_downloadFile(ftp, pathToPluginJar, localJarFileName)
else:
pluginPath = configValues.pathToPluginFolder
pathToPluginJar = Path(f"{pluginPath}/{localJarFileName}")
pluginVersion = ''
pluginName = ''
with ZipFile(pathToPluginJar, 'r') as pluginJar:
def ask_update_confirmation(input_selected_object : str) -> bool:
"""
Prints confirmation message of plugins which get updated and ask for confirmation
:param input_selected_object: Command line input
:returns: True or False if plugins should be udpated
"""
rich_console = Console()
rich_console.print("Selected plugins with available Updates:")
for plugin_file in INSTALLEDPLUGINLIST:
if plugin_file.plugin_is_outdated == False:
continue
if input_selected_object != "all" and input_selected_object != "*":
if re.search(input_selected_object, plugin_file.plugin_file_name, re.IGNORECASE):
rich_console.print(f"[not bold][bright_magenta]{plugin_file.plugin_name}", end=' ')
break
rich_console.print(f"[not bold][bright_magenta]{plugin_file.plugin_name}", end=' ')
rich_console.print()
update_confirmation = input("Update these plugins [y/n] ? ")
if str.lower(update_confirmation) != "y":
rich_print_error("Aborting the update process")
return False
return True
def egg_cracking_jar(plugin_file_name: str) -> str:
"""
Opens the plugin file as an archive and searches the plugin.yml file for the name and version entry
:param plugin_file_name: Filename of the plugin which should be openend
:returns: Plugin name in plugin.yml file
:returns: Plugin version in plugin.yml file
"""
config_values = config_value()
match config_values.connection:
case "sftp":
path_temp_plugin_folder = create_temp_plugin_folder()
connection = sftp_create_connection()
sftp_download_file(connection, plugin_file_name)
path_plugin_jar = Path(f"{path_temp_plugin_folder}/{plugin_file_name}")
case "ftp":
path_temp_plugin_folder = create_temp_plugin_folder()
connection = ftp_create_connection()
ftp_download_file(connection, plugin_file_name)
path_plugin_jar = Path(f"{path_temp_plugin_folder}/{plugin_file_name}")
case _:
path_plugin_folder = config_values.path_to_plugin_folder
path_plugin_jar = Path(f"{path_plugin_folder}/{plugin_file_name}")
# later used to escape for-loop
plugin_name = plugin_version = ""
# open plugin if it is an archive and read plugin.yml line for line to find name & version
try:
with io.TextIOWrapper(pluginJar.open('plugin.yml', 'r'), encoding="utf-8") as pluginYml:
pluginYmlContentLine = pluginYml.readlines()
for line in pluginYmlContentLine:
if searchMode == 'version':
if re.match(r'^\s*?version: ', line):
pluginVersion = re.sub(r'^\s*?version: ', '', line)
pluginVersion = pluginVersion.replace('\n', '')
pluginVersion = pluginVersion.replace("'", '')
pluginVersion = pluginVersion.replace('"', '')
elif searchMode == 'name':
if re.match(r'^\s*?name: ', line):
pluginName = re.sub(r'^\s*?name: ', '', line)
pluginName = pluginName.replace('\n', '')
pluginName = pluginName.replace("'", '')
pluginName = pluginName.replace('"', '')
with ZipFile(path_plugin_jar, "r") as plugin_jar:
with io.TextIOWrapper(plugin_jar.open("plugin.yml", "r"), encoding="utf-8") as plugin_yml:
for line in plugin_yml:
if plugin_name != "" and plugin_version != "":
break
if re.match(r"^\s*?name: ", line):
plugin_name = re.sub(r'^\s*?name: ', '', line)
plugin_name = plugin_name.replace("\n", "").replace("'", "").replace('"', "")
if re.match(r"^\s*?version: ", line):
plugin_version = re.sub(r'^\s*?version: ', "", line)
plugin_version = plugin_version.replace("\n", "").replace("'", "").replace('"', "")
except FileNotFoundError:
pluginVersion = ''
pluginName = ''
plugin_name = plugin_version = ""
except KeyError:
pluginVersion = ''
pluginName = ''
if not configValues.localPluginFolder:
deleteTempPluginFolder(tempPluginFolderPath)
if searchMode == 'version':
return pluginVersion
if searchMode == 'name':
return pluginName
plugin_name = plugin_version = ""
except zipfile.BadZipFile:
plugin_name = plugin_version = ""
# remove temp plugin folder if plugin was downloaded from sftp/ftp server
if config_values.connection != "local":
remove_temp_plugin_folder()
return plugin_name, plugin_version
def checkInstalledPackage(inputSelectedObject="all", inputOptionalParam=None):
configValues = configurationValues()
createPluginList()
pluginFolderPath = configValues.pathToPluginFolder
if not configValues.localPluginFolder:
if configValues.sftp_useSftp:
connection = createSFTPConnection()
pluginList = sftp_listAll(connection)
else:
connection = createFTPConnection()
pluginList = ftp_listAll(connection)
else:
pluginList = os.listdir(pluginFolderPath)
def check_update_available_installed_plugins(input_selected_object: str, config_values: config_value) -> str:
"""
Gets installed plugins and checks it against the apis if there are updates for the plugins available
i = 0
oldPlugins = 0
print(oColors.brightBlack + f"Checking: {inputSelectedObject}" + oColors.standardWhite)
if inputOptionalParam != "changelog":
print(oColors.brightBlack + f"Use 'check {inputSelectedObject} changelog' to get the latest changelog from plugins" + oColors.standardWhite)
print("┌─────┬────────────────────────────────┬──────────────┬──────────────┐")
print("│ No. │ Name │ Installed V. │ Latest V. │")
print("└─────┴────────────────────────────────┴──────────────┴──────────────┘")
:param input_selected_object: Command line input (default: all)
:param config_values: Config values from config file
:returns: Count of plugins, Count of plugins with available updates
"""
Plugin.create_plugin_list()
match config_values.connection:
case "sftp":
connection = sftp_create_connection()
plugin_list = sftp_list_all(connection)
case "ftp":
connection = ftp_create_connection()
plugin_list = ftp_list_all(connection)
case _:
plugin_folder_path = config_values.path_to_plugin_folder
plugin_list = os.listdir(plugin_folder_path)
plugin_count = plugins_with_udpates = 0
# create simple progress bar from rich
for plugin_file in track(plugin_list, description="[cyan]Checking...", transient=True, style="bright_yellow"):
plugin_attributes = True
match config_values.connection:
case "sftp":
plugin_attributes = sftp_validate_file_attributes(
connection, f"{config_values.remote_plugin_folder_on_server}/{plugin_file}"
)
case "ftp":
plugin_attributes = ftp_validate_file_attributes(
connection, f"{config_values.remote_plugin_folder_on_server}/{plugin_file}"
)
case _:
if not os.path.isfile(Path(f"{plugin_folder_path}/{plugin_file}")):
plugin_attributes = False
if not re.search(r'.jar$', plugin_file):
plugin_attributes = False
# skip plugin if no attributes were found to skip not valid plugin files
if plugin_attributes == False:
continue
plugin_file_name = get_plugin_file_name(plugin_file)
# supports command 'check pluginname' and skip the checking of every other plugin to speed things up a bit
if input_selected_object != "all" and input_selected_object != "*":
if not re.search(input_selected_object, plugin_file_name, re.IGNORECASE):
continue
plugin_file_version = get_plugin_file_version(plugin_file)
# check repository of plugin
plugin_spigot_id = search_plugin_spiget(plugin_file, plugin_file_name, plugin_file_version) # plugin_spigot_id isn't needed
# TODO add more plugin repositories here
# plugin wasn't found and not added to global plugin list so add
try:
for plugin in track(pluginList, description="Checking for updates" ,transient=True, complete_style="bright_yellow"):
if not configValues.localPluginFolder:
pluginFile = f"{configValues.sftp_folderPath}/{plugin}"
if configValues.sftp_useSftp:
pluginAttributes = sftp_validateFileAttributes(connection, pluginFile)
if pluginAttributes == False:
continue
else:
pluginAttributes = ftp_validateFileAttributes(connection, pluginFile)
if pluginAttributes == False:
continue
else:
if not os.path.isfile(Path(f"{pluginFolderPath}/{plugin}")):
continue
if not re.search(r'.jar$', plugin):
continue
try:
fileName = getFileName(plugin)
fileVersion = getFileVersion(plugin)
pluginId = getInstalledPlugin(fileName, fileVersion, plugin)
except TypeError:
continue
pluginIdStr = str(pluginId)
if fileVersion == '':
fileVersion = 'N/A'
try:
pluginLatestVersion = INSTALLEDPLUGINLIST[i][3]
if plugin_file not in INSTALLEDPLUGINLIST[-1].plugin_file_name:
Plugin.add_to_plugin_list(plugin_file, plugin_file_name, plugin_file_version, 'N/A', False, 'N/A', ())
except IndexError:
pluginLatestVersion = 'N/A'
Plugin.add_to_plugin_list(plugin_file, plugin_file_name, plugin_file_version, 'N/A', False, 'N/A', ())
if INSTALLEDPLUGINLIST[-1].plugin_is_outdated == True:
plugins_with_udpates += 1
plugin_count += 1
return plugin_count, plugins_with_udpates
if pluginLatestVersion == None:
pluginLatestVersion = 'N/A'
try:
pluginIsOutdated = INSTALLEDPLUGINLIST[i][4]
except IndexError:
pluginIsOutdated = 'N/A'
def check_installed_plugins(input_selected_object : str="all", input_parameter : str=None) -> None:
"""
Prints table overview of installed plugins with versions and available updates
if pluginIsOutdated == None:
pluginIsOutdated = 'N/A'
:param input_selected_object: Which plugin should be checked
:param input_parameter: Optional parameters
if pluginIsOutdated == True:
oldPlugins = oldPlugins + 1
:returns: None
"""
config_values = config_value()
plugin_count, plugins_with_udpates = check_update_available_installed_plugins(input_selected_object, config_values)
if re.search(r'.jar$', fileName):
fileName = eggCrackingJar(plugin, "name")
if inputSelectedObject != "all" and inputSelectedObject != "*":
if inputSelectedObject != pluginIdStr or not re.search(inputSelectedObject, fileName, re.IGNORECASE):
# print rich table of found plugins and result
rich_table = Table(box=None)
rich_table.add_column("No.", justify="right", style="cyan", no_wrap=True)
rich_table.add_column("Name", style="bright_magenta")
rich_table.add_column("Installed V.", justify="right", style="green")
rich_table.add_column("Latest V.", justify="right", style="bright_green")
rich_table.add_column("Update available", justify="left", style="white")
rich_table.add_column("Repository", justify="left", style="white")
# start counting at 1 for all my non-programming friends :)
i = 1
for plugin in INSTALLEDPLUGINLIST:
rich_table.add_row(
str(i),
plugin.plugin_name,
plugin.plugin_file_version,
plugin.plugin_latest_version,
str(plugin.plugin_is_outdated),
plugin.plugin_repository
)
i += 1
continue
if inputSelectedObject == "all" or inputSelectedObject != "*" or inputSelectedObject != "all":
if inputSelectedObject == pluginIdStr or re.search(inputSelectedObject, fileName, re.IGNORECASE):
if pluginLatestVersion == 'N/A':
print(oColors.brightBlack + f" [{1}]".rjust(6), end='')
rich_console = Console()
rich_console.print(rich_table)
rich_console.print()
if plugins_with_udpates != 0:
rich_console.print(
"[not bold][bright_yellow]Plugins with available updates: [bright_green]" +
f"{plugins_with_udpates}[bright_yellow]/[green]{plugin_count}"
)
else:
print(f" [{1}]".rjust(6), end='')
else:
if pluginLatestVersion == 'N/A':
print(oColors.brightBlack + f" [{i+1}]".rjust(6), end='')
else:
print(f" [{i+1}]".rjust(6), end='')
print(" ", end='')
if pluginIsOutdated == True:
print(oColors.brightRed + f"{fileName}".ljust(33) + oColors.standardWhite, end='')
elif pluginIsOutdated == False:
print(oColors.brightGreen + f"{fileName}".ljust(33) + oColors.standardWhite, end='')
else:
print(f"{fileName}".ljust(33), end='')
print(f"{fileVersion}".ljust(15), end='')
print(f"{pluginLatestVersion}".ljust(15))
if (inputOptionalParam == "changelog" and pluginLatestVersion != 'N/A'):
print(oColors.brightYellow + f"CHANGELOG {fileName}:" + oColors.standardWhite)
description = getUpdateDescription(pluginId)
print(description)
print()
if inputSelectedObject == pluginIdStr or re.search(inputSelectedObject, fileName, re.IGNORECASE):
break
else:
print(oColors.brightRed + "Wrong input! Use 'check all' to check every plugin for updates!" + oColors.standardWhite)
break
i += 1
except TypeError:
print(oColors.brightRed + "Error occured: Aborted checking for updates." + oColors.standardWhite)
print(oColors.brightYellow + f"Outdated plugins: [{oldPlugins}/{i}]" + oColors.standardWhite)
rich_console.print(f"[bright_green]All found plugins are on the newest version!")
return None
def updateInstalledPackage(inputSelectedObject='all'):
configValues = configurationValues()
if not configValues.localPluginFolder:
if configValues.sftp_useSftp:
connection = createSFTPConnection()
else:
connection = createFTPConnection()
def update_installed_plugins(input_selected_object : str="all", no_confirmation : bool=False) -> None:
"""
Checks if a plugin list exists and if so updates the selected plugins if there is an update available
:param input_selected_object: Plugin name to update (use 'all' or '*' for everything)
:param no_confirmation: Don't ask for confirmation if pluGET was called with param: --no-confirmation
:returns: None
"""
rich_console = Console()
config_values = config_value()
match config_values.connection:
case "sftp":
connection = sftp_create_connection()
case "ftp":
connection = ftp_create_connection()
# if INSTALLEDPLUGINLIST was not previously filled by 'check' command call the command to fill plugin list
try:
print(oColors.brightBlack + "Selected plugins with available Updates:" + oColors.standardWhite)
if inputSelectedObject == "all" or inputSelectedObject == "*":
for pluginIndex in range(len(INSTALLEDPLUGINLIST)):
if INSTALLEDPLUGINLIST[pluginIndex][4] == True:
fileName = getFileName(INSTALLEDPLUGINLIST[pluginIndex][0])
print(fileName, end=' ')
else:
print(inputSelectedObject, end=' ')
print()
updateConfirmation = input("Update these plugins [y/n] ? ")
if str.lower(updateConfirmation) != "y":
print(oColors.brightRed + "Aborting the update process."+ oColors.standardWhite)
return False
if len(INSTALLEDPLUGINLIST) == 0:
check_update_available_installed_plugins(input_selected_object, config_values)
except NameError:
print(oColors.brightRed + "Check for updates before updating plugins with: 'check all'" + oColors.standardWhite)
print(oColors.brightRed + "Started checking for updates..." + oColors.standardWhite)
checkInstalledPackage()
print(oColors.brightRed + f"Please input 'update {inputSelectedObject}' again!" + oColors.standardWhite)
return False
check_update_available_installed_plugins(input_selected_object, config_values)
i = 0
pluginsUpdated = 0
indexNumberUpdated = 0
print(oColors.brightBlack + f"Updating: {inputSelectedObject}" + oColors.standardWhite)
print("┌─────┬────────────────────────────────┬────────────┬──────────┐")
print("│ No. │ Name │ Old V. │ New V. │")
print("└─────┴────────────────────────────────┴────────────┴──────────┘")
# if argument 'all' was given recheck all plugins to avoid having only a few plugins from previously cached checks
if input_selected_object == "all" or input_selected_object == "*":
check_update_available_installed_plugins(input_selected_object, config_values)
# skip confirmation message if pluGET was called with --no-confirmation
if no_confirmation == False:
if ask_update_confirmation(input_selected_object) == False:
return None
# used later for output as stats
plugins_updated = plugins_skipped = 0
#for plugin in track(INSTALLEDPLUGINLIST, description="[cyan]Updating...", transient=True, style="bright_yellow"):
for plugin in INSTALLEDPLUGINLIST:
# supports command 'update pluginname' and skip the updating of every other plugin to speed things up a bit
if input_selected_object != "all" and input_selected_object != "*":
if not re.search(input_selected_object, plugin.plugin_file_name, re.IGNORECASE):
plugins_skipped += 1
continue
if plugin.plugin_is_outdated == False:
plugins_skipped += 1
continue
rich_console.print(
"\n [not bold][bright_white]● [bright_magenta]" +
f"{plugin.plugin_name} [green]{plugin.plugin_file_version}" + \
f" [cyan]→ [bright_green]{plugin.plugin_latest_version}"
)
plugins_updated += 1
plugin_path = get_download_path(config_values)
match config_values.connection:
# local plugin folder
case "local":
match (plugin.plugin_repository):
case "spigot":
try:
for pluginArray in track(INSTALLEDPLUGINLIST, description="Updating" ,transient=True, complete_style="bright_magenta", ):
plugin = INSTALLEDPLUGINLIST[i][0]
if not configValues.localPluginFolder:
pluginFile = f"{configValues.sftp_folderPath}/{plugin}"
if configValues.sftp_useSftp:
pluginAttributes = sftp_validateFileAttributes(connection, pluginFile)
if pluginAttributes == False:
i += 1
continue
else:
pluginAttributes = ftp_validateFileAttributes(connection, pluginFile)
if pluginAttributes == False:
i += 1
continue
else:
pluginFolderPath = configValues.pathToPluginFolder
if not os.path.isfile(Path(f"{pluginFolderPath}/{plugin}")):
i += 1
continue
if not re.search(r'.jar$', plugin):
i += 1
continue
try:
fileName = getFileName(plugin)
fileVersion = getFileVersion(plugin)
pluginId = INSTALLEDPLUGINLIST[i][1]
latestVersion = INSTALLEDPLUGINLIST[i][3]
except (TypeError, ValueError):
i += 1
continue
if re.search(r'.jar$', fileName):
fileName = eggCrackingJar(plugin, "name")
pluginIdStr = str(pluginId)
if pluginId == None or pluginId == '':
i += 1
continue
if inputSelectedObject != 'all' and inputSelectedObject != pluginIdStr and not re.search(inputSelectedObject, fileName, re.IGNORECASE):
i += 1
continue
if INSTALLEDPLUGINLIST[i][4] != True:
i += 1
continue
if INSTALLEDPLUGINLIST[i][4] == False and inputSelectedObject != 'all':
print(oColors.brightGreen + f"{fileName} is already on {latestVersion}" + oColors.standardWhite)
print(oColors.brightRed + "Aborting the update process."+ oColors.standardWhite)
break
print(f" [{indexNumberUpdated+1}]".rjust(6), end='')
print(" ", end='')
print(f"{fileName}".ljust(33), end='')
print(f"{fileVersion}".ljust(13), end='')
print(f"{latestVersion}".ljust(13))
if not configValues.localPluginFolder:
if configValues.sftp_seperateDownloadPath is True:
pluginPath = configValues.sftp_pathToSeperateDownloadPath
else:
pluginPath = configValues.sftp_folderPath
pluginPath = f"{pluginPath}/{plugin}"
indexNumberUpdated += 1
pluginsUpdated += 1
if configValues.sftp_useSftp:
sftp = createSFTPConnection()
try:
getSpecificPackage(pluginId, pluginPath)
if configValues.sftp_seperateDownloadPath is False:
sftp.remove(pluginPath)
get_specific_plugin_spiget(plugin.plugin_repository_data[0])
except HTTPError as err:
print(oColors.brightRed + f"HTTPError: {err.code} - {err.reason}" + oColors.standardWhite)
pluginsUpdated -= 1
rich_print_error(f"HTTPError: {err.code} - {err.reason}")
plugins_updated -= 1
except TypeError:
print(oColors.brightRed + f"TypeError: Couldn't download new version. Is the file available on spigotmc?" + oColors.standardWhite)
pluginsUpdated -= 1
except FileNotFoundError:
print(oColors.brightRed + f"FileNotFoundError: Old plugin file coulnd't be deleted" + oColors.standardWhite)
rich_print_error(
f"Error: TypeError > Couldn't download new version. Is the file available on spigotmc?"
)
plugins_updated -= 1
else:
ftp = createFTPConnection()
case _:
rich_print_error("Error: Plugin repository wasn't found")
return None
# don't delete files if they are downloaded to a seperate download path
if config_values.local_seperate_download_path == False:
try:
getSpecificPackage(pluginId, pluginPath)
if configValues.sftp_seperateDownloadPath is False:
ftp.delete(pluginPath)
except HTTPError as err:
print(oColors.brightRed + f"HTTPError: {err.code} - {err.reason}" + oColors.standardWhite)
pluginsUpdated -= 1
except TypeError:
print(oColors.brightRed + f"TypeError: Couldn't download new version. Is the file available on spigotmc?" + oColors.standardWhite)
pluginsUpdated -= 1
os.remove(Path(f"{plugin_path}/{plugin.plugin_file_name}"))
rich_console.print(
" [not bold][bright_green]Deleted old plugin file [cyan]→ [white]" +
f"{plugin.plugin_file_name}"
)
except FileNotFoundError:
print(oColors.brightRed + f"FileNotFoundError: Old plugin file coulnd't be deleted" + oColors.standardWhite)
rich_print_error("Error: Old plugin file couldn't be deleted")
else:
if configValues.seperateDownloadPath is True:
pluginPath = configValues.pathToSeperateDownloadPath
else:
pluginPath = configValues.pathToPluginFolder
indexNumberUpdated += 1
pluginsUpdated += 1
# plugin folder is on sftp or ftp server
case _:
plugin_path = f"{plugin_path}/{plugin.plugin_file_name}"
match (plugin.plugin_repository):
case "spigot":
try:
getSpecificPackage(pluginId, pluginPath)
if configValues.seperateDownloadPath is False:
pluginPath = f"{pluginPath}/{plugin}"
os.remove(pluginPath)
get_specific_plugin_spiget(plugin.plugin_repository_data[0])
except HTTPError as err:
print(oColors.brightRed + f"HTTPError: {err.code} - {err.reason}" + oColors.standardWhite)
pluginsUpdated -= 1
rich_print_error(f"HTTPError: {err.code} - {err.reason}")
plugins_updated -= 1
except TypeError:
print(oColors.brightRed + f"TypeError: Couldn't download new version. Is the file available on spigotmc?" + oColors.standardWhite)
pluginsUpdated -= 1
rich_print_error(
f"Error: TypeError > Couldn't download new version. Is the file available on spigotmc?"
)
plugins_updated -= 1
case _:
rich_print_error("Error: Plugin repository wasn't found")
return None
# don't delete old plugin files if they are downloaded to a seperate download path
if config_values.remote_seperate_download_path == False:
match config_values.connection:
case "sftp":
try:
connection.remove(plugin_path)
rich_console.print(
" [not bold][bright_green]Deleted old plugin file [cyan]→ [white]" +
f"{plugin.plugin_file_name}"
)
except FileNotFoundError:
print(oColors.brightRed + f"FileNotFoundError: Old plugin file coulnd't be deleted" + oColors.standardWhite)
if inputSelectedObject != 'all':
break
elif inputSelectedObject != 'all':
print(oColors.brightGreen + f"{fileName} is already on {latestVersion}" + oColors.standardWhite)
print(oColors.brightRed + "Aborting the update process."+ oColors.standardWhite)
break
rich_print_error("Error: Old plugin file couldn't be deleted")
case "ftp":
try:
connection.delete(plugin_path)
rich_console.print(
" [not bold][bright_green]Deleted old plugin file [cyan]→ [white]" +
f"{plugin.plugin_file_name}"
)
except FileNotFoundError:
rich_print_error("Error: Old plugin file couldn't be deleted")
i += 1
except TypeError:
print(oColors.brightRed + "Error occured: Aborted updating plugins." + oColors.standardWhite)
except NameError:
print(oColors.brightRed + "Check for updates before updating plugins with: 'check all'" + oColors.standardWhite)
print(oColors.brightRed + "Started checking for updates..." + oColors.standardWhite)
checkInstalledPackage()
print(oColors.brightRed + f"Please input 'update {inputSelectedObject}' again!" + oColors.standardWhite)
if i != 0:
print(oColors.brightYellow + f"Plugins updated: [{pluginsUpdated}/{i}]" + oColors.standardWhite)
if inputSelectedObject =='all' and pluginsUpdated == 0 and i != 0:
print(oColors.brightGreen + "All found plugins are on the latest version!" + oColors.standardWhite)
rich_console.print(
f"\n[not bold][bright_green]Plugins updated: {plugins_updated}/{(len(INSTALLEDPLUGINLIST) - plugins_skipped)}"
)
return None
def getInstalledPlugin(localFileName, localFileVersion, localPluginFullName):
url = "https://api.spiget.org/v2/search/resources/" + localFileName + "?field=name&sort=-downloads"
packageName = doAPIRequest(url)
plugin_match_found = False
pluginID = None
localFileVersionNew = localFileVersion
i = 0
for i in range(0, 3):
if plugin_match_found == True:
break
def search_plugin_spiget(plugin_file: str, plugin_file_name: str, plugin_file_version: str) -> int:
"""
Search the spiget api for the installed plugin and add it to the installed plugin list
:param plugin_file: Full file name of plugin
:param plugin_file_name: Name of plugin file
:param plugin_file_version: Version of plugin file
:returns: Plugin ID of Spigot Plugin
"""
url = f"https://api.spiget.org/v2/search/resources/{plugin_file_name}?field=name&sort=-downloads"
plugin_list = api_do_request(url)
plugin_file_version2 = plugin_file_version
for i in range(4):
if i == 1:
localFileVersionNew = re.sub(r'(\-\w*)', '', localFileVersion)
plugin_file_version2 = re.sub(r'(\-\w*)', '', plugin_file_version)
if i == 2:
pluginNameinYML = eggCrackingJar(localPluginFullName, 'name')
url = "https://api.spiget.org/v2/search/resources/" + pluginNameinYML + "?field=name&sort=-downloads"
plugin_name_in_yml, plugin_version_in_yml = egg_cracking_jar(plugin_file)
url = f"https://api.spiget.org/v2/search/resources/{plugin_name_in_yml}?field=name&sort=-downloads"
try:
packageName = doAPIRequest(url)
plugin_list = api_do_request(url)
except ValueError:
continue
localFileVersion = localFileVersionNew
for resource in packageName:
if plugin_match_found == True:
# if no plugin name was found with egg_cracking_jar() skip this round
if plugin_list is None:
continue
pID = resource["id"]
url2 = f"https://api.spiget.org/v2/resources/{pID}/versions?size=100&sort=-name"
# search with version which is in plugin.yml for the plugin
if i == 3:
plugin_file_version2 = plugin_version_in_yml
for plugin in plugin_list:
plugin_id = plugin["id"]
url2 = f"https://api.spiget.org/v2/resources/{plugin_id}/versions?size=100&sort=-name"
try:
packageVersions = doAPIRequest(url2)
plugin_versions = api_do_request(url2)
except ValueError:
continue
for updates in packageVersions:
updateVersion = updates["name"]
if localFileVersionNew in updateVersion:
plugin_match_found = True
pluginID = pID
updateId = updates["id"]
plugin_latest_version = getLatestPluginVersion(pID)
plugin_is_outdated = compareVersions(plugin_latest_version, updateVersion)
addToPluginList(localPluginFullName, pID, updateId, plugin_latest_version , plugin_is_outdated)
return pluginID
else:
if plugin_match_found != True:
pID = updateId = plugin_latest_version = plugin_is_outdated = None
addToPluginList(localPluginFullName, pID, updateId, plugin_latest_version , plugin_is_outdated)
return pluginID
if plugin_versions is None:
continue
for updates in plugin_versions:
update_version_name = updates["name"]
if plugin_file_version2 in update_version_name:
#spigot_update_id = updates["id"]
plugin_latest_version = get_latest_plugin_version_spiget(plugin_id)
plugin_is_outdated = compare_plugin_version(plugin_latest_version, update_version_name)
Plugin.add_to_plugin_list(
plugin_file,
plugin_file_name,
plugin_file_version,
plugin_latest_version,
plugin_is_outdated,
"spigot",
[plugin_id]
)
return plugin_id
return None

114
src/serverjar/serverjar_checker.py
View File

@ -1,114 +0,0 @@
import os
import sys
from urllib.error import HTTPError
from pathlib import Path
from handlers.handle_sftp import createSFTPConnection, sftp_listFilesInServerRoot
from handlers.handle_ftp import createFTPConnection, ftp_listFilesInServerRoot
from handlers.handle_config import configurationValues
from utils.consoleoutput import oColors
from serverjar.serverjar_paper import paperCheckForUpdate, papermc_downloader
def checkInstalledServerjar():
configValues = configurationValues()
if not configValues.localPluginFolder:
if not configValues.sftp_useSftp:
ftp = createFTPConnection()
serverRootList = ftp_listFilesInServerRoot(ftp)
else:
sftp = createSFTPConnection()
serverRootList = sftp_listFilesInServerRoot(sftp)
else:
serverRootList = os.path.dirname(configValues.pathToPluginFolder)
serverRootList = os.listdir(serverRootList)
installedServerjarFullName = None
try:
for files in serverRootList:
try:
if '.jar' in files:
installedServerjarFullName = files
break
except TypeError:
continue
except TypeError:
print(oColors.brightRed + "Serverjar couldn't be found." + oColors.standardWhite)
print(oColors.brightRed + "Aborting the process." + oColors.standardWhite)
if installedServerjarFullName == None:
print(oColors.brightRed + "Serverjar couldn't be found." + oColors.standardWhite)
print(oColors.brightRed + "Aborting the process." + oColors.standardWhite)
input("Press any key + enter to exit...")
sys.exit()
print(oColors.brightBlack + f"Checking: {installedServerjarFullName}" + oColors.standardWhite)
if 'paper' in installedServerjarFullName:
paperCheckForUpdate(installedServerjarFullName)
else:
print(oColors.brightRed + f"{installedServerjarFullName} isn't supported.")
print(oColors.brightRed + "Aborting the process." + oColors.standardWhite)
def updateServerjar(serverJarBuild='latest'):
configValues = configurationValues()
try:
if serverJarBuild == None:
serverJarBuild = 'latest'
if not configValues.localPluginFolder:
sftp = createSFTPConnection()
serverRootPath = configValues.sftp_folderPath
serverRootPath = Path(str(serverRootPath).replace(r'/plugins', ''))
serverRootList = sftp_listFilesInServerRoot(sftp)
else:
serverRoot = os.path.dirname(configValues.pathToPluginFolder)
serverRootList = os.listdir(serverRoot)
serverRootPath = configValues.pathToPluginFolder
helpPath = Path('/plugins')
helpPathstr = str(helpPath)
serverRootPath = Path(str(serverRootPath).replace(helpPathstr, ''))
installedServerjarFullName = None
except FileNotFoundError:
print(oColors.brightRed + "Path couldn't be found!" + oColors.standardWhite)
print(oColors.brightRed + "Check your config!" + oColors.standardWhite)
print(oColors.brightRed + "Aborting the process." + oColors.standardWhite)
input("Press any key + enter to exit...")
sys.exit()
try:
for files in serverRootList:
try:
if '.jar' in files:
installedServerjarFullName = files
break
except TypeError:
continue
except TypeError:
print(oColors.brightRed + "Serverjar couldn't be found." + oColors.standardWhite)
print(oColors.brightRed + "Aborting the process." + oColors.standardWhite)
if installedServerjarFullName == None:
print(oColors.brightRed + "Serverjar couldn't be found." + oColors.standardWhite)
print(oColors.brightRed + "Aborting the process." + oColors.standardWhite)
input("Press any key + enter to exit...")
sys.exit()
serverJarPath = Path(f"{serverRootPath}/{installedServerjarFullName}")
if 'paper' in installedServerjarFullName:
print(oColors.brightBlack + f"Updating Paper to build: {serverJarBuild}" + oColors.standardWhite)
try:
papermc_downloader(serverJarBuild, None, installedServerjarFullName)
if not configValues.localPluginFolder:
sftp.remove(serverJarPath)
else:
os.remove(serverJarPath)
except HTTPError as err:
print(oColors.brightRed + f"Error: {err.code} - {err.reason}" + oColors.standardWhite)
except FileNotFoundError:
print(oColors.brightRed + "Error: Old serverjar file coulnd't be deleted" + oColors.standardWhite)
else:
print(oColors.brightRed + f"{installedServerjarFullName} isn't supported.")
print(oColors.brightRed + "Aborting the process." + oColors.standardWhite)

218
src/serverjar/serverjar_paper.py
View File

@ -1,218 +0,0 @@
import re
import urllib.request
from pathlib import Path
from rich.console import Console
from utils.consoleoutput import oColors
from utils.web_request import doAPIRequest
from handlers.handle_sftp import createSFTPConnection, sftp_upload_server_jar
from handlers.handle_ftp import createFTPConnection, ftp_upload_server_jar
from handlers.handle_config import configurationValues
from utils.utilities import createTempPluginFolder, deleteTempPluginFolder, calculateFileSizeMb
def getInstalledPaperMinecraftVersion(localPaperName):
if localPaperName is None:
return False
mcVersionFull = re.search(r'(\d*\.*\d)+', localPaperName)
try:
mcVersion = mcVersionFull.group()
except AttributeError:
mcVersion = mcVersionFull
return mcVersion
def getInstalledPaperVersion(localPaperName):
if localPaperName is None:
return False
paperBuildFull = re.search(r'([\d]*.jar)', localPaperName)
try:
paperBuild = paperBuildFull.group()
except AttributeError:
paperBuild = paperBuildFull
paperBuild = paperBuild.replace('.jar', '')
return paperBuild
def findVersionGroup(mcVersion):
versionGroups = ['1.17', '1.16', '1.15']
if mcVersion is None:
return False
for versionGroup in versionGroups:
url = f"https://papermc.io/api/v2/projects/paper/version_group/{versionGroup}/builds"
papermcdetails = doAPIRequest(url)
papermcVersionForMc = papermcdetails["versions"]
for versions in papermcVersionForMc:
if versions == mcVersion:
paperVersionGroup = versionGroup
return paperVersionGroup
if versionGroup == mcVersion:
paperVersionGroup = versionGroup
return paperVersionGroup
return False # Not found
def findBuildVersion(wantedPaperBuild):
versionGroups = ['1.17', '1.16', '1.15']
if wantedPaperBuild is None:
return False
for versionGroup in versionGroups:
url = f"https://papermc.io/api/v2/projects/paper/version_group/{versionGroup}/builds"
papermcdetails = doAPIRequest(url)
paperMcBuilds = papermcdetails["builds"]
for build in paperMcBuilds:
paperBuild = str(build["build"])
if paperBuild == wantedPaperBuild:
paperVersionGroup = build["version"]
return paperVersionGroup
return False # Not found
def findLatestBuild(paperVersionGroup):
if paperVersionGroup is None:
return False
url = f"https://papermc.io/api/v2/projects/paper/version_group/{paperVersionGroup}/builds"
papermcbuilds = doAPIRequest(url)
if "status" in papermcbuilds: # Checks if the API returns a status. This means that there was an error.
return False
latestPaperBuild = papermcbuilds["builds"][-1]["build"]
return latestPaperBuild
def findLatestBuildForVersion(mcVersion):
if mcVersion is None:
return False
url = f"https://papermc.io/api/v2/projects/paper/versions/{mcVersion}"
papermcbuilds = doAPIRequest(url)
latestPaperBuild = papermcbuilds["builds"][-1]
return latestPaperBuild
def versionBehind(installedPaperBuild, latestPaperBuild):
if installedPaperBuild is None or latestPaperBuild is None:
return False
installedPaperBuildint = int(installedPaperBuild)
latestPaperBuildint = int(latestPaperBuild)
versionsBehind = latestPaperBuildint - installedPaperBuildint
return versionsBehind
def getDownloadFileName(paperMcVersion, paperBuild):
if paperMcVersion is None or paperBuild is None:
return False
url = f"https://papermc.io/api/v2/projects/paper/versions/{paperMcVersion}/builds/{paperBuild}"
buildDetails = doAPIRequest(url)
downloadName = buildDetails["downloads"]["application"]["name"]
return downloadName
def paperCheckForUpdate(installedServerjarFullName):
mcVersion = getInstalledPaperMinecraftVersion(installedServerjarFullName)
# Report an error if getInstalledPaperMinecraftVersion encountered an issue.
if not mcVersion:
print(oColors.brightRed + "ERR: An error was encountered while detecting the server's Minecraft version." +
oColors.standardWhite)
return False
paperInstalledBuild = getInstalledPaperVersion(installedServerjarFullName)
# Report an error if getInstalledPaperVersion encountered an issue.
if not paperInstalledBuild:
print(oColors.brightRed + "ERR: An error was encountered while detecting the server's Paper version." +
oColors.standardWhite)
return False
versionGroup = findVersionGroup(mcVersion)
# Report an error if findVersionGroup encountered an issue.
if not versionGroup:
print(oColors.brightRed + "ERR: An error was encountered while fetching the server's version group." +
oColors.standardWhite)
return False
paperLatestBuild = findLatestBuild(versionGroup)
# Report an error if findLatestBuild encountered an issue.
if not paperLatestBuild:
print(oColors.brightRed + "ERR: An error was encountered while fetching the latest version of PaperMC." +
oColors.standardWhite)
return False # Not currently handled, but can be at a later date. Currently just stops the following from
# being printed.
paperVersionBehind = versionBehind(paperInstalledBuild, paperLatestBuild)
# Report an error if getInstalledPaperVersion encountered an issue.
if not paperVersionBehind:
print(oColors.brightRed + "ERR: An error was encountered while detecting how many versions behind you are. "
f"Will display as 'N/A'." + oColors.standardWhite)
paperVersionBehind = "N/A" # Sets paperVersionBehind to N/A while still letting the versionBehind check return
# False for error-handing reasons.
# Does not return false as versions behind doesn't break things. It is just helpful information.
# paperVersionBehind will just display as "N/A"
print("┌─────┬────────────────────────────────┬──────────────┬──────────────┐")
print("│ No. │ Name │ Installed V. │ Latest V. │")
print("└─────┴────────────────────────────────┴──────────────┴──────────────┘")
print(" [1]".rjust(6), end='')
print(" ", end='')
if paperVersionBehind != 0:
print(oColors.brightRed + "paper".ljust(33) + oColors.standardWhite, end='')
else:
print(oColors.brightGreen + "paper".ljust(33) + oColors.standardWhite, end='')
print(f"{paperInstalledBuild}".ljust(15), end='')
print(f"{paperLatestBuild}".ljust(15))
print(oColors.brightYellow + f"Versions behind: [{paperVersionBehind}]" + oColors.standardWhite)
# https://papermc.io/api/docs/swagger-ui/index.html?configUrl=/api/openapi/swagger-config#/
def papermc_downloader(paperBuild='latest', mcVersion=None, installedServerjarName=None):
configValues = configurationValues()
if configValues.localPluginFolder == False:
downloadPath = createTempPluginFolder()
else:
downloadPath = configValues.pathToPluginFolder
helpPath = Path('/plugins')
helpPathstr = str(helpPath)
downloadPath = Path(str(downloadPath).replace(helpPathstr, ''))
if mcVersion == None:
if paperBuild == 'latest':
mcVersion = '1.17.1'
else:
mcVersion = findBuildVersion(paperBuild)
if installedServerjarName != None:
mcVersion = getInstalledPaperMinecraftVersion(installedServerjarName)
if paperBuild == 'latest':
paperBuild = findLatestBuildForVersion(mcVersion)
try:
downloadFileName = getDownloadFileName(mcVersion, paperBuild)
except KeyError:
print(oColors.brightRed + f"This version wasn't found for {mcVersion}" + oColors.standardWhite)
print(oColors.brightRed + f"Reverting to latest version for {mcVersion}" + oColors.standardWhite)
paperBuild = findLatestBuildForVersion(mcVersion)
downloadFileName = getDownloadFileName(mcVersion, paperBuild)
downloadPackagePath = Path(f"{downloadPath}/{downloadFileName}")
if configValues.localPluginFolder == False:
downloadPath = createTempPluginFolder()
url = f"https://papermc.io/api/v2/projects/paper/versions/{mcVersion}/builds/{paperBuild}/downloads/{downloadFileName}"
remotefile = urllib.request.urlopen(url)
filesize = remotefile.info()['Content-Length']
print(f"Getting Paper {paperBuild} for {mcVersion}")
console = Console()
with console.status("Downloading...", spinner='line', spinner_style='bright_magenta') as status:
urllib.request.urlretrieve(url, downloadPackagePath)
filesizeData = calculateFileSizeMb(filesize)
print("Downloaded " + (str(filesizeData)).rjust(9) + f" MB here {downloadPackagePath}")
if not configValues.localPluginFolder:
if not configValues.sftp_useSftp:
ftpSession = createFTPConnection()
ftp_upload_server_jar(ftpSession, downloadPackagePath)
else:
sftpSession = createSFTPConnection()
sftp_upload_server_jar(sftpSession, downloadPackagePath)
deleteTempPluginFolder(downloadPath)
print(oColors.brightGreen + "Downloaded successfully " + oColors.standardWhite + f"Paper {paperBuild}" + \
oColors.brightGreen + " for " + oColors.standardWhite + f"{mcVersion}" + oColors.standardWhite)

300
src/serverjar/serverjar_paper_velocity_waterfall.py Normal file
View File

@ -0,0 +1,300 @@
"""
Handles the update checking and downloading of these serverjars:
Paper, Velocity, Waterfall
All are from the PaperMC Team and use the same api structure which is the reason these are together
"""
import re
import requests
from pathlib import Path
from rich.table import Table
from rich.console import Console
from rich.progress import Progress
from src.handlers.handle_config import config_value
from src.utils.console_output import rich_print_error
from src.handlers.handle_sftp import sftp_create_connection, sftp_upload_server_jar
from src.handlers.handle_ftp import ftp_create_connection, ftp_upload_server_jar
from src.utils.utilities import \
api_do_request, create_temp_plugin_folder, remove_temp_plugin_folder, convert_file_size_down
def get_installed_mc_version(file_server_jar_full_name) -> str:
"""
Gets the used minecraft version from the file name
:param file_server_jar_full_name: Full file name of the installed serverjar
:returns: Used minecraft version
"""
mc_version_full = re.search(r"(\d*\.*\d)+", file_server_jar_full_name)
try:
mc_version = mc_version_full.group()
except AttributeError:
mc_version = mc_version_full
return mc_version
def get_installed_serverjar_version(file_server_jar_full_name) -> str:
"""
Gets the installed version of the installed serverjar
:param file_server_jar_full_name: Full file name fo the installed serverjar
:returns: Used serverjar version
"""
serverjar_version_full = re.search(r"([\d]*.jar)", file_server_jar_full_name)
try:
serverjar_version = serverjar_version_full.group()
except AttributeError:
serverjar_version = serverjar_version_full
serverjar_version = serverjar_version.replace('.jar', '')
return serverjar_version
def get_version_group(file_server_jar_full_name) -> str:
"""
Gets the version group which is used for the papermc api
:param mc_version: Version of Minecraft in use
:returns: Version group of api
"""
version_group = re.sub(r"-\d*.jar$", "", file_server_jar_full_name)
version_group = re.sub(r"^(\w*\-)", "", version_group)
return version_group
def find_latest_available_version(file_server_jar_full_name, version_group) -> int:
"""
Gets the latest available version of the installed serverjar version
:param version_group: Minecraft version group of the serverjar
:returns: Latest available version as int
"""
if "paper" in file_server_jar_full_name:
url = f"https://papermc.io/api/v2/projects/paper/versions/{version_group}/builds"
elif "waterfall" in file_server_jar_full_name:
url = f"https://papermc.io/api/v2/projects/waterfall/versions/{version_group}/builds"
elif "velocity" in file_server_jar_full_name:
url = f"https://papermc.io/api/v2/projects/velocity/versions/{version_group}/builds"
versions = api_do_request(url)
if "status" in versions: # Checks if the API returns a status. This means that there was an error.
return None
latest_version = versions["builds"][-1]["build"]
return latest_version
def get_versions_behind(serverjar_version, latest_version) -> int:
"""
Gets the number diffference between the two versions
:param serverjar_version: Installed serverjar version
:param latest_version: Latest avaialable serverjar version
:returns: Number difference between the two versions
"""
versions_behind = int(latest_version) - int(serverjar_version)
return versions_behind
def get_papermc_download_file_name(mc_version, serverjar_version, file_server_jar_full_name) -> str:
"""
Gets the download name from the papermc api
:param mc_version: Minecraft version
:param serverjar_version: Version of the serverjar
:param file_server_jar_full_name: Serverjar name
:returns: Download name of the file
"""
if "paper" in file_server_jar_full_name:
url = f"https://papermc.io/api/v2/projects/paper/versions/{mc_version}/builds/{serverjar_version}"
elif "waterfall" in file_server_jar_full_name:
url = f"https://papermc.io/api/v2/projects/waterfall/versions/{mc_version}/builds/{serverjar_version}"
elif "velocity" in file_server_jar_full_name:
url = f"https://papermc.io/api/v2/projects/velocity/versions/{mc_version}/builds/{serverjar_version}"
build_details = api_do_request(url)
download_name = build_details["downloads"]["application"]["name"]
return download_name
def serverjar_papermc_check_update(file_server_jar_full_name) -> None:
"""
Checks the installed paper serverjar if an update is available
:param file_server_jar_full_name: Full name of the paper server jar file name
:returns: None
"""
"""
mc_version = get_installed_mc_version(file_server_jar_full_name)
if mc_version == None:
rich_print_error("Error: An error occured while checking the mc version of the serverjar")
return None
print(mc_version)
"""
serverjar_version = get_installed_serverjar_version(file_server_jar_full_name)
if serverjar_version == None:
rich_print_error("Error: An error occured while checking the installed serverjar version")
return None
version_group = get_version_group(file_server_jar_full_name)
if version_group == None:
rich_print_error(
"Error: An error occured while checking the installed version group of the installed serverjar"
)
return None
latest_version = find_latest_available_version(file_server_jar_full_name, version_group)
if latest_version == None:
rich_print_error("Error: An error occured while checking for the latest available version of the serverjar")
return None
versions_behind = get_versions_behind(serverjar_version, latest_version)
rich_table = Table(box=None)
rich_table.add_column("Name", style="bright_magenta")
rich_table.add_column("Installed V.", justify="right", style="green")
rich_table.add_column("Latest V.", justify="right", style="bright_green")
rich_table.add_column("Versions behind", justify="right", style="cyan")
rich_table.add_row(
file_server_jar_full_name,
serverjar_version,
str(latest_version),
str(versions_behind)
)
rich_console = Console()
rich_console.print(rich_table)
return None
def serverjar_papermc_update(
server_jar_version: str="latest",
mc_version: str=None,
file_server_jar_full_name: str=None,
serverjar_to_download: str=None
) -> bool:
"""
Handles the downloading of the papermc serverjar
:param server_jar_version: Version of the serverjar which should get downloaded
:param mc_version: Minecraft version
:param no_confirmation: If no confirmation message should pop up
:param file_server_jar_full_name: The old serverjar file
:param serverjar_to_download: The serverjar to download because it supports: paper, velocity, waterfall
This is used in the handle_input function
:returns: True/False if the serverjar was downloaded successfully
"""
config_values = config_value()
match config_values.connection:
case "local":
path_server_root = config_values.path_to_plugin_folder
# need help_path or else TypeError will be thrown
help_path = Path('/plugins')
help_path_str = str(help_path)
path_server_root = Path(str(path_server_root).replace(help_path_str, ''))
case _:
path_server_root = create_temp_plugin_folder()
# exit if the mc version can't be found
if file_server_jar_full_name == None and mc_version == None:
rich_print_error("Error: Please specifiy the minecraft version as third argument!")
return False
# if both the file name and the serverjar_to_download are emtpy then exit
if file_server_jar_full_name == None and serverjar_to_download == None:
rich_print_error("Error: Couldn't get serverjar name to download")
return False
if mc_version == None:
mc_version = get_version_group(file_server_jar_full_name)
if file_server_jar_full_name == None:
papermc_serverjar = serverjar_to_download
else:
papermc_serverjar = file_server_jar_full_name
if server_jar_version == "latest" or server_jar_version == None:
server_jar_version = find_latest_available_version(papermc_serverjar, mc_version)
rich_console = Console()
rich_console.print(
f"\n [not bold][bright_white]● [bright_magenta]{papermc_serverjar.capitalize()}" + \
f" [cyan]→ [bright_green]{server_jar_version}"
)
if file_server_jar_full_name != None:
serverjar_version = get_installed_serverjar_version(file_server_jar_full_name)
if get_versions_behind(serverjar_version, server_jar_version) == 0:
rich_console.print(" [not bold][bright_green]No updates currently available!")
return False
try:
download_file_name = get_papermc_download_file_name(mc_version, server_jar_version, papermc_serverjar)
except KeyError:
rich_print_error(f" Error: This version wasn't found for {mc_version}")
rich_print_error(f" Reverting to latest version for {mc_version}")
try:
server_jar_version = find_latest_available_version(papermc_serverjar, mc_version)
download_file_name = get_papermc_download_file_name(mc_version, server_jar_version, papermc_serverjar)
except KeyError:
rich_print_error(
f" Error: Version {mc_version} wasn't found for {papermc_serverjar.capitalize()} in the papermc api"
)
return False
if "paper" in papermc_serverjar:
url = f"https://papermc.io/api/v2/projects/paper/versions/{mc_version}" + \
f"/builds/{server_jar_version}/downloads/{download_file_name}"
elif "waterfall" in papermc_serverjar:
url = f"https://papermc.io/api/v2/projects/waterfall/versions/{mc_version}" + \
f"/builds/{server_jar_version}/downloads/{download_file_name}"
elif "velocity" in papermc_serverjar:
url = f"https://papermc.io/api/v2/projects/velocity/versions/{mc_version}" + \
f"/builds/{server_jar_version}/downloads/{download_file_name}"
download_path = Path(f"{path_server_root}/{download_file_name}")
with Progress(transient=True) as progress:
header = {'user-agent': 'pluGET/1.0'}
r = requests.get(url, headers=header, stream=True)
try:
file_size = int(r.headers.get('Content-Length'))
# create progress bar
download_task = progress.add_task(" [cyan]Downloading...", total=file_size)
except TypeError:
# Content-lenght returned nothing
file_size = 0
with open(download_path, 'wb') as f:
# split downloaded data in chunks of 65536
for data in r.iter_content(chunk_size=65536):
f.write(data)
# don't show progress bar if no content-length was returned
if file_size == 0:
continue
progress.update(download_task, advance=len(data))
#f.flush()
# use rich console for nice colors
file_size_data = convert_file_size_down(convert_file_size_down(file_size))
rich_console.print(" [not bold][bright_green]Downloaded[bright_magenta] " + (str(file_size_data)).rjust(9) + \
f" MB [cyan]→ [white]{download_path}")
if config_values.connection == "sftp":
sftp_session = sftp_create_connection()
sftp_upload_server_jar(sftp_session, download_path)
remove_temp_plugin_folder()
elif config_values.connection == "ftp":
ftp_session = ftp_create_connection()
ftp_upload_server_jar(ftp_session, download_path)
remove_temp_plugin_folder()
return True

125
src/serverjar/serverjar_updatechecker.py Normal file
View File

@ -0,0 +1,125 @@
"""
Checks the installed serverjar for updates
"""
import os
from pathlib import Path
from rich.console import Console
from src.handlers.handle_config import config_value
from src.utils.console_output import rich_print_error
from src.handlers.handle_sftp import sftp_create_connection, sftp_list_files_in_server_root
from src.handlers.handle_ftp import ftp_create_connection, ftp_list_files_in_server_root
from src.serverjar.serverjar_paper_velocity_waterfall import serverjar_papermc_check_update, serverjar_papermc_update
def get_installed_server_jar_file(config_values) -> str:
"""
Gets the file name of the installed server jar
:param config_values: Configuration values from pluGET config
:returns: Full file name of installed server jar
"""
match config_values.connection:
case "sftp":
connection = sftp_create_connection()
file_list_server_root = sftp_list_files_in_server_root(connection)
case "ftp":
connection = ftp_create_connection()
file_list_server_root = ftp_list_files_in_server_root(connection)
case _:
file_list_server_root = os.path.dirname(config_values.path_to_plugin_folder)
file_list_server_root = os.listdir(file_list_server_root)
file_server_jar_full_name = None
try:
for file in file_list_server_root:
try:
if ".jar" in file:
file_server_jar_full_name = file
break
except TypeError:
continue
except TypeError:
rich_print_error("Error: Serverjar couldn't be found")
return None
return file_server_jar_full_name
def check_update_available_installed_server_jar() -> None:
"""
Handles the checking of available updates of the installed server jar
:returns: None
"""
config_values = config_value()
file_server_jar_full_name = get_installed_server_jar_file(config_values)
if file_server_jar_full_name == None:
# print error and exit function
rich_print_error("Error: Serverjar couldn't be found")
return None
# TODO: Add other serverjars here
if "paper" in file_server_jar_full_name or \
"waterfall" in file_server_jar_full_name or \
"velocity" in file_server_jar_full_name:
serverjar_papermc_check_update(file_server_jar_full_name)
else:
rich_print_error(f"{file_server_jar_full_name} isn't supported")
return None
def update_installed_server_jar(server_jar_version: str="latest") -> None:
"""
Handles the updating of the installed server jar
:returns: None
"""
config_values = config_value()
file_server_jar_full_name = get_installed_server_jar_file(config_values)
if file_server_jar_full_name == None:
# print error and exit function
rich_print_error("Error: Serverjar couldn't be found")
return None
# finding path which is used for deleting old server jar
match config_values.connection:
case "local":
path_server_root = config_values.path_to_plugin_folder
# need help_path or else TypeError will be thrown
help_path = Path('/plugins')
help_path_str = str(help_path)
path_server_root = Path(str(path_server_root).replace(help_path_str, ''))
case _:
path_server_root = config_values.remote_plugin_folder_on_server
path_server_root = str(path_server_root).replace(r'/plugins', '')
server_jar_path = f"{path_server_root}/{file_server_jar_full_name}"
rich_console = Console()
# TODO: Add other serverjars here
if "paper" in file_server_jar_full_name or \
"waterfall" in file_server_jar_full_name or \
"velocity" in file_server_jar_full_name:
download_successfull = serverjar_papermc_update(server_jar_version, None, file_server_jar_full_name, None)
if download_successfull is True:
match config_values.connection:
case "local":
os.remove(Path(server_jar_path))
case "sftp":
connection = sftp_create_connection()
connection.remove(server_jar_path)
case "ftp":
connection = ftp_create_connection()
connection.delete(server_jar_path)
rich_console.print(
" [not bold][bright_green]Deleted old server file [cyan]→ [white]" +
f"{file_server_jar_full_name}"
)
else:
rich_print_error(f"{file_server_jar_full_name} isn't supported")
return None

8
src/settings.py Normal file
View File

@ -0,0 +1,8 @@
"""
Constant values
PLUGETVERSION = current version of pluGET
"""
# constant values
PLUGETVERSION = "1.7.0"

0
src/tests/__init__.py Normal file
View File

31
src/tests/test_plugin_updatechecker.py Normal file
View File

@ -0,0 +1,31 @@
import unittest
from src.plugin import plugin_updatechecker
class TestCases(unittest.TestCase):
def test_get_plugin_file_name(self):
plugin_file_name = "LuckPerms-5.4.30.jar"
plugin_file_name_cropped = "LuckPerms"
result = plugin_updatechecker.get_plugin_file_name(plugin_file_name)
self.assertEqual(result, plugin_file_name_cropped)
def test_get_plugin_file_version(self):
plugin_file_name = "LuckPerms-5.4.30.jar"
plugin_version_cropped = "5.4.30"
result = plugin_updatechecker.get_plugin_file_version(plugin_file_name)
self.assertEqual(result, plugin_version_cropped)
def test_get_plugin_version_without_letters(self):
plugin_version = "VERSIONv5.4.30"
plugin_version_cropped = "5.4.30"
result = plugin_updatechecker.get_plugin_version_without_letters(plugin_version)
self.assertEqual(result, plugin_version_cropped)
def test_compare_plugin_version(self):
result = plugin_updatechecker.compare_plugin_version("5.4.30", "5.4.0")
result2 = plugin_updatechecker.compare_plugin_version("5.4.30", "8.7.60")
result3 = plugin_updatechecker.compare_plugin_version("5.4.30", "5.4.30")
self.assertEqual(result, True)
self.assertEqual(result2, False)
self.assertEqual(result3, False)

61
src/tests/test_plugin_utils.py Normal file
View File

@ -0,0 +1,61 @@
import unittest
from src.plugin import plugin_downloader
from src.utils import utilities
class TestCases(unittest.TestCase):
def test_handle_regex_plugin_name(self):
# Cropped name -> 'SUPERBPlugin'
plugin_name = "[1.13-5.49 ❤] >|> SUPERB Plugin <<💥| Now 150% OFF IN WINTER SALE IN SUMMER???"
plugin_name_cropped = "SUPERBPlugin"
result = plugin_downloader.handle_regex_plugin_name(plugin_name)
self.assertEqual(result, plugin_name_cropped)
def test_get_version_id_spiget(self):
# 21840 -> "Luckperms" in Version 5.4.30
result = plugin_downloader.get_version_id_spiget("28140", "5.4.30")
self.assertEqual(result, 455966)
def test_get_version_name_spiget(self):
# 455966 -> "5.4.30" from Luckperms
result = plugin_downloader.get_version_name_spiget("28140", 455966)
self.assertEqual(result, "5.4.30")
def test_get_download_path(self):
# local plugin folder
class config_values_local:
connection = "local"
local_seperate_download_path = True
local_path_to_seperate_download_path = "/local/path/plugins"
result=plugin_downloader.get_download_path(config_values_local)
self.assertEqual(result, config_values_local.local_path_to_seperate_download_path)
# plugin folder over sftp
class config_values_sftp:
connection = "sftp"
remote_seperate_download_path = True
remote_path_to_seperate_download_path = "/sftp/path/plugins"
result=plugin_downloader.get_download_path(config_values_sftp)
self.assertEqual(result, config_values_sftp.remote_path_to_seperate_download_path)
# plugin folder over ftp
class config_values_ftp:
connection = "ftp"
remote_seperate_download_path = True
remote_path_to_seperate_download_path = "/ftp/path/plugins"
result=plugin_downloader.get_download_path(config_values_ftp)
self.assertEqual(result, config_values_ftp.remote_path_to_seperate_download_path)
def test_convert_file_size_down(self):
# 100000 / 1024 = 97.66
result= utilities.convert_file_size_down(100000)
self.assertEqual(result, 97.66)
if __name__ == "__main__":
unittest.main()

50
src/tests/test_serverjar_paper_velocity_waterfall.py Normal file
View File

@ -0,0 +1,50 @@
import unittest
from src.serverjar import serverjar_paper_velocity_waterfall
class TestCases(unittest.TestCase):
def test_get_installed_mc_version(self):
# paper-1.19-40.jar -> 1.19
serverjar_file_name = "paper-1.19-40.jar"
mc_version = "1.19"
result = serverjar_paper_velocity_waterfall.get_installed_mc_version(serverjar_file_name)
self.assertEqual(result, mc_version)
def test_get_installed_serverjar_version(self):
# paper-1.19-40.jar -> 40
serverjar_file_name = "paper-1.19-40.jar"
serverjar_version = "40"
result = serverjar_paper_velocity_waterfall.get_installed_serverjar_version(serverjar_file_name)
self.assertEqual(result, serverjar_version)
def test_get_version_group(self):
# 1.18.2 -> 1.18
mc_version = "1.18.2"
mc_version_group = "1.18.2"
result = serverjar_paper_velocity_waterfall.get_version_group(mc_version)
self.assertEqual(result, mc_version_group)
def test_find_latest_available_version(self):
# Get latest available paper version for 1.15.2 which should be 393
file_server_jar_full_name = "paper-1.15.2-40.jar"
version_group = "1.15.2"
result = serverjar_paper_velocity_waterfall.find_latest_available_version(
file_server_jar_full_name,
version_group
)
self.assertEqual(result, 393)
def test_get_versions_behind(self):
# 161 - 157 = 4
serverjar_version = 157
latest_version = 161
result = serverjar_paper_velocity_waterfall.get_versions_behind(serverjar_version, latest_version)
self.assertEqual(result, 4)
if __name__ == "__main__":
unittest.main()

151
src/utils/console_output.py Normal file
View File

@ -0,0 +1,151 @@
"""
Handles the console on first startup of pluGET and prints logo and sets title
"""
import os
from rich.console import Console
from src.settings import PLUGETVERSION
def rich_print_error(error_message) -> None:
"""
Prints a formatted error message from rich
"""
console = Console()
console.print(error_message, style="bright_red")
return None
def rename_console_title() -> None:
"""
Renames the console title on first startup
"""
os.system("title " + "pluGET │ By Neocky")
return None
def clear_console() -> None:
"""
Clears the console on first startup
"""
os.system('cls' if os.name=='nt' else 'clear')
return None
def print_logo() -> None:
"""
Prints the logo of pluGET and the link to the github repo
"""
# use rich console
console = Console()
# line 1
console.print()
# line 2
console.print(" ██████",style="bright_magenta", end='')
console.print("╗ ", style="bright_yellow", end='')
console.print("██", style="bright_magenta", end='')
console.print("╗ ", style="bright_yellow", end='')
console.print("██", style="bright_magenta", end='')
console.print("╗ ", style="bright_yellow", end='')
console.print("██", style="bright_magenta", end='')
console.print("╗ ", style="bright_yellow", end='')
console.print("██████", style="bright_magenta", end='')
console.print("╗ ", style="bright_yellow", end='')
console.print("███████", style="bright_magenta", end='')
console.print("╗", style="bright_yellow", end='')
console.print("████████", style="bright_magenta", end='')
console.print("╗", style="bright_yellow")
# line 3
console.print(" ██", style="bright_magenta", end='')
console.print("╔══", style="bright_yellow", end='')
console.print("██", style="bright_magenta", end='')
console.print("╗", style="bright_yellow", end='')
console.print("██", style="bright_magenta", end='')
console.print("║ ", style="bright_yellow", end='')
console.print("██", style="bright_magenta", end='')
console.print("║ ", style="bright_yellow", end='')
console.print("██", style="bright_magenta", end='')
console.print("║", style="bright_yellow", end='')
console.print("██", style="bright_magenta", end='')
console.print("╔════╝ ", style="bright_yellow", end='')
console.print("██", style="bright_magenta", end='')
console.print("╔════╝╚══", style="bright_yellow", end='')
console.print("██", style="bright_magenta", end='')
console.print("╔══╝", style="bright_yellow")
# line 4
console.print(" ██████", style="bright_magenta", end='')
console.print("╔╝", style="bright_yellow", end='')
console.print("██", style="bright_magenta", end='')
console.print("║ ", style="bright_yellow", end='')
console.print("██", style="bright_magenta", end='')
console.print("║ ", style="bright_yellow", end='')
console.print("██", style="bright_magenta", end='')
console.print("║", style="bright_yellow", end='')
console.print("██", style="bright_magenta", end='')
console.print("║ ", style="bright_yellow", end='')
console.print("███", style="bright_magenta", end='')
console.print("╗", style="bright_yellow", end='')
console.print("█████", style="bright_magenta", end='')
console.print("╗ ", style="bright_yellow", end='')
console.print("██", style="bright_magenta", end='')
console.print("║ ", style="bright_yellow")
# line 5
console.print(" ██", style="bright_magenta", end='')
console.print("╔═══╝ ", style="bright_yellow", end='')
console.print("██", style="bright_magenta", end='')
console.print("║ ", style="bright_yellow", end='')
console.print("██", style="bright_magenta", end='')
console.print("║ ", style="bright_yellow", end='')
console.print("██", style="bright_magenta", end='')
console.print("║", style="bright_yellow", end='')
console.print("██", style="bright_magenta", end='')
console.print("║ ", style="bright_yellow", end='')
console.print("██", style="bright_magenta", end='')
console.print("║", style="bright_yellow", end='')
console.print("██", style="bright_magenta", end='')
console.print("╔══╝ ", style="bright_yellow", end='')
console.print("██", style="bright_magenta", end='')
console.print("║ ", style="bright_yellow")
# line 6
console.print(" ██", style="bright_magenta", end='')
console.print("║ ", style="bright_yellow", end='')
console.print("███████", style="bright_magenta", end='')
console.print("╗╚", style="bright_yellow", end='')
console.print("██████", style="bright_magenta", end='')
console.print("╔╝╚", style="bright_yellow", end='')
console.print("██████", style="bright_magenta", end='')
console.print("╔╝", style="bright_yellow", end='')
console.print("███████", style="bright_magenta", end='')
console.print("╗ ", style="bright_yellow", end='')
console.print("██", style="bright_magenta", end='')
console.print("║ ", style="bright_yellow")
# line 7
console.print(" ╚═╝ ╚══════╝ ╚═════╝ ╚═════╝ ╚══════╝ ╚═╝ ", style="bright_yellow")
# line 8
console.print()
# line 9
console.print(" ┌────────────────────────────────────┐", style="bright_black")
# line 10
console.print(" │ [", style="bright_black", end='')
console.print("By Neocky", style="bright_magenta", end='')
console.print("] │ ", style="bright_black")
# line 11
console.print(" │ ", style="bright_black", end='')
console.print("https://github.com/Neocky/pluGET", style="link https://github.com/Neocky/pluGET", end='')
console.print(" │ ", style="bright_black")
# line 12
console.print(" └────────────────────────────────────┘", style="bright_black")
console.print(" ───────────────────────────────────────────────────")
return None
def print_console_logo() -> None:
"""
Prints the logo of pluGET if it is called from console
"""
console = Console()
console.print(f"[not bold][bright_magenta]pluGET [bright_green]{PLUGETVERSION} ", end='')
console.print("created by Neocky → ", end='')
console.print("https://github.com/Neocky/pluGET", style="link https://github.com/Neocky/pluGET")
return None

89
src/utils/consoleoutput.py
View File

@ -1,89 +0,0 @@
from os import system
from os import name
def consoleTitle():
system("title " + "pluGET │ By Neocky")
def clearConsole():
system('cls' if name=='nt' else 'clear')
# https://docs.microsoft.com/en-us/windows/console/console-virtual-terminal-sequences
class oColors:
standardWhite = "\033[0m"
brightBlack = "\033[90m"
brightRed = "\033[91m"
brightGreen = "\033[92m"
brightYellow = "\033[93m"
brightMagenta = "\033[95m"
def printLogo():
print()
print(oColors.brightYellow + r" ___ ___ ___ ___ ___ ___ ")
print(oColors.brightMagenta + r" /\ " + oColors.brightYellow + r"\ " + oColors.brightMagenta + r"/"'\\' +
oColors.brightYellow + r"__\ " + oColors.brightMagenta + r"/"'\\' + oColors.brightYellow + r"__\ " +
oColors.brightMagenta + r"/\ "+ oColors.brightYellow + r"\ " + oColors.brightMagenta + r"/\ "+
oColors.brightYellow + r"\ " + oColors.brightMagenta + r"/\ "+ oColors.brightYellow + r""'\\ ')
print(oColors.brightMagenta + r" /::\ "+ oColors.brightYellow + r"\ " + oColors.brightMagenta + r"/:/ " +
oColors.brightYellow + r"/ " + oColors.brightMagenta + r"/:/ "+ oColors.brightYellow + r"/ " +
oColors.brightMagenta + r"/::\ "+ oColors.brightYellow + r"\ " + oColors.brightMagenta + r"/::\ "+
oColors.brightYellow + r"\ " + oColors.brightMagenta + r"\:\ "+ oColors.brightYellow + r""'\\ ')
print(oColors.brightMagenta + r" /:/\:\ "+ oColors.brightYellow + r"\ " + oColors.brightMagenta + r"/:/ "+
oColors.brightYellow + r"/ " + oColors.brightMagenta + r"/:/ "+ oColors.brightYellow + r"/ " +
oColors.brightMagenta + r"/:/\:\ "+ oColors.brightYellow + r"\ " + oColors.brightMagenta + r"/:/\:\ "+
oColors.brightYellow + r"\ " + oColors.brightMagenta + r"\:\ "+ oColors.brightYellow + r""'\\ ')
print(oColors.brightMagenta + r" /::"'\\' + oColors.brightYellow + r"~" + oColors.brightMagenta + r"\:\ "+
oColors.brightYellow + r"\ " + oColors.brightMagenta + r"/:/ "+ oColors.brightYellow + r"/ " +
oColors.brightMagenta + r"/:/ "+ oColors.brightYellow + r"/ ___ " + oColors.brightMagenta + r"/:/ \:\ "+
oColors.brightYellow + r"\ " + oColors.brightMagenta + r"/::"'\\'+ oColors.brightYellow + r"~" +
oColors.brightMagenta + r"\:\ "+ oColors.brightYellow + r"\ " + oColors.brightMagenta + r"/::\ "+
oColors.brightYellow + r""'\\')
print(oColors.brightMagenta + r" /:/\:\ \:"'\\'+ oColors.brightYellow + r"__\ " + oColors.brightMagenta + r"/:/"+
oColors.brightYellow + r"__/ " + oColors.brightMagenta + r"/:/"+ oColors.brightYellow + r"__/ " +
oColors.brightMagenta + r"/"'\\'+ oColors.brightYellow + r"__\ " + oColors.brightMagenta + r"/:/"+
oColors.brightYellow + r"__/_" + oColors.brightMagenta + r"\:"'\\'+ oColors.brightYellow + r"__\ " +
oColors.brightMagenta + r"/:/\:\ \:"'\\' + oColors.brightYellow + r"__\ " +
oColors.brightMagenta + r"/:/\:"'\\'+ oColors.brightYellow + r"__\ ")
print(oColors.brightMagenta + r" " + oColors.brightMagenta + r"\/"+ oColors.brightYellow + r"__" +
oColors.brightMagenta + r"\:\/:/"+ oColors.brightYellow + r" / " + oColors.brightMagenta + r"\:"'\\'+
oColors.brightYellow + r" \ " + oColors.brightMagenta + r"\:"'\\' + oColors.brightYellow + r" \ " +
oColors.brightMagenta + r"/:/ "+ oColors.brightYellow + r"/ " + oColors.brightMagenta + r"\:\ /\ \/" +
oColors.brightYellow + r"__/ " + oColors.brightMagenta + r"\:"'\\' + oColors.brightYellow + r"~" +
oColors.brightMagenta + r"\:\ \/" + oColors.brightYellow + r"__/ " + oColors.brightMagenta + r"/:/ \/"+
oColors.brightYellow + r"__/")
print(oColors.brightMagenta + r" \::/ "+ oColors.brightYellow + r"/ " + oColors.brightMagenta + r"\:\ "+
oColors.brightYellow + r"\ " + oColors.brightMagenta + r"\:\ /:/ "+ oColors.brightYellow + r"/ " +
oColors.brightMagenta + r"\:\ \:"'\\'+ oColors.brightYellow + r"__\ " + oColors.brightMagenta + r"\:\ \:"'\\'+
oColors.brightYellow + r"__\ " + oColors.brightMagenta + r"/:/ "+ oColors.brightYellow + r"/ ")
print(oColors.brightMagenta + r" \/"+ oColors.brightYellow + r"__/ " +
oColors.brightMagenta + r"\:\ " + oColors.brightYellow + r"\ " + oColors.brightMagenta + r"\:\/:/ "+
oColors.brightYellow + r"/ " + oColors.brightMagenta + r"\:\/:/ "+ oColors.brightYellow + r"/ " +
oColors.brightMagenta + r"\:\ \/"+ oColors.brightYellow + r"__/ " + oColors.brightMagenta + r"\/"+
oColors.brightYellow + r"__/ ")
print(oColors.brightMagenta + r" \:"'\\' + oColors.brightYellow + r"__\ " +
oColors.brightMagenta + r"\::/ " + oColors.brightYellow + r"/ " + oColors.brightMagenta + r"\::/ " +
oColors.brightYellow + r"/ " + oColors.brightMagenta + r"\:"'\\' + oColors.brightYellow + r"__\ ")
print(oColors.brightMagenta + r" \/" + oColors.brightYellow + r"__/ " +
oColors.brightMagenta + r"\/" + oColors.brightYellow + r"__/ " + oColors.brightMagenta + r"\/" +
oColors.brightYellow + r"__/ " + oColors.brightMagenta + r"\/" + oColors.brightYellow + r"__/ " +
oColors.standardWhite)
print()
print()
print(oColors.brightBlack + " ┌────────────────────────────────────┐" + oColors.standardWhite)
print(oColors.brightBlack + " │ [" + oColors.brightMagenta + "By Neocky" +oColors.brightBlack +
"] │ " + oColors.standardWhite)
print(oColors.brightBlack + " │ " + oColors.brightMagenta + "https://github.com/Neocky/pluGET" + oColors.brightBlack +
" │ " + oColors.standardWhite)
print(oColors.brightBlack + " └────────────────────────────────────┘" + oColors.standardWhite)
def printHorizontalLine():
print(" ─────────────────────────────────────────────────────────────────────────────────")
def printMainMenu():
printLogo()
printHorizontalLine()

262
src/utils/utilities.py
View File

@ -1,175 +1,143 @@
# misc functions
"""
Holds all the utilitie code for pluGET and the webrequests function
"""
import os
import sys
import shutil
import requests
import shutil
import re
from pathlib import Path
from rich.console import Console
from src.handlers.handle_sftp import sftp_create_connection
from src.handlers.handle_ftp import ftp_create_connection
from utils.consoleoutput import oColors
from handlers.handle_config import configurationValues
from handlers.handle_sftp import createSFTPConnection
from handlers.handle_ftp import createFTPConnection
from src.utils.console_output import rich_print_error
from src.handlers.handle_config import config_value
from src.settings import PLUGETVERSION
def getHelp():
print(oColors.brightYellow+ "Need help?" + oColors.standardWhite)
print("For a list of all commands: 'help command'")
print("Or check the docs here:")
print("https://github.com/Neocky/pluGET")
print("Or go to the official discord.")
print("The link for discord can also be found on Github!")
def getCommandHelp(optionalParams):
if optionalParams == None:
optionalParams = 'all'
print(oColors.brightBlack + f"Help for command: {optionalParams}" +oColors.standardWhite)
print("┌────────────────┬─────────────────┬─────────────────┬────────────────────────────────────────────────────────┐")
print("│ Command │ Selected Object │ Optional Params │ Function │")
print("└────────────────┴─────────────────┴─────────────────┴────────────────────────────────────────────────────────┘")
while True:
if optionalParams == 'all':
print(oColors.brightBlack + " GENERAL:" + oColors.standardWhite)
print(" exit ./anything Exit pluGET")
print(" help ./anything Get general help")
print(" help command all/command Get specific help to the commands of pluGET")
print(oColors.brightBlack + " PLUGIN MANAGEMENT:" + oColors.standardWhite)
print(" get Name/ID Version Downloads the latest version of a plugin")
print(" check Name/ID/all changelog Check for an update of an installed plugin")
print(" update Name/ID/all Update installed plugins to the latest version")
print(" search Name Search for a plugin and download the latest version")
print(" remove Name/ID Delete an installed plugin")
print(oColors.brightBlack + " SERVER SOFTWARE MANAGEMENT:" + oColors.standardWhite)
print(" check serverjar Check installed server software for an update")
print(" update serverjar Version/Latest Update installed server software to a specific version")
print(" get-paper PaperVersion McVersion Downloads a specific PaperMc version")
break
if optionalParams == 'exit':
print(oColors.brightBlack + " GENERAL:" + oColors.standardWhite)
print(" exit ./anything Exit pluGET")
break
if optionalParams == 'help':
print(oColors.brightBlack + " GENERAL:" + oColors.standardWhite)
print(" help ./anything Get general help")
print(" help command all/command Get specific help to the commands of pluGET")
break
if optionalParams == 'get':
print(oColors.brightBlack + " PLUGIN MANAGEMENT:" + oColors.standardWhite)
print(print(" get Name/ID Version Downloads the latest version of a plugin"))
break
if optionalParams == 'check':
print(oColors.brightBlack + " PLUGIN MANAGEMENT:" + oColors.standardWhite)
print(" check Name/ID/all Check for an update of an installed plugin")
print(oColors.brightBlack + " SERVER SOFTWARE MANAGEMENT:" + oColors.standardWhite)
print(" check serverjar Check installed server software for an update")
break
if optionalParams == 'update':
print(oColors.brightBlack + " PLUGIN MANAGEMENT:" + oColors.standardWhite)
print(" update Name/ID/all Update installed plugins to the latest version")
print(oColors.brightBlack + " SERVER SOFTWARE MANAGEMENT:" + oColors.standardWhite)
print(" update serverjar Version/Latest Update installed server software to a specific version")
break
if optionalParams == 'search':
print(oColors.brightBlack + " PLUGIN MANAGEMENT:" + oColors.standardWhite)
print(" search Name Search for a plugin and download the latest version")
break
if optionalParams == 'remove':
print(oColors.brightBlack + " PLUGIN MANAGEMENT:" + oColors.standardWhite)
print(" remove Name/ID Delete an installed plugin")
break
if optionalParams == 'get-paper':
print(oColors.brightBlack + " SERVER SOFTWARE MANAGEMENT:" + oColors.standardWhite)
print(" get-paper PaperVersion McVersion Downloads a specific PaperMc version")
break
else:
print(oColors.brightRed + "Error: Help for Command not found. Please try again. :(" + oColors.standardWhite)
break
def check_local_plugin_folder():
configValues = configurationValues()
if configValues.localPluginFolder:
if configValues.seperateDownloadPath:
pluginFolderPath = configValues.pathToSeperateDownloadPath
else:
pluginFolderPath = configValues.pathToPluginFolder
if not os.path.isdir(pluginFolderPath):
print(oColors.brightRed + "Plugin folder coulnd*t be found. Creating one..." + oColors.standardWhite)
def check_for_pluGET_update() -> None:
"""
Check with the github api if there is a new version for pluGET available and print download message if this is
the case
"""
response = api_do_request("https://api.github.com/repos/Neocky/pluGET/releases/latest")
# get '.1.6.10' as output
full_version_string = re.search(r"[\.?\d]*$", response["name"])
# remove '.' to get '1.6.10' as output
version = re.sub(r"^\.*", "", full_version_string.group())
console = Console()
try:
os.mkdir(pluginFolderPath)
except OSError:
print(oColors.brightRed + "Creation of directory %s failed" % pluginFolderPath)
print(oColors.brightRed + "Please check the config file!" + oColors.standardWhite)
input("Press any key + enter to exit...")
sys.exit()
else:
print("Created directory %s" % pluginFolderPath)
pluget_installed_version_tuple = tuple(map(int, (PLUGETVERSION.split("."))))
plugin_latest_version_tuple = tuple(map(int, (version.split("."))))
except ValueError:
console.print("Couldn't check if new version of pluGET is available")
return None
if pluget_installed_version_tuple < plugin_latest_version_tuple:
print(f"A new version of pluGET is available: {version}")
console.print("Download it here: ", end='')
console.print("https://github.com/Neocky/pluGET", style="link https://github.com/Neocky/pluGET")
return None
def apiTest():
apiStatusUrl = 'https://api.spiget.org/v2/status'
def api_do_request(url) -> list:
"""
Handles the webrequest and returns a json list
"""
webrequest_header = {'user-agent': 'pluGET/1.0'}
try:
r = requests.get(apiStatusUrl)
response = requests.get(url, headers=webrequest_header)
except:
rich_print_error("Error: Couldn't create webrequest")
# return None to make functions quit
return None
try:
api_json_data = response.json()
except:
rich_print_error("Error: Couldn't parse json of webrequest")
return None
return api_json_data
def api_test_spiget() -> None:
"""
Test if the Spiget api sends a 200 status code back
"""
try:
r = requests.get('https://api.spiget.org/v2/status')
except (requests.exceptions.HTTPError, requests.exceptions.ConnectionError):
print(oColors.brightRed + "Couldn't make a connection to the API. Check you connection to the internet!" + oColors.standardWhite)
input("Press any key + enter to exit...")
rich_print_error("Error: Couldn't make a connection to the API. Check you connection to the internet!")
sys.exit()
if r.status_code != 200:
print(oColors.brightRed + "Problems with the API detected. Plese try it again later!" + oColors.standardWhite)
input("Press any key + enter to exit...")
rich_print_error("Error: Problems with the API detected. Plese try it again later!")
sys.exit()
return None
def check_requirements():
configValues = configurationValues()
apiTest()
check_local_plugin_folder()
if not configValues.localPluginFolder:
if configValues.sftp_useSftp:
createSFTPConnection()
else:
createFTPConnection()
def create_temp_plugin_folder() -> Path:
"""
Creates a temporary folder to store plugins inside
Returns full path of temporary folder
"""
path_temp_plugin_folder = Path("./TempSFTPFolder")
if os.path.isdir(path_temp_plugin_folder):
return path_temp_plugin_folder
def createTempPluginFolder():
configValues = configurationValues()
tempPluginFolder = Path("./TempSFTPFolder")
if not os.path.isdir(tempPluginFolder):
try:
os.mkdir(tempPluginFolder)
os.mkdir(path_temp_plugin_folder)
except OSError:
print(oColors.brightRed + "Creation of directory %s failed" % configValues.pathToPluginFolder)
print(oColors.brightRed + "Please check the config file!" + oColors.standardWhite)
input("Press any key + enter to exit...")
rich_print_error(f"Error: Creation of directory {path_temp_plugin_folder} failed")
rich_print_error(" Please check for missing permissions in folder tree!")
sys.exit()
return tempPluginFolder
return path_temp_plugin_folder
def deleteTempPluginFolder(tempPluginFolder):
def remove_temp_plugin_folder() -> None:
"""
Removes the temporary plugin folder and all content inside it
"""
try:
shutil.rmtree(tempPluginFolder)
shutil.rmtree(Path("./TempSFTPFolder"))
except OSError as e:
print ("Error: %s - %s." % (e.filename, e.strerror))
rich_print_error(f"Error: {e.filename} - {e.strerror}")
return
def calculateFileSizeMb(downloadFileSize):
fileSizeDownload = int(downloadFileSize)
fileSizeMb = fileSizeDownload / 1024 / 1024
roundedFileSize = round(fileSizeMb, 2)
return roundedFileSize
def convert_file_size_down(file_size) -> float:
"""
Convert the size of the number one down. E.g. MB -> KB through division with 1024
"""
converted_file_size = (int(file_size)) / 1024
converted_file_size = round(converted_file_size, 2)
return converted_file_size
def calculateFileSizeKb(downloadFileSize):
fileSizeDownload = int(downloadFileSize)
fileSizeKb = fileSizeDownload / 1024
roundedFileSize = round(fileSizeKb, 2)
return roundedFileSize
def check_local_plugin_folder(config_values) -> None:
"""
Check if a local plugin folder exists and if not exit the programm
"""
if config_values.local_seperate_download_path:
plugin_folder_path = config_values.local_path_to_seperate_download_path
else:
plugin_folder_path = config_values.path_to_plugin_folder
if not os.path.isdir(plugin_folder_path):
rich_print_error(f"Error: Local plugin folder '{plugin_folder_path}' couldn't be found! \
\n Check the config and try again!")
sys.exit()
return None
def check_requirements() -> None:
"""
Check if the plugin folders are available
"""
config_values = config_value()
match config_values.connection:
case "local":
check_local_plugin_folder(config_values)
case "sftp":
sftp_create_connection()
case "ftp":
ftp_create_connection()
return None

9
src/utils/web_request.py
View File

@ -1,9 +0,0 @@
# Handles the web requests
import requests
def doAPIRequest(url):
headers = {'user-agent': 'pluGET/1.0'}
response = requests.get(url, headers=headers)
packageDetails = response.json()
return packageDetails