85 lines
2.6 KiB
Python
85 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
import tomllib
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
def load_env(path: Path | None = None) -> None:
|
|
env_path = path or ROOT / ".env"
|
|
if not env_path.exists():
|
|
return
|
|
for raw in env_path.read_text(encoding="utf-8-sig").splitlines():
|
|
line = raw.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
key, value = line.split("=", 1)
|
|
os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'"))
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ClientConfig:
|
|
domain: str
|
|
google_ads_customer_id: str
|
|
adspro_client_id: str | None = None
|
|
skip_in_all: tuple[str, ...] = ()
|
|
rules: dict | None = None
|
|
|
|
@property
|
|
def safe_customer_id(self) -> str:
|
|
return re.sub(r"\D", "", self.google_ads_customer_id)
|
|
|
|
def effective_rules(self, global_rules: dict, section: str) -> dict:
|
|
rules = dict(global_rules.get(section, {}))
|
|
rules.update((self.rules or {}).get(section, {}))
|
|
return rules
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AppConfig:
|
|
clients: dict[str, ClientConfig]
|
|
global_rules: dict
|
|
|
|
|
|
def _string_tuple(value) -> tuple[str, ...]:
|
|
if value is None:
|
|
return ()
|
|
if isinstance(value, str):
|
|
values = [value]
|
|
elif isinstance(value, (list, tuple, set)):
|
|
values = value
|
|
else:
|
|
return ()
|
|
return tuple(str(item).strip() for item in values if str(item).strip())
|
|
|
|
|
|
def load_config(path: Path | None = None) -> AppConfig:
|
|
config_path = path or ROOT / "config" / "clients.toml"
|
|
if not config_path.exists():
|
|
example = ROOT / "config" / "clients.example.toml"
|
|
raise FileNotFoundError(
|
|
f"Brak {config_path}. Utworz ten plik na podstawie {example}."
|
|
)
|
|
data = tomllib.loads(config_path.read_text(encoding="utf-8"))
|
|
clients = {}
|
|
for domain, row in data.get("clients", {}).items():
|
|
clients[domain] = ClientConfig(
|
|
domain=domain,
|
|
google_ads_customer_id=str(row["google_ads_customer_id"]),
|
|
adspro_client_id=str(row.get("adspro_client_id")) if row.get("adspro_client_id") else None,
|
|
skip_in_all=_string_tuple(row.get("skip_in_all")),
|
|
rules={key: value for key, value in row.items() if isinstance(value, dict)},
|
|
)
|
|
return AppConfig(clients=clients, global_rules=data.get("global_rules", {}))
|
|
|
|
|
|
def client_dir(domain: str) -> Path:
|
|
path = ROOT / "clients" / domain
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
return path
|