253 lines
8.3 KiB
Python
253 lines
8.3 KiB
Python
from flask_sqlalchemy import SQLAlchemy
|
|
from datetime import datetime
|
|
import secrets
|
|
from argon2 import PasswordHasher
|
|
from argon2.exceptions import VerifyMismatchError
|
|
|
|
db = SQLAlchemy()
|
|
ph = PasswordHasher()
|
|
|
|
|
|
class MedodsAPI(db.Model):
|
|
__tablename__ = "medods_api"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
url = db.Column(db.Text)
|
|
identity = db.Column(db.Text)
|
|
secretKey = db.Column(db.Text)
|
|
created_at = db.Column(db.DateTime, default=datetime.now)
|
|
updated_at = db.Column(db.DateTime, default=datetime.now, onupdate=datetime.now)
|
|
|
|
|
|
class ApiEndpoint(db.Model):
|
|
__tablename__ = "api_endpoints"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
|
|
method = db.Column(db.String(10), nullable=False) # GET / POST
|
|
title = db.Column(db.String(255), nullable=False) # Человеческое описание
|
|
url_path = db.Column(db.String(255), nullable=False) # /users, /scheduler
|
|
|
|
payload = db.Column(db.JSON, default=dict) # Тело запроса
|
|
query_params = db.Column(db.JSON, default=dict) # Query-параметры
|
|
|
|
created_at = db.Column(db.DateTime, default=datetime.now)
|
|
updated_at = db.Column(db.DateTime, default=datetime.now, onupdate=datetime.now)
|
|
|
|
def __repr__(self):
|
|
return f"<ApiEndpoint {self.method} {self.url_path}>"
|
|
|
|
def toDict(self):
|
|
return {
|
|
"id": self.id,
|
|
"method": self.method,
|
|
"title": self.title,
|
|
"url_path": self.url_path,
|
|
"payload": self.payload,
|
|
"query_params": self.query_params,
|
|
"created_at": self.created_at.strftime("%Y-%m-%d %H:%M:%S"),
|
|
"updated_at": self.updated_at.strftime("%Y-%m-%d %H:%M:%S"),
|
|
}
|
|
|
|
|
|
class VkAPI(db.Model):
|
|
__tablename__ = "vk_settings"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
group_id = db.Column(db.Integer)
|
|
access_token = db.Column(db.Text)
|
|
base_photo_url = db.Column(db.Integer)
|
|
created_at = db.Column(db.DateTime, default=datetime.now)
|
|
updated_at = db.Column(db.DateTime, default=datetime.now, onupdate=datetime.now)
|
|
|
|
def toDict(self):
|
|
return {
|
|
"id": self.id,
|
|
"group_id": self.group_id,
|
|
"access_token": self.access_token,
|
|
"base_photo_url": self.base_photo_url,
|
|
"created_at": self.created_at.strftime("%Y-%m-%d %H:%M:%S"),
|
|
"updated_at": self.updated_at.strftime("%Y-%m-%d %H:%M:%S"),
|
|
}
|
|
|
|
|
|
class UsersBirthdate(db.Model):
|
|
__tablename__ = "users_birthdate"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
enabled = db.Column(db.Boolean)
|
|
name = db.Column(db.Text)
|
|
short_name = db.Column(db.Text)
|
|
sex = db.Column(db.Text)
|
|
birthdate = db.Column(db.Date)
|
|
photo_link = db.Column(db.Text, nullable=True)
|
|
congratulations = db.Column(db.Text, nullable=True)
|
|
specialties = db.Column(db.JSON)
|
|
post_link = db.Column(db.Text, nullable=True)
|
|
publish_at = db.Column(db.DateTime, nullable=True)
|
|
created_at = db.Column(db.DateTime, default=datetime.now)
|
|
updated_at = db.Column(db.DateTime, default=datetime.now, onupdate=datetime.now)
|
|
|
|
def toDict(self):
|
|
return {
|
|
"id": self.id,
|
|
"enabled": self.enabled,
|
|
"name": self.name,
|
|
"shortName": self.short_name,
|
|
"sex": self.sex,
|
|
"birthdate": self.birthdate.strftime("%Y-%m-%d"),
|
|
"photoLink": self.photo_link,
|
|
"congratulations": self.congratulations,
|
|
"specialties": self.specialties,
|
|
"postLink": self.post_link,
|
|
"publishAt": (
|
|
self.publish_at.strftime("%Y-%m-%d %H:%M:%S")
|
|
if self.publish_at
|
|
else None
|
|
),
|
|
"created_at": self.created_at.strftime("%Y-%m-%d %H:%M:%S"),
|
|
"updated_at": self.updated_at.strftime("%Y-%m-%d %H:%M:%S"),
|
|
}
|
|
|
|
|
|
class BirthdateScheduler(db.Model):
|
|
__tablename__ = "birthdate_scheduler"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
hour = db.Column(db.Integer)
|
|
minute = db.Column(db.Integer)
|
|
enabled = db.Column(db.Boolean)
|
|
created_at = db.Column(db.DateTime, default=datetime.now)
|
|
updated_at = db.Column(db.DateTime, default=datetime.now, onupdate=datetime.now)
|
|
|
|
def toDict(self):
|
|
return {
|
|
"id": self.id,
|
|
"hour": self.hour,
|
|
"minute": self.minute,
|
|
"enabled": self.enabled,
|
|
"created_at": self.created_at.strftime("%Y-%m-%d %H:%M:%S"),
|
|
"updated_at": self.updated_at.strftime("%Y-%m-%d %H:%M:%S"),
|
|
}
|
|
|
|
|
|
class UsersMedods(db.Model):
|
|
__tablename__ = "users_medods"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
name = db.Column(db.Text)
|
|
short_name = db.Column(db.Text)
|
|
sex = db.Column(db.Text)
|
|
step = db.Column(db.Integer)
|
|
specialties = db.Column(db.JSON)
|
|
created_at = db.Column(db.DateTime, default=datetime.now)
|
|
updated_at = db.Column(db.DateTime, default=datetime.now, onupdate=datetime.now)
|
|
|
|
def toDict(self):
|
|
return {
|
|
"id": self.id,
|
|
"name": self.name,
|
|
"shortName": self.short_name,
|
|
"sex": self.sex,
|
|
"step": self.step,
|
|
"specialties": self.specialties,
|
|
"created_at": self.created_at.strftime("%Y-%m-%d %H:%M:%S"),
|
|
"updated_at": self.updated_at.strftime("%Y-%m-%d %H:%M:%S"),
|
|
}
|
|
|
|
|
|
class VkPost(db.Model):
|
|
__tablename__ = "vk_post"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
dynamic_text = db.Column(db.Text, nullable=True)
|
|
static_text = db.Column(db.Text)
|
|
selected_users = db.Column(db.JSON)
|
|
full_name = db.Column(db.Boolean, default=True)
|
|
post_id = db.Column(db.Integer, nullable=True)
|
|
publish_at = db.Column(db.DateTime, nullable=True)
|
|
created_at = db.Column(db.DateTime, default=datetime.now)
|
|
updated_at = db.Column(db.DateTime, default=datetime.now, onupdate=datetime.now)
|
|
|
|
def toDict(self):
|
|
return {
|
|
"id": self.id,
|
|
"dynamic_text": self.dynamic_text,
|
|
"static_text": self.static_text,
|
|
"selected_users": self.selected_users,
|
|
"full_name": self.full_name,
|
|
"post_id": self.post_id,
|
|
"publish_at": (
|
|
self.publish_at.strftime("%Y-%m-%d %H:%M:%S")
|
|
if self.publish_at
|
|
else None
|
|
),
|
|
"created_at": self.created_at.strftime("%Y-%m-%d %H:%M:%S"),
|
|
"updated_at": self.updated_at.strftime("%Y-%m-%d %H:%M:%S"),
|
|
}
|
|
|
|
|
|
class PostScheduler(db.Model):
|
|
__tablename__ = "post_scheduler"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
hour = db.Column(db.Integer)
|
|
minute = db.Column(db.Integer)
|
|
enabled = db.Column(db.Boolean)
|
|
created_at = db.Column(db.DateTime, default=datetime.now)
|
|
updated_at = db.Column(db.DateTime, default=datetime.now, onupdate=datetime.now)
|
|
|
|
def toDict(self):
|
|
return {
|
|
"id": self.id,
|
|
"hour": self.hour,
|
|
"minute": self.minute,
|
|
"enabled": self.enabled,
|
|
"created_at": self.created_at.strftime("%Y-%m-%d %H:%M:%S"),
|
|
"updated_at": self.updated_at.strftime("%Y-%m-%d %H:%M:%S"),
|
|
}
|
|
|
|
|
|
class Protection(db.Model):
|
|
__tablename__ = "protection"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
password = db.Column(db.Text, nullable=False) # hash
|
|
token = db.Column(db.Text, nullable=False)
|
|
|
|
# =========================
|
|
# Пароль
|
|
# =========================
|
|
|
|
def set_password(self, raw_password: str) -> None:
|
|
"""
|
|
Хэширует и сохраняет пароль (Argon2)
|
|
"""
|
|
self.password = ph.hash(raw_password)
|
|
|
|
def verify_password(self, raw_password: str) -> bool:
|
|
"""
|
|
Проверяет пароль
|
|
"""
|
|
try:
|
|
return ph.verify(self.password, raw_password)
|
|
except VerifyMismatchError:
|
|
return False
|
|
|
|
# =========================
|
|
# Token
|
|
# =========================
|
|
|
|
def generate_token(self) -> str:
|
|
"""
|
|
Генерирует новый token и сохраняет его
|
|
"""
|
|
self.token = secrets.token_urlsafe(32)
|
|
return self.token
|
|
|
|
def verify_token(self, token: str) -> bool:
|
|
"""
|
|
Проверяет token
|
|
"""
|
|
return bool(token) and secrets.compare_digest(self.token, token)
|