try:
bot.send_message(user_id, text)
except telebot.apihelper.ApiException:
logging.exception(f'Send Notification ERROR - {telebot.apihelper.ApiException}')
delete_user(user_id)def delete_user(user_id):
del_user_query = f'DELETE FROM {users_t} WHERE {users_col["user_id"]} = {user_id};'
post_sql_query(del_user_query)import telebot
from telebot import types
API_TOKEN = ""
bot = telebot.TeleBot(API_TOKEN)
@bot.message_handler(commands=['start'])
def start(message):
some_data_from_sqlite = [('one',), ('two',), ('ten',)] # список кортежей из БД
buttons_dict = {i: x[0] for i, x in enumerate(some_data_from_sqlite)}
keyboard = types.InlineKeyboardMarkup()
back_button = types.InlineKeyboardButton(text="Back", callback_data="MainMenu")
button_list = [types.InlineKeyboardButton(text=x, callback_data=x) for x in buttons_dict.values()] # callback дата понадобится при обработке нажатий на кнопку, но это уже совсем другая история :)
keyboard.add(*button_list, back_button)
bot.send_message(chat_id=message.chat.id, text=message.text, reply_markup=keyboard)
if __name__ == "__main__":
try:
bot.polling(none_stop=True)
except Exception as e:
print(e)
cursor.execute('SELECT id FROM users
WHERE game_is_true = "True"
AND game_is_true2 = "False"
AND id != ?', (my_id,))create table `rules` (`id` INT PRIMARY KEY, `rule` VARCHAR(50));
insert into `rules` (`id`, `rule`) VALUES (1, 'user');
insert into `rules` (`id`,`rule`) VALUES (2, 'moderator');
insert into `rules` (`id`,`rule`) VALUES (3,'admin');
create table `users` (`telegram_id` INT PRIMARY KEY, `username` VARCHAR(50), `status` INT, FOREIGN KEY (`status`) REFERENCES `rules` (`id`));
insert into `users` (`telegram_id`, `username`, `status`) VALUES (555, 'user555', '1');
insert into `users` (`telegram_id`, `username`, `status`) VALUES (666, 'user666', '2');
insert into `users` (`telegram_id`, `username`, `status`) VALUES (777, 'user777', '3');
insert into `users` (`telegram_id`, `username`, `status`) VALUES (888, 'user888', '3');select u.username from users u
left join rules r
ON u.status=r.id
where id=3;
import sqlite3
blob = bytearray(b'{"param1":"1","param2":"2"}')
with sqlite3.connect('np.db') as connection:
cursor = connection.cursor()
cursor.execute("CREATE TABLE IF NOT EXISTS t_info_t (blob_Value BLOB )")
cursor.execute("INSERT OR IGNORE INTO t_info_t (blob_Value) values (?)", (blob,))
cursor.execute("SELECT * FROM t_info_t")
data = cursor.fetchall()
print(data)
print(type(data[0][0]))
out = data[0][0].decode()
print(out)
print(type(out))CREATE TABLE IF NOT EXISTS games
(game TEXT NOT NULL, play_date TEXT NOT NULL, play_time TEXT NOT NULL,
team_a TEXT NOT NULL, team_b TEXT NOT NULL, cup TEXT,
PRIMARY KEY (game , play_date , play_time ));with sqlite3.connect('my.db') as connection:
cursor = connection.cursor()
cursor.execute(cmd)def is_user_warned(user_id: int) -> bool:if not is_user_warned(event.obj.reply_message['from_id']):def post_sql_query(sql_query):
with sqlite3.connect('my.db') as connection:
cursor = connection.cursor()
try:
cursor.execute(sql_query)
except Error:
pass
result = cursor.fetchall()
return result
def is_user_warned(user_id: int):
cmd = "select count(user_id) from warnlist where user_id = %d" % (user_id)
result = post_sql_query(cmd)
return result
if not is_user_warned(event.obj.reply_message['from_id']):
print('добавить')
else:
print('+ 1/3 варн')def post_sql_query(sql_query):
with sqlite3.connect(database) as connection:
cursor = connection.cursor()
try:
cursor.execute(sql_query)
except Error:
print(Error)
result = cursor.fetchall()
return result
def delete_user(user_id):
del_user_query = f'DELETE FROM users WHERE user_id = {user_id};'
post_sql_query(del_user_query)
try:
bot.send_message(user_id, text)
except telebot.apihelper.ApiException:
delete_user(user_id)import sqlite3
from sqlite3 import Error
from time import sleep, ctimedef post_sql_query(sql_query):
with sqlite3.connect('my.db') as connection:
cursor = connection.cursor()
try:
cursor.execute(sql_query)
except Error:
pass
result = cursor.fetchall()
return resultdef create_tables():
users_query = '''CREATE TABLE IF NOT EXISTS USERS
(user_id INTEGER PRIMARY KEY NOT NULL,
username TEXT,
first_name TEXT,
last_name TEXT,
reg_date TEXT);'''
post_sql_query(users_query)def register_user(user, username, first_name, last_name):
user_check_query = f'SELECT * FROM USERS WHERE user_id = {user};'
user_check_data = post_sql_query(user_check_query)
if not user_check_data:
insert_to_db_query = f'INSERT INTO USERS (user_id, username, first_name, last_name, reg_date) VALUES ({user}, "{username}", "{first_name}", "{last_name}", "{ctime()}");'
post_sql_query(insert_to_db_query )create_tables() # вызываем функцию создания таблицы users
@bot.message_handler(commands=['start'])
def start(message):
register_user(message.from_user.id, message.from_user.username,
message.from_user.first_name, message.from_user.last_name)
bot.send_message(message.from_user.id, f'Welcome {message.from_user.first_name}' )city = message.text
with db.connect('your.db') as connection:
cursor = connection.cursor()
cursor.execute("SELECT * FROM table WHERE row = ?", (city,))
result = cursor.fetchall()def check_city(city ):
with db.connect('your.db') as connection:
cursor = connection.cursor()
cursor.execute("SELECT * FROM table WHERE row = ? ", (city,))
data = cursor.fetchone()
if data is None:
return False
else:
return data
test = check_city(message.text)
if test:
bot.send_message(user_id, test[0]) # например первое поле из таблицы