[+] Conventions

This commit is contained in:
Charles Le Maux 2024-06-20 17:57:04 +01:00
parent ac330232ed
commit 9f07a97fc9
5 changed files with 66 additions and 77 deletions

50
api.py
View File

@ -8,110 +8,118 @@ app = Flask(__name__)
CORS(app)
cors = CORS(app, origins=['*'])
@app.route('/AccountCreate', methods=['POST'])
def account_create() -> Response:
data: Dict[str, str] = request.get_json()
if not data:
return jsonify({"error": "No JSON payload"}), 415
email: str = data.get('email')
port: str = data.get('port')
print(f'Email received: {email}')
print(f'Port received: {port}')
generic_executor.AccountCreate(port)
generic_executor.account_create(port)
return jsonify({'message': 'OK'}), 200
@app.route('/ServerCreate', methods=['POST'])
def server_create() -> Response:
data: Dict[str, str] = request.get_json()
if not data:
return jsonify({"error": "No JSON payload"}), 415
port: str = data.get('port')
name: str = data.get('name')
version: str = data.get('version')
print(f'Port received for server creation: {port}')
print(f'Server name: {name}')
print(f'Server version: {version}')
generic_executor.ServerCreate(port, name, version)
generic_executor.server_create(port, name, version)
return jsonify({'message': 'OK'}), 200
@app.route('/ServerDelete', methods=['POST'])
def server_delete() -> Response:
def server_delete() -> tuple[Response, int]:
data: Dict[str, str] = request.get_json()
if not data:
return jsonify({"error": "No JSON payload"}), 415
port: str = data.get('port')
name: str = data.get('name')
print(f'Port received for server deletion: {port}')
print(f'Server name to delete: {name}')
generic_executor.ServerDelete(port, name)
generic_executor.server_delete(port, name)
return jsonify({'message': 'OK'}), 200
@app.route('/AccountDelete', methods=['POST'])
def account_delete() -> Response:
def account_delete() -> tuple[Response, int]:
data: Dict[str, str] = request.get_json()
if not data:
return jsonify({"error": "No JSON payload"}), 415
port: str = data.get('port')
email: str = data.get('email')
print(f'Port received for account deletion: {port}')
print(f'Email of account to delete: {email}')
generic_executor.AccountDelete(port)
generic_executor.account_delete(port)
return jsonify({'message': 'OK'}), 200
@app.route('/ServerRun', methods=['POST'])
def server_run() -> Response:
def server_run() -> tuple[Response, int]:
data: Dict[str, str] = request.get_json()
if not data:
return jsonify({"error": "No JSON payload"}), 415
port: str = data.get('port')
name: str = data.get('name')
print(f'Port received for server launch: {port}')
print(f'Server name to launch: {name}')
generic_executor.ServerRun(port, name)
generic_executor.server_run(port, name)
return jsonify({'message': 'OK'}), 200
@app.route('/ServerStop', methods=['POST'])
def server_stop() -> Response:
def server_stop() -> tuple[Response, int]:
data: Dict[str, str] = request.get_json()
if not data:
return jsonify({"error": "No JSON payload"}), 415
port: str = data.get('port')
name: str = data.get('name')
print(f'Port received for server shutdown: {port}')
print(f'Server name to shut down: {name}')
generic_executor.ServerStop(port, name)
generic_executor.server_stop(port, name)
return jsonify({'message': 'OK'}), 200
@app.route('/UpdateProperty', methods=['POST'])
def update_property() -> Response:
def update_property() -> tuple[Response, int]:
data: Dict[str, str, str, str] = request.get_json()
if not data:
return jsonify({"error": "No JSON payload"}), 415
port: str = data.get('port')
name: str = data.get('name')
prop: str = data.get('property')
value: str = data.get('value')
print(f'The server {name} at port {port} changed {prop} to {value}')
generic_executor.UpdateProperty(port, name, prop, value)
generic_executor.update_property(port, name, prop, value)
return jsonify({'message': 'OK'}), 200
@app.route('/Command', methods=['POST'])
def command() -> tuple[Response, int]:
data: Dict[str, str, str] = request.get_json()
if not data:
return jsonify({"error": "No JSON payload"}), 415
port: str = data.get('port')
_command: str = data.get('command')
print(f'Server {port} executed command {_command}')
generic_executor.RunCommand(port, _command)
generic_executor.run_command(port, _command)
return jsonify({'message': 'OK'}), 200

View File

@ -4,24 +4,30 @@ import re
supported_versions = ["bukkit", "paper", "spigot"]
def get_all_versions(folder_path="servers/paper"):
return [name for name in os.listdir(folder_path) if os.path.isdir(os.path.join(folder_path, name))]
def version_exists(version, framework = "paper")->bool:
def version_exists(version, framework="paper") -> bool:
if framework not in supported_versions:
return False
folder_names = get_all_versions("servers/"+framework)
folder_names = get_all_versions("servers/" + framework)
return any(version in name for name in folder_names)
def create_folder(path):
os.makedirs(path, exist_ok=True)
def check_if_exists(path):
return os.path.exists(path)
def delete_non_empty_folder(path):
shutil.rmtree(path)
def copy_folder_contents(source_dir, destination_dir):
if not os.path.exists(destination_dir):
os.makedirs(destination_dir)
@ -34,12 +40,13 @@ def copy_folder_contents(source_dir, destination_dir):
if os.path.exists(destination_path):
for filename in os.listdir(source_path):
file_source_path = os.path.join(source_path, filename)
file_destination_path = os.path.join(destination_path, filename)
file_destination_path = os.path.join(destination_path, filename)
if not os.path.exists(file_destination_path):
shutil.copy2(file_source_path, file_destination_path)
else:
shutil.copytree(source_path, destination_path)
def update_server_property(file_path, property_name, new_value):
pattern = rf'^{property_name}=.*$'
with open(file_path, 'r') as file:
@ -53,5 +60,6 @@ def update_server_property(file_path, property_name, new_value):
with open(file_path, 'w') as file:
file.writelines(content)
if __name__ == "__main__":
pass

View File

@ -3,36 +3,44 @@ import file_manager
mc_manager: MinecraftServerManager = MinecraftServerManager()
def AccountCreate(port : str) -> bool:
file_manager.create_folder("users/"+port)
def ServerCreate(port : str, name : str, version : str) -> bool:
def account_create(port: str) -> bool:
file_manager.create_folder("users/" + port)
def server_create(port: str, name: str, version: str) -> bool:
server_path: str = f"users/{port}/{name}"
server_template_path: str = "servers/paper/"+version
server_template_path: str = "servers/paper/" + version
file_manager.create_folder(server_path)
file_manager.copy_folder_contents(server_template_path, server_path)
file_manager.copy_folder_contents("servers/shared", server_path)
file_manager.update_server_property(server_path+"/server.properties", "server-port", port)
file_manager.update_server_property(server_path + "/server.properties", "server-port", port)
def ServerDelete(port : str, name : str) -> bool:
def server_delete(port: str, name: str) -> bool:
server_path: str = f"users/{port}/{name}"
file_manager.delete_non_empty_folder(server_path)
def AccountDelete(port : str) -> bool:
file_manager.delete_non_empty_folder("users/"+port)
def ServerRun(port : str, name : str) -> bool:
def account_delete(port: str) -> bool:
file_manager.delete_non_empty_folder("users/" + port)
def server_run(port: str, name: str) -> bool:
server_id = mc_manager.start_server(f"users/{port}/{name}")
mc_manager.servers[server_id]['port'] = int(port)
def ServerStop(port : str, name : str) -> bool:
def server_stop(port: str, name: str) -> bool:
server_id = mc_manager.get_server_id_by_port(int(port))
mc_manager.stop_server(server_id)
def UpdateProperty(port : str, name : str, prop : str, value : str) -> bool:
def update_property(port: str, name: str, prop: str, value: str) -> bool:
property_file_path: str = f"users/{port}/{name}/server.properties"
file_manager.update_server_property(property_file_path, prop, value)
def RunCommand(port : str, command : str) -> bool:
def run_command(port: str, command: str) -> bool:
server_id = mc_manager.get_server_id_by_port(int(port))
mc_manager.execute_server_command(server_id, command)
mc_manager.execute_server_command(server_id, command)

View File

@ -1,6 +1,7 @@
import subprocess
import shlex
class MinecraftServerManager:
def __init__(self):
self.servers = {}
@ -15,10 +16,10 @@ class MinecraftServerManager:
'directory': server_directory,
'port': None
}
print(f"Started server {server_id} in directory {server_directory}")
return server_id
def execute_server_command(self, server_id, command):
if server_id in self.servers:
process = self.servers[server_id]['process']
@ -36,7 +37,7 @@ class MinecraftServerManager:
else:
print(f"No server found with ID {server_id}")
def stop_server_focefully(self, server_id):
def stop_server_forcefully(self, server_id):
if server_id in self.servers:
process = self.servers[server_id]['process']
process.terminate()
@ -47,7 +48,7 @@ class MinecraftServerManager:
def get_servers(self):
return self.servers.values()
def get_server_id_by_port(self, port):
for server_id, server_info in self.servers.items():
if server_info['port'] == port:

View File

@ -1,36 +0,0 @@
import re
def update_server_property(file_path, property_name, new_value):
"""
Updates a specified property in a Minecraft server configuration file.
Parameters:
- file_path: Path to the Minecraft server configuration file.
- property_name: Name of the property to update.
- new_value: New value to set for the property.
"""
# Regular expression pattern to match lines containing the property name
pattern = rf'^{property_name}=.*$'
# Read the file content
with open(file_path, 'r') as file:
content = file.readlines()
# Find the line containing the property and update its value
for i, line in enumerate(content):
if re.match(pattern, line):
content[i] = f"{property_name}={new_value}\n"
break
else:
raise ValueError(f"Property '{property_name}' not found in the file.")
# Write the updated content back to the file
with open(file_path, 'w') as file:
file.writelines(content)
# Example usage
file_path = 'folder/server.properties'
property_name = 'server-ip'
new_value = '127.0.0.1'
update_server_property(file_path, property_name, new_value)