dekun
e5a586f903
Move business code under modules/, env template to config/, PM2 single qihuo process, and _legacy shims for old imports. Co-authored-by: Cursor <cursoragent@cursor.com>
75 lines
2.2 KiB
Python
75 lines
2.2 KiB
Python
# Copyright (c) 2025-2026 马建军. All rights reserved.
|
|
# 专有软件 — 未经授权禁止复制、传播、转售。
|
|
# 严禁用于：带单/代客理财、向他人推荐期货品种或买卖建议、融资配资等业务。
|
|
# 详见 LICENSE.zh-CN.txt 与 docs/软件购买与使用协议.md
|
|
|
|
"""读写项目根目录 .env 文件（更新指定键，保留其余行）。"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
|
|
from modules.core.paths import ENV_FILE, LEGACY_ENV_FILE
|
|
|
|
|
|
def _default_env_path() -> str:
|
|
if ENV_FILE.is_file():
|
|
return str(ENV_FILE)
|
|
if LEGACY_ENV_FILE.is_file():
|
|
return str(LEGACY_ENV_FILE)
|
|
return str(ENV_FILE)
|
|
|
|
|
|
ENV_PATH = _default_env_path()
|
|
_KEY_RE = re.compile(r"^([A-Za-z_][A-Za-z0-9_]*)\s*=")
|
|
|
|
|
|
def env_file_path(path: str | None = None) -> str:
|
|
if path:
|
|
return path
|
|
from modules.core.paths import resolve_env_file
|
|
return resolve_env_file()
|
|
|
|
|
|
def _quote_env_value(value: str) -> str:
|
|
if value == "":
|
|
return '""'
|
|
if re.search(r'[\s#"\'\\=]', value):
|
|
escaped = value.replace("\\", "\\\\").replace('"', '\\"')
|
|
return f'"{escaped}"'
|
|
return value
|
|
|
|
|
|
def update_env_vars(updates: dict[str, str], path: str | None = None) -> None:
|
|
"""更新或追加 KEY=value，不改动注释与其他配置项。"""
|
|
if not updates:
|
|
return
|
|
env_path = env_file_path(path)
|
|
lines: list[str] = []
|
|
if os.path.isfile(env_path):
|
|
with open(env_path, encoding="utf-8") as f:
|
|
lines = f.read().splitlines()
|
|
|
|
seen: set[str] = set()
|
|
out: list[str] = []
|
|
for line in lines:
|
|
stripped = line.strip()
|
|
if not stripped or stripped.startswith("#"):
|
|
out.append(line)
|
|
continue
|
|
m = _KEY_RE.match(stripped)
|
|
if m and m.group(1) in updates:
|
|
key = m.group(1)
|
|
out.append(f"{key}={_quote_env_value(updates[key])}")
|
|
seen.add(key)
|
|
else:
|
|
out.append(line)
|
|
|
|
for key, val in updates.items():
|
|
if key not in seen:
|
|
out.append(f"{key}={_quote_env_value(val)}")
|
|
|
|
with open(env_path, "w", encoding="utf-8", newline="\n") as f:
|
|
if out:
|
|
f.write("\n".join(out) + "\n")
|