server/telegram_utils.py

"""Telegram утилиты: отправка сообщений и файлов"""
import os
import json
import urllib.request
import time
import logging
import certifi
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from .config import CF_URL, TOKEN

class TelegramUtils:
    def __init__(self):
        self.logger = logging.getLogger(__name__)
        self.session = requests.Session()
        retry = Retry(
            total=3,
            backoff_factor=1,
            status_forcelist=[500, 502, 503, 504]
        )
        self.session.mount("https://", HTTPAdapter(max_retries=retry))

    def send_tg(self, chat_id, text):
        if not CF_URL or not TOKEN:
            self.logger.warning("TG send aborted: CF_URL or TOKEN not set")
            return
        safe_text = str(text)[-4000:]
        url = f"{CF_URL}/bot{TOKEN}/sendMessage"
        data = json.dumps({
            "chat_id": chat_id,
            "text": safe_text,
            "parse_mode": "Markdown"
        }).encode('utf-8')
        headers = {
            "Content-Type": "application/json",
            "User-Agent": "PinkSky/7.0 (Linux; Python)"
        }
        try:
            response = self.session.post(url, data=data, headers=headers, timeout=15, verify=certifi.where())
            response.raise_for_status()
        except Exception as e:
            self.logger.error(f"TG send error: {e}")

    def download_tg_file(self, file_id, file_name, retries=3):
        if not TOKEN:
            self.logger.warning("TG download aborted: TOKEN not set")
            return None
        base_api = CF_URL if CF_URL else "https://api.telegram.org"
        get_file_url = f"{base_api}/bot{TOKEN}/getFile?file_id={file_id}"
        for attempt in range(retries):
            try:
                response = self.session.get(get_file_url, timeout=15, verify=certifi.where())
                response.raise_for_status()
                res = response.json()
                if not res.get("ok"):
                    return None
                file_path = res["result"]["file_path"]
                dl_url = f"{base_api}/file/bot{TOKEN}/{file_path}"
                os.makedirs("downloads", exist_ok=True)
                safe_name = os.path.basename(file_name)
                local_path = os.path.join("downloads", safe_name)

                response = self.session.get(dl_url, stream=True, timeout=60, verify=certifi.where())
                response.raise_for_status()
                with open(local_path, "wb") as f:
                    for chunk in response.iter_content(chunk_size=8192):
                        if chunk:
                            f.write(chunk)
                return local_path
            except Exception as e:
                self.logger.warning(f"TG download attempt {attempt + 1} failed: {e}")
                time.sleep(2)
        return None

    def send_tg_file(self, chat_id, file_path):
        if not os.path.exists(file_path):
            self.send_tg(chat_id, f"▸ ✗ ᴇʀʀᴏʀ\nFile not found: `{file_path}`")
            return
        base_api = CF_URL if CF_URL else "https://api.telegram.org"
        url = f"{base_api}/bot{TOKEN}/sendDocument"
        self.send_tg(chat_id, f"╭───────────────────╮\n │ 📤 ꜰɪʟᴇ ᴜᴘʟᴏᴀᴅɪɴɢ │\n ├───────────────────┤\n │ `{os.path.basename(file_path)}` │\n╰───────────────────╯")
        try:
            with open(file_path, 'rb') as f:
                files = {'document': f}
                response = self.session.post(url, data={'chat_id': chat_id}, files=files, timeout=60, verify=certifi.where())
                response.raise_for_status()
        except Exception as e:
            self.send_tg(chat_id, f"▸ ⚠️ ꜰɪʟᴇ ɪssᴜᴇ\n{str(e)[:200]}")

# Singleton instance
TG_UTILS = TelegramUtils()
send_tg = TG_UTILS.send_tg
download_tg_file = TG_UTILS.download_tg_file
send_tg_file = TG_UTILS.send_tg_file