-
Notifications
You must be signed in to change notification settings - Fork 2
/
tg_bot.py
170 lines (150 loc) · 7.88 KB
/
tg_bot.py
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
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
#!/usr/bin/env python3
import logging
import telegram
import config
import tweepy
import time
import nawab_logger
import pandas as pd
import twitter_bot
from datetime import timedelta, datetime, date
from telegram import InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import Updater, CommandHandler, CallbackQueryHandler, ConversationHandler
KILL_SIGNAL = 0
class Telegram_Bot(object):
def __init__(self, twitter_api, dirpath, data, level, auto_retweet):
self.dirpath = dirpath
self.data = data
self.twitter_api = twitter_api
self.level = level
self.nw_logger = nawab_logger.Nawab_Logging(dirpath, level)
self.auto_retweet = auto_retweet
self.tw_bot = twitter_bot.Twitter_Bot(self.dirpath, self.data, self.level)
def nawab_tg_authenticate(self):
updater = Updater(token=config.tg_token, use_context=True)
return updater
def help(self, update, context):
text = 'The retweet option is only for the bot admins. However normal users can view the tweet.'\
'\nSee the available commands in the keyboard below.'
url = 'https://github.com/Aniketh01/NaWaB/'
keyboard = [[InlineKeyboardButton("Source Code", url=url)]]
reply_markup = InlineKeyboardMarkup(keyboard)
update.message.reply_text(text=text, reply_markup=reply_markup)
def display_tweet(self, context):
job = context.job
global KILL_SIGNAL
if self.level == logging.CRITICAL or self.level == logging.WARNING:
with open(self.dirpath + "results.log", "a") as fp:
fp.write(time.strftime("%Y-%m-%d %I:%M:%S %p") + ',' + ' INFO ' + 'Telegram_Bot '
+ 'starting display parameter' + '\n')
self.nw_logger.logger('Telegram_Bot starting display parameter', 'info', 'Results')
# the previous date from which the tweets need to be sent
previous_date = self.tw_bot.nawab_find_prev_date()
previous_datetime = datetime(previous_date.year, previous_date.month, previous_date.day)
tid = pd.read_csv(self.dirpath + 'tid_store.csv')
tid['Date_time'] = pd.to_datetime(tid['Date_time'])
for index, tid_store in tid[::-1].iterrows():
scrape_date = tid_store['Date_time']
if scrape_date <= previous_datetime:
break
try:
u = self.twitter_api.get_status(id=tid_store['Id'])
username = u.author.screen_name
except tweepy.TweepError as e:
if self.level == logging.CRITICAL:
with open(self.dirpath + "error.log", "a") as fp:
fp.write(time.strftime("%Y-%m-%d %I:%M:%S %p") + ',' + ' ERROR ' + 'Telegram_Bot ' +
"Tweepy failed to get the status of the user from the " +
str(tid_store['Id']) + ' because of ' + e.reason + "\n")
self.nw_logger.logger('Telegram_Bot Tweepy failed to get the status of the user from the ' +
str(tid_store['Id']) + ' because of ' + e.reason, 'error', 'Error')
pass
url = 'https://twitter.com/' + \
username + '/status/' + str(tid_store['Id'])
if (job.context in config.tg_admin_id) and (self.auto_retweet == False or self.auto_retweet == None):
keyboard = [[InlineKeyboardButton("Retweet", callback_data=int(
tid_store['Id'])), InlineKeyboardButton("View", url=url)]]
reply_markup = InlineKeyboardMarkup(keyboard)
else:
keyboard = [[InlineKeyboardButton(
"View", url=url)]]
reply_markup = InlineKeyboardMarkup(keyboard)
if KILL_SIGNAL == 0:
context.bot.send_message(job.context, text=str(
url), reply_markup=reply_markup)
time.sleep(1)
else:
KILL_SIGNAL = 0
break
def start(self, update, context):
global KILL_SIGNAL
KILL_SIGNAL = 0
chat_id = int(update.message.chat_id)
try:
if 'job' in context.chat_data:
old_job = context.chat_data['job']
old_job.schedule_removal()
new_job = context.job_queue.run_once(
self.display_tweet, 2, context=chat_id)
context.chat_data['job'] = new_job
update.message.reply_text('Successfully started!')
if self.level == logging.CRITICAL or self.level == logging.WARNING:
with open(self.dirpath + "results.log", "a") as fp:
fp.write(time.strftime("%Y-%m-%d %I:%M:%S %p") + ',' + ' INFO ' + 'Telegram_Bot '
+ 'new job ' + str(new_job) + '\n')
self.nw_logger.logger('Telegram_Bot new job ' + str(new_job) , 'info', 'Results')
except (IndexError, ValueError):
update.message.reply_text('Did you /start yet?')
def button(self, update, context):
query = update.callback_query
# CallbackQueries need to be answered, even if no notification to the user is needed
# Some clients may have trouble otherwise. See https://core.telegram.org/bots/api#callbackquery
query.answer()
data = query.data
try:
self.twitter_api.retweet(data)
except tweepy.TweepError as e:
if self.level == logging.CRITICAL:
with open(self.dirpath + "error.log", "a") as fp:
fp.write(time.strftime("%Y-%m-%d %I:%M:%S %p") + ',' + ' ERROR ' + 'Telegram_Bot '
+ "Tweepy failed to retweet after reading from the store of id " +
str(data) + ' because of ' + e.reason + "\n")
self.nw_logger.logger('Telegram_Bot' + ' Tweepy failed to retweet after reading from the store of id ' +
str(data) + ' because of ' + e.reason, 'error', 'Error')
pass
try:
u = self.twitter_api.get_status(id=int(data))
username = u.author.screen_name
except tweepy.TweepError as e:
fp.write(time.strftime("%Y-%m-%d %I:%M:%S %p") + ',' + ' ERROR ' + 'Telegram_Bot ' +
"Tweepy failed to get the status of the user from the " +
str(data) + ' because of ' + e.reason + '\n')
self.nw_logger.logger('Telegram_Bot' + ' Tweepy failed to get the status of the user from the ' +
str(data) + ' because of ' + e.reason, 'error', 'Error')
pass
url = 'https://twitter.com/' + \
username + '/status/' + str(data)
keyboard = [[InlineKeyboardButton(
"View", url=url)]]
reply_markup = InlineKeyboardMarkup(keyboard)
query.edit_message_text(text="Retweeted: {}".format(
url), reply_markup=reply_markup)
def error(self, update, context):
"""Log Errors caused by Updates."""
if self.level == logging.CRITICAL:
with open(self.dirpath + "error.log", "a") as fp:
fp.write( time.strftime("%Y-%m-%d %I:%M:%S %p") + ',' + ' ERROR ' + 'Telegram_Bot ' +
'Update ' + update + ' caused error ' + context.error + '\n')
self.nw_logger.logger('Telegram_Bot' +
' Update' + update + 'caused error ' + context.error, 'error', 'Error')
def stop(self, update, context):
if 'job' not in context.chat_data:
update.message.reply_text('You have not activated the bot yet!')
return
job = context.chat_data['job']
job.schedule_removal()
del context.chat_data['job']
global KILL_SIGNAL
KILL_SIGNAL = 1
update.message.reply_text(
'Nawab Telegram bot has been stopped successfully!')