from datetime import datetime, timezone, timedelta from app.extensions import db from sqlalchemy.sql import func from sqlalchemy.exc import IntegrityError import uuid from app.utils.logger import logger class Members(db.Model): __tablename__ = 'members' id = db.Column(db.Integer, primary_key=True, autoincrement=True) uid = db.Column(db.String(150), nullable=False) username = db.Column(db.String(25), nullable=False) password = db.Column(db.String(100), nullable=True) loc = db.Column(db.String(20), nullable=True) status = db.Column(db.Integer, default=1) added = db.Column(db.DateTime(timezone=False), server_default=func.now()) updated = db.Column(db.DateTime(timezone=False), server_default=func.now(), onupdate=func.now()) email = db.Column(db.String(100), nullable=False) account_name = db.Column(db.String(100), nullable=True) firstname = db.Column(db.String(25), nullable=False) lastname = db.Column(db.String(25), nullable=True) country = db.Column(db.String(3), nullable=True) profile_completed = db.Column(db.DateTime(timezone=False)) stripe_customer_id = db.Column(db.String(100), nullable=True) option_name = db.Column(db.String(100), nullable=True) next_billing = db.Column(db.DateTime(timezone=False)) trial_end = db.Column(db.DateTime(timezone=False)) last_login = db.Column(db.DateTime(timezone=False), server_default=func.now(), onupdate=func.now()) phone = db.Column(db.String(25), nullable=True) full_address = db.Column(db.String(150), nullable=True) profile_picture = db.Column(db.String(100), nullable=True) # "account_id": self.account_id, def to_dict(self): return { "id": self.id, "member_id": self.id, "uid": str(self.uid), "profile_completed": self.profile_completed, "username": self.username, "password": self.password, "country": self.country, "phone": self.phone, "full_address": self.full_address, "loc": self.loc, "status": self.status, "added": self.added.isoformat() if self.added else None, "updated": self.updated.isoformat() if self.updated else None, "last_login": self.last_login.isoformat() if self.last_login else None, "email": self.email, "account_name": self.account_name, "firstname": self.firstname, "lastname": self.lastname, 'option_name': self.option_name, "next_billing": self.next_billing, "trial_end": self.trial_end, "stripe_customer_id": self.stripe_customer_id, "profile_picture": self.profile_picture, } def __repr__(self): return f'' @classmethod def get_member_by_email(cls, email): member = cls.query.filter_by(email=str(email)).first() if not member: return None return member @classmethod def get_member_by_username(cls, username): member = cls.query.filter_by(username=str(username)).first() if not member: return None return member @classmethod def get_member_by_uid(cls, uid): member = cls.query.filter_by(uid=str(uid)).first() if not member: logger.info(f"User UID = {uid} found") return None return member @classmethod def set_member_profile_completed(cls, uid): member = cls.query.filter_by(uid=str(uid)).first() if not member: logger.info(f"User UID = {uid} found") raise ValueError(f"Member with UID {uid} does not exist.") member.profile_completed = datetime.now(timezone.utc) return member.profile_completed @classmethod def set_member_profile_picture(cls, uid, picture_uid): logger.info(f"Set_Member_Profile_Picture UID = {uid} Picture_UID = {picture_uid} ") member = cls.query.filter_by(uid=str(uid)).first() if not member: logger.info(f"Set_Member_Profile_Picture UID = {uid} Not Found**") raise ValueError(f"Member with UID {uid} does not exist.") member.profile_picture = picture_uid logger.info(f"*** Member_Profile_Picture UID = {uid} Picture_Picture = {member.profile_picture} ") return member @classmethod def set_member_update_profile(cls, uid, profile_data): member = cls.query.filter_by(uid=str(uid)).first() if not member: logger.info(f"User UID = {uid} found") raise ValueError(f"Member with UID {uid} does not exist.") member.full_address = profile_data["full_address"] member.account_name = profile_data["account_name"] member.firstname = profile_data["firstname"] member.lastname = profile_data["lastname"] member.phone = profile_data["phone"] member.updated = datetime.now(timezone.utc) return member @classmethod def set_office_update_profile(cls, uid, profile_data): member = cls.query.filter_by(uid=str(uid)).first() if not member: logger.info(f"User UID = {uid} found") raise ValueError(f"Member with UID {uid} does not exist.") member.full_address = profile_data["member_full_address"] member.account_name = profile_data["member_account_name"] member.firstname = profile_data["member_firstname"] member.lastname = profile_data["member_lastname"] member.phone = profile_data["member_phone"] member.email = profile_data["member_email"] member.updated = datetime.now(timezone.utc) return member @classmethod def set_member_last_login(cls, uid): member = cls.query.filter_by(uid=str(uid)).first() if not member: logger.info(f"User UID = {uid} found") return None member.last_login = datetime.now(timezone.utc) return member.last_login @classmethod def add_member(cls, firstname, lastname, email, username, password, country, trials_days=90): # Get the current date and time current_date = datetime.now() # Calculate the date "next_billing_days" days from now trial_end = current_date + timedelta(days=trials_days) # Save the response member_data = cls( uid=str(uuid.uuid4()), firstname=firstname, lastname=lastname, username=username, email=email, country=country, password=password, trial_end=trial_end, added=datetime.now(timezone.utc), updated=datetime.now(timezone.utc) ) try: db.session.add(member_data) except IntegrityError as err: raise ValueError(f"Database integrity error: {err}") return member_data @classmethod def update_user_password(cls, reset_id, member_uid, member_id, new_passwprd_hash): member = cls.query.filter_by(uid=str(member_uid), id=member_id).first() if not member: raise ValueError(f"Reset with UID {member_uid} does not exist.") # Update reset_p status and the updated_at timestamp member.password = new_passwprd_hash @classmethod def set_user_stripe_id(cls, member_uid, member_id, stripe_customer_id): member = cls.query.filter_by(uid=str(member_uid), id=member_id).first() if not member: raise ValueError(f"Member with UID {member_uid} does not exist.") # Update stripe_customer_id member.stripe_customer_id = stripe_customer_id return stripe_customer_id @classmethod def set_user_option_name(cls, member_id, option_name): member = cls.query.filter_by(id=member_id).first() if not member: raise ValueError(f"Member with ID {member_id} does not exist.") # Update stripe_customer_id member.option_name = option_name return option_name @classmethod def set_user_next_billing(cls, member_id, next_billing): member = cls.query.filter_by(id=member_id).first() if not member: raise ValueError(f"Member with ID {member_id} does not exist.") # Update next_billing member.next_billing = next_billing return next_billing @classmethod def get_all_member(cls, email=None, username=None, page=1, limit=20): query = cls.query logger.info(f"Get all customer back") if username: query = query.filter(cls.username == username) if email: query = query.filter(cls.email == email) # Order by created_at descending (newest first) query = query.order_by(cls.added.desc()) # Get total count before pagination total_count = query.count() # Apply pagination offset = (page - 1) * limit query = query.limit(limit).offset(offset) return query.all(), total_count @classmethod def get_recent_member_login(cls, limit=10): query = cls.query logger.info(f"Get Recent Member Login") # Order by created_at descending (newest first) query = query.order_by(cls.last_login.desc()) query = query.limit(limit).offset(0) return query.all()