virtualguard101/webscripts
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[Feat] Add manager handle as a new script

Browse Source
This commit is contained in:
VIRTUALGUARD
2025-07-10 23:08:15 +08:00
parent 2ef893f072
commit 4089f70574
6 changed files with 234 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@ -1 +1,3 @@
.venv/
dist/
WebServicesManager.egg-info/

29
requirements.txt
View File

@ -1 +1,30 @@
build==1.2.2.post1
certifi==2025.7.9
cffi==1.17.1
charset-normalizer==3.4.2
colorlog==6.9.0
cryptography==45.0.5
docutils==0.21.2
id==1.5.0
idna==3.10
jaraco-classes==3.4.0
jaraco-context==6.0.1
jaraco-functools==4.2.1
jeepney==0.9.0
keyring==25.6.0
markdown-it-py==3.0.0
mdurl==0.1.2
more-itertools==10.7.0
nh3==0.2.21
packaging==25.0
pycparser==2.22
pygments==2.19.2
pyproject-hooks==1.2.0
readme-renderer==44.0
requests==2.32.4
requests-toolbelt==1.0.0
rfc3986==2.0.0
rich==14.0.0
secretstorage==3.3.3
twine==6.1.0
urllib3==2.5.0

102
src/ServicesManager.py
View File

@ -1,102 +0,0 @@
import subprocess
import sys
import os
import logging
import colorlog
# Initialize color logging
handler = colorlog.StreamHandler()
handler.setFormatter(colorlog.ColoredFormatter(
'%(log_color)s%(asctime)s - %(levelname)s - %(message)s',
log_colors={
'DEBUG': 'cyan',
'INFO': 'green',
'WARNING': 'yellow',
'ERROR': 'red',
'CRITICAL': 'bold_red',
}
))
logger = colorlog.getLogger(__name__)
logger.setLevel(logging.INFO)
logger.addHandler(handler)
def get_operation():
while True:
choice = input("Please select an operation (0 to stop, 1 to restart, q to quit): ").strip().lower()
if choice == 'q':
return 'q'
try:
op = int(choice)
if op in (0, 1):
return op
except ValueError:
pass
logger.warning("Invalid input, please enter 0, 1, or q.")
class Service:
"""A template management for web services
Attributes:
tag (str): The tag that marks the service instance to use which way to deploy.
name (str): The name of the service instance.
dir (str): The configuration and data directory of the service instance.
"""
def __init__(self, tag: str, name: str, dir=None):
self.tag = tag
self.name = name
self.dir = dir
def command_gen(self):
"""Generate service management commands based on the service's tag.
"""
system_services_command = "sudo systemctl"
docker_services_command = f"cd {os.path.expanduser(f'{self.dir}')} && docker compose"
if self.tag == "sys":
return system_services_command
elif self.tag == "docker":
return docker_services_command
else:
raise ValueError(f"The service tag {self.tag} was not included.")
def manage_service(self):
"""Manage the service based on user input.
"""
operation = get_operation()
if operation == 'q':
logger.info("User chose to quit the service management.")
return
command = self.command_gen()
full_command = None # Initialize full_command
if operation == 0:
# Stop the service
full_command = f"{command} stop {self.name}"
logger.info(f"Stopping service: {self.name}")
elif operation == 1:
# Restart the service
full_command = f"{command} restart {self.name}"
logger.info(f"Restarting service: {self.name}")
else:
logger.warning("Invalid operation; no service management executed.")
return # Exit if the operation is invalid
if full_command: # Ensure full_command is defined
try:
subprocess.run(full_command, check=True, shell=True)
logger.info(f"Service {self.name} operation completed successfully.")
except subprocess.CalledProcessError as e:
logger.error(f"Failed to manage service {self.name}: {e}")
# Example call
if __name__ == "__main__":
service = Service(tag="sys", name="nginx")
service.manage_service()

5
src/__init__.py
View File

@ -1 +1,4 @@
# __all__ = [
# "services",
# "manager"
# ]

83
src/manager.py Executable file
View File

@ -0,0 +1,83 @@
#!/usr/bin/env python3
from src.services import *
import logging
import colorlog
# Initialize color logging
handler = colorlog.StreamHandler()
handler.setFormatter(colorlog.ColoredFormatter(
'%(log_color)s%(asctime)s - %(levelname)s - %(message)s',
log_colors={
'DEBUG': 'cyan',
'INFO': 'green',
'WARNING': 'yellow',
'ERROR': 'red',
'CRITICAL': 'bold_red',
}
))
logger = colorlog.getLogger(__name__)
logger.setLevel(logging.INFO)
logger.addHandler(handler)
# nginx = Service(tag="sys", name="nginx")
# minecraft = Service(tag="sys", name="minecraft", path="~/web/minecraftService")
# homepage = Service(tag="docker", name="docker", path="~/web/homepageService")
# status = Service(tag="docker", name="status", path="~/web/statusService")
# gitea = Service(tag="docker", name="gitea", path="~/web/giteaService")
class Manager:
"""Interface set of services management operations.
Attributes:
service_list (list): List use for storing service instances.
"""
def __init__(self):
"""Initialize the Manager with an empty services list."""
self.services_list = []
def append_service(self, service_instance: Service) -> None:
"""Append a service instance to the services list.
Args:
service_instance (Service): The service instance to append.
"""
self.services_list.append(service_instance)
def register_service(self, service_tag: str, service_name: str, service_path: str) -> Service:
"""Register a new service.
Args:
service_tag ('sys' | 'docker'): The tag that marks the service instance to use which way to deploy ('sys' or 'docker').
service_name (str): The name of the service instance.
service_path (str): The configuration and data path of the service instance.
Returns:
service: A web service instance.
"""
if service_tag == "docker":
if not os.path.exists(service_path):
raise ValueError(f"Invalid service path: {service_path}")
service = Service(tag=service_tag, name=service_name, path=service_path)
# service.code = len(self.services_list) + 1
self.append_service(service_instance=service)
return service
service = Service(tag=service_tag, name=service_name)
# service.code = len(self.services_list) + 1
self.append_service(service_instance=service)
return service
def list_services(self) -> None:
"""Method use for count the num of service instances and list the names of them.
"""
sum_of_service = 0
service_name_list = []
for services in self.services_list:
service_name_list.append(services.name)
sum_of_service += 1
logger.info(f"The manager has registered {sum_of_service} services: {', '.join(service_name_list)}")

116
src/services.py Normal file
View File

@ -0,0 +1,116 @@
#!/usr/bin/env python3
import subprocess
import os
import logging
import colorlog
# Initialize color logging
handler = colorlog.StreamHandler()
handler.setFormatter(colorlog.ColoredFormatter(
'%(log_color)s%(asctime)s - %(levelname)s - %(message)s',
log_colors={
'DEBUG': 'cyan',
'INFO': 'green',
'WARNING': 'yellow',
'ERROR': 'red',
'CRITICAL': 'bold_red',
}
))
logger = colorlog.getLogger(__name__)
logger.setLevel(logging.INFO)
logger.addHandler(handler)
def get_operation():
while True:
choice = input("Please select an operation (0 to stop, 1 to restart, q to quit): ").strip().lower()
if choice == 'q':
return 'q'
try:
op = int(choice)
if op in (0, 1):
return op
except ValueError:
pass
logger.warning("Invalid input, please enter 0, 1, or q.")
class Service:
"""A template management for web services
Attributes:
tag ('sys' | 'docker'): The tag that marks the service instance to use which way to deploy.
name (str): The name of the service instance.
path (str): The configuration and data path of the service instance.
"""
def __init__(self, tag: str, name: str, path=None):
self._tag = tag
self._name = name
self._path = path
def command_gen(self):
"""Generate service management commands based on the service's tag.
"""
system_services_command = "sudo systemctl"
docker_services_command = f"cd {os.path.expanduser(f'{self._path}')} && docker compose"
if self._tag == "sys":
return system_services_command
elif self._tag == "docker":
return docker_services_command
else:
raise ValueError(f"The service tag {self._tag} was not included.")
def service_operation(self):
"""Manage the service based on user input.
"""
operation = get_operation()
if operation == 'q':
logger.info("User chose to quit the service management.")
return
command = self.command_gen()
full_command = None # Initialize full_command
if self._tag == "sys":
if operation == 0:
# Stop the service
full_command = f"{command} stop {self._name}"
logger.info(f"Stopping service: {self._name}")
elif operation == 1:
# Restart the service
full_command = f"{command} restart {self._name}"
logger.info(f"Restarting service: {self._name}")
else:
logger.warning("Invalid operation; no service management executed.")
return # Exit if the operation is invalid
if self._tag == "docker":
if operation == 0:
# Stop the service
full_command = f"{command} down"
logger.info(f"Stopping service: {self._name}")
elif operation == 1:
# Restart the service
full_command = f"{command} up -d"
logger.info(f"Restarting service: {self._name}")
else:
logger.warning("Invalid operation; no service management executed.")
return # Exit if the operation is invalid
if full_command: # Ensure full_command is defined
try:
subprocess.run(full_command, check=True, shell=True)
logger.info(f"Service {self._name} operation completed successfully.")
except subprocess.CalledProcessError as e:
logger.error(f"Failed to manage service {self._name}: {e}")
# Example call
if __name__ == "__main__":
service = Service(tag="docker", name="homepage", path="~/web/homepageService")
service.service_operation()