dekun
125 lines
3.9 KiB
Python
125 lines
3.9 KiB
Python
"""读写仓库根 .env.control（仅补缺，不覆盖已有值）。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from pathlib import Path
|
|
|
|
from .config import _repo_root, get_control_settings
|
|
|
|
|
|
def env_control_path() -> Path:
|
|
return _repo_root() / ".env.control"
|
|
|
|
|
|
def _read_text(path: Path) -> str:
|
|
if not path.is_file():
|
|
return ""
|
|
return path.read_text(encoding="utf-8")
|
|
|
|
|
|
def _write_text(path: Path, text: str) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
if text and not text.endswith("\n"):
|
|
text += "\n"
|
|
path.write_text(text, encoding="utf-8")
|
|
try:
|
|
path.chmod(0o600)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def get_env_value(key: str, text: str | None = None) -> str | None:
|
|
raw = text if text is not None else _read_text(env_control_path())
|
|
m = re.search(rf"(?m)^{re.escape(key)}=(.*)$", raw)
|
|
if not m:
|
|
return None
|
|
val = m.group(1).strip()
|
|
if len(val) >= 2 and val[0] == val[-1] and val[0] in ("'", '"'):
|
|
val = val[1:-1]
|
|
return val
|
|
|
|
|
|
def upsert_env_control(key: str, value: str, *, overwrite: bool = True) -> Path:
|
|
"""写入/更新单个键。overwrite=False 时若已有非空值则跳过。"""
|
|
if "\n" in value or "\r" in value:
|
|
raise ValueError(f"{key} 值不能包含换行")
|
|
path = env_control_path()
|
|
text = _read_text(path)
|
|
existing = get_env_value(key, text)
|
|
if not overwrite and existing is not None and existing.strip() != "":
|
|
return path
|
|
safe = value.replace("\\", "\\\\").replace('"', '\\"')
|
|
line = f'{key}="{safe}"'
|
|
pattern = re.compile(rf"(?m)^{re.escape(key)}=.*$")
|
|
if pattern.search(text):
|
|
if not overwrite:
|
|
return path
|
|
text = pattern.sub(line, text)
|
|
else:
|
|
if text and not text.endswith("\n"):
|
|
text += "\n"
|
|
text += line + "\n"
|
|
_write_text(path, text)
|
|
return path
|
|
|
|
|
|
# 一键部署默认项：仅在缺失或为空时写入
|
|
_DEPLOY_DEFAULTS: dict[str, str] = {
|
|
"CONTROL_AUTH_USERNAME": "admin",
|
|
"CONTROL_AUTH_PASSWORD": "admin123",
|
|
"CONTROL_AUTH_SECRET": "change-me-control-secret-please",
|
|
"CONTROL_TOKEN_TTL_SEC": "604800",
|
|
"CONTROL_POLL_INTERVAL_SEC": "8",
|
|
"CONTROL_SSE_INTERVAL_SEC": "1",
|
|
"CONTROL_SSE_PROBE_TIMEOUT_SEC": "2.5",
|
|
"CONTROL_HTTP_TIMEOUT_SEC": "12",
|
|
"CONTROL_AUTH_TOKEN_VERSION": "1",
|
|
"CONTROL_LAN_AUTH_BYPASS": "0",
|
|
}
|
|
|
|
|
|
def ensure_env_control_defaults() -> dict[str, bool]:
|
|
"""
|
|
确保 .env.control 存在且关键键有值。
|
|
已有非空值绝不覆盖。返回 {key: written?}。
|
|
"""
|
|
written: dict[str, bool] = {}
|
|
path = env_control_path()
|
|
before = _read_text(path)
|
|
for key, default in _DEPLOY_DEFAULTS.items():
|
|
old = get_env_value(key, before)
|
|
if old is not None and old.strip() != "":
|
|
written[key] = False
|
|
continue
|
|
upsert_env_control(key, default, overwrite=False)
|
|
# 若文件原先无该键，before 里也没有；重新读确认
|
|
after = get_env_value(key)
|
|
written[key] = after == default or (old is None or old.strip() == "")
|
|
get_control_settings.cache_clear()
|
|
return written
|
|
|
|
|
|
def update_control_credentials(
|
|
*,
|
|
new_username: str,
|
|
new_password: str,
|
|
bump_token_version: bool = True,
|
|
) -> None:
|
|
user = new_username.strip()
|
|
pwd = new_password
|
|
if not user or not pwd:
|
|
raise ValueError("用户名和密码不能为空")
|
|
if len(pwd) < 6:
|
|
raise ValueError("密码至少 6 位")
|
|
upsert_env_control("CONTROL_AUTH_USERNAME", user, overwrite=True)
|
|
upsert_env_control("CONTROL_AUTH_PASSWORD", pwd, overwrite=True)
|
|
if bump_token_version:
|
|
cur = get_env_value("CONTROL_AUTH_TOKEN_VERSION") or "1"
|
|
try:
|
|
ver = int(cur) + 1
|
|
except ValueError:
|
|
ver = 2
|
|
upsert_env_control("CONTROL_AUTH_TOKEN_VERSION", str(ver), overwrite=True)
|
|
get_control_settings.cache_clear()
|