193 lines
7.1 KiB
Python
193 lines
7.1 KiB
Python
"""SQLAlchemy models."""
|
|
from datetime import datetime, timezone
|
|
from flask_sqlalchemy import SQLAlchemy
|
|
from flask_login import UserMixin
|
|
import bcrypt
|
|
import json
|
|
|
|
db = SQLAlchemy()
|
|
|
|
|
|
def utcnow():
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
class User(UserMixin, db.Model):
|
|
__tablename__ = "users"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
username = db.Column(db.String(64), unique=True, nullable=False, index=True)
|
|
email = db.Column(db.String(255), unique=True, nullable=False)
|
|
password_hash = db.Column(db.String(255), nullable=False)
|
|
role = db.Column(db.String(16), nullable=False, default="user") # 'user' | 'admin'
|
|
is_active = db.Column(db.Boolean, default=True, nullable=False)
|
|
|
|
mfa_secret = db.Column(db.String(64), nullable=True)
|
|
mfa_enabled = db.Column(db.Boolean, default=False, nullable=False)
|
|
|
|
# Optional user avatar (stored as normalised PNG bytes in the DB so it
|
|
# survives an "unzip over the top" upgrade without any external files).
|
|
avatar_data = db.Column(db.LargeBinary, nullable=True)
|
|
avatar_mime = db.Column(db.String(64), nullable=True)
|
|
|
|
created_at = db.Column(db.DateTime, default=utcnow, nullable=False)
|
|
last_login_at = db.Column(db.DateTime, nullable=True)
|
|
failed_login_count = db.Column(db.Integer, default=0, nullable=False)
|
|
locked_until = db.Column(db.DateTime, nullable=True)
|
|
|
|
# ---- password helpers ----
|
|
def set_password(self, password: str) -> None:
|
|
self.password_hash = bcrypt.hashpw(
|
|
password.encode("utf-8"), bcrypt.gensalt(rounds=12)
|
|
).decode("utf-8")
|
|
|
|
def check_password(self, password: str) -> bool:
|
|
try:
|
|
return bcrypt.checkpw(
|
|
password.encode("utf-8"), self.password_hash.encode("utf-8")
|
|
)
|
|
except (ValueError, TypeError):
|
|
return False
|
|
|
|
@property
|
|
def is_admin(self) -> bool:
|
|
return self.role == "admin"
|
|
|
|
@property
|
|
def is_locked(self) -> bool:
|
|
return self.locked_until is not None and self.locked_until > utcnow()
|
|
|
|
@property
|
|
def has_avatar(self) -> bool:
|
|
return self.avatar_data is not None and len(self.avatar_data) > 0
|
|
|
|
@property
|
|
def initials(self) -> str:
|
|
return (self.username or "?")[:2].upper()
|
|
|
|
|
|
class Alert(db.Model):
|
|
__tablename__ = "alerts"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
title = db.Column(db.String(200), nullable=False)
|
|
description = db.Column(db.Text, nullable=True)
|
|
category = db.Column(db.String(64), nullable=True) # e.g. 'License', 'Renewal'
|
|
expiration_date = db.Column(db.Date, nullable=False)
|
|
# JSON-encoded list of integer days, e.g. "[30, 14, 7, 1]"
|
|
reminder_days_json = db.Column(db.String(255), nullable=False, default="[30,14,7,1]")
|
|
# JSON-encoded list of days for which the reminder has already been sent (avoid dupes)
|
|
reminders_sent_json = db.Column(db.String(255), nullable=False, default="[]")
|
|
is_active = db.Column(db.Boolean, default=True, nullable=False)
|
|
created_by = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=True)
|
|
created_at = db.Column(db.DateTime, default=utcnow, nullable=False)
|
|
updated_at = db.Column(db.DateTime, default=utcnow, onupdate=utcnow, nullable=False)
|
|
|
|
creator = db.relationship("User", foreign_keys=[created_by])
|
|
|
|
@property
|
|
def reminder_days(self) -> list:
|
|
try:
|
|
return sorted(set(int(x) for x in json.loads(self.reminder_days_json)), reverse=True)
|
|
except (ValueError, TypeError):
|
|
return []
|
|
|
|
@reminder_days.setter
|
|
def reminder_days(self, value) -> None:
|
|
self.reminder_days_json = json.dumps(sorted(set(int(x) for x in value), reverse=True))
|
|
|
|
@property
|
|
def reminders_sent(self) -> list:
|
|
try:
|
|
return list(json.loads(self.reminders_sent_json))
|
|
except (ValueError, TypeError):
|
|
return []
|
|
|
|
@reminders_sent.setter
|
|
def reminders_sent(self, value) -> None:
|
|
self.reminders_sent_json = json.dumps(list(value))
|
|
|
|
|
|
class AppSetting(db.Model):
|
|
"""Generic key/value settings store (mail config + misc)."""
|
|
__tablename__ = "app_settings"
|
|
|
|
key = db.Column(db.String(64), primary_key=True)
|
|
value = db.Column(db.Text, nullable=True)
|
|
updated_at = db.Column(db.DateTime, default=utcnow, onupdate=utcnow, nullable=False)
|
|
|
|
@classmethod
|
|
def get(cls, key: str, default=None):
|
|
row = db.session.get(cls, key)
|
|
return row.value if row is not None else default
|
|
|
|
@classmethod
|
|
def set(cls, key: str, value: str) -> None:
|
|
row = db.session.get(cls, key)
|
|
if row is None:
|
|
row = cls(key=key, value=value)
|
|
db.session.add(row)
|
|
else:
|
|
row.value = value
|
|
db.session.commit()
|
|
|
|
@classmethod
|
|
def get_all(cls) -> dict:
|
|
return {row.key: row.value for row in cls.query.all()}
|
|
|
|
|
|
class AuditLog(db.Model):
|
|
__tablename__ = "audit_log"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
timestamp = db.Column(db.DateTime, default=utcnow, nullable=False, index=True)
|
|
user_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=True)
|
|
username = db.Column(db.String(64), nullable=True) # denormalised for survival after delete
|
|
action = db.Column(db.String(64), nullable=False)
|
|
details = db.Column(db.Text, nullable=True)
|
|
ip_address = db.Column(db.String(45), nullable=True)
|
|
|
|
@staticmethod
|
|
def record(action: str, user=None, details: str = "", ip: str = ""):
|
|
entry = AuditLog(
|
|
action=action,
|
|
user_id=getattr(user, "id", None),
|
|
username=getattr(user, "username", None),
|
|
details=details[:1000] if details else None,
|
|
ip_address=ip,
|
|
)
|
|
db.session.add(entry)
|
|
db.session.commit()
|
|
|
|
|
|
class MailLog(db.Model):
|
|
"""Record of every outbound email dispatched by the platform."""
|
|
__tablename__ = "mail_log"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
timestamp = db.Column(db.DateTime, default=utcnow, nullable=False, index=True)
|
|
kind = db.Column(db.String(32), nullable=False, default="general") # reminder | test | log-export | general
|
|
recipients = db.Column(db.Text, nullable=True)
|
|
subject = db.Column(db.String(500), nullable=True)
|
|
status = db.Column(db.String(16), nullable=False, default="unknown") # sent | failed
|
|
info = db.Column(db.Text, nullable=True)
|
|
sent_copy = db.Column(db.String(24), nullable=False, default="n/a") # saved | failed | disabled | n/a
|
|
triggered_by = db.Column(db.String(64), nullable=True) # username or "system"
|
|
|
|
@staticmethod
|
|
def record(kind, recipients, subject, status, info="", sent_copy="n/a", triggered_by="system"):
|
|
if isinstance(recipients, (list, tuple)):
|
|
recipients = ", ".join(str(r) for r in recipients)
|
|
entry = MailLog(
|
|
kind=(kind or "general")[:32],
|
|
recipients=(recipients or "")[:2000],
|
|
subject=(subject or "")[:500],
|
|
status=(status or "unknown")[:16],
|
|
info=(info or "")[:2000],
|
|
sent_copy=(sent_copy or "n/a")[:24],
|
|
triggered_by=(triggered_by or "system")[:64],
|
|
)
|
|
db.session.add(entry)
|
|
db.session.commit()
|
|
return entry
|