99 lines
3.5 KiB
Python
99 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Пример скрипта для отправки сообщений через Telegram бота.
|
|
|
|
Использование (из корня проекта):
|
|
python tools/dev/send_telegram_example.py <user_id> <message>
|
|
"""
|
|
|
|
import sys
|
|
import requests
|
|
import json
|
|
import os
|
|
|
|
from dotenv import load_dotenv
|
|
|
|
_PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
|
|
load_dotenv(os.path.join(_PROJECT_ROOT, ".env"))
|
|
load_dotenv()
|
|
|
|
def send_message(user_id, message_text, bot_url=None, api_token=None):
|
|
"""
|
|
Отправляет сообщение пользователю через Telegram бота
|
|
|
|
Args:
|
|
user_id (int): ID пользователя в Telegram
|
|
message_text (str): Текст сообщения
|
|
bot_url (str): URL бота (если не указан, берется из переменных окружения)
|
|
api_token (str): API токен для аутентификации (если не указан, берется из переменных окружения)
|
|
|
|
Returns:
|
|
tuple: (success: bool, message: str)
|
|
"""
|
|
if bot_url is None:
|
|
# Пытаемся получить URL из переменных окружения
|
|
webhook_url = os.getenv('WEBHOOK_URL')
|
|
webhook_port = os.getenv('WEBHOOK_PORT', '5005')
|
|
|
|
if webhook_url:
|
|
# Используем внешний URL через reverse proxy
|
|
bot_url = f"{webhook_url}/send_message"
|
|
else:
|
|
bot_url = f"http://localhost:{webhook_port}/send_message"
|
|
|
|
if api_token is None:
|
|
api_token = os.getenv('WEBHOOK_SECRET')
|
|
|
|
try:
|
|
url = f"{bot_url}/send_message"
|
|
data = {
|
|
"user_id": user_id,
|
|
"message": message_text
|
|
}
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'X-API-Secret-Token': api_token
|
|
}
|
|
|
|
response = requests.post(url, json=data, headers=headers, timeout=10)
|
|
result = response.json()
|
|
|
|
if response.status_code == 200 and result.get("success"):
|
|
return True, result.get("message", "Сообщение отправлено успешно")
|
|
else:
|
|
return False, result.get("error", "Неизвестная ошибка")
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
return False, f"Ошибка сети: {e}"
|
|
except json.JSONDecodeError as e:
|
|
return False, f"Ошибка парсинга ответа: {e}"
|
|
except Exception as e:
|
|
return False, f"Неожиданная ошибка: {e}"
|
|
|
|
def main():
|
|
if len(sys.argv) != 3:
|
|
print("Использование: python tools/dev/send_telegram_example.py <user_id> <message>")
|
|
print("Пример: python tools/dev/send_telegram_example.py 123456789 'Привет! Это тестовое сообщение.'")
|
|
sys.exit(1)
|
|
|
|
try:
|
|
user_id = int(sys.argv[1])
|
|
except ValueError:
|
|
print("Ошибка: user_id должен быть числом")
|
|
sys.exit(1)
|
|
|
|
message_text = sys.argv[2]
|
|
|
|
print(f"Отправка сообщения пользователю {user_id}: {message_text}")
|
|
|
|
success, result_message = send_message(user_id, message_text)
|
|
|
|
if success:
|
|
print(f"✅ {result_message}")
|
|
else:
|
|
print(f"❌ {result_message}")
|
|
sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|