118 lines
3.4 KiB
Python
118 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from .paths import PROJECT_ROOT
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DbConfig:
|
|
host: str
|
|
port: str
|
|
user: str
|
|
password: str
|
|
name: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AppConfig:
|
|
env_path: Path | None
|
|
feishu_app_id: str
|
|
feishu_app_secret: str
|
|
feishu_base_url: str
|
|
feishu_app_token: str | None
|
|
analytics_db_host: str
|
|
analytics_db_port: str
|
|
analytics_db_user: str
|
|
analytics_db_password: str
|
|
analytics_db_name: str
|
|
|
|
|
|
def load_env_file(env_path: Path) -> dict[str, str]:
|
|
values: dict[str, str] = {}
|
|
if not env_path.exists():
|
|
return values
|
|
|
|
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)
|
|
values[key.strip()] = value.strip().strip("'").strip('"')
|
|
return values
|
|
|
|
|
|
def load_env_candidates() -> tuple[dict[str, str], Path | None]:
|
|
candidates = [
|
|
PROJECT_ROOT / ".env",
|
|
PROJECT_ROOT / "project (1)" / ".env",
|
|
PROJECT_ROOT.parent / ".env",
|
|
Path.cwd() / ".env",
|
|
]
|
|
seen: set[Path] = set()
|
|
for candidate in candidates:
|
|
candidate = candidate.resolve()
|
|
if candidate in seen:
|
|
continue
|
|
seen.add(candidate)
|
|
values = load_env_file(candidate)
|
|
if values:
|
|
return values, candidate
|
|
return {}, None
|
|
|
|
|
|
def get_config(config: dict[str, str], *keys: str) -> str | None:
|
|
for key in keys:
|
|
value = config.get(key) or os.getenv(key)
|
|
if value:
|
|
return value
|
|
return None
|
|
|
|
|
|
def require_config(config: dict[str, str], *keys: str) -> str:
|
|
value = get_config(config, *keys)
|
|
if value:
|
|
return value
|
|
raise KeyError(f"缺少必填配置: {' / '.join(keys)}")
|
|
|
|
|
|
def load_db_config(prefix: str) -> tuple[DbConfig, Path | None]:
|
|
config_map, env_path = load_env_candidates()
|
|
return (
|
|
DbConfig(
|
|
host=require_config(config_map, f"{prefix}_DB_HOST"),
|
|
port=require_config(config_map, f"{prefix}_DB_PORT"),
|
|
user=require_config(config_map, f"{prefix}_DB_USER"),
|
|
password=require_config(config_map, f"{prefix}_DB_PASSWORD"),
|
|
name=require_config(config_map, f"{prefix}_DB_NAME"),
|
|
),
|
|
env_path,
|
|
)
|
|
|
|
|
|
def load_raw_db_config() -> tuple[DbConfig, Path | None]:
|
|
return load_db_config("RAW")
|
|
|
|
|
|
def load_analytics_db_config() -> tuple[DbConfig, Path | None]:
|
|
return load_db_config("ANALYTICS")
|
|
|
|
|
|
def build_app_config() -> AppConfig:
|
|
config, env_path = load_env_candidates()
|
|
return AppConfig(
|
|
env_path=env_path,
|
|
feishu_app_id=require_config(config, "FEISHU_APP_ID", "APP_ID"),
|
|
feishu_app_secret=require_config(config, "FEISHU_APP_SECRET", "FEISHU_APP_SECRE", "APP_SECRET"),
|
|
feishu_base_url=require_config(config, "FEISHU_BASE_URL", "BASE_URL").rstrip("/"),
|
|
feishu_app_token=get_config(config, "FEISHU_APP_TOKEN"),
|
|
analytics_db_host=require_config(config, "ANALYTICS_DB_HOST"),
|
|
analytics_db_port=require_config(config, "ANALYTICS_DB_PORT"),
|
|
analytics_db_user=require_config(config, "ANALYTICS_DB_USER"),
|
|
analytics_db_password=require_config(config, "ANALYTICS_DB_PASSWORD"),
|
|
analytics_db_name=require_config(config, "ANALYTICS_DB_NAME"),
|
|
)
|
|
|