mirror of
https://github.com/djimboy/djimbo_template_aio3.git
synced 2026-07-25 09:44:29 +00:00
Update
This commit is contained in:
@@ -0,0 +1,2 @@
|
||||
from .db_settings import Settingsx, ModelBase as ModelSettings
|
||||
from .db_users import Usersx, ModelBase as ModelUsers
|
||||
|
||||
@@ -0,0 +1,494 @@
|
||||
# - *- coding: utf- 8 - *-
|
||||
import sqlite3
|
||||
from typing import Any, Generic, TypeVar, Optional
|
||||
|
||||
from pydantic.fields import FieldInfo
|
||||
from pydantic_core import PydanticUndefinedType
|
||||
|
||||
from tgbot.data.config import PATH_DATABASE
|
||||
from tgbot.database.adb_migration import CONST_MIGRATE_TABLES, CONST_MIGRATE_COLUMNS
|
||||
from tgbot.utils.const_functions import ded
|
||||
|
||||
|
||||
# noinspection PyProtectedMember
|
||||
|
||||
################################################################################
|
||||
############################### SUPPORT FUNCTIONS ##############################
|
||||
# Converting the resulting list into a dictionary
|
||||
def dict_factory(cursor, row) -> dict:
|
||||
save_dict = {}
|
||||
|
||||
for idx, col in enumerate(cursor.description):
|
||||
save_dict[col[0]] = row[idx]
|
||||
|
||||
return save_dict
|
||||
|
||||
|
||||
# Formatting a query without arguments
|
||||
def update_format(sql, parameters: dict) -> tuple[str, list]:
|
||||
values = ", ".join([
|
||||
f"{item} = ?" for item in parameters
|
||||
])
|
||||
sql += f" {values}"
|
||||
|
||||
return sql, list(parameters.values())
|
||||
|
||||
|
||||
# Formatting a query with arguments
|
||||
def update_format_where(sql, parameters: dict) -> tuple[str, list]:
|
||||
sql += " WHERE "
|
||||
|
||||
sql += " AND ".join([
|
||||
f"{item} = ?" for item in parameters
|
||||
])
|
||||
|
||||
return sql, list(parameters.values())
|
||||
|
||||
|
||||
################################################################################
|
||||
################# BASIC CLASS FOR WORKING WITH DATABASE TABLES #################
|
||||
ModelTranslator = TypeVar("ModelTranslator", bound="ModelBase")
|
||||
|
||||
|
||||
class Databasex(Generic[ModelTranslator]):
|
||||
def __init__(self):
|
||||
self.storage_name = "storage"
|
||||
self.column_one = False
|
||||
self.table_model = None
|
||||
|
||||
# Delete entry
|
||||
def delete(self, **kwargs):
|
||||
with sqlite3.connect(PATH_DATABASE) as con:
|
||||
con.row_factory = dict_factory
|
||||
sql = f"DELETE FROM {self.storage_name}"
|
||||
sql, parameters = update_format_where(sql, kwargs)
|
||||
|
||||
con.execute(sql, parameters)
|
||||
|
||||
# Clear all entries
|
||||
def clear(self):
|
||||
with sqlite3.connect(PATH_DATABASE) as con:
|
||||
con.row_factory = dict_factory
|
||||
sql = f"DELETE FROM {self.storage_name}"
|
||||
|
||||
con.execute(sql)
|
||||
|
||||
# Get entry
|
||||
def get(self, **kwargs) -> Optional[ModelTranslator]:
|
||||
with sqlite3.connect(PATH_DATABASE) as con:
|
||||
con.row_factory = dict_factory
|
||||
sql = f"SELECT * FROM {self.storage_name}"
|
||||
sql, parameters = update_format_where(sql, kwargs)
|
||||
|
||||
response = con.execute(sql, parameters).fetchone()
|
||||
|
||||
if response is not None:
|
||||
response = self.table_model(**response)
|
||||
|
||||
return response
|
||||
|
||||
# Get entries
|
||||
def gets(self, **kwargs) -> list[Optional[ModelTranslator]]:
|
||||
with sqlite3.connect(PATH_DATABASE) as con:
|
||||
con.row_factory = dict_factory
|
||||
sql = f"SELECT * FROM {self.storage_name}"
|
||||
sql, parameters = update_format_where(sql, kwargs)
|
||||
|
||||
response = con.execute(sql, parameters).fetchall()
|
||||
|
||||
return [self.table_model(**cache_object) for cache_object in response]
|
||||
|
||||
# Get all entries
|
||||
def get_all(self) -> list[Optional[ModelTranslator]]:
|
||||
with sqlite3.connect(PATH_DATABASE) as con:
|
||||
con.row_factory = dict_factory
|
||||
sql = f"SELECT * FROM {self.storage_name}"
|
||||
|
||||
response = con.execute(sql).fetchall()
|
||||
|
||||
return [self.table_model(**cache_object) for cache_object in response]
|
||||
|
||||
# Edit entry
|
||||
def update(self, **kwargs):
|
||||
...
|
||||
|
||||
# Export column names, default data, and type hints
|
||||
def fields(self) -> dict[str, list[dict[str, Any]]]:
|
||||
for field_name, field_data in self.table_model.model_fields.items():
|
||||
field_name: str
|
||||
field_data: FieldInfo
|
||||
|
||||
column_data = [
|
||||
{
|
||||
'column': field_name,
|
||||
'type': field_data.annotation,
|
||||
'value': field_data.default,
|
||||
} for field_name, field_data in self.table_model.model_fields.items()
|
||||
]
|
||||
|
||||
return {'column_data': column_data, 'column_one': self.column_one}
|
||||
|
||||
|
||||
################################################################################
|
||||
################################################################################
|
||||
# Export all tables from the Database to a list
|
||||
def database_export_fields_db() -> list:
|
||||
with sqlite3.connect(PATH_DATABASE) as con:
|
||||
con.row_factory = dict_factory
|
||||
|
||||
database_tables = con.execute("SELECT * FROM sqlite_master where type='table'").fetchall()
|
||||
database_tables_db = [table['name'] for table in database_tables]
|
||||
|
||||
# If there are service tables, remove them from the list of available tables.
|
||||
if "sqlite_sequence" in database_tables_db:
|
||||
database_tables_db.remove("sqlite_sequence")
|
||||
if "sqlite_master" in database_tables_db:
|
||||
database_tables_db.remove("sqlite_master")
|
||||
|
||||
return database_tables_db
|
||||
|
||||
|
||||
# Export all tables, columns, and typehints from structures
|
||||
def database_export_fields_structure() -> dict:
|
||||
# Import all table models
|
||||
import tgbot.database as import_modeles
|
||||
|
||||
# Storing all structure files (db_X.py) in a single list
|
||||
all_tables = [
|
||||
getattr(import_modeles, name)
|
||||
for name in dir(import_modeles)
|
||||
if (
|
||||
isinstance(getattr(import_modeles, name), type) and
|
||||
(name.istitle() and name.endswith("x"))
|
||||
)
|
||||
]
|
||||
|
||||
save_tables = {} # Storage of all sorted and converted tables/columns
|
||||
|
||||
# Iterating through tables and columns
|
||||
for table in all_tables:
|
||||
table_fields = table().fields()
|
||||
|
||||
column_data = table_fields['column_data']
|
||||
column_one = table_fields['column_one']
|
||||
|
||||
cache_columns = []
|
||||
|
||||
# If the table name is not in the dictionary, add it.
|
||||
if table().storage_name not in save_tables:
|
||||
save_tables[table().storage_name] = {
|
||||
'column_one': column_one,
|
||||
'column_datas': [],
|
||||
}
|
||||
|
||||
# Column sorting
|
||||
for table_field in column_data:
|
||||
column_name = table_field['column']
|
||||
column_type = table_field['type'].__name__
|
||||
column_value = table_field['value']
|
||||
|
||||
# Replacing python date types with SQL
|
||||
column_type = column_type.replace(
|
||||
"str", "TEXT",
|
||||
).replace(
|
||||
"int", "INTEGER",
|
||||
).replace(
|
||||
"float", "REAL",
|
||||
)
|
||||
|
||||
# If there is no default value, replace it with - None
|
||||
if isinstance(column_value, PydanticUndefinedType):
|
||||
column_value = None
|
||||
|
||||
# Default values based on type hints
|
||||
if column_type == "TEXT":
|
||||
if "status" in column_name.lower():
|
||||
column_value = "False"
|
||||
else:
|
||||
column_value = "None"
|
||||
elif column_type == "INTEGER":
|
||||
column_value = 0
|
||||
elif column_type == "REAL":
|
||||
column_value = 0.0
|
||||
else:
|
||||
column_value = table_field['value']
|
||||
|
||||
save_tables[table().storage_name]['column_datas'].append(
|
||||
{
|
||||
'column': column_name,
|
||||
'type': column_type,
|
||||
'value': column_value,
|
||||
}
|
||||
)
|
||||
|
||||
return save_tables
|
||||
|
||||
|
||||
################################################################################
|
||||
################################################################################
|
||||
# Database migration
|
||||
def database_migration():
|
||||
with sqlite3.connect(PATH_DATABASE) as con:
|
||||
con.row_factory = dict_factory
|
||||
|
||||
# If there are values for tables or columns, perform migration.
|
||||
if len(CONST_MIGRATE_TABLES) > 0 or len(CONST_MIGRATE_COLUMNS) > 0:
|
||||
print(f"|| Migrating | Data was found")
|
||||
|
||||
# Export existing tables in the Database
|
||||
database_tables_db = database_export_fields_db()
|
||||
|
||||
# Renaming tables - iterating through tables
|
||||
for table_name_later, table_name_new in CONST_MIGRATE_TABLES.items():
|
||||
if table_name_later in database_tables_db:
|
||||
print(f"|| Migrating | Table '{table_name_later}' renamed to '{table_name_new}' (✓)")
|
||||
con.execute(f"ALTER TABLE {table_name_later} RENAME TO {table_name_new}")
|
||||
else:
|
||||
print(f"|| Migrating | Table '{table_name_later}' for renamed not found (X)")
|
||||
|
||||
# Renaming columns - iterating through tables with columns inside
|
||||
for table_name in CONST_MIGRATE_COLUMNS:
|
||||
if table_name in database_tables_db:
|
||||
# Unloading existing columns from the current table
|
||||
columns_export = [
|
||||
column['name'] for column in con.execute(f"PRAGMA table_info({table_name})").fetchall()
|
||||
]
|
||||
|
||||
# Browsing columns to rename them
|
||||
for column_name_later, column_name_new in CONST_MIGRATE_COLUMNS[table_name].items():
|
||||
if column_name_later in columns_export:
|
||||
print(
|
||||
f"|| Migrating | Column '{column_name_later}' renamed to '{column_name_new}' from table '{table_name}' (✓)"
|
||||
)
|
||||
con.execute(
|
||||
f"ALTER TABLE {table_name} RENAME COLUMN {column_name_later} TO {column_name_new}"
|
||||
)
|
||||
else:
|
||||
print(
|
||||
f"|| Migrating | Column '{column_name_later}' from table '{table_name}' not found (X)"
|
||||
)
|
||||
else:
|
||||
print(f"|| Migrating | Table '{table_name}' for renamed column(s) not found (X)")
|
||||
|
||||
else:
|
||||
print(f"|| Migrating | Data not found")
|
||||
|
||||
print(f"||")
|
||||
|
||||
|
||||
# Checking the relevance of columns (adding new ones, deleting old ones)
|
||||
def database_check_columns():
|
||||
with sqlite3.connect(PATH_DATABASE) as con:
|
||||
con.row_factory = dict_factory
|
||||
|
||||
symbol_log = True
|
||||
|
||||
# Exporting tables and columns from structures
|
||||
fields_structure = database_export_fields_structure()
|
||||
|
||||
for number, field in enumerate(fields_structure.items()):
|
||||
table_name: str = field[0] # Table name
|
||||
table_data: dict = field[1] # Table data
|
||||
|
||||
column_datas = table_data['column_datas'] # Column name, type, and value
|
||||
|
||||
columns_cache_add = [] # Columns that will be added to the database
|
||||
columns_cache_edit_columns = [] # Columns whose date types will be changed in the database
|
||||
columns_cache_edit_data = None # Which table columns will change date types in the database?
|
||||
|
||||
# Unloading existing columns from the current table
|
||||
column_datas_types_db = [column for column in con.execute(f"PRAGMA table_info({table_name})").fetchall()]
|
||||
|
||||
# Unloading existing column names from the current table
|
||||
columns_export_db = [column['name'] for column in column_datas_types_db]
|
||||
|
||||
# Iterating through columns from a structure
|
||||
for column_structure in column_datas:
|
||||
# If a column from the structure is present in the database table
|
||||
if column_structure['column'] in columns_export_db:
|
||||
columns_export_db.remove(column_structure['column'])
|
||||
|
||||
# Date type of the current column in the Structure
|
||||
column_type_structure = column_structure['type']
|
||||
|
||||
# Date type of the current column in the database
|
||||
column_type_database = [
|
||||
column for column in con.execute(f"PRAGMA table_info({table_name})").fetchall() if
|
||||
column_structure['column'] == column['name']
|
||||
]
|
||||
|
||||
# If the date column type (DB) is in the list, extract it.
|
||||
if len(column_type_database) > 0:
|
||||
column_type_database = column_type_database[0]['type']
|
||||
|
||||
# If the date types differ and have not been saved before, add them to the dictionary.
|
||||
if column_type_structure != column_type_database:
|
||||
if columns_cache_edit_data is None:
|
||||
columns_cache_edit_data = column_datas
|
||||
|
||||
if column_structure['column'] not in columns_cache_edit_columns:
|
||||
columns_cache_edit_columns.append(
|
||||
f"{column_structure['column']}: {column_type_database} -> {column_type_structure} "
|
||||
)
|
||||
else:
|
||||
# If a column from the structure is not present in the database, save it for addition.
|
||||
columns_cache_add.append(column_structure)
|
||||
|
||||
# The remaining columns that are present in the database table but are missing from the structure
|
||||
columns_cache_delete = columns_export_db
|
||||
|
||||
# Adding new columns
|
||||
for column in columns_cache_add:
|
||||
print(f"|| Modification | Column '{column['column']}' added to table '{table_name}'")
|
||||
con.execute(
|
||||
f"ALTER TABLE {table_name} ADD COLUMN {column['column']} {column['type']} DEFAULT '{column['value']}'"
|
||||
)
|
||||
|
||||
# Deleting existing obsolete columns
|
||||
for column in columns_cache_delete:
|
||||
print(f"|| Modification | Column '{column}' deleted from table '{table_name}'")
|
||||
con.execute(f"ALTER TABLE {table_name} DROP COLUMN {column}")
|
||||
|
||||
# Changing the date types of columns in a table
|
||||
sql_edit_columns_create_table = []
|
||||
sql_edit_columns_insert_values = []
|
||||
columns_cache_edit_columns = ", ".join(columns_cache_edit_columns)
|
||||
|
||||
if columns_cache_edit_data is not None:
|
||||
for column in columns_cache_edit_data:
|
||||
if column['type'] == "TEXT":
|
||||
sql_edit_columns_create_table.append(
|
||||
f"{column['column']} {column['type']} DEFAULT '{column['value']}'"
|
||||
)
|
||||
else:
|
||||
sql_edit_columns_create_table.append(
|
||||
f"{column['column']} {column['type']} DEFAULT {column['value']}"
|
||||
)
|
||||
|
||||
sql_edit_columns_insert_values.append(column['column'])
|
||||
|
||||
sql_edit_columns_create_table = ", ".join(sql_edit_columns_create_table)
|
||||
sql_edit_columns_create_table = f"CREATE TABLE {table_name}_new ({sql_edit_columns_create_table})"
|
||||
|
||||
sql_edit_columns_insert_values = ", ".join(sql_edit_columns_insert_values)
|
||||
sql_edit_columns_insert_values = f"INSERT INTO {table_name}_new ({sql_edit_columns_insert_values}) SELECT {sql_edit_columns_insert_values} FROM {table_name}"
|
||||
|
||||
con.execute(sql_edit_columns_create_table) # Creating a new table
|
||||
con.execute(sql_edit_columns_insert_values) # Transferring data from old columns to new ones
|
||||
con.execute(f"DROP TABLE {table_name};")
|
||||
con.execute(f"ALTER TABLE {table_name}_new RENAME TO {table_name};")
|
||||
|
||||
print(
|
||||
f"|| Modification | Columns '{columns_cache_edit_columns}' was edit data type for table '{table_name}'"
|
||||
)
|
||||
|
||||
|
||||
# Creating/deleting tables, columns, and records
|
||||
def database_initialization():
|
||||
print("======== DATABASE PROCESSING ========")
|
||||
database_migration() # Migration check
|
||||
|
||||
with sqlite3.connect(PATH_DATABASE) as con:
|
||||
con.row_factory = dict_factory
|
||||
|
||||
# List of exported tables and columns from structures
|
||||
database_fields = database_export_fields_structure()
|
||||
|
||||
# List of exported tables from the structure
|
||||
database_tables_import = [table for table in database_fields.keys()]
|
||||
|
||||
# List of tables available in the Database
|
||||
database_tables_db = database_export_fields_db()
|
||||
|
||||
# List of tables to be deleted that are not present in imports but are present in the database
|
||||
database_tables_delete = [
|
||||
table for table in database_tables_db if table not in database_tables_import
|
||||
]
|
||||
|
||||
# Browse the list of imported tables, columns, and default records to create them.
|
||||
for number, db_field in enumerate(database_fields.items()):
|
||||
table_name: str = db_field[0] # Table name
|
||||
table_data: dict = db_field[1] # Table data
|
||||
|
||||
column_one = table_data['column_one'] # Will the table have one record or more
|
||||
column_datas = table_data['column_datas'] # Column name, data type, and value
|
||||
column_len = len(column_datas) # Length (number) of all columns in the table
|
||||
|
||||
# Checking for the required number of records in a table
|
||||
if len(con.execute(f"PRAGMA table_info({table_name})").fetchall()) == column_len:
|
||||
print(f"|| Creating | Table '{table_name}' was found ({number + 1}/{len(database_fields)})")
|
||||
else:
|
||||
print(f"|| Creating | Table '{table_name}' not found ({number + 1}/{len(database_fields)})")
|
||||
|
||||
# Column names with type hints (user_id INTEGER, user_name TEXT)
|
||||
columns_cache_create = []
|
||||
|
||||
# Column names without type hints (user_id, user_name)
|
||||
columns_cache_insert = []
|
||||
|
||||
# Default data for automatic entry into columns (settings, payment systems)
|
||||
columns_values = []
|
||||
|
||||
############################################################
|
||||
##################### CREATING TABLES ######################
|
||||
for column in column_datas:
|
||||
column_name = column['column'] # Table name
|
||||
column_value = column['value'] # Default value in column
|
||||
column_type = column['type'] # Entry data type (INTEGER, TEXT, REAL, ...)
|
||||
|
||||
# Incrementable entry check
|
||||
if column_name == "increment":
|
||||
column_execute_create = "increment INTEGER PRIMARY KEY AUTOINCREMENT"
|
||||
else:
|
||||
column_execute_create = f"{column_name} {column_type}"
|
||||
|
||||
columns_cache_insert.append(column_name)
|
||||
columns_cache_create.append(column_execute_create)
|
||||
|
||||
# If default data is available, add it.
|
||||
if column_one and column_value is not None:
|
||||
columns_values.append(column_value)
|
||||
|
||||
# Collecting a request with all added columns
|
||||
column_execute_create = ",\n".join(columns_cache_create)
|
||||
column_execute_create = ded(f"""
|
||||
CREATE TABLE IF NOT EXISTS {table_name}(
|
||||
{column_execute_create}
|
||||
)
|
||||
""")
|
||||
|
||||
# Request to create a table and columns
|
||||
con.execute(column_execute_create)
|
||||
|
||||
############################################################
|
||||
####################### ADDING ENTRY #######################
|
||||
# If there is a record to add and the table supports only one record
|
||||
if column_one and len(columns_values) > 0:
|
||||
column_check_have = con.execute(f"SELECT * FROM {table_name}").fetchall()
|
||||
|
||||
# If there is no entry in the table, add it.
|
||||
if len(column_check_have) == 0:
|
||||
# Request collection
|
||||
column_execute_insert = ",\n".join(columns_cache_insert)
|
||||
column_execute_insert = ded(f"""
|
||||
INSERT INTO {table_name}(
|
||||
{column_execute_insert}
|
||||
)
|
||||
VALUES ({"?," * (len(columns_cache_insert) - 1) + "?"})
|
||||
""")
|
||||
|
||||
# Request to add entries to the table
|
||||
con.execute(column_execute_insert, columns_values)
|
||||
|
||||
print(f"|| Creating | Table '{table_name}' was created ({number + 1}/{len(database_fields)})")
|
||||
|
||||
# If there are tables to be deleted, delete them.
|
||||
if len(database_tables_delete) > 0:
|
||||
for table_name in database_tables_delete:
|
||||
print(f"|| Database | Table '{table_name}' was deleted")
|
||||
con.execute(f"DROP TABLE {table_name}")
|
||||
|
||||
print("||")
|
||||
database_check_columns() # Checking the relevance of columns
|
||||
print("=====================================")
|
||||
print()
|
||||
@@ -0,0 +1,36 @@
|
||||
# Example with migration - Tables
|
||||
#
|
||||
# CONST_MIGRATE_TABLES = {
|
||||
# 'old_table_name1': 'new_table_name1',
|
||||
# 'old_table_name2': 'new_table_name2',
|
||||
# }
|
||||
|
||||
|
||||
# Example with migration - Columns
|
||||
# CONST_MIGRATE_COLUMNS = {
|
||||
# 'name_of_the_table_in_which_the_column_is_located1': {
|
||||
# 'old_column_name11': 'new_column_name11',
|
||||
# 'old_column_name12': 'new_column_name12',
|
||||
# },
|
||||
# 'name_of_the_table_in_which_the_column_is_located2': {
|
||||
# 'old_column_name21': 'new_column_name21',
|
||||
# 'old_column_name22': 'new_column_name22',
|
||||
# },
|
||||
# }
|
||||
|
||||
|
||||
################################################################################
|
||||
############################### MIGRATION TABLES ###############################
|
||||
CONST_MIGRATE_TABLES = {
|
||||
# 'storage_refillme': 'storage_refill',
|
||||
# 'storage_purchasesme': 'storage_purchases',
|
||||
}
|
||||
|
||||
################################################################################
|
||||
############################### MIGRATION COLUMNS ##############################
|
||||
CONST_MIGRATE_COLUMNS = {
|
||||
# 'storage_refill': {
|
||||
# 'refill_methodme': 'refill_method',
|
||||
# 'refill_amountme': 'refill_amount',
|
||||
# },
|
||||
}
|
||||
@@ -1,88 +0,0 @@
|
||||
# - *- coding: utf- 8 - *-
|
||||
import sqlite3
|
||||
|
||||
from tgbot.data.config import PATH_DATABASE
|
||||
from tgbot.utils.const_functions import ded
|
||||
|
||||
|
||||
# Преобразование полученного списка в словарь
|
||||
def dict_factory(cursor, row) -> dict:
|
||||
save_dict = {}
|
||||
|
||||
for idx, col in enumerate(cursor.description):
|
||||
save_dict[col[0]] = row[idx]
|
||||
|
||||
return save_dict
|
||||
|
||||
|
||||
# Форматирование запроса без аргументов
|
||||
def update_format(sql, parameters: dict) -> tuple[str, list]:
|
||||
values = ", ".join([
|
||||
f"{item} = ?" for item in parameters
|
||||
])
|
||||
sql += f" {values}"
|
||||
|
||||
return sql, list(parameters.values())
|
||||
|
||||
|
||||
# Форматирование запроса с аргументами
|
||||
def update_format_where(sql, parameters: dict) -> tuple[str, list]:
|
||||
sql += " WHERE "
|
||||
|
||||
sql += " AND ".join([
|
||||
f"{item} = ?" for item in parameters
|
||||
])
|
||||
|
||||
return sql, list(parameters.values())
|
||||
|
||||
|
||||
################################################################################
|
||||
# Создание всех таблиц для БД
|
||||
def create_dbx():
|
||||
with sqlite3.connect(PATH_DATABASE) as con:
|
||||
con.row_factory = dict_factory
|
||||
|
||||
############################################################
|
||||
# Создание таблицы с хранением - пользователей
|
||||
if len(con.execute("PRAGMA table_info(storage_users)").fetchall()) == 7:
|
||||
print("DB was found(1/2)")
|
||||
else:
|
||||
con.execute(
|
||||
ded(f"""
|
||||
CREATE TABLE storage_users(
|
||||
increment INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
user_id INTEGER,
|
||||
user_login TEXT,
|
||||
user_name TEXT,
|
||||
user_surname TEXT,
|
||||
user_fullname TEXT,
|
||||
user_unix INTEGER
|
||||
)
|
||||
""")
|
||||
)
|
||||
print("DB was not found(1/2) | Creating...")
|
||||
|
||||
# Создание таблицы с хранением - настроек
|
||||
if len(con.execute("PRAGMA table_info(storage_settings)").fetchall()) == 1:
|
||||
print("DB was found(2/2)")
|
||||
else:
|
||||
con.execute(
|
||||
ded(f"""
|
||||
CREATE TABLE storage_settings(
|
||||
status_work TEXT
|
||||
)
|
||||
""")
|
||||
)
|
||||
|
||||
con.execute(
|
||||
ded(f"""
|
||||
INSERT INTO storage_settings(
|
||||
status_work
|
||||
)
|
||||
VALUES (?)
|
||||
"""),
|
||||
[
|
||||
'True',
|
||||
]
|
||||
)
|
||||
print("DB was not found(2/2) | Creating...")
|
||||
@@ -4,33 +4,40 @@ import sqlite3
|
||||
from pydantic import BaseModel
|
||||
|
||||
from tgbot.data.config import PATH_DATABASE
|
||||
from tgbot.database.db_helper import dict_factory, update_format
|
||||
from tgbot.database.adb_helper import Databasex, dict_factory, update_format
|
||||
|
||||
|
||||
# Модель таблицы
|
||||
class SettingsModel(BaseModel):
|
||||
status_work: str # Статус работы бота
|
||||
# Table model
|
||||
class ModelBase(BaseModel):
|
||||
status_work: str # Status bot work
|
||||
|
||||
class Config:
|
||||
json_schema_extra = {
|
||||
'storage_name': 'storage_settings'
|
||||
}
|
||||
|
||||
|
||||
# Работа с настройками
|
||||
class Settingsx:
|
||||
storage_name = "storage_settings"
|
||||
# Settings
|
||||
class Settingsx(Databasex[ModelBase]):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.table_model = ModelBase
|
||||
self.storage_name = self.table_model.model_json_schema()['storage_name']
|
||||
self.column_one = True
|
||||
|
||||
# Получение записи
|
||||
@staticmethod
|
||||
def get() -> SettingsModel:
|
||||
# Get entry
|
||||
def get(self) -> ModelBase:
|
||||
with sqlite3.connect(PATH_DATABASE) as con:
|
||||
con.row_factory = dict_factory
|
||||
sql = f"SELECT * FROM {Settingsx.storage_name}"
|
||||
sql = f"SELECT * FROM {self.storage_name}"
|
||||
|
||||
return SettingsModel(**con.execute(sql).fetchone())
|
||||
return ModelBase(**con.execute(sql).fetchone())
|
||||
|
||||
# Редактирование записи
|
||||
@staticmethod
|
||||
def update(**kwargs):
|
||||
# Edit entry
|
||||
def update(self, **kwargs):
|
||||
with sqlite3.connect(PATH_DATABASE) as con:
|
||||
con.row_factory = dict_factory
|
||||
sql = f"UPDATE {Settingsx.storage_name} SET"
|
||||
sql = f"UPDATE {self.storage_name} SET"
|
||||
sql, parameters = update_format(sql, kwargs)
|
||||
|
||||
con.execute(sql, parameters)
|
||||
|
||||
+28
-83
@@ -4,28 +4,37 @@ import sqlite3
|
||||
from pydantic import BaseModel
|
||||
|
||||
from tgbot.data.config import PATH_DATABASE
|
||||
from tgbot.database.db_helper import dict_factory, update_format_where, update_format
|
||||
from tgbot.database.adb_helper import Databasex, dict_factory, update_format
|
||||
from tgbot.utils.const_functions import get_unix, ded
|
||||
|
||||
|
||||
# Модель таблицы
|
||||
class UserModel(BaseModel):
|
||||
increment: int # Инкремент записи
|
||||
user_id: int # Айди пользователя
|
||||
user_login: str # Юзернейм пользователя
|
||||
user_name: str # Имя пользователя
|
||||
user_surname: str # Фамилия пользователя
|
||||
user_fullname: str # Полное имя + фамилия пользователя
|
||||
user_unix: int # Дата регистрации пользователя в боте (в UNIX времени)
|
||||
# Table model
|
||||
class ModelBase(BaseModel):
|
||||
increment: int # increment
|
||||
user_id: int # User id
|
||||
user_login: str # User username
|
||||
user_name: str # User name
|
||||
user_surname: str # User surname
|
||||
user_fullname: str # User fullname (name + surname)
|
||||
user_unix: int # Date of user registration in the bot (in UNIX time)
|
||||
|
||||
class Config:
|
||||
json_schema_extra = {
|
||||
'storage_name': 'storage_users'
|
||||
}
|
||||
|
||||
|
||||
# Работа с юзером
|
||||
class Userx:
|
||||
storage_name = "storage_users"
|
||||
# Users
|
||||
class Usersx(Databasex[ModelBase]):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.table_model = ModelBase
|
||||
self.storage_name = self.table_model.model_json_schema()['storage_name']
|
||||
self.column_one = False
|
||||
|
||||
# Добавление записи
|
||||
@staticmethod
|
||||
# Add entry
|
||||
def add(
|
||||
self,
|
||||
user_id: int,
|
||||
user_login: str,
|
||||
user_name: str,
|
||||
@@ -39,7 +48,7 @@ class Userx:
|
||||
|
||||
con.execute(
|
||||
ded(f"""
|
||||
INSERT INTO {Userx.storage_name} (
|
||||
INSERT INTO {self.storage_name} (
|
||||
user_id,
|
||||
user_login,
|
||||
user_name,
|
||||
@@ -58,76 +67,12 @@ class Userx:
|
||||
],
|
||||
)
|
||||
|
||||
# Получение записи
|
||||
@staticmethod
|
||||
def get(**kwargs) -> UserModel:
|
||||
# Edit entry
|
||||
def update(self, user_id: int, **kwargs):
|
||||
with sqlite3.connect(PATH_DATABASE) as con:
|
||||
con.row_factory = dict_factory
|
||||
sql = f"SELECT * FROM {Userx.storage_name}"
|
||||
sql, parameters = update_format_where(sql, kwargs)
|
||||
|
||||
response = con.execute(sql, parameters).fetchone()
|
||||
|
||||
if response is not None:
|
||||
response = UserModel(**response)
|
||||
|
||||
return response
|
||||
|
||||
# Получение записей
|
||||
@staticmethod
|
||||
def gets(**kwargs) -> list[UserModel]:
|
||||
with sqlite3.connect(PATH_DATABASE) as con:
|
||||
con.row_factory = dict_factory
|
||||
sql = f"SELECT * FROM {Userx.storage_name}"
|
||||
sql, parameters = update_format_where(sql, kwargs)
|
||||
|
||||
response = con.execute(sql, parameters).fetchall()
|
||||
|
||||
if len(response) >= 1:
|
||||
response = [UserModel(**cache_object) for cache_object in response]
|
||||
|
||||
return response
|
||||
|
||||
# Получение всех записей
|
||||
@staticmethod
|
||||
def get_all() -> list[UserModel]:
|
||||
with sqlite3.connect(PATH_DATABASE) as con:
|
||||
con.row_factory = dict_factory
|
||||
sql = f"SELECT * FROM {Userx.storage_name}"
|
||||
|
||||
response = con.execute(sql).fetchall()
|
||||
|
||||
if len(response) >= 1:
|
||||
response = [UserModel(**cache_object) for cache_object in response]
|
||||
|
||||
return response
|
||||
|
||||
# Редактирование записи
|
||||
@staticmethod
|
||||
def update(user_id, **kwargs):
|
||||
with sqlite3.connect(PATH_DATABASE) as con:
|
||||
con.row_factory = dict_factory
|
||||
sql = f"UPDATE {Userx.storage_name} SET"
|
||||
sql = f"UPDATE {self.storage_name} SET"
|
||||
sql, parameters = update_format(sql, kwargs)
|
||||
parameters.append(user_id)
|
||||
|
||||
con.execute(sql + "WHERE user_id = ?", parameters)
|
||||
|
||||
# Удаление записи
|
||||
@staticmethod
|
||||
def delete(**kwargs):
|
||||
with sqlite3.connect(PATH_DATABASE) as con:
|
||||
con.row_factory = dict_factory
|
||||
sql = f"DELETE FROM {Userx.storage_name}"
|
||||
sql, parameters = update_format_where(sql, kwargs)
|
||||
|
||||
con.execute(sql, parameters)
|
||||
|
||||
# Очистка всех записей
|
||||
@staticmethod
|
||||
def clear():
|
||||
with sqlite3.connect(PATH_DATABASE) as con:
|
||||
con.row_factory = dict_factory
|
||||
sql = f"DELETE FROM {Userx.storage_name}"
|
||||
|
||||
con.execute(sql)
|
||||
|
||||
@@ -2,25 +2,34 @@
|
||||
from aiogram.types import InlineKeyboardMarkup
|
||||
from aiogram.utils.keyboard import InlineKeyboardBuilder
|
||||
|
||||
from tgbot.data.config import get_admins
|
||||
from tgbot.utils.const_functions import ikb
|
||||
|
||||
|
||||
# Кнопки инлайн меню
|
||||
def menu_finl(user_id: int) -> InlineKeyboardMarkup:
|
||||
# Inline keyboards for users
|
||||
def user_finl() -> InlineKeyboardMarkup:
|
||||
keyboard = InlineKeyboardBuilder()
|
||||
|
||||
keyboard.row(
|
||||
ikb("User X", data="user_inline_x"),
|
||||
ikb("User 1", data="user_inline:user_btn"),
|
||||
ikb("User 2", data="..."),
|
||||
).row(
|
||||
ikb("User Unknown", data="unknown"),
|
||||
)
|
||||
|
||||
if user_id in get_admins():
|
||||
keyboard.row(
|
||||
ikb("Admin X", data="admin_inline_x"),
|
||||
ikb("Admin 1", data="admin_inline:admin_btn"),
|
||||
ikb("Admin 2", data="unknown"),
|
||||
)
|
||||
return keyboard.as_markup()
|
||||
|
||||
|
||||
# Inline keyboards for admins
|
||||
def admin_finl() -> InlineKeyboardMarkup:
|
||||
keyboard = InlineKeyboardBuilder()
|
||||
|
||||
keyboard.row(
|
||||
ikb("Admin X", data="admin_inline_x"),
|
||||
ikb("Admin 1", data="admin_inline:admin_btn"),
|
||||
ikb("Admin 2", data="..."),
|
||||
).row(
|
||||
ikb("Admin Unknown", data="unknown"),
|
||||
)
|
||||
|
||||
return keyboard.as_markup()
|
||||
|
||||
@@ -1,22 +0,0 @@
|
||||
# - *- coding: utf- 8 - *-
|
||||
from aiogram.utils.keyboard import InlineKeyboardBuilder
|
||||
|
||||
from tgbot.utils.const_functions import ikb
|
||||
|
||||
# Тестовые админ инлайн кнопки
|
||||
admin_inl = InlineKeyboardBuilder(
|
||||
).row(
|
||||
ikb("Admin Inline 1", data="..."),
|
||||
ikb("Admin Inline 2", data="..."),
|
||||
).row(
|
||||
ikb("Admin Inline 3", data="..."),
|
||||
).as_markup()
|
||||
|
||||
# Тестовые юзер инлайн кнопки
|
||||
user_inl = InlineKeyboardBuilder(
|
||||
).row(
|
||||
ikb("User Inline 1", data="..."),
|
||||
ikb("User Inline 2", data="..."),
|
||||
).row(
|
||||
ikb("User Inline 3", data="..."),
|
||||
).as_markup()
|
||||
@@ -11,12 +11,12 @@ def menu_frep(user_id: int) -> ReplyKeyboardMarkup:
|
||||
keyboard = ReplyKeyboardBuilder()
|
||||
|
||||
keyboard.row(
|
||||
rkb("User Inline"), rkb("User Reply"),
|
||||
rkb("User button"),
|
||||
)
|
||||
|
||||
if user_id in get_admins():
|
||||
keyboard.row(
|
||||
rkb("Admin Inline"), rkb("Admin Reply"),
|
||||
rkb("Admin button"),
|
||||
)
|
||||
|
||||
return keyboard.as_markup(resize_keyboard=True)
|
||||
|
||||
@@ -1,22 +0,0 @@
|
||||
# - *- coding: utf- 8 - *-
|
||||
from aiogram.utils.keyboard import ReplyKeyboardBuilder
|
||||
|
||||
from tgbot.utils.const_functions import rkb
|
||||
|
||||
# Тестовые админ реплай кнопки
|
||||
admin_rep = ReplyKeyboardBuilder(
|
||||
).row(
|
||||
rkb("Admin Reply 1"),
|
||||
rkb("Admin Reply 2"),
|
||||
).row(
|
||||
rkb("🔙 Main menu"),
|
||||
).as_markup(resize_keyboard=True)
|
||||
|
||||
# Тестовые юзер реплай кнопки
|
||||
user_rep = ReplyKeyboardBuilder(
|
||||
).row(
|
||||
rkb("User Reply 1"),
|
||||
rkb("User Reply 2"),
|
||||
).row(
|
||||
rkb("🔙 Main menu"),
|
||||
).as_markup(resize_keyboard=True)
|
||||
@@ -1,11 +1,11 @@
|
||||
# - *- coding: utf- 8 - *-
|
||||
from aiogram import Dispatcher
|
||||
|
||||
from tgbot.middlewares.middleware_user import ExistsUserMiddleware
|
||||
from tgbot.middlewares.middleware_throttling import ThrottlingMiddleware
|
||||
from tgbot.middlewares.middleware_user import ExistsUserMiddleware
|
||||
|
||||
|
||||
# Регистрация всех миддлварей
|
||||
# Register all middlewares
|
||||
def register_all_middlwares(dp: Dispatcher):
|
||||
dp.callback_query.outer_middleware(ExistsUserMiddleware())
|
||||
dp.message.outer_middleware(ExistsUserMiddleware())
|
||||
|
||||
@@ -8,7 +8,7 @@ from aiogram.types import Message, User
|
||||
from cachetools import TTLCache
|
||||
|
||||
|
||||
# Антиспам
|
||||
# Antiflood
|
||||
class ThrottlingMiddleware(BaseMiddleware):
|
||||
def __init__(self, default_rate: Union[int, float] = 0.5) -> None:
|
||||
self.default_rate = default_rate
|
||||
@@ -49,15 +49,7 @@ class ThrottlingMiddleware(BaseMiddleware):
|
||||
self.users[this_user.id]['count_throttled'] = 2
|
||||
self.users[this_user.id]['now_rate'] = self.default_rate + 3
|
||||
|
||||
await event.reply(
|
||||
"<b>❗ Пожалуйста, не спамьте.\n"
|
||||
"❗ Please, do not spam.</b>",
|
||||
)
|
||||
await event.reply("<b>❗ Please, do not spam.")
|
||||
elif self.users[this_user.id]['count_throttled'] == 2:
|
||||
self.users[this_user.id]['count_throttled'] = 3
|
||||
self.users[this_user.id]['now_rate'] = self.default_rate + 5
|
||||
|
||||
await event.reply(
|
||||
"<b>❗ Бот не будет отвечать до прекращения спама.\n"
|
||||
"❗ The bot will not respond until the spam stops.</b>",
|
||||
)
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
# - *- coding: utf- 8 - *-
|
||||
from aiogram import BaseMiddleware
|
||||
|
||||
from tgbot.database.db_users import Userx
|
||||
from tgbot.database.db_users import Usersx
|
||||
from tgbot.utils.const_functions import clear_html
|
||||
|
||||
|
||||
@@ -11,7 +11,7 @@ class ExistsUserMiddleware(BaseMiddleware):
|
||||
this_user = data.get("event_from_user")
|
||||
|
||||
if not this_user.is_bot:
|
||||
get_user = Userx.get(user_id=this_user.id)
|
||||
get_user = Usersx().get(user_id=this_user.id)
|
||||
|
||||
user_id = this_user.id
|
||||
user_login = this_user.username
|
||||
@@ -29,20 +29,20 @@ class ExistsUserMiddleware(BaseMiddleware):
|
||||
if len(user_surname) >= 1: user_fullname += f" {user_surname}"
|
||||
|
||||
if get_user is None:
|
||||
Userx.add(user_id, user_login.lower(), user_name, user_surname, user_fullname)
|
||||
Usersx().add(user_id, user_login.lower(), user_name, user_surname, user_fullname)
|
||||
else:
|
||||
if user_name != get_user.user_name:
|
||||
Userx.update(get_user.user_id, user_name=user_name)
|
||||
Usersx().update(get_user.user_id, user_name=user_name)
|
||||
|
||||
if user_surname != get_user.user_surname:
|
||||
Userx.update(get_user.user_id, user_surname=user_surname)
|
||||
Usersx().update(get_user.user_id, user_surname=user_surname)
|
||||
|
||||
if user_fullname != get_user.user_fullname:
|
||||
Userx.update(get_user.user_id, user_fullname=user_fullname)
|
||||
Usersx().update(get_user.user_id, user_fullname=user_fullname)
|
||||
|
||||
if user_login.lower() != get_user.user_login:
|
||||
Userx.update(get_user.user_id, user_login=user_login.lower())
|
||||
Usersx().update(get_user.user_id, user_login=user_login.lower())
|
||||
|
||||
data['User'] = Userx.get(user_id=user_id)
|
||||
data['User'] = Usersx().get(user_id=user_id)
|
||||
|
||||
return await handler(event, data)
|
||||
|
||||
+10
-10
@@ -7,9 +7,9 @@ from tgbot.routers.user import user_menu
|
||||
from tgbot.utils.misc.bot_filters import IsAdmin
|
||||
|
||||
|
||||
# Регистрация всех роутеров
|
||||
# Register all routers
|
||||
def register_all_routers(dp: Dispatcher):
|
||||
# Подключение фильтров
|
||||
# Connect filters
|
||||
main_errors.router.message.filter(F.chat.type == "private")
|
||||
main_start.router.message.filter(F.chat.type == "private")
|
||||
|
||||
@@ -18,13 +18,13 @@ def register_all_routers(dp: Dispatcher):
|
||||
|
||||
main_missed.router.message.filter(F.chat.type == "private")
|
||||
|
||||
# Подключение обязательных роутеров
|
||||
dp.include_router(main_errors.router) # Роутер ошибки
|
||||
dp.include_router(main_start.router) # Роутер основных команд
|
||||
# Connect needed routers
|
||||
dp.include_router(main_errors.router) # Router - error
|
||||
dp.include_router(main_start.router) # Router - main
|
||||
|
||||
# Подключение пользовательских роутеров (юзеров и админов)
|
||||
dp.include_router(user_menu.router) # Юзер роутер
|
||||
dp.include_router(admin_menu.router) # Админ роутер
|
||||
# Connect usebles routers (users и admins)
|
||||
dp.include_router(user_menu.router) # Router - user
|
||||
dp.include_router(admin_menu.router) # Router - admin
|
||||
|
||||
# Подключение обязательных роутеров
|
||||
dp.include_router(main_missed.router) # Роутер пропущенных апдейтов
|
||||
# Connect needed routers
|
||||
dp.include_router(main_missed.router) # Router - missed handlers
|
||||
|
||||
@@ -8,38 +8,32 @@ from aiogram.types import FSInputFile, Message, CallbackQuery
|
||||
from aiogram.utils.media_group import MediaGroupBuilder
|
||||
|
||||
from tgbot.data.config import PATH_DATABASE, PATH_LOGS
|
||||
from tgbot.database.db_users import UserModel
|
||||
from tgbot.keyboards.inline_misc import admin_inl
|
||||
from tgbot.keyboards.reply_misc import admin_rep
|
||||
from tgbot.database.db_users import BaseModel as UserModel
|
||||
from tgbot.keyboards.inline_main import admin_finl
|
||||
from tgbot.utils.const_functions import get_date
|
||||
from tgbot.utils.misc.bot_models import FSM, ARS
|
||||
|
||||
router = Router(name=__name__)
|
||||
|
||||
|
||||
# Кнопка - Admin Inline
|
||||
@router.message(F.text == 'Admin Inline')
|
||||
# Message - Admin button
|
||||
@router.message(F.text == 'Admin button')
|
||||
async def admin_button_inline(message: Message, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
|
||||
await state.clear()
|
||||
|
||||
await message.answer("Click Button - Admin Inline", reply_markup=admin_inl)
|
||||
await message.answer(
|
||||
"Inline admin keyboards",
|
||||
reply_markup=admin_finl()
|
||||
)
|
||||
|
||||
|
||||
# Кнопка - Admin Reply
|
||||
@router.message(F.text == 'Admin Reply')
|
||||
async def admin_button_reply(message: Message, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
|
||||
await state.clear()
|
||||
|
||||
await message.answer("Click Button - Admin Reply", reply_markup=admin_rep)
|
||||
|
||||
|
||||
# Колбэк - Admin X
|
||||
# Callback - Admin X
|
||||
@router.callback_query(F.data == 'admin_inline_x')
|
||||
async def admin_callback_inline_x(call: CallbackQuery, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
|
||||
await call.answer(f"Click Admin X")
|
||||
|
||||
|
||||
# Колбэк - Admin
|
||||
# Callback - Admin
|
||||
@router.callback_query(F.data.startswith('admin_inline:'))
|
||||
async def admin_callback_inline(call: CallbackQuery, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
|
||||
get_data = call.data.split(":")[1]
|
||||
@@ -47,7 +41,7 @@ async def admin_callback_inline(call: CallbackQuery, bot: Bot, state: FSM, arSes
|
||||
await call.answer(f"Click Admin - {get_data}", True)
|
||||
|
||||
|
||||
# Получение Базы Данных
|
||||
# Get Database file
|
||||
@router.message(Command(commands=['db', 'database']))
|
||||
async def admin_database(message: Message, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
|
||||
await state.clear()
|
||||
@@ -58,7 +52,7 @@ async def admin_database(message: Message, bot: Bot, state: FSM, arSession: ARS,
|
||||
)
|
||||
|
||||
|
||||
# Получение логов
|
||||
# Get Logs file
|
||||
@router.message(Command(commands=['log', 'logs']))
|
||||
async def admin_log(message: Message, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
|
||||
await state.clear()
|
||||
@@ -79,7 +73,7 @@ async def admin_log(message: Message, bot: Bot, state: FSM, arSession: ARS, User
|
||||
await message.answer_media_group(media=media_group.build())
|
||||
|
||||
|
||||
# Очистить логи
|
||||
# Clear logs file
|
||||
@router.message(Command(commands=['clear_log', 'clear_logs', 'log_clear', 'logs_clear']))
|
||||
async def admin_logs_clear(message: Message, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
|
||||
await state.clear()
|
||||
|
||||
@@ -8,14 +8,14 @@ from tgbot.utils.misc.bot_logging import bot_logger
|
||||
router = Router(name=__name__)
|
||||
|
||||
|
||||
# Ошибка с блокировкой бота пользователем
|
||||
# Error with block forbidden user
|
||||
# @router.errors(ExceptionTypeFilter(TelegramForbiddenError))
|
||||
# class MyHandler(ErrorHandler):
|
||||
# async def handle(self):
|
||||
# ...
|
||||
|
||||
|
||||
# Ошибка с редактированием одинакового сообщения
|
||||
# Error with edit duplicate message
|
||||
@router.errors(ExceptionMessageFilter(
|
||||
"Bad Request: message is not modified: specified new message content and reply markup are exactly the same as a current content and reply markup of the message")
|
||||
)
|
||||
|
||||
@@ -2,34 +2,34 @@
|
||||
from aiogram import Router, Bot, F
|
||||
from aiogram.types import CallbackQuery, Message
|
||||
|
||||
from tgbot.database.db_users import UserModel
|
||||
from tgbot.database import ModelUsers
|
||||
from tgbot.utils.const_functions import del_message
|
||||
from tgbot.utils.misc.bot_models import FSM, ARS
|
||||
|
||||
router = Router(name=__name__)
|
||||
|
||||
|
||||
# Колбэк с удалением сообщения
|
||||
# Callback with delete message
|
||||
@router.callback_query(F.data == 'close_this')
|
||||
async def main_callback_close(call: CallbackQuery, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
|
||||
async def main_callback_close(call: CallbackQuery, bot: Bot, state: FSM, arSession: ARS, User: ModelUsers):
|
||||
await del_message(call.message)
|
||||
|
||||
|
||||
# Колбэк с обработкой кнопки
|
||||
# Callback with processing miss button
|
||||
@router.callback_query(F.data == '...')
|
||||
async def main_callback_answer(call: CallbackQuery, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
|
||||
async def main_callback_answer(call: CallbackQuery, bot: Bot, state: FSM, arSession: ARS, User: ModelUsers):
|
||||
await call.answer(cache_time=30)
|
||||
|
||||
|
||||
# Колбэк с обработкой удаления сообщений потерявших стейт
|
||||
# Callback with processing miss callback data
|
||||
@router.callback_query()
|
||||
async def main_callback_missed(call: CallbackQuery, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
|
||||
async def main_callback_missed(call: CallbackQuery, bot: Bot, state: FSM, arSession: ARS, User: ModelUsers):
|
||||
await call.answer(f"❗️ Miss callback: {call.data}", True)
|
||||
|
||||
|
||||
# Обработка всех неизвестных команд
|
||||
# Processing all unknowns callback datas
|
||||
@router.message()
|
||||
async def main_message_missed(message: Message, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
|
||||
async def main_message_missed(message: Message, bot: Bot, state: FSM, arSession: ARS, User: ModelUsers):
|
||||
await message.answer(
|
||||
"♦️ Unknown command.\n"
|
||||
"♦️ Enter /start",
|
||||
|
||||
@@ -3,7 +3,7 @@ from aiogram import Router, Bot, F
|
||||
from aiogram.filters import Command
|
||||
from aiogram.types import Message
|
||||
|
||||
from tgbot.database.db_users import UserModel
|
||||
from tgbot.database.db_users import BaseModel as UserModel
|
||||
from tgbot.keyboards.reply_main import menu_frep
|
||||
from tgbot.utils.const_functions import ded
|
||||
from tgbot.utils.misc.bot_models import FSM, ARS
|
||||
@@ -11,8 +11,8 @@ from tgbot.utils.misc.bot_models import FSM, ARS
|
||||
router = Router(name=__name__)
|
||||
|
||||
|
||||
# Открытие главного меню
|
||||
@router.message(F.text.in_(('🔙 Main menu', '🔙 Return')))
|
||||
# Main menu
|
||||
@router.message(F.text.in_(('menu', 'return', 'start')))
|
||||
@router.message(Command(commands=['start']))
|
||||
async def main_start(message: Message, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
|
||||
await state.clear()
|
||||
@@ -20,7 +20,7 @@ async def main_start(message: Message, bot: Bot, state: FSM, arSession: ARS, Use
|
||||
await message.answer(
|
||||
ded(f"""
|
||||
🔸 Hello, {User.user_name}
|
||||
🔸 Enter /start or /inline
|
||||
🔸 Enter /start or /menu
|
||||
"""),
|
||||
reply_markup=menu_frep(message.from_user.id),
|
||||
)
|
||||
|
||||
@@ -3,55 +3,43 @@ from aiogram import Router, Bot, F
|
||||
from aiogram.filters import Command
|
||||
from aiogram.types import Message, CallbackQuery
|
||||
|
||||
from tgbot.database.db_users import UserModel
|
||||
from tgbot.keyboards.inline_main import menu_finl
|
||||
from tgbot.keyboards.inline_misc import user_inl
|
||||
from tgbot.keyboards.reply_misc import user_rep
|
||||
from tgbot.database.db_users import BaseModel as UserModel
|
||||
from tgbot.keyboards.inline_main import user_finl
|
||||
from tgbot.keyboards.reply_main import menu_frep
|
||||
from tgbot.utils.misc.bot_models import FSM, ARS
|
||||
|
||||
router = Router(name=__name__)
|
||||
|
||||
|
||||
# Кнопка - User Inline
|
||||
@router.message(F.text == 'User Inline')
|
||||
# Message - User button
|
||||
@router.message(F.text == 'User button')
|
||||
async def user_button_inline(message: Message, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
|
||||
await state.clear()
|
||||
|
||||
await message.answer(
|
||||
"Click Button - User Inline",
|
||||
reply_markup=user_inl,
|
||||
"Inline user keyboards",
|
||||
reply_markup=user_finl()
|
||||
)
|
||||
|
||||
|
||||
# Кнопка - User Reply
|
||||
@router.message(F.text == 'User Reply')
|
||||
async def user_button_reply(message: Message, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
|
||||
# Command - /menu
|
||||
@router.message(Command(commands="menu"))
|
||||
async def user_command_menu(message: Message, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
|
||||
await state.clear()
|
||||
|
||||
await message.answer(
|
||||
"Click Button - User Reply",
|
||||
reply_markup=user_rep,
|
||||
"Enter command - /menu",
|
||||
reply_markup=menu_frep(message.from_user.id),
|
||||
)
|
||||
|
||||
|
||||
# Команда - /inline
|
||||
@router.message(Command(commands="inline"))
|
||||
async def user_command_inline(message: Message, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
|
||||
await state.clear()
|
||||
|
||||
await message.answer(
|
||||
"Click command - /inline",
|
||||
reply_markup=menu_finl(message.from_user.id),
|
||||
)
|
||||
|
||||
|
||||
# Колбэк - User X
|
||||
# Callback - User X
|
||||
@router.callback_query(F.data == 'user_inline_x')
|
||||
async def user_callback_inline_x(call: CallbackQuery, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
|
||||
await call.answer(f"Click User X")
|
||||
|
||||
|
||||
# Колбэк - User
|
||||
# Callback - User
|
||||
@router.callback_query(F.data.startswith('user_inline:'))
|
||||
async def user_callback_inline(call: CallbackQuery, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
|
||||
get_data = call.data.split(":")[1]
|
||||
|
||||
@@ -7,19 +7,19 @@ from typing import Union
|
||||
|
||||
import pytz
|
||||
from aiogram import Bot
|
||||
from aiogram.types import InlineKeyboardButton, KeyboardButton, WebAppInfo, Message, InlineKeyboardMarkup, \
|
||||
ReplyKeyboardMarkup
|
||||
from aiogram.types import (InlineKeyboardButton, KeyboardButton, WebAppInfo, Message, InlineKeyboardMarkup,
|
||||
ReplyKeyboardMarkup)
|
||||
|
||||
from tgbot.data.config import get_admins, BOT_TIMEZONE
|
||||
|
||||
|
||||
######################################## AIOGRAM ########################################
|
||||
# Генерация реплай кнопки
|
||||
#################################### AIOGRAM ###################################
|
||||
# Generate replay button
|
||||
def rkb(text: str) -> KeyboardButton:
|
||||
return KeyboardButton(text=text)
|
||||
|
||||
|
||||
# Генерация инлайн кнопки
|
||||
# Generate inline button
|
||||
def ikb(text: str, data: str = None, url: str = None, switch: str = None, web: str = None) -> InlineKeyboardButton:
|
||||
if data is not None:
|
||||
return InlineKeyboardButton(text=text, callback_data=data)
|
||||
@@ -29,9 +29,11 @@ def ikb(text: str, data: str = None, url: str = None, switch: str = None, web: s
|
||||
return InlineKeyboardButton(text=text, switch_inline_query=switch)
|
||||
elif web is not None:
|
||||
return InlineKeyboardButton(text=text, web_app=WebAppInfo(url=web))
|
||||
else:
|
||||
raise "Unknown data"
|
||||
|
||||
|
||||
# Удаление сообщения с обработкой ошибки от телеграма
|
||||
# Deleting a message with error handling from Telegram
|
||||
async def del_message(message: Message):
|
||||
try:
|
||||
await message.delete()
|
||||
@@ -39,7 +41,7 @@ async def del_message(message: Message):
|
||||
...
|
||||
|
||||
|
||||
# Умная отправка сообщений (автоотправка сообщения с фото или без)
|
||||
# Smart messaging (automatic sending of messages with or without photos)
|
||||
async def smart_message(
|
||||
bot: Bot,
|
||||
user_id: int,
|
||||
@@ -62,7 +64,7 @@ async def smart_message(
|
||||
)
|
||||
|
||||
|
||||
# Отправка сообщения всем админам
|
||||
# Send a message to all administrators
|
||||
async def send_admins(bot: Bot, text: str, markup=None, not_me=0):
|
||||
for admin in get_admins():
|
||||
try:
|
||||
@@ -77,8 +79,8 @@ async def send_admins(bot: Bot, text: str, markup=None, not_me=0):
|
||||
...
|
||||
|
||||
|
||||
######################################## ПРОЧЕЕ ########################################
|
||||
# Удаление отступов в многострочной строке ("""text""")
|
||||
##################################### MISC #####################################
|
||||
# Removing indents in a multi-line string ("""text""")
|
||||
def ded(get_text: str) -> str:
|
||||
if get_text is not None:
|
||||
split_text = get_text.split("\n")
|
||||
@@ -98,7 +100,7 @@ def ded(get_text: str) -> str:
|
||||
return get_text
|
||||
|
||||
|
||||
# Очистка текста от HTML тэгов ('<b>test</b>' -> *b*test*/b*)
|
||||
# Cleaning text of HTML tags ('<b>test</b>' -> *b*test*/b*)
|
||||
def clear_html(get_text: str) -> str:
|
||||
if get_text is not None:
|
||||
if "</" in get_text: get_text = get_text.replace("<", "*")
|
||||
@@ -110,7 +112,7 @@ def clear_html(get_text: str) -> str:
|
||||
return get_text
|
||||
|
||||
|
||||
# Очистка пробелов в списке (['', 1, ' ', 2] -> [1, 2])
|
||||
# Cleaning up gaps in the list (['', 1, ' ', 2] -> [1, 2])
|
||||
def clear_list(get_list: list) -> list:
|
||||
while "" in get_list: get_list.remove("")
|
||||
while " " in get_list: get_list.remove(" ")
|
||||
@@ -122,12 +124,12 @@ def clear_list(get_list: list) -> list:
|
||||
return get_list
|
||||
|
||||
|
||||
# Разбив списка на несколько частей ([1, 2, 3, 4] 2 -> [[1, 2], [3, 4]])
|
||||
# Split the list into several parts ([1, 2, 3, 4] 2 -> [[1, 2], [3, 4]])
|
||||
def split_list(get_list: list, count: int) -> list[list]:
|
||||
return [get_list[i:i + count] for i in range(0, len(get_list), count)]
|
||||
|
||||
|
||||
# Получение текущей даты (True - дата с временем, False - дата без времени)
|
||||
# Get the current date (True - date with time, False - date without time)
|
||||
def get_date(full: bool = True) -> str:
|
||||
if full:
|
||||
return datetime.now(pytz.timezone(BOT_TIMEZONE)).strftime("%d.%m.%Y %H:%M:%S")
|
||||
@@ -135,7 +137,7 @@ def get_date(full: bool = True) -> str:
|
||||
return datetime.now(pytz.timezone(BOT_TIMEZONE)).strftime("%d.%m.%Y")
|
||||
|
||||
|
||||
# Получение текущего unix времени (True - время в наносекундах, False - время в секундах)
|
||||
# Get the current Unix time (True - time in nanoseconds, False - time in seconds)
|
||||
def get_unix(full: bool = False) -> int:
|
||||
if full:
|
||||
return time.time_ns()
|
||||
@@ -143,7 +145,7 @@ def get_unix(full: bool = False) -> int:
|
||||
return int(time.time())
|
||||
|
||||
|
||||
# Конвертация unix в дату и даты в unix
|
||||
# Converting Unix to date and dates to Unix
|
||||
def convert_date(from_time, full=True, second=True) -> Union[str, int]:
|
||||
from tgbot.data.config import BOT_TIMEZONE
|
||||
|
||||
@@ -194,7 +196,7 @@ def convert_date(from_time, full=True, second=True) -> Union[str, int]:
|
||||
return to_time
|
||||
|
||||
|
||||
# Генерация уникального айди
|
||||
# Generation of a unique ID
|
||||
def gen_id(len_id: int = 16) -> int:
|
||||
mac_address = uuid.getnode()
|
||||
time_unix = int(str(time.time_ns())[:len_id])
|
||||
@@ -203,8 +205,10 @@ def gen_id(len_id: int = 16) -> int:
|
||||
return mac_address + time_unix + random_int
|
||||
|
||||
|
||||
# Генерация пароля | default, number, letter, onechar
|
||||
# Password generation | default, number, letter, onechar
|
||||
def gen_password(len_password: int = 16, type_password: str = "default") -> str:
|
||||
char_password = list("1234567890abcdefghigklmnopqrstuvyxwzABCDEFGHIGKLMNOPQRSTUVYXWZ")
|
||||
|
||||
if type_password == "default":
|
||||
char_password = list("1234567890abcdefghigklmnopqrstuvyxwzABCDEFGHIGKLMNOPQRSTUVYXWZ")
|
||||
elif type_password == "letter":
|
||||
@@ -223,7 +227,7 @@ def gen_password(len_password: int = 16, type_password: str = "default") -> str:
|
||||
return random_chars
|
||||
|
||||
|
||||
# Дополнение к числу корректного времени (1 -> 1 день, 3 -> 3 дня)
|
||||
# Addition to the correct time (1 -> 1 day, 3 -> 3 days)
|
||||
def convert_times(get_time: int, get_type: str = "day") -> str:
|
||||
get_time = int(get_time)
|
||||
if get_time < 0: get_time = 0
|
||||
@@ -251,7 +255,7 @@ def convert_times(get_time: int, get_type: str = "day") -> str:
|
||||
return f"{get_time} {get_list[count]}"
|
||||
|
||||
|
||||
# Проверка на булевый тип
|
||||
# Boolean type check
|
||||
def is_bool(value: Union[bool, str, int]) -> bool:
|
||||
value = str(value).lower()
|
||||
|
||||
@@ -263,8 +267,8 @@ def is_bool(value: Union[bool, str, int]) -> bool:
|
||||
raise ValueError(f"invalid truth value {value}")
|
||||
|
||||
|
||||
######################################## ЧИСЛА ########################################
|
||||
# Преобразование экспоненциальных чисел в читаемый вид (1e-06 -> 0.000001)
|
||||
################################### NUMBERS ####################################
|
||||
# Converting exponential numbers to a readable form (1e-06 -> 0.000001)
|
||||
def snum(amount: Union[int, float], remains: int = 2) -> str:
|
||||
format_str = "{:." + str(remains) + "f}"
|
||||
str_amount = format_str.format(float(amount))
|
||||
@@ -284,7 +288,7 @@ def snum(amount: Union[int, float], remains: int = 2) -> str:
|
||||
return str(str_amount)
|
||||
|
||||
|
||||
# Конвертация любого числа в вещественное, с удалением нулей в конце (remains - округление)
|
||||
# Convert any number to a real number, removing trailing zeros (remains - rounding)
|
||||
def to_float(get_number, remains: int = 2) -> Union[int, float]:
|
||||
if "," in str(get_number):
|
||||
get_number = str(get_number).replace(",", ".")
|
||||
@@ -313,7 +317,7 @@ def to_float(get_number, remains: int = 2) -> Union[int, float]:
|
||||
return get_number
|
||||
|
||||
|
||||
# Конвертация вещественного числа в целочисленное
|
||||
# Converting a real number to an integer
|
||||
def to_int(get_number: float) -> int:
|
||||
if "," in get_number:
|
||||
get_number = str(get_number).replace(",", ".")
|
||||
@@ -323,7 +327,7 @@ def to_int(get_number: float) -> int:
|
||||
return get_number
|
||||
|
||||
|
||||
# Проверка ввода на число
|
||||
# Data validation for numbers
|
||||
def is_number(get_number: Union[str, int, float]) -> bool:
|
||||
if str(get_number).isdigit():
|
||||
return True
|
||||
@@ -337,7 +341,7 @@ def is_number(get_number: Union[str, int, float]) -> bool:
|
||||
return False
|
||||
|
||||
|
||||
# Преобразование числа в читаемый вид (123456789 -> 123 456 789)
|
||||
# Converting a number to a readable form (123456789 -> 123,456,789)
|
||||
def format_rate(amount: Union[float, int], around: int = 2) -> str:
|
||||
if "," in str(amount): amount = float(str(amount).replace(",", "."))
|
||||
if " " in str(amount): amount = float(str(amount).replace(" ", ""))
|
||||
|
||||
@@ -4,22 +4,22 @@ from aiogram.types import BotCommand, BotCommandScopeChat, BotCommandScopeDefaul
|
||||
|
||||
from tgbot.data.config import get_admins
|
||||
|
||||
# Команды для юзеров
|
||||
# Commands for users
|
||||
user_commands = [
|
||||
BotCommand(command="start", description="♻️ Restart bot"),
|
||||
BotCommand(command="inline", description="🌀 Get Inline keyboard"),
|
||||
BotCommand(command="menu", description="🌀 Get keyboards"),
|
||||
]
|
||||
|
||||
# Команды для админов
|
||||
# Commands for admins
|
||||
admin_commands = [
|
||||
BotCommand(command="start", description="♻️ Restart bot"),
|
||||
BotCommand(command="inline", description="🌀 Get Inline keyboard"),
|
||||
BotCommand(command="menu", description="🌀 Get keyboards"),
|
||||
BotCommand(command="log", description="🖨 Get Logs"),
|
||||
BotCommand(command="db", description="📦 Get Database"),
|
||||
]
|
||||
|
||||
|
||||
# Установка команд
|
||||
# Set commands
|
||||
async def set_commands(bot: Bot):
|
||||
await bot.set_my_commands(user_commands, scope=BotCommandScopeDefault())
|
||||
|
||||
|
||||
@@ -5,7 +5,7 @@ from aiogram.types import Message
|
||||
from tgbot.data.config import get_admins
|
||||
|
||||
|
||||
# Проверка на админа
|
||||
# Filter on admin right
|
||||
class IsAdmin(BaseFilter):
|
||||
async def __call__(self, message: Message) -> bool:
|
||||
if message.from_user.id in get_admins():
|
||||
|
||||
@@ -5,24 +5,24 @@ import colorlog
|
||||
|
||||
from tgbot.data.config import PATH_LOGS
|
||||
|
||||
# Формат логгирования
|
||||
# Logging format
|
||||
log_formatter_file = bot_logger.Formatter("%(levelname)s | %(asctime)s | %(filename)s:%(lineno)d | %(message)s")
|
||||
log_formatter_console = colorlog.ColoredFormatter(
|
||||
"%(purple)s%(levelname)s %(blue)s|%(purple)s %(asctime)s %(blue)s|%(purple)s %(filename)s:%(lineno)d %(blue)s|%(purple)s %(message)s%(red)s",
|
||||
datefmt="%d-%m-%Y %H:%M:%S",
|
||||
)
|
||||
|
||||
# Логгирование в файл logs.log
|
||||
# Logging in file logs.log
|
||||
file_handler = bot_logger.FileHandler(PATH_LOGS, "w", "utf-8")
|
||||
file_handler.setFormatter(log_formatter_file)
|
||||
file_handler.setLevel(bot_logger.INFO)
|
||||
|
||||
# Логгирование в консоль
|
||||
# Logging in console
|
||||
console_handler = bot_logger.StreamHandler()
|
||||
console_handler.setFormatter(log_formatter_console)
|
||||
console_handler.setLevel(bot_logger.CRITICAL)
|
||||
|
||||
# Подключение настроек логгирования
|
||||
# Connect logging settings
|
||||
bot_logger.basicConfig(
|
||||
format="%(levelname)s | %(asctime)s | %(filename)s:%(lineno)d | %(message)s",
|
||||
handlers=[
|
||||
|
||||
@@ -6,13 +6,13 @@ from tgbot.data.config import get_admins, PATH_DATABASE, BOT_STATUS_NOTIFICATION
|
||||
from tgbot.utils.const_functions import get_date, send_admins
|
||||
|
||||
|
||||
# Выполнение функции после запуска бота (рассылка админам о запуске бота)
|
||||
# Notification after run bot (for all admins)
|
||||
async def startup_notify(bot: Bot):
|
||||
if len(get_admins()) >= 1 and BOT_STATUS_NOTIFICATION:
|
||||
await send_admins(bot, "<b>✅ Bot was started</b>")
|
||||
|
||||
|
||||
# Автоматические бэкапы БД
|
||||
# Autobackup Database
|
||||
async def autobackup_admin(bot: Bot):
|
||||
for admin in get_admins():
|
||||
try:
|
||||
|
||||
Reference in New Issue
Block a user