Nexus Agent
f9934bd4f1
保存 Telegram 时同步至 .env，health_monitor 支持 MySQL 回退， 设置页展示巡检状态，部署时自动安装 cron。 Co-authored-by: Cursor <cursoragent@cursor.com>
66 lines
1.9 KiB
Python
66 lines
1.9 KiB
Python
"""Shared .env read/write helpers (UTF-8 LF, quoted values)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from pathlib import Path
|
|
|
|
from server.utils.text_io import read_utf8_text, write_utf8_lf
|
|
|
|
_ENV_KEY_RE = re.compile(r"^([A-Z0-9_]+)=(.*)$")
|
|
|
|
|
|
def escape_env_value(value: str) -> str:
|
|
escaped = value.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n")
|
|
return f'"{escaped}"'
|
|
|
|
|
|
def read_env_var(path: Path, key: str) -> str | None:
|
|
"""Return unquoted value for key, or None if missing/empty."""
|
|
if not path.is_file():
|
|
return None
|
|
for line in read_utf8_text(path).splitlines():
|
|
m = _ENV_KEY_RE.match(line.strip())
|
|
if not m or m.group(1) != key:
|
|
continue
|
|
raw = m.group(2).strip()
|
|
if not raw:
|
|
return None
|
|
if len(raw) >= 2 and raw[0] == raw[-1] == '"':
|
|
inner = raw[1:-1]
|
|
return inner.replace("\\n", "\n").replace('\\"', '"').replace("\\\\", "\\")
|
|
return raw
|
|
return None
|
|
|
|
|
|
def upsert_env_vars(path: Path, updates: dict[str, str]) -> bool:
|
|
"""Upsert keys in a .env file. Creates file if missing. Returns True if written."""
|
|
if not updates:
|
|
return False
|
|
|
|
lines: list[str] = []
|
|
if path.is_file():
|
|
lines = read_utf8_text(path).splitlines()
|
|
|
|
seen: set[str] = set()
|
|
out: list[str] = []
|
|
for line in lines:
|
|
m = _ENV_KEY_RE.match(line.strip())
|
|
if m and m.group(1) in updates:
|
|
key = m.group(1)
|
|
out.append(f"{key}={escape_env_value(updates[key])}")
|
|
seen.add(key)
|
|
else:
|
|
out.append(line)
|
|
|
|
for key, value in updates.items():
|
|
if key not in seen:
|
|
out.append(f"{key}={escape_env_value(value)}")
|
|
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
body = "\n".join(out)
|
|
if body and not body.endswith("\n"):
|
|
body += "\n"
|
|
write_utf8_lf(path, body)
|
|
return True
|