cm_bot_v2/app/telegram_notifier.py

import logging
import os
from pathlib import Path
from typing import Optional

import requests


logger = logging.getLogger(__name__)

TELEGRAM_API_URL = "https://api.telegram.org"


def _load_env_from_file() -> None:
    """
    Load key=value pairs from a .env-style file to populate os.environ.
    Existing environment variables take precedence.
    """
    candidate_paths = []
    env_file = os.getenv("ENV_FILE_PATH")
    if env_file:
        candidate_paths.append(Path(env_file))
    candidate_paths.append(Path(__file__).resolve().parents[1] / ".env")
    candidate_paths.append(Path.cwd() / ".env")

    for path in candidate_paths:
        if not path:
            continue
        if not path.exists() or not path.is_file():
            continue

        try:
            with path.open("r") as env_file_handle:
                for line in env_file_handle:
                    stripped = line.strip()
                    if not stripped or stripped.startswith("#") or "=" not in stripped:
                        continue
                    key, value = stripped.split("=", 1)
                    key = key.strip()
                    value = value.strip().strip('"').strip("'")
                    if key and key not in os.environ:
                        os.environ[key] = value
        except OSError as exc:
            logger.warning("Unable to load env file %s: %s", path, exc)


_load_env_from_file()


def get_telegram_bot_token() -> str:
    """
    Fetch the Telegram bot token from the TELEGRAM_BOT_TOKEN environment variable.
    Raises RuntimeError if missing so the application fails fast during startup.
    """
    token = os.getenv("TELEGRAM_BOT_TOKEN")
    if not token:
        _load_env_from_file()
        token = os.getenv("TELEGRAM_BOT_TOKEN")
    if not token:
        raise RuntimeError("Missing TELEGRAM_BOT_TOKEN environment variable")
    return token


class TelegramNotifier:
    """
    Utility wrapper to send proactive notifications to a dedicated Telegram chat.
    Requires TELEGRAM_ALERT_CHAT_ID and TELEGRAM_ALERT_BOT_TOKEN (or TELEGRAM_BOT_TOKEN) env vars.
    """

    def __init__(self):
        self.chat_id = os.getenv("TELEGRAM_ALERT_CHAT_ID")
        self.bot_token = os.getenv("TELEGRAM_ALERT_BOT_TOKEN") or os.getenv("TELEGRAM_BOT_TOKEN")
        if not self.chat_id:
            logger.warning(
                "TELEGRAM_ALERT_CHAT_ID environment variable missing; skipping Telegram notification setup"
            )
        if not self.bot_token:
            logger.warning(
                "TELEGRAM_ALERT_BOT_TOKEN/TELEGRAM_BOT_TOKEN env var missing; skipping Telegram notification setup"
            )

    def _can_notify(self) -> bool:
        return bool(self.chat_id and self.bot_token)

    def _send_message(self, text: str, parse_mode: Optional[str] = None) -> None:
        if not self._can_notify():
            logger.warning(
                "Skipping Telegram notification: bot_token/chat_id not configured. Message: %s",
                text,
            )
            return

        url = f"{TELEGRAM_API_URL}/bot{self.bot_token}/sendMessage"
        payload = {"chat_id": self.chat_id, "text": text}
        if parse_mode:
            payload["parse_mode"] = parse_mode

        try:
            response = requests.post(url, json=payload, timeout=10)
            if response.status_code != 200:
                logger.error(
                    "Failed to send Telegram notification. Status: %s Body: %s",
                    response.status_code,
                    response.text,
                )
        except requests.RequestException as exc:
            logger.exception("Error sending Telegram notification: %s", exc)

    def notify_login_failure(self, details: str) -> None:
        message = f"⚠️ CM Telegram Bot login issue detected.\nDetails: {details}\nAutomatic account creation paused until the next retry window."
        self._send_message(message)

    def notify_generic_error(self, details: str) -> None:
        message = f"⚠️ CM Telegram Bot encountered an issue:\n{details}"
        self._send_message(message)