feat: option added to enable/disable notification alert

Author: codeperfectplus

src/models.py

@@ -86,6 +86,7 @@ class User(db.Model, UserMixin):
     email = db.Column(db.String(150), unique=True, nullable=False)
     password = db.Column(db.String(150), nullable=False)
     user_level = db.Column(db.String(50), nullable=False, default='user')
+    receive_email_alerts = db.Column(db.Boolean, default=True)
 
 
 class EmailPassword(db.Model):
@@ -108,7 +109,8 @@ class EmailPassword(db.Model):
     # Create admin user if not exists
     if not User.query.filter_by(username='admin').first():
         hashed_password = generate_password_hash('adminpassword')
-        admin_user = User(username='admin', email="codeperfectplus@gmail.com", password=hashed_password, user_level='admin')
+        admin_user = User(username='admin', email="codeperfectplus@gmail.com", password=hashed_password, user_level='admin',
+                          receive_email_alerts=True)
 
         db.session.add(admin_user)
         db.session.commit()
@@ -117,7 +119,7 @@ class EmailPassword(db.Model):
     if not User.query.filter_by(username='user').first():
         hashed_password = generate_password_hash('userpassword')
         user = User(username='user', email="test@mail.com",
-                    password=hashed_password, user_level='user')
+                    password=hashed_password, user_level='user', receive_email_alerts=False)
 
         db.session.add(user)
         db.session.commit()

src/routes/auth.py

@@ -20,6 +20,24 @@
 def load_user(user_id):
     return User.query.get(int(user_id))
 
+
+def get_admin_emails_for_email_alerts():
+    with app.app_context():
+        admin_emails = User.query.filter_by(user_level='admin', receive_email_alerts=True).all()
+        print(admin_emails)
+        if not admin_emails:
+            return None
+        return [admin.email for admin in admin_emails]
+    
+def get_all_users_emails(receive_email_alerts=True):
+    with app.app_context():
+        users = User.query.filter_by(receive_email_alerts=receive_email_alerts).all()
+        if not users:
+            return None
+        return [user.email for user in users]
+
+
+
 @app.route('/login', methods=['GET', 'POST'])
 def login():
     if request.method == 'POST':
@@ -28,6 +46,10 @@ def login():
         user = User.query.filter_by(username=username).first()
         if user and check_password_hash(user.password, password):
             login_user(user)
+            admin_email_address = get_admin_emails_for_email_alerts()
+            if admin_email_address:
+                send_email(admin_email_address, 'Login Alert', f'{user.username} logged in to the system.')
+            
             return redirect(url_for('dashboard'))
         flash('Invalid username or password', 'danger')
     return render_template('login.html')
@@ -50,6 +72,15 @@ def signup():
 
         hashed_password = generate_password_hash(password)
         new_user = User(username=username, password=hashed_password)
+        
+        admin_email_address = get_admin_emails_for_email_alerts()
+        # extends the signup user to send an email to the admin
+        if admin_email_address:
+            send_email(admin_email_address, 'New User Alert', f'{username} has signed up to the system.')
+        
+        # send email to the new user
+        send_email([new_user.email], 'Welcome to the system', f'Hello {new_user.username}, welcome to the system.')
+
         db.session.add(new_user)
         db.session.commit()
         flash('Account created successfully, please log in.')
@@ -90,25 +121,55 @@ def logout():
 #         return f(*args, **kwargs)
 #     return decorated_function
 
+@app.route('/users')
+@login_required
+def view_users():
+    if current_user.user_level != 'admin':
+        flash("Your account does not have permission to view this page.", "danger")
+        return render_template("error/permission_denied.html")
+
+    # Fetch all users from the database
+    users = User.query.all()
+
+    return render_template('view_users.html', users=users)
+
 @app.route('/user/<username>', methods=['GET', 'POST'])
 @login_required
 def change_user_settings(username):
     user = User.query.filter_by(username=username).first_or_404()
 
     if request.method == 'POST':
         new_username = request.form['username']
+        new_email = request.form['email']
         new_user_level = request.form['user_level']
-        
+        receive_email_alerts = 'receive_email_alerts' in request.form
+
         # Update user details
         user.username = new_username
+        user.email = new_email
         user.user_level = new_user_level
+        user.receive_email_alerts = receive_email_alerts
+
         db.session.commit()
-        
+
         flash('User settings updated successfully!', 'success')
         return redirect(url_for('change_user_settings', username=user.username))
 
     return render_template('change_user.html', user=user)
 
+@app.route('/delete_user/<username>', methods=['POST'])
+@login_required
+def delete_user(username):
+    if current_user.user_level != 'admin':
+        flash("Your account does not have permission to perform this action.", "danger")
+        return redirect(url_for('view_users'))  # Redirect to the users page
+
+    user = User.query.filter_by(username=username).first_or_404()
+    db.session.delete(user)
+    db.session.commit()
+    
+    flash(f'User {username} has been deleted successfully!', 'success')
+    return redirect(url_for('view_users'))
 
 @app.route("/update-email-password", methods=["GET", "POST"])
 @login_required
@@ -134,17 +195,30 @@ def update_email_password():
 @login_required
 def send_email_page():
     if request.method == "POST":
-        receiver_email = request.form.get("receiver_email")
+ 
+        receiver_email = request.form.get("recipient")
         subject = request.form.get("subject")
         body = request.form.get("body")
         attachment = request.files.get("attachment")
 
+        if not receiver_email or not subject or not body:
+            flash("Please provide recipient, subject, and body.", "danger")
+            return redirect(url_for('send_email_page'))
+        
+        if receiver_email == "all_users":
+            receiver_email = get_all_users_emails()
+        elif receiver_email == "admin_users":
+            receiver_email = get_admin_emails_for_email_alerts()
+
+        if not receiver_email:
+            flash("No users found to send email to.", "danger")
+            return redirect(url_for('send_email_page'))
+        
         # Save attachment if any
         attachment_path = None
         if attachment:
             attachment_path = f"/tmp/{attachment.filename}"
             attachment.save(attachment_path)
-
         try:
             send_email(receiver_email, subject, body, attachment_path)
             flash("Email sent successfully!", "success")

src/routes/speedtest.py

@@ -1,13 +1,15 @@
 import datetime
 from flask import render_template, blueprints
-
+from flask_login import login_required, current_user
 from src.config import app, db
 from src.models import DashboardSettings, SpeedTestResult
 from src.utils import run_speedtest
+from src.scripts.email_me import send_email
 
 speedtest_bp = blueprints.Blueprint("speedtest", __name__)
 
 @app.route("/speedtest")
+@login_required
 def speedtest():
     settings = DashboardSettings.query.first()
     SPEEDTEST_COOLDOWN_IN_HOURS = settings.speedtest_cooldown
@@ -34,6 +36,10 @@ def speedtest():
             )
             db.session.add(new_result)
             db.session.commit()
+            receiver_email = current_user.email
+            subject = "Speedtest Result"
+            body = f"Download Speed: {speedtest_result['download_speed']} \nUpload Speed: {speedtest_result['upload_speed']} \nPing: {speedtest_result['ping']} "
+            send_email(receiver_email, subject, body)
             return render_template(
                 "speedtest_result.html",
                 speedtest_result=speedtest_result,

src/scripts/email_me.py

@@ -4,7 +4,6 @@
 from email.mime.text import MIMEText
 from email.mime.base import MIMEBase
 from email import encoders
-
 from dotenv import load_dotenv
 
 # Load environment variables from .env file
@@ -14,48 +13,49 @@
 EMAIL_ADDRESS = os.getenv("EMAIL_ADDRESS")
 EMAIL_PASSWORD = os.getenv("EMAIL_PASSWORD")
 
-
 def send_email(receiver_email, subject, body, attachment_path=None):
-    try:
-        # Create a multipart message
-        msg = MIMEMultipart()
-        msg['From'] = EMAIL_ADDRESS
-        msg['To'] = receiver_email
-        msg['Subject'] = subject
-        # append message to the body
-        append_message = "This is an automated email from the SystemGuru application. Please do not reply to this email."
-        body = body + "\n\n" + append_message
-
-        # Attach the body with the msg instance
-        msg.attach(MIMEText(body, 'plain'))
-
-        # Attach a file if provided
-        if attachment_path:
-            with open(attachment_path, "rb") as attachment:
-                part = MIMEBase("application", "octet-stream")
-                part.set_payload(attachment.read())
-                encoders.encode_base64(part)
-                part.add_header(
-                    "Content-Disposition",
-                    f"attachment; filename= {attachment_path}",
-                )
-                msg.attach(part)
-
-        # Create an SMTP session
-        server = smtplib.SMTP('smtp.gmail.com', 587)  # Use Gmail's SMTP server
-        server.starttls()  # Enable security
-
-        # Login with sender's email and password
-        server.login(EMAIL_ADDRESS, EMAIL_PASSWORD)
-
-        # Send the email
-        text = msg.as_string()
-        server.sendmail(EMAIL_ADDRESS, receiver_email, text)
-
-        # Close the SMTP session
-        server.quit()
-
-        print(f"Email sent successfully to {receiver_email}")
-
-    except Exception as e:
-        print(f"Failed to send email. Error: {str(e)}")
+    if isinstance(receiver_email, str):
+        receiver_email = [receiver_email]  # Convert single address to list
+
+    for email in receiver_email:
+        try:
+            # Create a multipart message
+            msg = MIMEMultipart()
+            msg['From'] = EMAIL_ADDRESS
+            msg['To'] = email
+            msg['Subject'] = subject
+            # Append message to the body
+            append_message = "This is an automated email from the SystemGuard application. Please do not reply to this email."
+            full_body = body + "\n\n" + append_message
+
+            # Attach the body with the msg instance
+            msg.attach(MIMEText(full_body, 'plain'))
+
+            # Attach a file if provided
+            if attachment_path:
+                with open(attachment_path, "rb") as attachment:
+                    part = MIMEBase("application", "octet-stream")
+                    part.set_payload(attachment.read())
+                    encoders.encode_base64(part)
+                    part.add_header(
+                        "Content-Disposition",
+                        f"attachment; filename= {os.path.basename(attachment_path)}",
+                    )
+                    msg.attach(part)
+
+            # Create an SMTP session
+            with smtplib.SMTP('smtp.gmail.com', 587) as server:  # Use Gmail's SMTP server
+                server.starttls()  # Enable security
+
+                # Login with sender's email and password
+                server.login(EMAIL_ADDRESS, EMAIL_PASSWORD)
+
+                # Send the email
+                text = msg.as_string()
+                server.sendmail(EMAIL_ADDRESS, email, text)
+
+            print(f"Email sent successfully to {email}")
+
+        except Exception as e:
+            print(f"Failed to send email to {email}. Error: {str(e)}")
+