Spaces:
Runtime error
Runtime error
File size: 4,095 Bytes
feef734 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 | """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
|