from datetime import datetime, timezone, timedelta from werkzeug.security import generate_password_hash, check_password_hash from flask_login import UserMixin from sqlalchemy import event, func from .. import db import secrets import logging logger = logging.getLogger(__name__) class User(UserMixin, db.Model): __tablename__ = 'enterprise_users' id = db.Column(db.Integer, primary_key=True) username = db.Column(db.String(25), unique=True, nullable=False) email = db.Column(db.String(120), unique=True, nullable=False) company = db.Column(db.String(100), nullable=False) password_hash = db.Column(db.String(128), nullable=False) created_at = db.Column(db.DateTime) password_changed_at = db.Column(db.DateTime) is_admin = db.Column(db.Boolean, default=False) last_login = db.Column(db.DateTime) is_active = db.Column(db.Boolean, default=True) failed_login_attempts = db.Column(db.Integer, default=0) last_failed_login = db.Column(db.DateTime) reset_token = db.Column(db.String(100), unique=True) reset_token_expires = db.Column(db.DateTime) # Relationships quotations = db.relationship('Quotation', backref='user', lazy=True) def set_password(self, password): try: self.password_hash = generate_password_hash(password, method='pbkdf2:sha256') self.password_changed_at = datetime.now(timezone.utc) self.reset_token = None self.reset_token_expires = None except Exception as e: logger.error(f'Error setting password: {str(e)}', exc_info=True) raise def check_password(self, password): return check_password_hash(self.password_hash, password) def generate_reset_token(self): token = secrets.token_urlsafe(16) # Shorter token for display self.reset_token = token self.reset_token_expires = datetime.now(timezone.utc) + timedelta(minutes=15) # 15-minute expiry return token def verify_reset_token(self, token): if not self.reset_token or not self.reset_token_expires: return False if datetime.now(timezone.utc) > self.reset_token_expires: return False return secrets.compare_digest(self.reset_token, token) @classmethod def authenticate(cls, email, password): user = cls.query.filter(func.lower(cls.email) == email.lower()).first() if user and user.check_password(password): if not user.is_active: return None user.failed_login_attempts = 0 user.last_failed_login = None db.session.commit() return user if user: user.failed_login_attempts = (user.failed_login_attempts or 0) + 1 user.last_failed_login = datetime.now(timezone.utc) db.session.commit() return None @classmethod def create(cls, username, email, company, password): try: logger.info(f'Checking if username {username} exists...') if cls.query.filter(func.lower(cls.username) == username.lower()).first(): logger.warning(f'Username {username} already taken') raise ValueError('Username already taken') logger.info(f'Checking if email {email} exists...') if cls.query.filter(func.lower(cls.email) == email.lower()).first(): logger.warning(f'Email {email} already registered') raise ValueError('Email already registered') logger.info('Creating new user...') user = cls( username=username, email=email, company=company, is_active=True, failed_login_attempts=0, is_admin=False ) logger.info('Setting password...') user.set_password(password) logger.info('Adding user to session...') db.session.add(user) logger.info('User created successfully') return user except Exception as e: logger.error(f'Error creating user: {str(e)}', exc_info=True) raise def __repr__(self): return f'' # Event listeners for model changes @event.listens_for(User, 'before_insert') def set_timestamps(mapper, connection, target): target.created_at = datetime.now(timezone.utc) target.password_changed_at = datetime.now(timezone.utc)