Update intranet tools and voucher comparison
This commit is contained in:
@@ -0,0 +1,92 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sqlite3
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
try:
|
||||
from dotenv import load_dotenv
|
||||
except ImportError: # pragma: no cover - keeps admin helpers usable before dependencies install.
|
||||
load_dotenv = None
|
||||
|
||||
|
||||
BASE_DIR = Path(__file__).resolve().parent
|
||||
MIN_SAFE_SQLITE_VERSION = (3, 51, 3)
|
||||
|
||||
|
||||
def load_runtime_env() -> None:
|
||||
env_path = BASE_DIR / ".env"
|
||||
if not env_path.exists():
|
||||
return
|
||||
if load_dotenv is not None:
|
||||
load_dotenv(env_path, override=False)
|
||||
return
|
||||
for raw_line in env_path.read_text(encoding="utf-8").splitlines():
|
||||
line = raw_line.strip()
|
||||
if not line or line.startswith("#") or "=" not in line:
|
||||
continue
|
||||
key, value = line.split("=", 1)
|
||||
key = key.strip()
|
||||
if not key or key in os.environ:
|
||||
continue
|
||||
value = value.strip().strip('"').strip("'")
|
||||
os.environ[key] = value
|
||||
|
||||
|
||||
load_runtime_env()
|
||||
|
||||
|
||||
def _path_from_env(name: str, default: Path) -> Path:
|
||||
raw = str(os.environ.get(name, "") or "").strip()
|
||||
return Path(raw).expanduser().resolve() if raw else default.resolve()
|
||||
|
||||
|
||||
def _env_flag(name: str, default: bool = False) -> bool:
|
||||
raw = str(os.environ.get(name, "") or "").strip().lower()
|
||||
if not raw:
|
||||
return default
|
||||
return raw in {"1", "true", "yes", "y", "on"}
|
||||
|
||||
|
||||
DB_PATH = _path_from_env("INTRANET_DB_PATH", BASE_DIR / "data.db")
|
||||
BACKUP_DIR = _path_from_env("INTRANET_BACKUP_DIR", BASE_DIR / "backups")
|
||||
HANMAC_EXPORT_DIR = _path_from_env("INTRANET_HANMAC_EXPORT_DIR", Path("/tmp") / "my_intranet_hanmac_exports")
|
||||
WEHAGO_COMPARE_EXPORT_ROOT = _path_from_env(
|
||||
"INTRANET_COMPARE_EXPORT_DIR",
|
||||
BASE_DIR / "static" / "exports" / "wehago_compare",
|
||||
)
|
||||
WEHAGO_SOURCE_ROOT = _path_from_env("WEHAGO_SOURCE_ROOT", BASE_DIR.parent / "WEHAGO_DB")
|
||||
CACHE_ROOT = _path_from_env("INTRANET_CACHE_ROOT", BASE_DIR / "runtime_cache")
|
||||
REQUIRE_SAFE_SQLITE = _env_flag("INTRANET_REQUIRE_SAFE_SQLITE", default=False)
|
||||
WAL_WARN_BYTES = int(os.environ.get("INTRANET_WAL_WARN_BYTES", str(256 * 1024 * 1024)))
|
||||
WAL_BLOCK_HEAVY_BYTES = int(os.environ.get("INTRANET_WAL_BLOCK_HEAVY_BYTES", str(512 * 1024 * 1024)))
|
||||
|
||||
|
||||
def sqlite_runtime_status() -> dict[str, Any]:
|
||||
version = tuple(int(part) for part in sqlite3.sqlite_version.split("."))
|
||||
return {
|
||||
"sqlite_version": sqlite3.sqlite_version,
|
||||
"minimum_safe_version": ".".join(str(part) for part in MIN_SAFE_SQLITE_VERSION),
|
||||
"meets_minimum_safe_version": version >= MIN_SAFE_SQLITE_VERSION,
|
||||
"safe_version_required": REQUIRE_SAFE_SQLITE,
|
||||
}
|
||||
|
||||
|
||||
def validate_sqlite_runtime() -> dict[str, Any]:
|
||||
status = sqlite_runtime_status()
|
||||
if REQUIRE_SAFE_SQLITE and not status["meets_minimum_safe_version"]:
|
||||
raise RuntimeError(
|
||||
"SQLite 3.51.3 이상이 필요합니다. "
|
||||
f"현재 런타임은 {status['sqlite_version']}입니다. "
|
||||
"Docker 안전 런타임 또는 최신 SQLite가 연결된 Python으로 실행하세요."
|
||||
)
|
||||
return status
|
||||
|
||||
|
||||
def ensure_runtime_directories() -> None:
|
||||
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
|
||||
BACKUP_DIR.mkdir(parents=True, exist_ok=True)
|
||||
HANMAC_EXPORT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
WEHAGO_COMPARE_EXPORT_ROOT.mkdir(parents=True, exist_ok=True)
|
||||
CACHE_ROOT.mkdir(parents=True, exist_ok=True)
|
||||
Reference in New Issue
Block a user