Files
Nexus/server/infrastructure/env_file.py
2026-07-08 22:31:31 +08:00

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