Files
natureinpots_community/plugins/auth/models.py
2025-07-09 01:05:45 -05:00

128 lines
4.7 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

from uuid import uuid4
from datetime import datetime
from werkzeug.security import generate_password_hash, check_password_hash
from flask_login import UserMixin
from sqlalchemy import event
from sqlalchemy.orm import object_session
from app import db, login_manager
from app.config import Config
class User(db.Model, UserMixin):
__tablename__ = 'users'
__table_args__ = {'extend_existing': True}
id = db.Column(db.Integer, primary_key=True)
email = db.Column(db.String(120), unique=True, nullable=False)
password_hash = db.Column(db.Text, nullable=False)
role = db.Column(db.String(50), default='user')
is_verified = db.Column(db.Boolean, default=False)
excluded_from_analytics = db.Column(db.Boolean, default=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
is_deleted = db.Column(db.Boolean, nullable=False, default=False)
is_banned = db.Column(db.Boolean, nullable=False, default=False)
suspended_until = db.Column(db.DateTime, nullable=True)
# ← New: how many invites the user may still send
invites_remaining = db.Column(
db.Integer,
nullable=False,
default=Config.INVITES_PER_USER
)
# relationships (existing)
submitted_submissions = db.relationship(
"Submission",
foreign_keys="Submission.user_id",
back_populates="submitter",
lazy=True
)
reviewed_submissions = db.relationship(
"Submission",
foreign_keys="Submission.reviewed_by",
back_populates="reviewer",
lazy=True
)
def set_password(self, password):
self.password_hash = generate_password_hash(password)
def check_password(self, password):
return check_password_hash(self.password_hash, password)
class Invitation(db.Model):
__tablename__ = 'invitation'
id = db.Column(db.Integer, primary_key=True)
code = db.Column(db.String(36), unique=True, nullable=False, default=lambda: str(uuid4()))
recipient_email = db.Column(db.String(120), nullable=False)
created_by = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
sender_ip = db.Column(db.String(45), nullable=False)
receiver_ip = db.Column(db.String(45), nullable=True)
is_used = db.Column(db.Boolean, default=False, nullable=False)
used_by = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=True)
used_at = db.Column(db.DateTime, nullable=True)
is_active = db.Column(db.Boolean, default=True, nullable=False)
creator = db.relationship(
'User',
foreign_keys=[created_by],
backref='invitations_sent'
)
user = db.relationship(
'User',
foreign_keys=[used_by],
backref='invitation_received'
)
def mark_used(self, user, ip_addr):
self.is_used = True
self.user = user
self.used_at = datetime.utcnow()
self.receiver_ip = ip_addr
# ─── Autorevoke invites when a user is banned/suspended/deleted ────────────────
@event.listens_for(User, 'after_update')
def revoke_user_invites(mapper, connection, target):
sess = object_session(target)
if not sess:
return
state = db.inspect(target)
banned_changed = state.attrs.is_banned.history.has_changes()
deleted_changed = state.attrs.is_deleted.history.has_changes()
suspended_changed = state.attrs.suspended_until.history.has_changes()
if (banned_changed and target.is_banned) \
or (deleted_changed and target.is_deleted) \
or (suspended_changed and target.suspended_until and target.suspended_until > datetime.utcnow()):
invs = sess.query(Invitation) \
.filter_by(created_by=target.id, is_active=True, is_used=False) \
.all()
refund = len(invs)
for inv in invs:
inv.is_active = False
target.invites_remaining = target.invites_remaining + refund
sess.add(target)
sess.flush()
# ─── Flask-Login user loader ────────────────────────────────────────────────────
@login_manager.user_loader
def load_user(user_id):
if not str(user_id).isdigit():
return None
return User.query.get(int(user_id))
def register_user_loader(lm):
"""
Called by the JSONdriven plugin loader to wire FlaskLogin.
"""
lm.user_loader(load_user)