dekun
53863559f4
Retarget git remote, install path, and deploy docs from crypto_monitor to crypto_monitor_user. Co-authored-by: Cursor <cursoragent@cursor.com>
122 lines
3.6 KiB
Python
122 lines
3.6 KiB
Python
"""读写实例目录 .env(行级 upsert,原子落盘)."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
import tempfile
|
|
from typing import Optional
|
|
|
|
_KEY_LINE = re.compile(r"^(\s*)([A-Za-z_][A-Za-z0-9_]*)\s*=\s*(.*)\s*$")
|
|
|
|
|
|
def parse_env_lines(text: str) -> list[str]:
|
|
return text.replace("\r\n", "\n").replace("\r", "\n").splitlines()
|
|
|
|
|
|
def read_env_lines(path: str) -> list[str]:
|
|
if not os.path.isfile(path):
|
|
return []
|
|
with open(path, "r", encoding="utf-8", errors="ignore") as f:
|
|
return parse_env_lines(f.read())
|
|
|
|
|
|
def env_get(lines: list[str], key: str) -> Optional[str]:
|
|
for line in lines:
|
|
m = _KEY_LINE.match(line)
|
|
if m and m.group(2) == key:
|
|
raw = m.group(3).strip()
|
|
if (raw.startswith('"') and raw.endswith('"')) or (raw.startswith("'") and raw.endswith("'")):
|
|
return raw[1:-1]
|
|
return raw
|
|
return None
|
|
|
|
|
|
def env_get_all(lines: list[str]) -> dict[str, str]:
|
|
out: dict[str, str] = {}
|
|
for line in lines:
|
|
m = _KEY_LINE.match(line)
|
|
if m:
|
|
key = m.group(2)
|
|
raw = m.group(3).strip()
|
|
if (raw.startswith('"') and raw.endswith('"')) or (raw.startswith("'") and raw.endswith("'")):
|
|
out[key] = raw[1:-1]
|
|
else:
|
|
out[key] = raw
|
|
return out
|
|
|
|
|
|
def upsert_env_line(lines: list[str], key: str, value: str) -> list[str]:
|
|
pat = re.compile(r"^\s*" + re.escape(key) + r"\s*=")
|
|
out: list[str] = []
|
|
replaced = False
|
|
safe = value if value is not None else ""
|
|
if any(c in safe for c in (' ', '#', '"', "'")):
|
|
safe = '"' + safe.replace("\\", "\\\\").replace('"', '\\"') + '"'
|
|
new_line = f"{key}={safe}"
|
|
for line in lines:
|
|
if pat.match(line):
|
|
if not replaced:
|
|
out.append(new_line)
|
|
replaced = True
|
|
continue
|
|
out.append(line)
|
|
if not replaced:
|
|
if out and out[-1].strip():
|
|
out.append("")
|
|
out.append(new_line)
|
|
return out
|
|
|
|
|
|
def write_env_lines_atomic(path: str, lines: list[str]) -> None:
|
|
directory = os.path.dirname(os.path.abspath(path)) or "."
|
|
os.makedirs(directory, exist_ok=True)
|
|
fd, tmp = tempfile.mkstemp(prefix=".env.", dir=directory, text=True)
|
|
try:
|
|
with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as f:
|
|
f.write("\n".join(lines))
|
|
if lines:
|
|
f.write("\n")
|
|
os.replace(tmp, path)
|
|
finally:
|
|
if os.path.exists(tmp):
|
|
try:
|
|
os.remove(tmp)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def apply_env_updates(path: str, updates: dict[str, str]) -> list[str]:
|
|
lines = read_env_lines(path)
|
|
changed: list[str] = []
|
|
for key, value in updates.items():
|
|
if value is None:
|
|
continue
|
|
old = env_get(lines, key)
|
|
if old == value:
|
|
continue
|
|
lines = upsert_env_line(lines, key, value)
|
|
changed.append(key)
|
|
if changed:
|
|
write_env_lines_atomic(path, lines)
|
|
return changed
|
|
|
|
|
|
def load_env_file_into_environ(path: str) -> None:
|
|
if not os.path.exists(path):
|
|
return
|
|
with open(path, "r", encoding="utf-8", errors="ignore") as f:
|
|
text = f.read()
|
|
if text.startswith("\ufeff"):
|
|
text = text[1:]
|
|
for line in parse_env_lines(text):
|
|
s = line.strip()
|
|
if not s or s.startswith("#"):
|
|
continue
|
|
if "=" not in s:
|
|
continue
|
|
k, _, v = s.partition("=")
|
|
clean_key = k.strip()
|
|
clean_val = v.strip().strip('"').strip("'")
|
|
if clean_key:
|
|
os.environ[clean_key] = clean_val
|