import asyncio import socket import time import logging import datetime from aiogram import Router, F from aiogram import Bot, Dispatcher, types from aiogram.filters.command import Command from aiogram.types import FSInputFile from aiogram.fsm.context import FSMContext from aiogram.fsm.storage.memory import MemoryStorage from aiogram.fsm.state import StatesGroup, State from aiogram.utils.chat_action import ChatActionSender #from aiogram.types import BufferedInputFile import configparser from asyncpg_lite import DatabaseManager from os import listdir from os.path import isfile, join import lib.OCer as OCer import serial router = Router() async def get_user_list(db_url, deletion_password, table_name): try: db_manager = DatabaseManager(db_url=db_url, deletion_password=deletion_password) data = [] async with db_manager as manager: one_data = await manager.select_data(table_name=table_name) for i in one_data: data.append(i) logging.info(f"Fetched {len(data)} records from the database.") return data except Exception as err: logging.error(err) pass def init(Bot): global config, bot, logger, some_open_ports, zxc bot = Bot #consts&confs config = configparser.ConfigParser() #bc_calculator = lomka.BCCCalculator() some_open_ports = serial.tools.list_ports.comports() #load config file config.read('config.ini') # device_hwid = config['DEFAULT']['device_hwid'] # baudrate = config['DEFAULT']['baudrate'] #log section logging.basicConfig( level=logging.INFO if config['DEFAULT']['debug'] == "false" else logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('lastest.log'), logging.StreamHandler()]) logger = logging.getLogger(socket.gethostname()) zxc = OCer.caser(device_hwid="PID=1A86:7523", baudrate=9600, config=config, logger=logger) zxc.set_controller_number(1) zxc.set_port_device() main_kb = [ [types.KeyboardButton(text="Выбрать контроллер")], [types.KeyboardButton(text="Открыть секцию")], [types.KeyboardButton(text="Статус секции")], [types.KeyboardButton(text="Статус ВСЕХ секций")], [types.KeyboardButton(text="Открыть ВСЕ секции")], [types.KeyboardButton(text="Контакты")]] keyboardos = types.ReplyKeyboardMarkup(keyboard=main_kb, resize_keyboard=True, input_field_placeholder="amogus life420") class Form(StatesGroup): input_controller = State() input_section = State() input_status_section = State() input_status_section_all = State() input_approve = State() @router.message(Command("start")) async def cmd_start(message: types.Message): keyboard = types.ReplyKeyboardMarkup(keyboard=main_kb, resize_keyboard=True, input_field_placeholder="Выберите действие...", is_persistent=True) await message.answer(f"{message.from_user.id}\n", reply_markup=keyboard) @router.message(F.text.lower() == "выбрать контроллер") async def choose_controller(message: types.Message, state: FSMContext): kb = [ [types.KeyboardButton(text="Домой")] ] keyboard = types.ReplyKeyboardMarkup(keyboard=kb, resize_keyboard=True, input_field_placeholder="amogus life420") await message.reply("Определите номер управляемого контроллера", reply_markup=keyboard) await state.set_state(Form.input_controller) @router.message(F.text, Form.input_controller) async def set_controller_number(message: types.Message, state: FSMContext): await state.update_data(confirm=message.text) user_data = await state.get_data() try: await zxc.set_controller_capacity(int(user_data["confirm"])) await message.reply(f"Контроллер ({user_data["confirm"]}) выбран успешно!", reply_markup=keyboardos) except Exception as e: await message.reply(f"Возникло исключение(ошибка): {e}", reply_markup=keyboardos) await state.clear() @router.message(F.text.lower() == "статус всех секций") async def status_all_sections(message: types.Message, state: FSMContext): cool_data = await zxc.READ_ALL_CELLS() await message.reply(f"{cool_data}", reply_markup=keyboardos) await state.clear() @router.message(F.text.lower() == "открыть все секции") async def open_all_sections(message: types.Message, state: FSMContext): kb = [ [types.KeyboardButton(text="Продолжить")], [types.KeyboardButton(text="Домой")] ] keyboard = types.ReplyKeyboardMarkup(keyboard=kb, resize_keyboard=True, input_field_placeholder="amogus life420") await message.reply("Данная операция откроет ВСЕ секции доступные контроллеру.\nПродолжить?", reply_markup=keyboard) await state.set_state(Form.input_approve) @router.message(F.text, Form.input_approve) async def open_all_sections_confirm(message: types.Message, state: FSMContext): await state.update_data(confirm=message.text) user_data = await state.get_data() if str(user_data["confirm"]).lower() == "продолжить": try: cool_data = await zxc.OPEN_ALL_CELLS_LEGACY(True) await message.reply(f"Статус ячеек:\n{cool_data}", reply_markup=keyboardos) except Exception as e: await message.reply(f"Возникло исключение(ошибка): {e}", reply_markup=keyboardos) await state.clear() @router.message(F.text.lower() == "контакты") async def contact_info(message: types.Message): keyboard = types.ReplyKeyboardMarkup(keyboard=main_kb, resize_keyboard=True, input_field_placeholder="amogus life420") await message.reply("Настучать админу: \n" \ "Discord: Valo#1234 | @valoker\n" \ "GitHub: @Valoker\n" \ "Telegram: @Valoka", reply_markup=keyboard) @router.message(F.text.lower() == "домой") async def home(message: types.Message, state: FSMContext): await message.answer(reply_markup=keyboardos, text="Меню") await state.clear() @router.message(F.text.lower() == "открыть секцию") async def open_section(message: types.Message, state: FSMContext): kb = [ [types.KeyboardButton(text="Домой")] ] keyboard = types.ReplyKeyboardMarkup(keyboard=kb, resize_keyboard=True, input_field_placeholder="amogus life420") await message.reply("Отправьте номер секции для открытия..", reply_markup=keyboard) await state.set_state(Form.input_section) @router.message(F.text, Form.input_section) async def open_section_number(message: types.Message, state: FSMContext): await state.update_data(confirm=message.text) user_data = await state.get_data() try: await zxc.open_cell(int(user_data["confirm"])) await message.reply(f"Секция ({user_data["confirm"]}) контроллера " \ f"({zxc.controller_number}) открыта.", reply_markup=keyboardos) except Exception as e: await message.reply(f"Возникло исключение(ошибка): {e}", reply_markup=keyboardos) await state.clear() @router.message(F.text.lower() == "статус секции") async def get_section_status(message: types.Message, state: FSMContext): kb = [ [types.KeyboardButton(text="Домой")] ] keyboard = types.ReplyKeyboardMarkup(keyboard=kb, resize_keyboard=True, input_field_placeholder="amogus life420") await message.reply("Отправьте номер желаемой секции..", reply_markup=keyboard) await state.set_state(Form.input_status_section) @router.message(F.text, Form.input_status_section) async def get_section_status_fin(message: types.Message, state: FSMContext): await state.update_data(confirm=message.text) user_data = await state.get_data() try: controller_data = await zxc.read_controller(cell_number=int(user_data["confirm"])) await message.reply(f"Секция ({user_data["confirm"]}) контроллера " \ f"({zxc.controller_number}) со статусом: {controller_data[0]["CURRENT_STATUS"]}. \n{controller_data}", reply_markup=keyboardos) except Exception as e: await message.reply(f"Возникло исключение(ошибка): {e}", reply_markup=keyboardos) await state.clear()