Tasks: add mc and edit-encrypted, unify base command set

This commit is contained in:
av
2026-07-13 12:57:42 +03:00
parent 93a0af7b94
commit a41f6bca8e
+54 -35
View File
@@ -14,6 +14,10 @@ VARS_FILE = "vars/vars.yml"
AUTHELIA_DOCKER = "docker run --rm -v $PWD:/data authelia/authelia:4.39.4 authelia" AUTHELIA_DOCKER = "docker run --rm -v $PWD:/data authelia/authelia:4.39.4 authelia"
# --- Базовый набор: единый интерфейс команд во всех репозиториях ---
# (набор и сигнатуры команд общие; реализация может отличаться по месту)
def _yq(query: str) -> str: def _yq(query: str) -> str:
result = subprocess.run( result = subprocess.run(
["yq", query, HOSTS_FILE], capture_output=True, text=True, check=True ["yq", query, HOSTS_FILE], capture_output=True, text=True, check=True
@@ -29,25 +33,6 @@ def _remote_host() -> str:
return _yq(".ungrouped.hosts.server.ansible_host") return _yq(".ungrouped.hosts.server.ansible_host")
def _application_dir() -> str:
"""Чтение application_dir: сначала из inventory (override), затем из vars/vars.yml."""
inv_value = _yq('.ungrouped.hosts.server.application_dir // ""')
if inv_value:
return inv_value
result = subprocess.run(
["yq", ".application_dir", VARS_FILE],
capture_output=True,
text=True,
check=True,
)
value = result.stdout.strip()
if not value or value == "null":
raise Exit(
f"application_dir не определён ни в inventory, ни в {VARS_FILE}", code=1
)
return value
def _rest_args() -> list[str]: def _rest_args() -> list[str]:
"""Возвращает аргументы после '--' из sys.argv""" """Возвращает аргументы после '--' из sys.argv"""
try: try:
@@ -74,7 +59,7 @@ def install_roles(ctx: Context) -> None:
@task @task
def pl(ctx: Context) -> None: def pl(ctx: Context) -> None:
"""Запустить плейбуки по имени: inv pl -- gitea miniflux""" """Запустить плейбуки по имени: inv pl -- <name> [name ...]"""
names = _rest_args() names = _rest_args()
if not names: if not names:
raise Exit("Укажи хотя бы один плейбук: inv pl -- <name> [name ...]", code=1) raise Exit("Укажи хотя бы один плейбук: inv pl -- <name> [name ...]", code=1)
@@ -88,34 +73,28 @@ def pl(ctx: Context) -> None:
@task @task
def ssh(ctx: Context) -> None: def ssh(ctx: Context) -> None:
"""SSH на удалённый сервер""" """SSH на удалённый сервер"""
subprocess.run(f"ssh {_remote_user()}@{_remote_host()}", shell=True) ctx.run(f"ssh {_remote_user()}@{_remote_host()}", pty=True)
@task @task
def zj(ctx: Context) -> None: def zj(ctx: Context) -> None:
"""Запуск zellij на удаленном сервере""" """Запуск zellij на удалённом сервере"""
subprocess.run( ctx.run(
f"ssh {_remote_user()}@{_remote_host()} -t zellij attach --create main", f"ssh {_remote_user()}@{_remote_host()} -t zellij attach --create main",
shell=True, pty=True,
) )
@task(aliases=["login"]) @task
def login_as_app(ctx: Context, app: str) -> None: def mc(ctx: Context) -> None:
"""SSH и переключиться на пользователя приложения: inv login gitea""" """Midnight Commander: слева локальная директория, справа удалённая домашняя"""
# sudo -i: login shell, -u: от имени пользователя ctx.run(f"mc . sh://{_remote_user()}@{_remote_host()}", pty=True)
# bash -i: интерактивный режим (job control), -l: login (читает профиль)
app_dir = f"{_application_dir()}/{app}"
subprocess.run(
f"""ssh {_remote_user()}@{_remote_host()} -t 'sudo -iu {app} bash -c "cd {app_dir} && exec bash -il"'""",
shell=True,
)
@task @task
def btop(ctx: Context) -> None: def btop(ctx: Context) -> None:
"""Запустить btop на удалённом сервере""" """Запустить btop на удалённом сервере"""
subprocess.run(f"ssh {_remote_user()}@{_remote_host()} -t btop", shell=True) ctx.run(f"ssh {_remote_user()}@{_remote_host()} -t btop", pty=True)
@task @task
@@ -130,6 +109,46 @@ def decrypt(ctx: Context, file: str) -> None:
ctx.run(f"uv run ansible-vault decrypt {file}") ctx.run(f"uv run ansible-vault decrypt {file}")
@task(aliases=["ee"])
def edit_encrypted(ctx: Context, path: str) -> None:
"""Редактировать vault-файл через ansible-vault edit (без расшифровки на диск)"""
ctx.run(f"uv run ansible-vault edit {path}", pty=True)
# --- Специфичные для этого репозитория задачи ---
def _application_dir() -> str:
"""Чтение application_dir: сначала из inventory (override), затем из vars/vars.yml."""
inv_value = _yq('.ungrouped.hosts.server.application_dir // ""')
if inv_value:
return inv_value
result = subprocess.run(
["yq", ".application_dir", VARS_FILE],
capture_output=True,
text=True,
check=True,
)
value = result.stdout.strip()
if not value or value == "null":
raise Exit(
f"application_dir не определён ни в inventory, ни в {VARS_FILE}", code=1
)
return value
@task(aliases=["login"])
def login_as_app(ctx: Context, app: str) -> None:
"""SSH и переключиться на пользователя приложения: inv login gitea"""
# sudo -i: login shell, -u: от имени пользователя
# bash -i: интерактивный режим (job control), -l: login (читает профиль)
app_dir = f"{_application_dir()}/{app}"
subprocess.run(
f"""ssh {_remote_user()}@{_remote_host()} -t 'sudo -iu {app} bash -c "cd {app_dir} && exec bash -il"'""",
shell=True,
)
@task @task
def authelia_cli(ctx: Context, args: str = "") -> None: def authelia_cli(ctx: Context, args: str = "") -> None:
"""Запустить authelia CLI в docker""" """Запустить authelia CLI в docker"""