"""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