This commit is contained in:
@@ -0,0 +1,98 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user