icons, main .env

app/env_file.py

@@ -0,0 +1,142 @@
+"""Bezpečné čtení a zápis hlavního AppFactory env souboru (/opt/appfactory/config/appfactory.env).
+
+Zdroj pravdy zůstává soubor na disku — nikdy se neukládá do DB. Modul pracuje výhradně
+s pevnou cestou APPFACTORY_ENV; žádná jiná cesta není povolená.
+"""
+
+import os
+import re
+import shutil
+from datetime import datetime
+
+from app.config import APPFACTORY_ENV
+
+ENV_PATH = APPFACTORY_ENV
+
+# Povolený tvar klíče (shell env konvence).
+KEY_RE = re.compile(r"^[A-Z_][A-Z0-9_]*$")
+
+# Hodnota, kterou lze zapsat bez uvozovek (žádné mezery ani shell-speciální znaky).
+_SIMPLE_VALUE_RE = re.compile(r"^[A-Za-z0-9_./:@%+,=-]*$")
+
+
+def _parse_line(line: str):
+    """Vrátí (key, value) pro řádek typu KEY=value, jinak None (komentář/prázdné/neparsovatelné)."""
+    stripped = line.strip()
+    if not stripped or stripped.startswith("#") or "=" not in line:
+        return None
+    key, _, raw_value = line.partition("=")
+    key = key.strip()
+    if not KEY_RE.match(key):
+        return None
+    return key, _parse_value(raw_value)
+
+
+def _unescape_double(value: str) -> str:
+    out = []
+    i = 0
+    while i < len(value):
+        char = value[i]
+        if char == "\\" and i + 1 < len(value) and value[i + 1] in '"\\$`':
+            out.append(value[i + 1])
+            i += 2
+        else:
+            out.append(char)
+            i += 1
+    return "".join(out)
+
+
+def _parse_value(raw: str) -> str:
+    value = raw.strip()
+    if len(value) >= 2 and value[0] == '"' and value[-1] == '"':
+        return _unescape_double(value[1:-1])
+    if len(value) >= 2 and value[0] == "'" and value[-1] == "'":
+        return value[1:-1]
+    return value
+
+
+def format_value(value: str) -> str:
+    """Naformátuje hodnotu pro shell env soubor — prosté hodnoty bez uvozovek,
+    hodnoty s mezerami/speciálními znaky v uvozovkách s bezpečným escapem."""
+    if value == "":
+        return ""
+    if _SIMPLE_VALUE_RE.match(value):
+        return value
+    escaped = (
+        value.replace("\\", "\\\\")
+        .replace('"', '\\"')
+        .replace("$", "\\$")
+        .replace("`", "\\`")
+    )
+    return f'"{escaped}"'
+
+
+def read_entries() -> list[tuple[str, str]]:
+    """Načte uspořádaný seznam (key, value) ze souboru. Duplicitní klíče bere jen poprvé."""
+    try:
+        with open(ENV_PATH, "r", encoding="utf-8") as handle:
+            lines = handle.readlines()
+    except FileNotFoundError:
+        return []
+
+    entries: list[tuple[str, str]] = []
+    seen: set[str] = set()
+    for line in lines:
+        parsed = _parse_line(line)
+        if not parsed:
+            continue
+        key, value = parsed
+        if key in seen:
+            continue
+        seen.add(key)
+        entries.append((key, value))
+    return entries
+
+
+def save_entries(desired_pairs: list[tuple[str, str]]) -> str | None:
+    """Zapíše nový obsah souboru z požadovaného uspořádaného seznamu (key, value).
+
+    Zachová komentáře, prázdné řádky a původní pořadí klíčů. Smazané klíče vypustí,
+    nové přidá na konec. Před zápisem vytvoří časově označený backup. Vrací cestu backupu.
+    """
+    desired_map = dict(desired_pairs)
+    desired_order = [key for key, _ in desired_pairs]
+
+    try:
+        with open(ENV_PATH, "r", encoding="utf-8") as handle:
+            original_lines = handle.readlines()
+    except FileNotFoundError:
+        original_lines = []
+
+    out_lines: list[str] = []
+    emitted: set[str] = set()
+    for line in original_lines:
+        parsed = _parse_line(line)
+        if not parsed:
+            out_lines.append(line.rstrip("\n"))
+            continue
+        key, _value = parsed
+        if key not in desired_map or key in emitted:
+            continue  # smazaný klíč nebo duplicitní původní řádek
+        out_lines.append(f"{key}={format_value(desired_map[key])}")
+        emitted.add(key)
+
+    for key in desired_order:
+        if key not in emitted:
+            out_lines.append(f"{key}={format_value(desired_map[key])}")
+            emitted.add(key)
+
+    content = "\n".join(out_lines) + "\n"
+
+    backup_path = None
+    if os.path.exists(ENV_PATH):
+        timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
+        backup_path = f"{ENV_PATH}.bak-{timestamp}"
+        shutil.copy2(ENV_PATH, backup_path)
+
+    tmp_path = f"{ENV_PATH}.tmp-{os.getpid()}"
+    with open(tmp_path, "w", encoding="utf-8", newline="\n") as handle:
+        handle.write(content)
+    os.replace(tmp_path, ENV_PATH)
+
+    return backup_path