Instantly share code, notes, and snippets.

Embed
What would you like to do?
Python Telegram Bot
import threading
import telepot
import sqlite3
import log
SETTINGS_TABLE = 'settings'
SUBSCRIBER_TABLE = 'subscriber'
LAST_UPDATE_ID = 'last_update_id'
UPDATE_INTERVAL_SEC = 5
class Bot:
def __init__(self, token, db_path):
self.active = False
self.token = token
self.bot = telepot.Bot(self.token)
info = self.bot.getMe()
self.id = info['id']
self.name = info['first_name']
self.username = info['username']
self.init_db(db_path)
def init_db(self, db_path):
log.debug('Init DB')
self.db_path = db_path
conn = sqlite3.connect(self.db_path)
# Init SETTINGS_TABLE
cur = conn.cursor()
q = 'create table if not exists %s(Name TEXT UNIQUE, Value TEXT)' % SETTINGS_TABLE
cur.execute(q)
# Init SUBSCRIBER_TABLE
q = 'create table if not exists %s(ID INT UNIQUE, Name TEXT, Realname TEXT)' % SUBSCRIBER_TABLE
cur.execute(q)
conn.commit()
conn.close()
def status(self):
log.info('id: %d' % self.id )
log.info('name: "%s"' % self.name)
log.info('username: "%s"' % self.username)
log.info('token: "%s"' % (self.token) )
def start(self):
log.info('Start %s' % self.username)
self.active = True
self.update()
def stop(self):
log.info('Stop %s' % self.username)
self.active = False
def broadcast(self, msg):
'구독자 전원에게 메시지를 송신한다'
conn = sqlite3.connect(self.db_path)
cur = conn.cursor()
q = 'select ID from %s' % SUBSCRIBER_TABLE
cur.execute(q)
rows = cur.fetchall()
conn.close()
for row in rows:
id = row[0]
self.bot.sendMessage(id, msg)
def get_last_update_id(self):
conn = sqlite3.connect(self.db_path)
cur = conn.cursor()
q = 'select Name, Value from %s where Name="%s"' % (SETTINGS_TABLE, LAST_UPDATE_ID)
value = -1
try:
cur.execute(q)
except sqlite3.OperationalError:
pass
else:
rows = cur.fetchall()
for row in rows:
if row[0] == LAST_UPDATE_ID:
value = (int)(row[1])
conn.close()
return value
def set_last_update_id(self, update_id):
conn = sqlite3.connect(self.db_path)
cur = conn.cursor()
q = 'insert or replace into %s (Name, Value) values (?, ?)' % SETTINGS_TABLE
r = cur.execute(q, (LAST_UPDATE_ID, update_id))
conn.commit()
conn.close()
def parse_update(self, response):
for r in response: # r = update
_from = r['message']['from']
date = r['message']['date']
user_id = _from['id']
user_name = _from['first_name']
user_realname = _from['username']
text = r['message']['text']
self.set_last_update_id(r['update_id'])
self.on_message(user_id, user_name, user_realname, text)
def on_join(self, user_id, user_name, user_realname):
conn = sqlite3.connect(self.db_path)
cur = conn.cursor()
q = 'insert or replace into %s (ID, Name, Realname) values (?, ?, ?)' % SUBSCRIBER_TABLE
conn.execute(q, (user_id, user_name, user_realname))
conn.commit()
conn.close()
def on_message(self, user_id, user_name, user_realname, text):
if text == '/join':
self.on_join(user_id, user_name, user_realname)
else:
log.info('%s: %s' % (user_name, text) )
def update(self):
if not self.active:
return
response = self.bot.getUpdates(self.get_last_update_id() + 1) # 마지막 업데이트 이후 내용만
self.parse_update(response)
threading.Timer(UPDATE_INTERVAL_SEC, self.update).start() # 다음 업데이트 예약
if __name__ == '__main__':
bot = Bot('000000000:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', 'd:/project/a/db/bot.db')
bot.status()
bot.start()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment