ПКС / ПКС. Материалы лекций

from wb_unread import *

def simple_activate(bot_driver, chat_num): chat_elem = None

try:

# клик по чату по его номеру params = '//*[@id="pane-

side"]/div[1]/div/div/div[{}]/div/div/div[2]/div[1]/div[1]'.format(chat_num) chat_elem = bot_driver.find_element(By.XPATH, params) chat_elem.click()

print("Чат {} активирован из списка!!!".format(chat_elem)) except:

print("Завершение функции активации чата {}, т.к. чат не удалось активировать!(".format(chat_elem))

return -11 try:

# получение контекста time.sleep(2)

chat_elem = bot_driver.find_element(By.XPATH, '//*[@id="main"]/div[3]/div/div[2]/div[2]')

print(chat_elem.text) except:

print("Ошибка считывания данных!(") return -12

return 0

if __name__ == "__main__": bot_driver = None

try:

botdriver = simple_prepare() if botdriver is not None:

num_chats = simple_chat_count(botdriver) if num_chats > 0:

num_select = 1 # номер чата для выбора

while int(num_select) >= 1 and int(num_select) <= num_chats: num_select = input("В списке {} чатов, введите номер чата для

считывания всех сообщений"

"(число от 1 до {} включительно); для выхода - любое другое значение".format(num_chats, num_chats))

ret_value = int(simple_activate(botdriver, num_select)) if ret_value >= 0:

print("Функция поиска последних сообщений для чата № {} выполнена

успешно!)") else:

print("Не найдены сообщения для чата № {}!(")

else: # завершение приложения, если ни один из вариантов не выбран print("Нажмите любую клавишу для завершения приложения...") pass

finally:

if botdriver is not None: botdriver.quit()

print("Драйвер браузера и связанные соединения успешно закрыты!!") print("Приложение успешно закрыто.")

import time

from datetime import date import json

import pymysql

from decimal import *

# для распознавания типа Decimal из формата JSON def decimal_default(obj):

if isinstance(obj, Decimal): return float(obj)

raise TypeError("Ошибка сериализации JSON: '%s'" % type(obj).__name__)

class dbase_mysql: def __init__(self):

db = self.mysql_connect() if db is not None:

print("Проверка чтения") self.select_data("1002")

def mysql_connect(self): db = None

try:

db = pymysql.connect( host='sutct.org', user='1_11', password=‘dsfsadfsdf', database='1_11')

print("Подключение к БД успешно установлено!) (время: {})".format(time.localtime())) except Exception as e:

print("Подключение к БД не установлено!( \n{}\n{}\n(время: {})". e.__repr__(), e.args, format(time.localtime()))

return db

def update_data(self, index, calorie): db = self.mysql_connect()

ret = None try:

if db is not None: cur = db.cursor()

sql = "UPDATE CALORIE_CONTENT SET calorie_content = {} WHERE product_id = {}".format(index, calorie)

print("Запрос {} успешно подготовлен!".format(sql)) else:

print("Не удалось подключиться к БД!(") return ret

except:

print("Запрос {} не удалось подготовить!(".format(sql)) return ret

try:

cur.execute(sql)

print("Выполнение: cur.execute(sql)") json_data = ["Изменение выполнено"] cur = db.cursor()

sql = "SELECT * FROM CALORIE_CONTENT WHERE product_id = {}".format(index) cur.execute(sql)

print("Выполнено: cur.execute(sql)") rows = cur.fetchall()

print("Выполнено: rows = cur.fetchall()") row_headers = [x[0] for x in cur.description]

print("Выполнено: row_headers = [x[0] for x in cur.description]") for row in rows:

json_data.append(dict(zip(row_headers, row)))

print("Выполнено: json_data.append(dict(zip(row_headers, row)))") print("json_data = \n{}".format(json_data))

ret = json.dumps(json_data, skipkeys=False, ensure_ascii=False, check_circular=True, allow_nan=True, cls=None, indent=None, separators=None, default=decimal_default, sort_keys=False)

print("Успешно получен результат на запрос:\n{}".format(ret)) except Exception as e:

print("Запрос {} не удалось выполнить на сервере!( Ошибка:\n{}\n{}". format(sql, e.__repr__(), e.args))

format(sql, e.__repr__(), e.args)) finally:

db.close() return ret

def select_data(self, index): db = self.mysql_connect() ret = None

try:

if db is not None: cur = db.cursor()

sql = "SELECT * FROM CALORIE_CONTENT WHERE product_id = {}".format(index) print("Запрос {} успешно подготовлен!)".format(sql))

else:

print("Не удалось подключиться к БД!(") return ret

except:

print("Запрос {} не удалось подготовить!(".format(sql)) return ret

try:

cur.execute(sql)

print("Выполнено: cur.execute(sql)") rows = cur.fetchall()

print("Выполнено: rows = cur.fetchall()") row_headers = [x[0] for x in cur.description]

print("Выполнено: row_headers = [x[0] for x in cur.description]")

json_data = []

for row in rows: json_data.append(dict(zip(row_headers, row)))

print("Выполнено: json_data.append(dict(zip(row_headers, row)))") print("json_data = \n{}".format(json_data))

ret = json.dumps(json_data, skipkeys=False, ensure_ascii=False, check_circular=True, allow_nan=True, cls=None, indent=None, separators=None, default=str, sort_keys=False)

print("Успешно получен результат на запрос:\n{}".format(ret)) except Exception as e:

print("Запрос {0} не удалось выполнить на сервере!( Ошибка:\n{1}\n{2}".format(sql, e.__repr__(), e.args))

finally:

db.close() return ret

def insert_data(self, data): db = self.mysql_connect() ret = None

try:

if db is not None: cur = db.cursor()

sql = "INSERT INTO EATING_LOGBOOK (log_date, user_name, mass, product_id) VALUES ('" + str(date.today()) +"'"

for d in data:

sql = sql + ", " + str(d) sql = sql + ")"

print("Запрос {} успешно подготовлен!)".format(sql)) else:

print("Не удалось подключиться к БД!(") return ret

except:

print("Запрос {} не удалось подготовить!(".format(sql)) return ret

try:

cur.execute(sql)

print("Выполнено: cur.execute(sql)") json_data = ["Данные успешно добавлены"] print("json_data = \n{}".format(json_data))

ret = json.dumps(json_data, skipkeys=False, ensure_ascii=False, check_circular=True,

allow_nan=True, cls=None, indent=None, separators=None, default=decimal_default, sort_keys=False)

print("Успешно получен результат на запрос:\n{}".format(ret)) except Exception as e:

print("Запрос {} не удалось выполнить на сервере!( Ошибка:\n{}\n{}". format(sql, e.__repr__(), e.args))

finally:

db.close() return ret

def delete_data (self, index): db = self.mysql_connect() ret = None

try:

if db is not None: cur = db.cursor()

sql = "DELETE FROM EATING_LOGBOOK WHERE log_id = {}".format(index) print("Запрос {} успешно подготовлен!)".format(sql))

else:

print("Не удалось подключиться к БД!(") return ret

except:

print("Запрос {} не удалось подготовить!(".format(sql)) return ret

try:

cur.execute(sql)

print("Выполнение: cur.execute(sql)") json_data = ["Изменение выполнено"] print("json_data = \n{}".format(json_data))

ret = json.dumps(json_data, skipkeys=False, ensure_ascii=False, check_circular=True, allow_nan=True, cls=None, indent=None, separators=None, default=decimal_default, sort_keys=False)

print("Успешно получен результат на запрос:\n{}".format(ret)) except Exception as e:

print("Запрос {} не удалось выполнить на сервере!( Ошибка:\n{}\n{}". format(sql), e.__repr__(), e.args)

finally:

db.close() return ret

import unittest

from db_mysql import dbase_mysql

class TestsCreate(unittest.TestCase): def test_create_success(self):

print("Выполнение test_create_success") db = dbase_mysql()

self.assertEqual('["Данные успешно добавлены"]', db.insert_data(['\'rozaroza\'',300,2])) # add assertion here

def test_create_fail(self): print("Выполнение test_create_fail") db = dbase_mysql()

self.assertNotEqual('["Данные успешно добавлены"]', db.insert_data(['\'rozaroza\'', 2]))

class TestsUpdate(unittest.TestCase): def test_update_success(self):

print("Выполнение test_create_success") db = dbase_mysql()

self.assertEqual('["Изменение выполнено", {"product_id": 1, "product_name": "tomato", "calorie_content": 24.0, "fats_content": 0.2, "proteins_content": 1.1, "carbohydrates_content": 3.8, "water_content": 92.0, "cellulose_content": 1.4}]', db.update_data(1, 24)) # add assertion here

def test_update_fail(self): print("Выполнение test_create_fail") db = dbase_mysql()

self.assertNotEqual('["Изменение выполнено", {"product_id": 1, "product_name": "tomato", "calorie_content": 25.0, "fats_content": 0.2, "proteins_content": 1.1, "carbohydrates_content": 3.8, "water_content": 92.0, "cellulose_content": 1.4}]',

db.update_data(1, 24))

class TestsRead(unittest.TestCase): def test_read_success(self):

print("Выполнение test_create_success") db = dbase_mysql()

self.assertEqual(['[{"log_id": 11, "log_date": "20.10.2021", "user_name": "ulitka0000", "mass": 23232, "product_id": 3}]'], db.select_data(11)) # add assertion here

def test_read_fail(self): print("Выполнение test_create_fail") db = dbase_mysql()

self.assertNotEqual('[{"log_id": 10, "log_date": "2021-10-19", "user_name": "rozaroza", "mass": 62, "product_id": 1}]',

db.select_data(10))

class TestsDelete(unittest.TestCase): def test_delete_success(self):

print("Выполнение test_create_success") db = dbase_mysql()

self.assertEqual('["Изменение выполнено"]', db.delete_data(4)) # add assertion here

def test_delete_fail(self): print("Выполнение test_create_fail") db = dbase_mysql()

self.assertNotEqual('["Изменение выполнено"]', db.delete_data(10))

if __name__ == '__main__': unittest.main()

from telegram.ext import Updater, CommandHandler, MessageHandler, Filters import config

from db_mysql import dbase_mysql

db = config.DataBase()

updater = Updater(token = config.TOKEN) dispatcher = updater.dispatcher

class tg_bot:

def __init__(self):

self.updater = Updater(token=config.TOKEN) self.dispatcher = self.updater.dispatcher self.dbase = dbase_mysql()

start_handler = CommandHandler('start', self.func_start) self.dispatcher.add_handler(start_handler)

text_handler = MessageHandler(Filters.text & (~Filters.command), self.func_text) self.dispatcher.add_handler(text_handler)

# обработчик команды '/insert'

insert_handler = CommandHandler('insert', self.func_insert) self.dispatcher.add_handler(insert_handler)

# обработчик команды '/select'

select_handler = CommandHandler('select', self.func_select) self.dispatcher.add_handler(select_handler)

# обработчик команды '/update'

select_handler = CommandHandler('update', self.func_update) self.dispatcher.add_handler(select_handler)

# обработчик не распознанных команд

unknown_handler = MessageHandler(Filters.command, self.func_unknown) self.dispatcher.add_handler(unknown_handler)

#запуск прослушивания сообщений self.updater.start_polling()

#обработчик нажатия Ctrl+C self.updater.idle()

def about(self):

return "Для извлечения исходных данных используйте команды:\n \

/insert ['имя пользователя', масса продукта без единиц измерения, id продукта], где index - номер параметра в таблице содержания калорий от 1 до 7;\n \

/select index, где index - номер параметра в таблице продуктов от 1 до 5\n \

/delete index, где index - номер параметра в таблице записей журнала потребления\n \ /update index calorie, где index - номер параметра в таблице продуктов от 1 до 5, calorie

- новая калорийность продукта"

def func_start(self, update, context):

context.bot.send_message(chat_id=update.effective_chat.id, text="CalorieCalculatingBot591 a.k.a mxkrsCalCalcBot")

# обработчик команды '/delete'

insert_handler = CommandHandler('delete', self.func_delete) self.dispatcher.add_handler(insert_handler)

def func_text(self, update, context):

text_out = 'Получено сообщение: ' + update.message.text + '\n' + self.about() context.bot.send_message(chat_id=update.effective_chat.id, text=text_out)