miband-bot/miband_tracker/config.py

# SPDX-License-Identifier: GPL-3.0-or-later
# Copyright (C) 2026 Alexey

from __future__ import annotations

import os
from dataclasses import dataclass
from pathlib import Path


def _load_local_env(filename: str = "secrets.env") -> None:
    """Load KEY=VALUE pairs from a local .env file into os.environ.

    Only sets variables that are NOT already present in the environment
    (explicit env always wins). Skips blank lines and comments (#).
    """
    env_file = Path(filename)
    if not env_file.exists():
        return
    try:
        for raw_line in env_file.read_text(encoding="utf-8").splitlines():
            line = raw_line.strip()
            if not line or line.startswith("#") or "=" not in line:
                continue
            key, _, value = line.partition("=")
            key = key.strip()
            if key and key not in os.environ:
                os.environ[key] = value.strip()
    except Exception:
        pass  # Never crash on env-file read failure


class ConfigError(ValueError):
    """Invalid runtime configuration."""


def _env_bool(name: str, default: bool = False) -> bool:
    value = os.environ.get(name)
    if value is None:
        return default
    return value.strip().lower() in {"1", "true", "yes", "on"}


def _env_int(name: str, default: int, *, min_value: int | None = None) -> int:
    raw = os.environ.get(name, str(default)).strip()
    try:
        value = int(raw)
    except ValueError as exc:
        raise ConfigError(f"{name} должен быть целым числом") from exc
    if min_value is not None and value < min_value:
        raise ConfigError(f"{name} должен быть не меньше {min_value}")
    return value


@dataclass(frozen=True)
class Settings:
    data_dir: Path
    db_path: Path
    status_path: Path
    bot_state_db_path: Path
    telegram_bot_token: str
    telegram_allowed_user_id: int | None
    sync_interval: int
    query_duration: int
    enable_fds_sleep_details: bool

    @classmethod
    def from_env(cls, *, require_bot: bool = False) -> Settings:
        # Load local secrets.env (if present) before reading env vars.
        # This makes Python the single source of truth on all platforms.
        _load_local_env()
        # Auto-detect local vs Docker mode: if DATA_DIR is not set, use ./data
        # when running locally (secrets.env present or ./data already exists),
        # otherwise fall back to the Docker default /opt/miband-tracker/data.
        _default_data = (
            "./data"
            if (Path("secrets.env").exists() or Path("data").is_dir())
            else "/opt/miband-tracker/data"
        )
        data_dir = Path(os.environ.get("DATA_DIR", _default_data))
        allowed_user_id = parse_single_user_id(
            os.environ.get("TELEGRAM_ALLOWED_USER_ID", ""), required=False
        )
        # Если ID не задан в env, пробуем загрузить из файла allowed_user.id
        allowed_user_file = data_dir / "allowed_user.id"
        if allowed_user_id is None and allowed_user_file.exists():
            try:
                allowed_user_id = int(allowed_user_file.read_text(encoding="utf-8").strip())
            except Exception:
                pass

        bot_token = os.environ.get("TELEGRAM_BOT_TOKEN", "").strip()
        if require_bot and not bot_token:
            raise ConfigError("TELEGRAM_BOT_TOKEN не задан")
        return cls(
            data_dir=data_dir,
            db_path=Path(os.environ.get("DB_PATH", str(data_dir / "miband.db"))),
            status_path=Path(os.environ.get("STATUS_PATH", str(data_dir / "status.json"))),
            bot_state_db_path=Path(
                os.environ.get(
                    "BOT_STATE_DB_PATH",
                    str(data_dir / "fitness_bot_state.db"),
                )
            ),
            telegram_bot_token=bot_token,
            telegram_allowed_user_id=allowed_user_id,
            sync_interval=_env_int("SYNC_INTERVAL", 900, min_value=0),
            query_duration=_env_int("QUERY_DURATION", 2, min_value=1),
            enable_fds_sleep_details=_env_bool("ENABLE_FDS_SLEEP_DETAILS", default=True),
        )

    def require_user_id(self, user_id: int | None = None) -> int:
        resolved = user_id if user_id is not None else self.telegram_allowed_user_id
        if resolved is None:
            raise ConfigError("TELEGRAM_ALLOWED_USER_ID должен содержать ровно один user id")
        return int(resolved)

    def token_path(self, user_id: int | None = None) -> Path:
        uid = self.require_user_id(user_id)
        preferred = self.data_dir / f"token_{uid}.json"
        legacy = self.data_dir / "token.json"
        if preferred.exists() or not legacy.exists():
            return preferred
        return legacy

    def user_db_path(self, user_id: int | None = None) -> Path:
        uid = self.require_user_id(user_id)
        preferred = self.data_dir / f"miband_{uid}.db"
        if preferred.exists():
            return preferred
        return self.db_path

    def user_status_path(self, user_id: int | None = None) -> Path:
        uid = self.require_user_id(user_id)
        preferred = self.data_dir / f"status_{uid}.json"
        if preferred.exists():
            return preferred
        return self.status_path

    def canonical_user_db_path(self, user_id: int | None = None) -> Path:
        return self.data_dir / f"miband_{self.require_user_id(user_id)}.db"

    def canonical_user_status_path(self, user_id: int | None = None) -> Path:
        return self.data_dir / f"status_{self.require_user_id(user_id)}.json"


def parse_single_user_id(raw: str, *, required: bool = False) -> int | None:
    values = [item.strip() for item in raw.split(",") if item.strip()]
    if not values:
        if required:
            raise ConfigError("TELEGRAM_ALLOWED_USER_ID не задан или пуст")
        return None
    if len(values) > 1:
        raise ConfigError("TELEGRAM_ALLOWED_USER_ID должен содержать ровно один user id")
    try:
        return int(values[0])
    except ValueError as exc:
        raise ConfigError("TELEGRAM_ALLOWED_USER_ID должен быть целым числом") from exc