Files
crypto_monitor/lib/instance/instance_pm2_lib.py
T
dekun b733e551a0 Normalize fullwidth punctuation to ASCII across codebase.
Add scripts/normalize_ambiguous_unicode.py; fix corrupted patch_instance_theme_templates.py. Preserves curly quotes in string literals; removes Git homoglyph warnings on .env.example.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-08 23:42:26 +08:00

50 lines
1.5 KiB
Python

"""PM2 重启当前实例(仅 Linux 部署环境)."""
from __future__ import annotations
import os
import subprocess
import sys
from typing import Any
def default_pm2_app_name(exchange_key: str) -> str:
mapping = {
"okx": "crypto_okx",
"binance": "crypto_binance",
"gate": "crypto_gate",
}
return mapping.get((exchange_key or "").strip().lower(), "crypto_okx")
def resolve_pm2_app_name(exchange_key: str) -> str:
explicit = (os.getenv("PM2_APP_NAME") or "").strip()
if explicit:
return explicit
return default_pm2_app_name(exchange_key)
def restart_instance_pm2(exchange_key: str) -> dict[str, Any]:
if not sys.platform.startswith("linux"):
return {"ok": False, "msg": "仅 Linux 服务器支持 PM2 重启", "app": None}
app_name = resolve_pm2_app_name(exchange_key)
try:
proc = subprocess.run(
["pm2", "restart", app_name, "--update-env"],
capture_output=True,
text=True,
timeout=120,
)
ok = proc.returncode == 0
return {
"ok": ok,
"app": app_name,
"msg": (proc.stdout or proc.stderr or "").strip()[:500],
"returncode": proc.returncode,
}
except FileNotFoundError:
return {"ok": False, "msg": "未找到 pm2 命令", "app": app_name}
except subprocess.TimeoutExpired:
return {"ok": False, "msg": "pm2 restart 超时", "app": app_name}
except Exception as e:
return {"ok": False, "msg": str(e), "app": app_name}