database/user_pw.py - patch - Esgalnor/TG_Bot

database/user_pw.py
256 строк8 KB
Esgalnor
upload files
18 авг 2025, 17:24
18 авг 2025, 17:2482ee7ff
import datetime
import json
import re

from time import strftime
from database.models_pw import User, db_user_path
from config.config import MASTER_ID
from utils.json_funcs import to_obj, to_str


class UserPW:

    def __init__(self):
        self.model = User
        self.__create_users_table()

    def __create_users_table(self) -> None:
        """
        ф-ция создания таблицы пользователей
        :return:
        """
        self.model.create_table()

    def new_user(self, id_user: int) -> None:
        """
        ф-ция создаёт запись в таблице с новым пользователем
        :param id_user: id - пользователя
        :return:
        """
        new_elem = json.dumps({"time": strftime("%d %h %Y %H:%M:%S")})
        self.model.create(id_user=id_user)

    def set_name(self, id_user: int, name: str = ''):
        """
        устанавливает имя для пользователя
        """
        user = self.model.get(self.model.id_user == id_user)
        user.name = name
        user.save()

    def get_name(self, id_user: int):
        """
        возвращает имя пользователя
        """
        user = self.model.get(self.model.id_user == id_user)
        return user.name

    def get_check(self, id_user: int):
        """
        возвращает доверенность к пользователю
        """
        user = self.model.get(self.model.id_user == id_user)
        return user.check_user

    def set_temp(self, id_user: int, text: str = ''):
        """
        записывает временную инфу
        """
        user = self.model.get(self.model.id_user == id_user)
        user.temp = text
        user.save()

    def get_temp(self, id_user: int):
        """
        возвращает временную инфу
        """
        user = self.model.get(self.model.id_user == id_user)
        return user.temp

    def set_check(self, id_user: int, check: str = ''):
        """
        устанавливает доверие к пользователю
        """
        user = self.model.get(self.model.id_user == id_user)
        if "да" in check.lower():
            user.check_user = True
        if "нет" in check.lower():
            user.check_user = False
        user.save()

    def set_curr_rec(self, id_user: int, date: str):
        """
        производит запись - {<номер записи> : <дата, время, процедуры>}
        """
        user = self.model.get(self.model.id_user == id_user)

        all_rec = user.curr_rec
        if all_rec:
            all_rec = to_obj(all_rec)
        else:
            all_rec = dict()
        all_rec[len(all_rec.keys())] = date

        user.curr_rec = to_str(all_rec)
        user.save()

    def del_rec(self, id_user):
        """
        удаляет запись по id
        """
        user = self.model.get(self.model.id_user == id_user)
        user.curr_rec = ''
        user.save()

    def del_rec_date(self, date):
        """
        удаляет запись по дате
        """
        ...

    def get_curr_rec(self, id_user: int):
        """
        возвращает записи по id
        """
        user = self.model.get(self.model.id_user == id_user)
        all_rec = user.curr_rec
        if all_rec:
            all_rec = ';\n'.join(to_obj(all_rec).values())
        return all_rec

    def get_list_user(self) -> list:
        """
        ф-ция возвращает список id пользователей
        :return:
        """
        result = [user.id_user for user in self.model.select()]
        return result

    def get_tg_name(self, id_user: int):
        """
        Возвращает ник из ТГ
        """
        user = self.model.get(self.model.id_user == id_user)
        if user.tg_name:
            return user.tg_name
        return None

    def set_tg_name(self, id_user: int, new_tg_name: str):
        """
        устанавливает ник из ТГ
        """
        user = self.model.get(self.model.id_user == id_user)
        user.tg_name = new_tg_name
        user.save()
        return "Имя пользователя {} удачно добавлено/изменено.".format(user.tg_name)

    def get_phone(self, id_user: int):
        """
        возвращает телефон пользователя
        """
        user = self.model.get(self.model.id_user == id_user)
        if user.phone:
            return user.phone
        return None

    def set_phone(self, id_user: int, phone: str):
        """
        устанавливает телефон пользователя
        """
        user = self.model.get(self.model.id_user == id_user)
        phone_re = re.compile(r"^[0-9]{10}$")
        if phone_re.search(phone):
            user.phone = int(phone)
            user.save()
            return "Телефон {} добавлен/изменён.".format(user.phone)
        else:
            return "Ошибка. Проверьте корректность ввода"

    def set_wait(self, id_user, value: bool):
        """
        устанавливает ожидание подтверждения записи
        """
        user = self.model.get(self.model.id_user == id_user)
        user.wait_check_rec = value
        user.save()

    def get_wait(self, id_user):
        """
        возвращает ожидание подтверждения записи
        """
        user = self.model.get(self.model.id_user == id_user)
        return user.wait_check_rec

    def del_user(self, id_user: int) -> None:
        """
        удаляет выбранного пользователя
        """
        if id_user in self.get_list_user():
            user = self.model.select().where(self.model.id_user == id_user).get()
            user.delete_instance()
        else:
            print('User not found')

    def del_all_user(self):
        """
        удаляет все записи
        """
        [self.del_user(idx) for idx in self.get_list_user()]

    def min_id(self):
        """
        получение минимального значения id
        """
        min_id = min(self.get_list_user())
        if min_id > 0:
            min_id = 0
        return min_id

    def get_temp_id_list(self):
        """
        получение списка временных id. Чтобы не засирать БД
        """

        return [user.id_user for user in self.model.select(self.model.id_user).where(self.model.id_user < 0)]

    def id_temp_user(self):
        """
        возвращает id временного юзера без записей,
        при необходимости создание юзера для временной записи
        """
        for id in self.get_temp_id_list():
            if not self.get_curr_rec(id):
                return id
        else:
            id = self.min_id() - 1
            self.new_user(id)
            return id


users = UserPW()

if __name__ == "__main__":

    #print(users.get_temp_id_list())

    id_1 = 6440979964
    id_2 = 5222312325
    id_3 = 1966003259

    idx = [id_1, id_2, id_3, 1169643603]

    #users.del_user(id_3)
    #users.set_tg_name(id_2, '')

    list_users = users.get_list_user()

    """for id in idx:
        if id in list_users:
            users.set_temp(id)
            print(id)"""

    """for id in list_users:
        users.set_temp(id)"""

    #users.set_curr_rec(id_1, "11.10.2024")
    ...