dekun
49 lines
1.3 KiB
Python
49 lines
1.3 KiB
Python
"""中控 .env 读写与 PM2 重启."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from lib.env.env_file_lib import apply_env_updates, load_env_file_into_environ
|
|
from lib.env.shared_env_lib import (
|
|
apply_ai_env_to_all,
|
|
build_ai_env_payload,
|
|
restart_instances_then_hub_pm2,
|
|
)
|
|
from lib.instance.instance_pm2_lib import schedule_pm2_restart
|
|
|
|
HUB_DIR = Path(__file__).resolve().parent
|
|
|
|
|
|
def hub_env_path() -> str:
|
|
return str(HUB_DIR / ".env")
|
|
|
|
|
|
def update_hub_credentials(*, new_password: str, new_username: str | None = None) -> list[str]:
|
|
updates: dict[str, str] = {"HUB_PASSWORD": new_password}
|
|
if new_username:
|
|
updates["HUB_USERNAME"] = new_username
|
|
path = hub_env_path()
|
|
changed = apply_env_updates(path, updates)
|
|
if changed:
|
|
load_env_file_into_environ(path)
|
|
return changed
|
|
|
|
|
|
def get_hub_ai_env_payload() -> dict[str, Any]:
|
|
return build_ai_env_payload(hub_env_path())
|
|
|
|
|
|
def save_hub_ai_env(updates: dict[str, str]) -> dict[str, Any]:
|
|
return apply_ai_env_to_all(updates)
|
|
|
|
|
|
def restart_all_pm2() -> dict[str, Any]:
|
|
return restart_instances_then_hub_pm2()
|
|
|
|
|
|
def restart_hub_pm2() -> dict[str, Any]:
|
|
app_name = (os.getenv("PM2_APP_NAME") or "").strip() or "manual-trading-hub"
|
|
return schedule_pm2_restart(app_name)
|