Compare commits
No commits in common. "3d4e9c3d7f8680abec64f937b7a85866b1537af0" and "2d5e09e54853cbbd3843cf3253cc51a024e0f3e8" have entirely different histories.
3d4e9c3d7f
...
2d5e09e548
|
@ -1,6 +0,0 @@
|
|||
.idea
|
||||
venv
|
||||
probe.*
|
||||
__pycache__
|
||||
*.egg-info
|
||||
dist
|
|
@ -1,187 +0,0 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
from hellmbot.env import ENV
|
||||
from sqlite3 import connect as sqlite
|
||||
from typing import Any, Iterable
|
||||
from enum import Enum
|
||||
|
||||
|
||||
##############################
|
||||
# sqlite Database Management #
|
||||
##############################
|
||||
|
||||
|
||||
class DBColumnsTypes(Enum):
|
||||
"""
|
||||
Data types for sqlite table columns
|
||||
"""
|
||||
stroke = "TEXT"
|
||||
integer_number = "INTEGER"
|
||||
float_number = "FLOAT"
|
||||
|
||||
|
||||
class DataBase(object):
|
||||
"""
|
||||
Describes the interaction with the database sqlite3
|
||||
"""
|
||||
DB_PATH = ENV.DB_PATH
|
||||
|
||||
@staticmethod
|
||||
def __getcolumns(columns: dict[str: DBColumnsTypes]) -> str:
|
||||
"""
|
||||
Converts dictionary to a string for a sqlite query
|
||||
|
||||
:param columns: Dict["column_title": ColumnType]
|
||||
:return: Part of sqlite query
|
||||
"""
|
||||
keys = tuple(columns.keys())
|
||||
values = tuple(columns.values())
|
||||
lst = zip(keys, values)
|
||||
lst = [f"{column_title.lower()} {column_type.value}" for column_title, column_type in lst]
|
||||
return ", ".join(lst)
|
||||
|
||||
def execute(self, request: str, params: list[Any] = None) -> None:
|
||||
"""
|
||||
Executes sqlite query
|
||||
|
||||
:param request: Query text
|
||||
:param params: Additional parameters for the request
|
||||
"""
|
||||
with sqlite(self.DB_PATH) as db:
|
||||
cursor = db.cursor()
|
||||
if params is None:
|
||||
cursor.execute(request)
|
||||
return
|
||||
cursor.execute(request, params)
|
||||
db.commit()
|
||||
|
||||
def gentable(self, table: str, columns: dict[str: DBColumnsTypes]) -> None:
|
||||
"""
|
||||
Generates table in the database
|
||||
|
||||
:param table: Name of the table
|
||||
:param columns: Columns in the table
|
||||
"""
|
||||
request = f"CREATE TABLE IF NOT EXISTS {table} (id INTEGER PRIMARY KEY, {self.__getcolumns(columns)})"
|
||||
self.execute(request)
|
||||
|
||||
def insert(self, table: str, items: dict[str: Any]) -> None:
|
||||
"""
|
||||
Inserts data into a field of a database table
|
||||
|
||||
:param table: Name of the table
|
||||
:param items: Dict["column_title": value]
|
||||
"""
|
||||
columns = [title.lower() for title in tuple(items.keys())]
|
||||
columns = ", ".join(columns)
|
||||
values = list(items.values())
|
||||
request = f"INSERT INTO {table} ({columns}) VALUES ({("?, " * len(values))[:-2]})"
|
||||
self.execute(request, values)
|
||||
|
||||
def get(self, table: str, column: list[str] = None, where: dict[str: Any] = None, order: str = None) -> list[Any]:
|
||||
"""
|
||||
Returns data from sqlite table
|
||||
|
||||
:param table: Name of the table
|
||||
:param column: List of the columns titles
|
||||
:param where: Dict['column_title': value]
|
||||
:param order: sqlite query order syntax stroke
|
||||
:return: List of values from table
|
||||
"""
|
||||
args = None
|
||||
if column is None: # if no column is specified, all columns are used by default
|
||||
column = "*"
|
||||
else:
|
||||
column = ", ".join(i.lower() for i in column)
|
||||
request = [f"SELECT {column} FROM {table}"]
|
||||
if where is not None:
|
||||
args = list(where.values())
|
||||
where = f"WHERE {", ".join(tuple(where.keys()))} = ({("?, " * len(args))[:-2]})"
|
||||
request.append(where)
|
||||
if order is not None:
|
||||
order = f"ORDER BY {order}"
|
||||
request.append(order)
|
||||
request = [" ".join(request)]
|
||||
if args is not None:
|
||||
request.append(args)
|
||||
with sqlite(self.DB_PATH) as db:
|
||||
cursor = db.cursor()
|
||||
cursor.execute(*request)
|
||||
return cursor.fetchall()
|
||||
|
||||
def delete(self, table: str, where: dict[str: Any]) -> None:
|
||||
values = list(where.values())
|
||||
request = f"DELETE FROM {table} WHERE {", ".join(tuple(where.keys()))} = ({("?, " * len(values))[:-2]})"
|
||||
self.execute(request, values)
|
||||
|
||||
|
||||
class ServersDB(DataBase):
|
||||
"""
|
||||
Describes the server channel table in the database
|
||||
|
||||
Attributes:
|
||||
- channels (tuple) - Tuple of server channel id's
|
||||
"""
|
||||
TABLE = "servers" # Table name constant
|
||||
|
||||
def __init__(self, server_id: int) -> None:
|
||||
"""
|
||||
Init Server DataBase object
|
||||
|
||||
:param server_id: id of discord server
|
||||
"""
|
||||
self.gentable(self.TABLE, {
|
||||
"server_id": DBColumnsTypes.integer_number, # id of discord server
|
||||
"channel_id": DBColumnsTypes.integer_number, # id of discord voice channel id
|
||||
"loop": DBColumnsTypes.integer_number # Channel sequence number
|
||||
})
|
||||
self.server = server_id
|
||||
|
||||
def check_server_exists(self) -> bool:
|
||||
"""
|
||||
Checks the existence of the server in the database table
|
||||
"""
|
||||
lst = self.get(self.TABLE, ["server_id"])
|
||||
if len(lst) > 0:
|
||||
return True
|
||||
return False
|
||||
|
||||
def add_channel(self, channel_id: int) -> None:
|
||||
"""
|
||||
Adds channel to the database
|
||||
|
||||
:param channel_id: id of discord channel
|
||||
"""
|
||||
self.insert(self.TABLE, {
|
||||
"server_id": self.server,
|
||||
"channel_id": channel_id
|
||||
})
|
||||
|
||||
@property
|
||||
def channels(self) -> tuple:
|
||||
"""
|
||||
Returns tuple of server channel id's
|
||||
|
||||
:return: tuple of server channel id's
|
||||
"""
|
||||
lst = self.get(self.TABLE, ["channel_id"], {"server_id": self.server}, "loop")
|
||||
if len(lst) == 0:
|
||||
raise IndexError("This server has no added channels")
|
||||
return tuple(map(lambda element: element[0], lst))
|
||||
|
||||
def clear_channels(self) -> None:
|
||||
"""
|
||||
Deletes all server channels from the database
|
||||
"""
|
||||
self.delete(self.TABLE, {"server_id": self.server})
|
||||
|
||||
def __iter__(self) -> Iterable[int]:
|
||||
"""
|
||||
Returns iterable object of server channel id's
|
||||
"""
|
||||
return iter(self.channels)
|
||||
|
||||
def __bool__(self) -> bool:
|
||||
"""
|
||||
Checks the existence of the server in the database table
|
||||
"""
|
||||
return self.check_server_exists()
|
|
@ -1,29 +0,0 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
###################
|
||||
# Actions on user #
|
||||
###################
|
||||
|
||||
class User(object):
|
||||
"""
|
||||
Describes Discord user
|
||||
|
||||
Attributes:
|
||||
- user (int): Discord id
|
||||
"""
|
||||
|
||||
def __init__(self, user_id: int) -> None:
|
||||
"""
|
||||
Init Discord user object
|
||||
|
||||
:param user_id: user discord id
|
||||
"""
|
||||
self.user = user_id
|
||||
|
||||
def move_to(self, channel_id: int) -> None: # ToDo
|
||||
"""
|
||||
Moves the user to the specified voice channel
|
||||
|
||||
:param channel_id: discord id of voice channel
|
||||
"""
|
||||
pass
|
|
@ -1,23 +0,0 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
from os import environ
|
||||
|
||||
|
||||
############################################################################
|
||||
# Getting values from the system environment / defining standard constants #
|
||||
############################################################################
|
||||
|
||||
|
||||
class ENV(object):
|
||||
|
||||
@property
|
||||
def DB_PATH(self) -> str:
|
||||
"""
|
||||
Sets the path to the sqlite database file
|
||||
|
||||
:return: Path to DataBase file
|
||||
"""
|
||||
default = "/etc/hellmbot/database.sqlite"
|
||||
env = environ.get("DB_PATH")
|
||||
if env is None:
|
||||
return default
|
||||
return env
|
Loading…
Reference in New Issue