Spaces:
Runtime error
Runtime error
| """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}") | |