blob: c985ff444e69925beaacd6b335c7940290b2cc23 (
plain)
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
|
import os
import sqlite3
from urllib import parse
import requests
from dotenv import load_dotenv
from util import setupLogger, getDbConn
load_dotenv()
logger = setupLogger()
class Bot:
def __init__(self):
self.token = os.getenv('BOT_TOKEN')
self.chatId = os.getenv('BOT_CHAT_ID')
self.isProcessingPolling = False
self.hasNewMessages = False
self.messages = list()
def sendMessage(self, message, log=True):
message = parse.quote_plus(message)
url = f'https://api.telegram.org/bot{self.token}/sendMessage?chat_id={self.chatId}&parse_mode=html&text={message}'
response = requests.get(url)
if log:
if response.status_code == 200:
logger.info(f'Message sent: {response.text}')
else:
logger.error(f'Error sending message: {response.text}')
def _dbUpdate(self, data):
conn = getDbConn()
cur = conn.cursor()
for data in data['result']:
updateId = data['update_id']
msg = data['message']
msgId = msg['message_id']
msgDate = msg['date']
msgFrom = msg['from']['username']
text = msg['text']
res = cur.execute(f"SELECT id, is_read FROM incoming WHERE id = {msgId}")
if not res.fetchall():
cur.execute(
f"INSERT INTO incoming VALUES ({msgId}, {msgDate}, '{msgFrom}', '{text}', False, {updateId})")
self.hasNewMessages = True
conn.commit()
conn.close()
def _checkUnreadDb(self):
try:
if not self.hasNewMessages:
conn = getDbConn()
cur = conn.cursor()
res = cur.execute(f"SELECT id, is_read FROM incoming WHERE is_read = 0")
if len(res.fetchall()) > 0:
self.hasNewMessages = True
conn.close()
except sqlite3.Error as e:
logger.error(e)
def pollUpdate(self):
self.isProcessingPolling = True
try:
url = f'https://api.telegram.org/bot{self.token}/getUpdates'
response = requests.get(url)
if response.status_code == 200:
self._dbUpdate(response.json())
# Clean queue after reading because apparently it has maximum size and can block newer messages
self.cleanTelegramApiQueue()
except requests.exceptions.ConnectionError as e:
logger.error("Connection error during polling update")
self._checkUnreadDb()
self.isProcessingPolling = False
def getNewMessages(self):
result = list()
try:
if not self.hasNewMessages:
return result
conn = getDbConn()
cur = conn.cursor()
resCheck = cur.execute(f"SELECT text FROM incoming WHERE is_read = False")
for row in resCheck.fetchall():
result.append(row[0])
cur.execute(f"UPDATE incoming SET is_read = True WHERE is_read = False")
conn.commit()
conn.close()
self.hasNewMessages = False
except sqlite3.Error as e:
logger.error(e)
finally:
return result
def cleanTelegramApiQueue(self):
try:
baseGetUpdateUrl = f'https://api.telegram.org/bot{self.token}/getUpdates'
response = requests.get(f'{baseGetUpdateUrl}?offset=-1')
data = response.json()
if data['result']:
lastUpdateId = data['result'][0]['update_id']
# dapet trik dari stackoverflow
# https://stackoverflow.com/questions/61976560/how-to-delete-queue-updates-in-telegram-api
requests.get(f'{baseGetUpdateUrl}?offset={lastUpdateId + 1}')
except requests.exceptions.ConnectionError as e:
logger.error("Connection error during polling update")
if __name__ == '__main__':
bot = Bot()
bot.getNewMessages()
|
