Add multi-account command ACL system
This commit is contained in:
Codex
205
app/config.py
Normal file
205
app/config.py
Normal file
@@ -0,0 +1,205 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from dotenv import load_dotenv
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class AccountConfig:
|
||||
name: str
|
||||
session: str
|
||||
api_id: int | None = None
|
||||
api_hash: str | None = None
|
||||
session_string: str | None = None
|
||||
phone_number: str | None = None
|
||||
password: str | None = None
|
||||
phone_code: str | None = None
|
||||
bot_token: str | None = None
|
||||
proxy: dict[str, Any] | str | None = None
|
||||
workdir: Path | None = None
|
||||
in_memory: bool = False
|
||||
no_updates: bool = False
|
||||
takeout: bool = False
|
||||
hide_password: bool = False
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class AppConfig:
|
||||
api_id: int | None
|
||||
api_hash: str | None
|
||||
session_dir: Path
|
||||
accounts: list[AccountConfig]
|
||||
|
||||
|
||||
def _read_json_file(path: Path) -> dict[str, Any]:
|
||||
return json.loads(path.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def _parse_int(value: Any, *, field_name: str) -> int:
|
||||
try:
|
||||
return int(value)
|
||||
except (TypeError, ValueError) as exc:
|
||||
raise ValueError(f"{field_name} must be an integer") from exc
|
||||
|
||||
|
||||
def _parse_accounts(raw_accounts: Any, *, default_api_id: int | None, default_api_hash: str | None, default_session_dir: Path) -> list[AccountConfig]:
|
||||
if raw_accounts is None:
|
||||
return []
|
||||
|
||||
if not isinstance(raw_accounts, list):
|
||||
raise ValueError("accounts must be a list")
|
||||
|
||||
accounts: list[AccountConfig] = []
|
||||
for index, raw_account in enumerate(raw_accounts, start=1):
|
||||
if isinstance(raw_account, str):
|
||||
raw_account = {"name": raw_account, "session": raw_account}
|
||||
if not isinstance(raw_account, dict):
|
||||
raise ValueError(f"accounts[{index}] must be a string or object")
|
||||
|
||||
name = str(
|
||||
raw_account.get("name")
|
||||
or raw_account.get("label")
|
||||
or raw_account.get("session")
|
||||
or raw_account.get("session_name")
|
||||
or ""
|
||||
).strip()
|
||||
session = str(raw_account.get("session") or raw_account.get("session_name") or name).strip()
|
||||
|
||||
if not name:
|
||||
raise ValueError(f"accounts[{index}] is missing name")
|
||||
if not session:
|
||||
raise ValueError(f"accounts[{index}] is missing session")
|
||||
|
||||
account_api_id = raw_account.get("api_id", default_api_id)
|
||||
account_api_hash = raw_account.get("api_hash", default_api_hash)
|
||||
|
||||
if account_api_id is not None:
|
||||
account_api_id = _parse_int(account_api_id, field_name=f"accounts[{index}].api_id")
|
||||
if account_api_hash is not None:
|
||||
account_api_hash = str(account_api_hash)
|
||||
|
||||
account_workdir = raw_account.get("workdir")
|
||||
if account_workdir is not None:
|
||||
account_workdir = Path(str(account_workdir)).expanduser()
|
||||
if not account_workdir.is_absolute():
|
||||
account_workdir = (default_session_dir.parent / account_workdir).resolve()
|
||||
|
||||
accounts.append(
|
||||
AccountConfig(
|
||||
name=name,
|
||||
session=session,
|
||||
api_id=account_api_id,
|
||||
api_hash=account_api_hash,
|
||||
session_string=raw_account.get("session_string"),
|
||||
phone_number=raw_account.get("phone_number"),
|
||||
password=raw_account.get("password"),
|
||||
phone_code=raw_account.get("phone_code"),
|
||||
bot_token=raw_account.get("bot_token"),
|
||||
proxy=raw_account.get("proxy"),
|
||||
workdir=account_workdir,
|
||||
in_memory=bool(raw_account.get("in_memory", False) or raw_account.get("session_string")),
|
||||
no_updates=bool(raw_account.get("no_updates", False)),
|
||||
takeout=bool(raw_account.get("takeout", False)),
|
||||
hide_password=bool(raw_account.get("hide_password", False)),
|
||||
)
|
||||
)
|
||||
|
||||
return accounts
|
||||
|
||||
|
||||
def _accounts_from_env() -> list[dict[str, Any]] | None:
|
||||
raw_accounts = os.environ.get("ACCOUNTS_JSON")
|
||||
if raw_accounts:
|
||||
parsed = json.loads(raw_accounts)
|
||||
if not isinstance(parsed, list):
|
||||
raise ValueError("ACCOUNTS_JSON must contain a JSON list")
|
||||
return parsed
|
||||
|
||||
raw_sessions = os.environ.get("ACCOUNT_SESSIONS")
|
||||
if raw_sessions:
|
||||
sessions = [item.strip() for item in raw_sessions.split(",") if item.strip()]
|
||||
return [{"name": session, "session": session} for session in sessions]
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def load_app_config(root: Path) -> AppConfig:
|
||||
dotenv_path = os.environ.get("BOT_DOTENV_PATH")
|
||||
if dotenv_path:
|
||||
candidate_dotenv = Path(dotenv_path).expanduser()
|
||||
if not candidate_dotenv.is_absolute():
|
||||
candidate_dotenv = (root / candidate_dotenv).resolve()
|
||||
else:
|
||||
candidate_dotenv = root / ".env"
|
||||
|
||||
load_dotenv(candidate_dotenv, override=False)
|
||||
|
||||
config_path = os.environ.get("BOT_CONFIG")
|
||||
if config_path:
|
||||
candidate = Path(config_path).expanduser()
|
||||
if not candidate.is_absolute():
|
||||
candidate = (root / candidate).resolve()
|
||||
else:
|
||||
candidate = root / "config.json"
|
||||
|
||||
raw: dict[str, Any] = {}
|
||||
if candidate.exists():
|
||||
raw = _read_json_file(candidate)
|
||||
|
||||
env_api_id = os.environ.get("TG_API_ID")
|
||||
env_api_hash = os.environ.get("TG_API_HASH")
|
||||
env_session_dir = os.environ.get("SESSION_DIR")
|
||||
|
||||
if env_api_id is not None:
|
||||
raw["api_id"] = env_api_id
|
||||
if env_api_hash is not None:
|
||||
raw["api_hash"] = env_api_hash
|
||||
if env_session_dir is not None:
|
||||
raw["session_dir"] = env_session_dir
|
||||
|
||||
env_accounts = _accounts_from_env()
|
||||
if env_accounts is not None:
|
||||
raw["accounts"] = env_accounts
|
||||
|
||||
if "accounts" not in raw:
|
||||
raise ValueError(
|
||||
"Missing accounts configuration. Provide accounts in config.json or set ACCOUNTS_JSON / ACCOUNT_SESSIONS."
|
||||
)
|
||||
|
||||
api_id = _parse_int(raw["api_id"], field_name="api_id") if "api_id" in raw else None
|
||||
api_hash = str(raw["api_hash"]) if "api_hash" in raw else None
|
||||
session_dir = Path(str(raw.get("session_dir", "sessions"))).expanduser()
|
||||
if not session_dir.is_absolute():
|
||||
session_dir = (root / session_dir).resolve()
|
||||
|
||||
accounts = _parse_accounts(
|
||||
raw["accounts"],
|
||||
default_api_id=api_id,
|
||||
default_api_hash=api_hash,
|
||||
default_session_dir=session_dir,
|
||||
)
|
||||
|
||||
if not accounts:
|
||||
raise ValueError("At least one account must be configured")
|
||||
|
||||
seen_names: set[str] = set()
|
||||
seen_sessions: set[str] = set()
|
||||
for account in accounts:
|
||||
if account.name in seen_names:
|
||||
raise ValueError(f"Duplicate account name: {account.name}")
|
||||
if account.session in seen_sessions:
|
||||
raise ValueError(f"Duplicate session name: {account.session}")
|
||||
seen_names.add(account.name)
|
||||
seen_sessions.add(account.session)
|
||||
|
||||
return AppConfig(
|
||||
api_id=api_id,
|
||||
api_hash=api_hash,
|
||||
session_dir=session_dir,
|
||||
accounts=accounts,
|
||||
)
|
||||
Reference in New Issue
Block a user