78 lines
2.1 KiB
Python
78 lines
2.1 KiB
Python
# Импорт сторонних модулей
|
|
import psycopg2
|
|
|
|
# Импорт модулей приложения
|
|
from . import logger
|
|
|
|
|
|
class DataBase(object):
|
|
"""..."""
|
|
|
|
def __init__(self, host: str, port: int, name: str, user: str, password: str):
|
|
super(DataBase, self).__init__()
|
|
self.host = host
|
|
self.port = port
|
|
self.name = name
|
|
self.user = user
|
|
self.password = password
|
|
|
|
self.conn = None
|
|
|
|
def connect(self) -> bool:
|
|
"""..."""
|
|
|
|
try:
|
|
self.conn = psycopg2.connect(
|
|
host=self.host,
|
|
port=self.port,
|
|
user=self.user,
|
|
password=self.password,
|
|
database=self.name,
|
|
)
|
|
|
|
return True
|
|
except:
|
|
return False
|
|
|
|
def check_issues_count(self, telegram_id: int) -> bool:
|
|
"""..."""
|
|
|
|
sql = 'SELECT id FROM issues WHERE telegram_id = %s;'
|
|
|
|
cur = self.conn.cursor()
|
|
|
|
cur.execute(sql, (telegram_id,))
|
|
|
|
result = cur.fetchall()
|
|
|
|
if result == None:
|
|
return 0
|
|
else:
|
|
return len(result)
|
|
|
|
def add_issue(self, telegram_id: int, message: str) -> dict:
|
|
"""..."""
|
|
|
|
sql = 'INSERT INTO issues ' '(telegram_id, message)' ' VALUES (%s, %s);'
|
|
|
|
total_issues = self.check_issues_count(telegram_id)
|
|
if total_issues >= 3:
|
|
return {
|
|
'status': False,
|
|
'message': '⛔ Произошла ошибка.\n' 'Вы оставили много запросов, чтобы я мог оперативно их обработать.',
|
|
}
|
|
|
|
cur = self.conn.cursor()
|
|
|
|
try:
|
|
cur.execute(sql, (telegram_id, message))
|
|
self.conn.commit()
|
|
|
|
return {'status': True, 'message': '🔔 Замечательно, Ваш запросов отправлен автору.\nЖдите ответа 🙃'}
|
|
except Exception as e:
|
|
logger.error(e)
|
|
return {'status': False, 'message': '⛔ Произошла ошибка'}
|
|
|
|
def close(self) -> None:
|
|
self.conn.close()
|