Initial standalone crypto_okx with one-click deploy.

Add deploy/manage.sh bootstrap for git.bz121.com/dekun/crypto_okx and point docs at this repo.

Co-authored-by: Cursor <cursoragent@cursor.com>

lib/env/env_file_lib.py

@@ -0,0 +1,121 @@
+"""读写实例目录 .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