mirror of
https://github.com/RoboSats/robosats.git
synced 2024-12-13 19:06:26 +00:00
1c696beb5d
* Update telegram_watcher.py It is used "startswith()" to check if the text start with "/start". "filter()" instead of "get()" to avoid an exception if there is no profile asociated with a token. "attempts -= 1" instead of "attempts = attempts - 1" An "error.log" file is created if "response = self.session.get" does not work * Update telegram_watcher.py Added instructions for the user if the format is not correct * Update telegram_watcher.py * Update telegram_watcher.py * Update telegram_watcher.py * Update telegram_watcher.py * Update telegram_watcher.py * Fix bugs --------- Co-authored-by: Kote <45825337+Gravity2106@users.noreply.github.com>
83 lines
3.3 KiB
Python
83 lines
3.3 KiB
Python
import time
|
|
import traceback
|
|
|
|
from decouple import config
|
|
from django.core.management.base import BaseCommand
|
|
from django.db import transaction
|
|
|
|
from api.messages import Telegram
|
|
from api.models import Profile
|
|
from api.utils import get_session
|
|
|
|
|
|
class Command(BaseCommand):
|
|
|
|
help = "Polls telegram /getUpdates method"
|
|
rest = 3 # seconds between consecutive polls
|
|
|
|
bot_token = config("TELEGRAM_TOKEN")
|
|
updates_url = f"https://api.telegram.org/bot{bot_token}/getUpdates"
|
|
session = get_session()
|
|
telegram = Telegram()
|
|
|
|
def handle(self, *args, **options):
|
|
offset = 0
|
|
while True:
|
|
time.sleep(self.rest)
|
|
params = {"offset": offset + 1, "timeout": 5}
|
|
try:
|
|
response = self.session.get(self.updates_url, params=params)
|
|
if response.status_code != 200:
|
|
with open("error.log", "a") as f:
|
|
f.write(
|
|
f"Error getting updates, status code: {response.status_code}\n"
|
|
)
|
|
continue
|
|
response = response.json()
|
|
response = self.session.get(self.updates_url, params=params).json()
|
|
except Exception as e:
|
|
with open("error.log", "a") as f:
|
|
f.write(f"Error getting updates: {e}\n{traceback.format_exc()}\n")
|
|
continue
|
|
|
|
if not response["result"]:
|
|
continue
|
|
for result in response["result"]:
|
|
if not result.get("message") or not result.get("message").get("text"):
|
|
continue
|
|
message = result["message"]["text"]
|
|
if not message or not message.startswith("/start"):
|
|
continue
|
|
parts = message.split(" ")
|
|
if len(parts) < 2:
|
|
self.telegram.send_message(
|
|
chat_id=result["message"]["from"]["id"],
|
|
text='You must enable the notifications bot using the RoboSats client. Click on your "Robot profile" -> "Enable Telegram" and follow the link or scan the QR code.',
|
|
)
|
|
continue
|
|
token = parts[-1]
|
|
profile = Profile.objects.filter(telegram_token=token).first()
|
|
if not profile:
|
|
self.telegram.send_message(
|
|
chat_id=result["message"]["from"]["id"],
|
|
text=f'Wops, invalid token! There is no Robot with telegram chat token "{token}"',
|
|
)
|
|
continue
|
|
|
|
attempts = 5
|
|
while attempts >= 0:
|
|
try:
|
|
with transaction.atomic():
|
|
profile.telegram_chat_id = result["message"]["from"]["id"]
|
|
profile.telegram_lang_code = result["message"]["from"][
|
|
"language_code"
|
|
]
|
|
self.telegram.welcome(profile.user)
|
|
profile.telegram_enabled = True
|
|
profile.save()
|
|
break
|
|
except Exception:
|
|
time.sleep(5)
|
|
attempts -= 1
|
|
offset = response["result"][-1]["update_id"]
|