Compare commits
1 Commits
renovate/p
...
8984e0ebb7
| Author | SHA1 | Date | |
|---|---|---|---|
| 8984e0ebb7 |
231
cli.py
231
cli.py
@@ -1,231 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
import os
|
||||
import secrets
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
from datetime import datetime
|
||||
from enum import StrEnum
|
||||
from pathlib import Path
|
||||
from typing import NoReturn
|
||||
|
||||
REQUIRED_VARS = ["DEEPSEEK_API_KEY", "TMDB_API_KEY", "QBITTORRENT_URL"]
|
||||
|
||||
# Size in bytes
|
||||
KEYS_TO_GENERATE = {
|
||||
"JWT_SECRET": 32,
|
||||
"JWT_REFRESH_SECRET": 32,
|
||||
"CREDS_KEY": 32,
|
||||
"CREDS_IV": 16,
|
||||
}
|
||||
|
||||
|
||||
class Style(StrEnum):
|
||||
"""ANSI codes for styling output.
|
||||
Usage: f"{Style.RED}Error{Style.RESET}"
|
||||
"""
|
||||
|
||||
RESET = "\033[0m"
|
||||
BOLD = "\033[1m"
|
||||
RED = "\033[31m"
|
||||
GREEN = "\033[32m"
|
||||
YELLOW = "\033[33m"
|
||||
CYAN = "\033[36m"
|
||||
DIM = "\033[2m"
|
||||
|
||||
|
||||
# Only for terminals and if not specified otherwise
|
||||
USE_COLORS = sys.stdout.isatty() and "NO_COLOR" not in os.environ
|
||||
|
||||
|
||||
def styled(text: str, color_code: str) -> str:
|
||||
"""Apply color only if supported by the terminal."""
|
||||
if USE_COLORS:
|
||||
return f"{color_code}{text}{Style.RESET}"
|
||||
return text
|
||||
|
||||
|
||||
def log(msg: str, color: str | None = None, prefix="") -> None:
|
||||
"""Print a formatted message."""
|
||||
formatted_msg = styled(msg, color) if color else msg
|
||||
print(f"{prefix}{formatted_msg}")
|
||||
|
||||
|
||||
def error_exit(msg: str) -> NoReturn:
|
||||
"""Print an error message in red and exit."""
|
||||
log(f"❌ {msg}", Style.RED)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def is_docker_running() -> bool:
|
||||
""" "Check if Docker is available and responsive."""
|
||||
if shutil.which("docker") is None:
|
||||
error_exit("Docker is not installed.")
|
||||
|
||||
result = subprocess.run(
|
||||
["docker", "info"],
|
||||
# Redirect stdout/stderr to keep output clean on success
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
# Prevent exception being raised
|
||||
check=False,
|
||||
)
|
||||
return result.returncode == 0
|
||||
|
||||
|
||||
def parse_env(content: str) -> dict[str, str]:
|
||||
"""Parses existing keys and values into a dict (ignoring comments)."""
|
||||
env_vars = {}
|
||||
for raw_line in content.splitlines():
|
||||
line = raw_line.strip()
|
||||
if line and not line.startswith("#") and "=" in line:
|
||||
key, value = line.split("=", 1)
|
||||
env_vars[key.strip()] = value.strip()
|
||||
|
||||
return env_vars
|
||||
|
||||
|
||||
def dump_env(content: str, data: dict[str, str]) -> str:
|
||||
new_content: list[str] = []
|
||||
processed_keys = set()
|
||||
|
||||
for raw_line in content.splitlines():
|
||||
line = raw_line.strip()
|
||||
# Fast line (empty, comment or not an assignation)
|
||||
if len(line) == 0 or line.startswith("#") or "=" not in line:
|
||||
new_content.append(raw_line)
|
||||
continue
|
||||
|
||||
# Slow line (inline comment to be kept)
|
||||
key_chunk, value_chunk = raw_line.split("=", 1)
|
||||
key = key_chunk.strip()
|
||||
|
||||
# Not in the update list
|
||||
if key not in data:
|
||||
new_content.append(raw_line)
|
||||
continue
|
||||
|
||||
processed_keys.add(key)
|
||||
new_value = data[key]
|
||||
|
||||
if " #" not in value_chunk:
|
||||
new_line = f"{key_chunk}={new_value}"
|
||||
else:
|
||||
_, comment = value_chunk.split(" #", 1)
|
||||
new_line = f"{key_chunk}={new_value} #{comment}"
|
||||
|
||||
new_content.append(new_line)
|
||||
|
||||
for key, value in data.items():
|
||||
if key not in processed_keys:
|
||||
new_content.append(f"{key}={value}")
|
||||
|
||||
return "\n".join(new_content) + "\n"
|
||||
|
||||
|
||||
def ensure_env() -> None:
|
||||
"""Manage .env lifecycle: creation, secret generation, prompts."""
|
||||
env_path = Path(".env")
|
||||
env_example_path = Path(".env.example")
|
||||
updated: bool = False
|
||||
|
||||
# Read .env if exists
|
||||
if env_path.exists():
|
||||
content: str = env_path.read_text(encoding="utf-8")
|
||||
else:
|
||||
content: str = env_example_path.read_text(encoding="utf-8")
|
||||
|
||||
existing_vars: dict[str, str] = parse_env(content)
|
||||
|
||||
# Generate missing secrets
|
||||
for key, length in KEYS_TO_GENERATE.items():
|
||||
if key not in existing_vars or not existing_vars[key]:
|
||||
log(f"Generating {key}...", Style.GREEN, prefix=" ")
|
||||
existing_vars[key] = secrets.token_hex(length)
|
||||
updated = True
|
||||
log("Done", Style.GREEN, prefix=" ")
|
||||
|
||||
# Prompt for missing mandatory keys
|
||||
color = Style.YELLOW if USE_COLORS else ""
|
||||
reset = Style.RESET if USE_COLORS else ""
|
||||
for key in REQUIRED_VARS:
|
||||
if key not in existing_vars or not existing_vars[key]:
|
||||
try:
|
||||
existing_vars[key] = input(
|
||||
f" {color}Enter value for {key}: {reset}"
|
||||
).strip()
|
||||
updated = True
|
||||
except KeyboardInterrupt:
|
||||
print()
|
||||
error_exit("Aborted by user.")
|
||||
|
||||
# Write to disk
|
||||
if updated:
|
||||
# But backup original first
|
||||
if env_path.exists():
|
||||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
backup_path = Path(f"{env_path}.{timestamp}.bak")
|
||||
shutil.copy(env_path, backup_path)
|
||||
log(f"Backup created: {backup_path}", Style.DIM)
|
||||
|
||||
new_content = dump_env(content, existing_vars)
|
||||
env_path.write_text(new_content, encoding="utf-8")
|
||||
log(".env updated successfully.", Style.GREEN)
|
||||
else:
|
||||
log("Configuration is up to date.", Style.GREEN)
|
||||
|
||||
|
||||
def setup() -> None:
|
||||
"""Orchestrate initialization."""
|
||||
is_docker_running()
|
||||
ensure_env()
|
||||
|
||||
|
||||
def status() -> None:
|
||||
"""Display simple dashboard."""
|
||||
# Hardcoded bold style for title if colors are enabled
|
||||
title_style = Style.BOLD if USE_COLORS else ""
|
||||
reset_style = Style.RESET if USE_COLORS else ""
|
||||
|
||||
print(f"\n{title_style}ALFRED STATUS{reset_style}")
|
||||
print(f"{title_style}==============={reset_style}\n")
|
||||
|
||||
# Docker Check
|
||||
if is_docker_running():
|
||||
print(f" Docker: {styled('✓ running', Style.GREEN)}")
|
||||
else:
|
||||
print(f" Docker: {styled('✗ stopped', Style.RED)}")
|
||||
|
||||
# Env Check
|
||||
if Path(".env").exists():
|
||||
print(f" .env: {styled('✓ present', Style.GREEN)}")
|
||||
else:
|
||||
print(f" .env: {styled('✗ missing', Style.RED)}")
|
||||
|
||||
print("")
|
||||
|
||||
|
||||
def check() -> None:
|
||||
"""Silent check for prerequisites (used by 'make up')."""
|
||||
setup()
|
||||
|
||||
|
||||
def main() -> None:
|
||||
if len(sys.argv) < 2:
|
||||
print("Usage: python cli.py [setup|check|status]")
|
||||
sys.exit(1)
|
||||
|
||||
cmd = sys.argv[1]
|
||||
|
||||
if cmd == "setup":
|
||||
setup()
|
||||
elif cmd == "check":
|
||||
check()
|
||||
elif cmd == "status":
|
||||
status()
|
||||
else:
|
||||
error_exit(f"Unknown command: {cmd}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user