Spaces:
Runtime error
Runtime error
File size: 2,883 Bytes
896124d | 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 | """Telegram утилиты: отправка сообщений и файлов"""
import os
import json
import urllib.request
import time
import requests
from .config import CF_URL, TOKEN
def send_tg(chat_id, text):
if not CF_URL or not TOKEN:
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": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"}
req = urllib.request.Request(url, data=data, headers=headers)
try:
urllib.request.urlopen(req, timeout=15)
except Exception as e:
print(f"TG send error: {e}")
def download_tg_file(file_id, file_name, retries=3):
if not TOKEN:
return None
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
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:
res = requests.get(get_file_url, timeout=15, verify=False).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 = requests.get(dl_url, stream=True, timeout=60, verify=False)
if response.status_code != 200:
direct_url = f"https://api.telegram.org/file/bot{TOKEN}/{file_path}"
response = requests.get(direct_url, stream=True, timeout=60, verify=False)
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:
time.sleep(2)
return None
def send_tg_file(chat_id, file_path):
if not os.path.exists(file_path):
send_tg(chat_id, f"File not found: {file_path}")
return
base_api = CF_URL if CF_URL else "https://api.telegram.org"
url = f"{base_api}/bot{TOKEN}/sendDocument"
send_tg(chat_id, f"Sending file: {os.path.basename(file_path)}...")
try:
with open(file_path, 'rb') as f:
res = requests.post(url, data={'chat_id': chat_id}, files={'document': f}, verify=False, timeout=60)
if res.status_code == 200:
print(f"File {file_path} sent to TG!")
else:
send_tg(chat_id, "Failed to send file.")
except Exception as e:
send_tg(chat_id, f"Send error: {e}")
|