working composectl, first attemp
This commit is contained in:
@@ -0,0 +1,18 @@
|
||||
#!/usr/bin/env python3
|
||||
# vim: set ft=python ts=4 sw=4 et: fileencoding=utf-8
|
||||
# SPDX-FileCopyrightText: 2025-present RenatoXSR <renatoxsr@gmail.com>
|
||||
# SPDX-License-Identifier: BSD-3-Clause-Clear
|
||||
"""RenatoXSR's Python Utilities (monorepo)"""
|
||||
import os, sys, platform
|
||||
|
||||
# Setup
|
||||
import renatoxsr.const
|
||||
import renatoxsr.logger
|
||||
logger = renatoxsr.logger.logger
|
||||
logger.debug("platform: %s", platform.platform())
|
||||
logger.debug("system: %s", platform.system())
|
||||
logger.debug("node: %s", platform.node())
|
||||
|
||||
# Import modules that should be exported
|
||||
import renatoxsr.utils
|
||||
import renatoxsr.scripts.composectl
|
||||
@@ -0,0 +1,6 @@
|
||||
#!/usr/bin/env python3
|
||||
# vim: set ft=python ts=4 sw=4 et: fileencoding=utf-8
|
||||
# SPDX-FileCopyrightText: 2025-present RenatoXSR <renatoxsr@gmail.com>
|
||||
# SPDX-License-Identifier: BSD-3-Clause-Clear
|
||||
from importlib.metadata import version
|
||||
__version__ = version('renatoxsr')
|
||||
@@ -0,0 +1,61 @@
|
||||
#!/usr/bin/env python3
|
||||
# vim: set ft=python ts=4 sw=4 et: fileencoding=utf-8
|
||||
# SPDX-FileCopyrightText: 2025-present RenatoXSR <renatoxsr@gmail.com>
|
||||
# SPDX-License-Identifier: BSD-3-Clause-Clear
|
||||
"""CONSTANTS"""
|
||||
|
||||
# Builtin imports
|
||||
from datetime import datetime, date, time
|
||||
from enum import Enum, StrEnum
|
||||
from pathlib import Path
|
||||
|
||||
import pandas as pd
|
||||
import numpy as np
|
||||
|
||||
class COLORS(StrEnum):
|
||||
# see: https://stackoverflow.com/questions/4842424/list-of-ansi-color-escape-sequences
|
||||
RESET = "\033[0m"
|
||||
RED = "\033[31m"
|
||||
BLUE = "\033[34m"
|
||||
GREY = "\033[90m"
|
||||
WHITE = "\033[37m"
|
||||
GREEN = "\033[32m"
|
||||
YELLOW = "\033[93m"
|
||||
ORANGE = "\033[33m"
|
||||
CYAN = "\033[36m"
|
||||
DEFAULT = WHITE
|
||||
TITLE = WHITE
|
||||
ERROR = RED
|
||||
DEBUG = CYAN
|
||||
WARNING = ORANGE
|
||||
WARN = YELLOW
|
||||
INFO = GREEN
|
||||
CRITICAL = WHITE
|
||||
|
||||
|
||||
@classmethod
|
||||
def _missing_(cls, value):
|
||||
#value = value.lower()
|
||||
for member in cls:
|
||||
if member.value == value:
|
||||
return member
|
||||
return cls.DEFAULT
|
||||
|
||||
|
||||
class INVALID (Enum):
|
||||
EMPTY = object()
|
||||
INT = object()
|
||||
NONE = None
|
||||
NAN = np.nan
|
||||
NAT = pd.NaT
|
||||
DF = pd.DataFrame().to_string()
|
||||
SERIES = pd.Series().to_string()
|
||||
STR = str()
|
||||
LIST = list()
|
||||
DICT = dict()
|
||||
TUPLE = tuple()
|
||||
SET = set()
|
||||
ERROR = Exception
|
||||
DATE = date(1900,1,1)
|
||||
|
||||
# EOF
|
||||
@@ -0,0 +1,242 @@
|
||||
#!/usr/bin/env python3
|
||||
# vim: set ft=python ts=4 sw=4 et: fileencoding=utf-8
|
||||
# SPDX-FileCopyrightText: 2025-present RenatoXSR <renatoxsr@gmail.com>
|
||||
# SPDX-License-Identifier: BSD-3-Clause-Clear
|
||||
"""Create a logger object (stdout) and optionally a logfile."""
|
||||
|
||||
from datetime import datetime
|
||||
from enum import StrEnum
|
||||
import logging
|
||||
import os
|
||||
from pathlib import Path
|
||||
from platform import node, platform, system
|
||||
import sys
|
||||
import colorlog
|
||||
|
||||
|
||||
from renatoxsr.const import COLORS, INVALID
|
||||
# Defaults
|
||||
FORMAT = f"[%(name)s:%(levelname)s:%(filename)s:%(funcName)s:%(lineno)d][%(process)d] %(message)s"
|
||||
now = datetime.now().strftime('%Y-%m-%d_%H-%M-%S')
|
||||
default_logfile = Path.cwd() / "logs" / f"{now}-{__name__}.log"
|
||||
|
||||
def get_formatter(format: str = FORMAT) -> logging.Formatter:
|
||||
"""Return a new logger formatter.
|
||||
|
||||
Args:
|
||||
format (str, optional): [description]. Defaults to None.
|
||||
|
||||
Returns:
|
||||
logging.Formatter: [description].
|
||||
|
||||
"""
|
||||
formatter = logging.Formatter(format)
|
||||
return formatter
|
||||
|
||||
|
||||
def set_logfile(
|
||||
*,
|
||||
name: str | None = None,
|
||||
formatter: logging.Formatter | None = None,
|
||||
format: str | None = None,
|
||||
logfile: Path | None = None,
|
||||
) -> logging.Handler:
|
||||
"""Log a message to a file .
|
||||
|
||||
Args:
|
||||
name (str, optional): [description]. Defaults to None.
|
||||
formatter (logging.Formatter, optional): [description]. Defaults to None.
|
||||
format (str, optional): [description]. Defaults to None.
|
||||
logfile (Path, optional): [description]. Defaults to None.
|
||||
|
||||
Returns:
|
||||
logging.Handler: [description].
|
||||
|
||||
"""
|
||||
# file handler
|
||||
now_str = now
|
||||
|
||||
if name is None:
|
||||
name = __name__
|
||||
|
||||
if formatter is None:
|
||||
if format is not None:
|
||||
formatter = get_formatter(format)
|
||||
else:
|
||||
formatter = get_formatter()
|
||||
|
||||
if logfile is None:
|
||||
logfile = default_logfile
|
||||
|
||||
if logfile.is_file():
|
||||
logfile.parent.mkdir(parents=True, exist_ok=True)
|
||||
file_handler = logging.FileHandler(logfile, mode="a", encoding="utf-8")
|
||||
|
||||
if logfile.is_dir():
|
||||
logfile.mkdir(parents=True, exist_ok=True)
|
||||
file_handler = logging.FileHandler(
|
||||
logfile / f"{name}.log", mode="a", encoding="utf-8"
|
||||
)
|
||||
|
||||
file_handler.setFormatter(formatter)
|
||||
|
||||
return file_handler
|
||||
|
||||
|
||||
def set_loglevel(logger_instance: logging.Logger | None = None, logLevel="") -> int:
|
||||
"""Set the loglevel of a logger instance .
|
||||
|
||||
Args:
|
||||
logger_instance (logging.Logger, optional): [description]. Defaults to None.
|
||||
logLevel (str, optional): logging.[DEBUG|INFO|etc]. Defaults to "".
|
||||
|
||||
Returns:
|
||||
logging.setLevel or None
|
||||
int: EffectiveLoglevel()
|
||||
|
||||
"""
|
||||
if logger_instance:
|
||||
logger_instance.setLevel(getattr(logging, logLevel, "INFO"))
|
||||
elif logLevel:
|
||||
logger.setLevel(getattr(logging, logLevel, "INFO"))
|
||||
return logger.getEffectiveLevel()
|
||||
|
||||
|
||||
def set_logger(
|
||||
*,
|
||||
name: str | None = None,
|
||||
formatter: logging.Formatter | None = None,
|
||||
format: str | None = None,
|
||||
loglevel: str | None = None,
|
||||
logfile: Path | None = None,
|
||||
log_to_file: bool = False,
|
||||
) -> logging.Logger:
|
||||
"""Create and return a logger.
|
||||
|
||||
Args:
|
||||
name (str): logger name.
|
||||
formatter (logging.Formatter): formatter to use.
|
||||
format (str): format string to create a formatter.
|
||||
loglevel (str): log level to select.
|
||||
logfile (Path): logfile path.
|
||||
log_to_file (bool): whether to log to file.
|
||||
|
||||
Returns:
|
||||
logging.Logger: configured logger.
|
||||
|
||||
"""
|
||||
if name is None:
|
||||
name = __name__
|
||||
|
||||
if formatter is None:
|
||||
if isinstance(format, str):
|
||||
formatter = get_formatter(format)
|
||||
else:
|
||||
formatter = get_formatter()
|
||||
|
||||
stdout_handler = colorlog.StreamHandler()
|
||||
stdout_handler.setFormatter(colorlog.ColoredFormatter(
|
||||
"%(log_color)s" + FORMAT,
|
||||
reset=True,
|
||||
log_colors={
|
||||
'DEBUG': 'cyan',
|
||||
'INFO': 'green',
|
||||
'WARNING': 'yellow',
|
||||
'ERROR': 'red',
|
||||
'CRITICAL': 'red,bg_white',
|
||||
},
|
||||
secondary_log_colors={},
|
||||
style ='%',
|
||||
))
|
||||
|
||||
# stdout handler
|
||||
#stdout_handler = logging.StreamHandler()
|
||||
#stdout_handler.setFormatter(formatter)
|
||||
|
||||
logger = logging.getLogger(name)
|
||||
logger.addHandler(stdout_handler)
|
||||
|
||||
# file handler
|
||||
if logfile:
|
||||
log_to_file = True
|
||||
|
||||
|
||||
|
||||
if log_to_file:
|
||||
file_handler = set_logfile(
|
||||
name=name,
|
||||
formatter=formatter,
|
||||
format=format,
|
||||
logfile=logfile,
|
||||
)
|
||||
|
||||
logger.addHandler(file_handler)
|
||||
|
||||
if loglevel is None:
|
||||
loglevel = os.getenv("LOGLEVEL", "INFO")
|
||||
# if logging.getLevelNamesMapping().get(loglevel.upper()) is None:
|
||||
if loglevel not in logging.getLevelNamesMapping():
|
||||
loglevel = "INFO"
|
||||
|
||||
logger.setLevel(loglevel)
|
||||
|
||||
logger.debug("Loglevel: %s", os.getenv("LOGLEVEL"))
|
||||
return logger
|
||||
|
||||
|
||||
# upon import:
|
||||
loglevel = "INFO"
|
||||
for arg in [a for a in sys.argv if a.startswith("--log")]:
|
||||
if arg.startswith("--loglevel="):
|
||||
loglevel = arg.split("=", 1)[1]
|
||||
if arg.startswith("--logfile="):
|
||||
log_to_file = arg.split("=", 1)[1]
|
||||
if arg == "--debug":
|
||||
loglevel = "DEBUG"
|
||||
logger = set_logger()
|
||||
logger.setLevel(loglevel)
|
||||
|
||||
|
||||
# CRITICAL = 50
|
||||
# FATAL = CRITICAL
|
||||
# ERROR = 40
|
||||
# WARNING = 30
|
||||
# WARN = WARNING
|
||||
# INFO = 20
|
||||
# DEBUG = 10
|
||||
# NOTSET = 0
|
||||
|
||||
# _levelToName = {
|
||||
# CRITICAL: 'CRITICAL',
|
||||
# ERROR: 'ERROR',
|
||||
# WARNING: 'WARNING',
|
||||
# INFO: 'INFO',
|
||||
# DEBUG: 'DEBUG',
|
||||
# NOTSET: 'NOTSET',
|
||||
# }
|
||||
# _nameToLevel = {
|
||||
# 'CRITICAL': CRITICAL,
|
||||
# 'FATAL': FATAL,
|
||||
# 'ERROR': ERROR,
|
||||
# 'WARN': WARNING,
|
||||
# 'WARNING': WARNING,
|
||||
# 'INFO': INFO,
|
||||
# 'DEBUG': DEBUG,
|
||||
# 'NOTSET': NOTSET,
|
||||
# }
|
||||
|
||||
|
||||
# def log(level, msg, *args, **kwargs):
|
||||
# logger.log(level, f"{COLORS.level}" + msg + f"{COLORS.RESET}", *args, **kwargs)
|
||||
|
||||
# def debug(msg, *args, **kwargs):
|
||||
# logger.log("DEBUG", msg, *args, **kwargs)
|
||||
|
||||
# def info(msg, *args, **kwargs):
|
||||
# logger.log("INFO", msg, *args, **kwargs)
|
||||
|
||||
# def error(msg, *args, **kwargs):
|
||||
# logger.log("ERROR", msg, *args, **kwargs)
|
||||
|
||||
# def warning(msg, *args, **kwargs):
|
||||
# logger.log("WARNING", msg, *args, **kwargs)
|
||||
@@ -0,0 +1,368 @@
|
||||
#!/usr/bin/env python3
|
||||
# vim: set ft=python ts=4 sw=4 et: fileencoding=utf-8
|
||||
# SPDX-FileCopyrightText: 2025-present RenatoXSR <renatoxsr@gmail.com>
|
||||
# SPDX-License-Identifier: BSD-3-Clause-Clear
|
||||
"""`composectl` script
|
||||
|
||||
$ composectl
|
||||
--all | --subdir <absolute/or/relative/to/path>
|
||||
--compose-file <relative-to-path-or-to-path-and-subdir.yml>
|
||||
--path /srv/containers
|
||||
<compose cmd>
|
||||
"""
|
||||
# python3 << '#EOF'
|
||||
|
||||
try:
|
||||
import argparse
|
||||
import asyncio
|
||||
from collections import namedtuple
|
||||
from collections.abc import Mapping, Sequence
|
||||
import datetime as dt
|
||||
from enum import Enum, IntEnum, StrEnum, property as enum_property
|
||||
import logging
|
||||
import os
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import sys
|
||||
from typing import TextIO, NamedTuple
|
||||
|
||||
except ImportError as e:
|
||||
args = [e.msg]
|
||||
msg = ["[%s]"]
|
||||
if e.name and e.path:
|
||||
args.append(e.name)
|
||||
args.append(e.path)
|
||||
msg.append("('%s' : '%s')")
|
||||
else:
|
||||
msg.append("('%s')")
|
||||
if e.name:
|
||||
args.append(e.name)
|
||||
if e.path:
|
||||
args.append(e.path)
|
||||
|
||||
#logging.basicConfig(level=logging.DEBUG)
|
||||
#logger = logging.getLogger(__name__)
|
||||
#logger.critical(msg, *args)
|
||||
print(msg % tuple(args))
|
||||
sys.exit(1)
|
||||
|
||||
#print("import success")
|
||||
|
||||
# ***
|
||||
|
||||
# Globals and constants
|
||||
|
||||
# ANSI CODES
|
||||
def CSI(code) -> str:
|
||||
return "\033[" + str(code) + "m"
|
||||
|
||||
COLOR_RESET = CSI(0)
|
||||
|
||||
class COLOR(IntEnum):
|
||||
BLACK = 0 #\033[40m\033[90m\033[100m
|
||||
RED = 1 #\033[41m\033[91m\033[101m
|
||||
GREEN = 2 #\033[42m\033[92m\033[102m
|
||||
YELLOW= 3 #\033[43m\033[93m\033[103m
|
||||
BLUE = 4 #\033[44m\033[94m\033[104m
|
||||
MAGENTA = 5 #\033[45m\033[95m\033[105m
|
||||
CYAN = 6 #\033[46m\033[96m\033[106m
|
||||
WHITE = 7 #\033[47m\033[97m\033[107m
|
||||
FG = 30
|
||||
BG = 40
|
||||
|
||||
@enum_property
|
||||
def reset(cls):
|
||||
return CSI(0)
|
||||
|
||||
@enum_property
|
||||
def fg(self):
|
||||
return CSI(COLOR.FG + self)
|
||||
|
||||
@enum_property
|
||||
def bg(self):
|
||||
return CSI(COLOR.BG + self)
|
||||
|
||||
|
||||
# ***
|
||||
|
||||
# Global Setup
|
||||
|
||||
# Setup logger
|
||||
#logging.basicConfig(level=logging.DEBUG, format=LOGFMT)
|
||||
#for handler in logging.root.handlers[:]:
|
||||
# logging.root.removeHandler(handler)
|
||||
# handler.close()
|
||||
|
||||
# Log utils
|
||||
LEVEL_COLOR: dict[int, str] = {
|
||||
logging.DEBUG: COLOR.CYAN.fg,
|
||||
logging.INFO: COLOR.MAGENTA.fg,
|
||||
logging.WARNING: COLOR.YELLOW.fg,
|
||||
logging.ERROR: COLOR.RED.fg,
|
||||
logging.CRITICAL: COLOR.RED.bg + COLOR.WHITE.fg,
|
||||
}
|
||||
|
||||
LEVEL_NAME: dict[int, str] = {
|
||||
logging.DEBUG: "DEBUG",
|
||||
logging.INFO: "INFO",
|
||||
logging.WARNING: "WARN",
|
||||
logging.ERROR: "ERR",
|
||||
logging.CRITICAL: "CRIT",
|
||||
}
|
||||
|
||||
|
||||
class LevelColorFormatter(logging.Formatter):
|
||||
def format(self, record):
|
||||
LOGFMT = ("[%(name)s][%(filename)s" +
|
||||
#":%(funcName)s" +
|
||||
"(%(lineno)04d)]" +
|
||||
"[%(relativeCreated)d]%(levelname)5s: %(message)s")
|
||||
record.relativeCreatedSeconds = record.relativeCreated / 1000.0
|
||||
record.levelname = LEVEL_NAME.get(record.levelno, "UNSET")
|
||||
formatter = logging.Formatter(f"{LEVEL_COLOR[record.levelno]}" +
|
||||
LOGFMT.replace("%(relativeCreated)d","%(relativeCreatedSeconds).03f")
|
||||
+ COLOR_RESET)
|
||||
return formatter.format(record)
|
||||
|
||||
|
||||
# ***
|
||||
# Get module logger
|
||||
logger = logging.getLogger(__name__)
|
||||
logger.setLevel(logging.DEBUG)
|
||||
|
||||
# Console Logger
|
||||
stderr_handler = logging.StreamHandler(stream=sys.stderr)
|
||||
stderr_handler.setFormatter(LevelColorFormatter())
|
||||
stderr_handler.setLevel(logging.DEBUG)
|
||||
logger.addHandler(stderr_handler)
|
||||
#print(logger.handlers)
|
||||
|
||||
# ***
|
||||
|
||||
|
||||
MIN_VERSION = NamedTuple("MIN_VERSION", [
|
||||
("major", int),
|
||||
("minor", int),
|
||||
("micro", int),
|
||||
("releaselevel", str),
|
||||
("serial", int)])(
|
||||
major=3,
|
||||
minor=11,
|
||||
micro=None,
|
||||
releaselevel=None,
|
||||
serial=None)
|
||||
|
||||
python_requires = f">={MIN_VERSION.major}{MIN_VERSION.minor}"
|
||||
|
||||
# Check python version
|
||||
def check_python_version(version_info, min_version):
|
||||
logger.debug(version_info)
|
||||
#print(version_info)
|
||||
if version_info < min_version:
|
||||
logger.critical("Insufficient Python version: %d.%d. (needs at least: %d.%d)",
|
||||
version_info[0],
|
||||
version_info[1],
|
||||
min_version[0],
|
||||
min_version[1])
|
||||
sys.exit(1)
|
||||
|
||||
# Run check
|
||||
check_python_version(sys.version_info, MIN_VERSION)
|
||||
|
||||
# Package version
|
||||
def set_package_version(file):
|
||||
mtime = Path(file).stat().st_mtime
|
||||
version = dt.datetime.fromtimestamp(mtime).strftime("v%Y-%m-%d.%H-%M-%S.%f")
|
||||
logger.debug("%s:%s", file, version)
|
||||
return version
|
||||
|
||||
# Set version
|
||||
__version__ = set_package_version(__file__)
|
||||
|
||||
# Global Functions
|
||||
|
||||
def cmd(compose_cmd: str | Path, compose_file: str | Path, *args, **kwargs):
|
||||
if isinstance(compose_cmd, str):
|
||||
compose_bin, compose_args = compose_cmd.split(" ", maxsplit=1)
|
||||
bin_path = shutil.which(compose_bin)
|
||||
logger.debug("BIN: %s", bin_path)
|
||||
full_cmd = [bin_path]
|
||||
if " " in compose_args:
|
||||
full_cmd.extend(compose_args.split(" "))
|
||||
else:
|
||||
full_cmd.append(compose_args)
|
||||
if not compose_file:
|
||||
compose_file = "compose.yml"
|
||||
compose_fpath = Path(compose_file).resolve()
|
||||
if not compose_fpath.exists():
|
||||
logger.error("Compose file does not exist: '%s'", compose_file)
|
||||
return None
|
||||
full_cmd.extend(["-f", compose_fpath])
|
||||
for a in args:
|
||||
if isinstance(a, str):
|
||||
full_cmd.append(a)
|
||||
elif isinstance(a, Mapping):
|
||||
for k,v in a.items():
|
||||
if " " in k:
|
||||
logger.error("Invalid key name: %s", k)
|
||||
continue
|
||||
if len(k) > 1:
|
||||
full_cmd.append(f"--{k}")
|
||||
else:
|
||||
full_cmd.append(f"-{k}")
|
||||
full_cmd.append(str(v))
|
||||
elif isinstance(a, Sequence):
|
||||
full_cmd.extend(a)
|
||||
else:
|
||||
full_cmd.append(a)
|
||||
logger.debug("CMD: %s", repr(full_cmd))
|
||||
return full_cmd
|
||||
|
||||
# def log_wait(i, td, outs, errs):
|
||||
# #if outs is None and errs is None:
|
||||
# logger.debug("[%d] %d.%03d", i, td.seconds, td.microseconds % 1000)
|
||||
# if outs is not None:
|
||||
# sys.stdout.write(
|
||||
# COLOR.GREEN.fg +
|
||||
# (outs.decode(encoding="utf-8") if isinstance(outs, bytes) else outs)
|
||||
# + COLOR_RESET)
|
||||
# sys.stdout.flush()
|
||||
# if errs is not None:
|
||||
# sys.stderr.write(
|
||||
# COLOR.RED.fg +
|
||||
# (errs.decode(encoding="utf-8") if isinstance(errs, bytes) else errs)
|
||||
# + COLOR_RESET)
|
||||
# sys.stderr.flush()
|
||||
|
||||
|
||||
async def runcmd(cmd, name: str = None):
|
||||
cmd_exec = cmd[0]
|
||||
cmd_args = cmd[1:]
|
||||
logger.debug("EXEC: '%s'", cmd_exec)
|
||||
logger.debug("ARGS: ['%s']", "', '".join([str(c) for c in cmd_args]))
|
||||
start = dt.datetime.now()
|
||||
i = 0
|
||||
async with asyncio.TaskGroup() as tg:
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
cmd_exec, *cmd_args,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE)
|
||||
|
||||
tg.create_task(read_stream(proc.stdout, sys.stdout,
|
||||
name=(name or "STDOUT"),
|
||||
color=COLOR.GREEN.fg,
|
||||
start_time=start))
|
||||
tg.create_task(read_stream(proc.stderr, sys.stderr,
|
||||
name=(name or "STDERR"),
|
||||
color=COLOR.RED.fg,
|
||||
start_time=start))
|
||||
return_code = await proc.wait()
|
||||
|
||||
|
||||
async def read_stream(
|
||||
stream_from: TextIO,
|
||||
stream_to: TextIO,
|
||||
name: str=None,
|
||||
color: str = None,
|
||||
start_time: dt.datetime = None,
|
||||
):
|
||||
while True:
|
||||
line = await stream_from.readline()
|
||||
if not line:
|
||||
break
|
||||
if start_time:
|
||||
td = dt.datetime.now() - start_time
|
||||
delta = "%d.%03d " % (td.seconds, td.microseconds % 1000)
|
||||
stream_to.write(
|
||||
(color or "") +
|
||||
(f"[{name}] " if name else "") +
|
||||
(delta if start_time else "") +
|
||||
line.decode("raw_unicode_escape") + COLOR_RESET)
|
||||
stream_to.flush()
|
||||
|
||||
|
||||
# Configured Actions
|
||||
class ACTION(Enum):
|
||||
UP = ("--progress=plain", "up", "-d", "--remove-orphans")
|
||||
UPDB = ("--progress=plain", "up","-d","--build", "--pull", "always", "--remove-orphans")
|
||||
DOWN = ("--progress=plain", "down","--remove-orphans")
|
||||
LOGS = ("--progress=plain", "logs",)
|
||||
LOGSF = ("--progress=plain", "logs","-f")
|
||||
LS = ("--progress=plain", "ls",)
|
||||
BUILD = ("--progress=plain", "build", "--pull", "--push")
|
||||
PUSH = ("--progress=plain", "push")
|
||||
|
||||
|
||||
|
||||
|
||||
def main():
|
||||
logger.debug(f"CWD:\t{os.getcwd()}")
|
||||
p = argparse.ArgumentParser()
|
||||
p.add_argument("--root", type=Path, default="/srv/containers")
|
||||
#p.add_argument("--prefix", default="", help="Prefix of compose files")
|
||||
#p.add_argument("--suffix", default="", help="Suffix to find compose files, before .yml extension")
|
||||
p.add_argument("--subdir", default="./", help="Subdir to find containers")
|
||||
p.add_argument("--name", default="*", help="Name of folder to find compose.yml file")
|
||||
p.add_argument("--cmd", default="docker compose", help="compose command")
|
||||
p.add_argument("--file", default="compose.yml")
|
||||
p.add_argument("--profiles", help="Comma separated list of compose profiles")
|
||||
#p.add_argument("--project", help="Project directories to find compose.yml and/or names")
|
||||
p.add_argument("--services", help="services of compose file")
|
||||
p.add_argument("action", choices=[a.name.lower() for a in ACTION])
|
||||
p.add_argument("project_and_or_services", nargs="*", help="project and/or services (this list will be parsed before used)")
|
||||
args = p.parse_args()
|
||||
logger.debug("Args:")
|
||||
for k,v in vars(args).items():
|
||||
logger.debug(" - %s:\t%s", k, repr(v))
|
||||
action = ACTION[args.action.upper()].value
|
||||
logger.debug("Action:\t('%s')", "', '".join(action))
|
||||
try:
|
||||
logger.debug("Starting runner context")
|
||||
with asyncio.Runner() as runner:
|
||||
i = 0
|
||||
root_path = args.root.joinpath(args.subdir)
|
||||
glob_pattern = args.name+"/*"+args.file
|
||||
logger.debug("Globbing '%s' in '%s'", glob_pattern, root_path)
|
||||
for path in root_path.glob(glob_pattern):
|
||||
i += 1
|
||||
logger.info("[%02d] File: '%s'", i, str(path))
|
||||
runner.run(
|
||||
runcmd(
|
||||
cmd(args.cmd, path, action),
|
||||
name=path.parent.name))
|
||||
logger.info("Found %d matches in '%s' (pattern: %s", i, root_path, glob_pattern)
|
||||
except asyncio.CancelledError as e:
|
||||
logger.critical("Cancel task: %s (%s)", str(e), path)
|
||||
asyncio.sleep(0.5)
|
||||
except RuntimeError as e:
|
||||
logger.critical(str(e))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
sys.exit(main())
|
||||
except KeyboardInterrupt as sigint:
|
||||
logger.critical("[KeyboardInterrupt] %s", str(sigint))
|
||||
try:
|
||||
loop = asyncio.get_event_loop()
|
||||
if loop.is_running():
|
||||
all_tasks = asyncio.all_tasks(loop)
|
||||
for task in all_tasks:
|
||||
task.cancel()
|
||||
loop.run_until_complete(
|
||||
asyncio.gather(
|
||||
*all_tasks,
|
||||
#return_exceptions=True
|
||||
return_exceptions=False
|
||||
)
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error("Error during task gathering: %s", str(e))
|
||||
finally:
|
||||
logger.info("Exit.")
|
||||
sys.exit(1)
|
||||
except RuntimeError as e:
|
||||
logger.critical("[RuntimeError] %s", str(e))
|
||||
#sys.exit(1)
|
||||
|
||||
#EOF
|
||||
@@ -0,0 +1,6 @@
|
||||
#!/usr/bin/env python3
|
||||
# vim: set ft=python ts=4 sw=4 et: fileencoding=utf-8
|
||||
# SPDX-FileCopyrightText: 2025-present RenatoXSR <renatoxsr@gmail.com>
|
||||
# SPDX-License-Identifier: BSD-3-Clause-Clear
|
||||
"""`utils` package of RenatoXSR's Python Utilities (monorepo)"""
|
||||
import renatoxsr.utils.uuid7_base58
|
||||
@@ -0,0 +1,38 @@
|
||||
import sys
|
||||
import uuid
|
||||
import base58
|
||||
|
||||
### Garante suporte nativo no Python 3.14+ ou usa fallback para versões anteriores
|
||||
|
||||
if sys.version_info >= (3, 14):
|
||||
from uuid import uuid7
|
||||
else:
|
||||
from uuid6 import uuid7
|
||||
|
||||
def uuid7_to_base58() -> tuple[str, str]:
|
||||
"""Gera um UUIDv7 e o codifica em uma string compacta Base58."""
|
||||
### 1. Gera o UUIDv7 padrão (128 bits / 16 bytes)
|
||||
u7 = uuid7()
|
||||
### 2. Transforma os 16 bytes do UUID em Base58 (sem o checksum do Bitcoin)
|
||||
# .zfill(22) garante que ele sempre tenha 22 caracteres se houver zeros à esquerda
|
||||
encoded_b58 = base58.b58encode(u7.bytes).decode('utf-8').zfill(22)
|
||||
return str(u7), encoded_b58
|
||||
|
||||
def base58_to_uuid7(b58_string: str) -> str:
|
||||
"""Decodifica uma string Base58 de volta para o formato original de UUIDv7."""
|
||||
### 1. Decodifica a string de volta para os 16 bytes originais
|
||||
decoded_bytes = base58.b58decode(b58_string)
|
||||
### 2. Reconstrói o objeto UUID a partir dos bytes
|
||||
return str(uuid.UUID(bytes=decoded_bytes))
|
||||
|
||||
def demo():
|
||||
""" --- Demonstração de Uso ---"""
|
||||
### Gerando o ID compactado
|
||||
uuid_original, id_base58 = uuid7_to_base58()
|
||||
print(f"UUIDv7 Original (36 chars): {uuid_original}")
|
||||
print(f"UUIDv7 em Base58 (22 chars): {id_base58}")
|
||||
### Revertendo para provar que é bidirecional (lossless)
|
||||
uuid_revertido = base58_to_uuid7(id_base58)
|
||||
print(f"Decodificado de volta: {uuid_revertido}")
|
||||
assert uuid_original == uuid_revertido, "Erro: Os UUIDs não coincidem!"
|
||||
print("\nSucesso! Conversão perfeita e segura para URLs.")
|
||||
@@ -0,0 +1,72 @@
|
||||
import base64
|
||||
|
||||
# Marcadores de fronteira (Não alteram o visual do texto)
|
||||
ZW_START = '\u200a' # Hair Space (usado como marcador de início)
|
||||
ZW_END = '\u202f' # Narrow No-Break Space (usado como marcador de fim)
|
||||
|
||||
# Lista com exatamente 32 caracteres invisíveis para os dados (Base32)
|
||||
ZW_CHARS = [
|
||||
'\u200b', '\u200c', '\u200d', '\ufeff', '\u200e', '\u200f', '\u202a', '\u202b',
|
||||
'\u202c', '\u202d', '\u202e', '\u2060', '\u2061', '\u2062', '\u2063', '\u2064',
|
||||
'\u2066', '\u2067', '\u2068', '\u2069', '\u206a', '\u206b', '\u206c', '\u206d',
|
||||
'\u206e', '\u206f', '\ufe00', '\ufe01', '\ufe02', '\ufe03', '\ufe04', '\ufe05',
|
||||
]
|
||||
|
||||
B32_ALPHABET = "ABCDEFGHIJKLMNOPQRSTUVWXYZ234567"
|
||||
|
||||
# Mapeamentos para conversão rápida
|
||||
B32_TO_ZW = {B32_ALPHABET[i]: ZW_CHARS[i] for i in range(32)}
|
||||
ZW_TO_B32 = {ZW_CHARS[i]: B32_ALPHABET[i] for i in range(32)}
|
||||
|
||||
def ocultar_texto(texto_visivel: str, texto_oculto: str) -> str:
|
||||
"""Codifica o texto oculto e o envelopa com marcadores de início e fim."""
|
||||
# Converte o texto oculto para Base32 sem padding
|
||||
b32_data = base64.b32encode(texto_oculto.encode('utf-8')).decode('utf-8').replace('=', '')
|
||||
|
||||
# Converte para a sequência invisível e adiciona as fronteiras START e END
|
||||
zw_encoded = ZW_START + "".join(B32_TO_ZW[char] for char in b32_data) + ZW_END
|
||||
|
||||
# Insere o bloco invisível de forma segura no início do texto visível
|
||||
return zw_encoded + texto_visivel
|
||||
|
||||
def revelar_texto(texto_com_marca: str) -> str:
|
||||
"""Localiza os marcadores de início e fim para extrair e decodificar o payload."""
|
||||
# Encontra os índices dos delimitadores
|
||||
try:
|
||||
inicio = texto_com_marca.index(ZW_START) + 1
|
||||
fim = texto_com_marca.index(ZW_END, inicio)
|
||||
except ValueError:
|
||||
return "" # Marcadores não encontrados ou incompletos
|
||||
|
||||
# Extrai apenas o bloco que está estritamente entre o início e o fim
|
||||
bloco_invisivel = texto_com_marca[inicio:fim]
|
||||
|
||||
# Filtra apenas os caracteres válidos do nosso alfabeto de dados
|
||||
zw_extraido = [char for char in bloco_invisivel if char in ZW_TO_B32]
|
||||
|
||||
# Reconstrói a string Base32
|
||||
b32_data = "".join(ZW_TO_B32[char] for char in zw_extraido)
|
||||
|
||||
if not b32_data:
|
||||
return ""
|
||||
|
||||
# Ajusta o padding do Base32
|
||||
mod = len(b32_data) % 8
|
||||
if mod:
|
||||
b32_data += '=' * (8 - mod)
|
||||
|
||||
return base64.b32decode(b32_data.encode('utf-8')).decode('utf-8')
|
||||
|
||||
# --- Exemplo de Uso ---
|
||||
if __name__ == "__main__":
|
||||
texto_original = "Mensagem importante que será enviada ao cliente."
|
||||
id_secreto = "ID-999-STRICT"
|
||||
|
||||
# Injetando
|
||||
texto_com_payload = ocultar_texto(texto_original, id_secreto)
|
||||
print(f"Texto gerado (Total de caracteres: {len(texto_com_payload)}):")
|
||||
print(f"-> {texto_com_payload} <-\n")
|
||||
|
||||
# Extraindo
|
||||
id_descoberto = revelar_texto(texto_com_payload)
|
||||
print(f"ID isolado e extraído com sucesso: '{id_descoberto}'")
|
||||
Reference in New Issue
Block a user