Files

routes/admin.py

@@ -0,0 +1,330 @@
+from flask import Blueprint, render_template, redirect, url_for, flash, abort, request
+from flask_login import login_required, current_user
+from models import db, User, Post, Friendship, Comment, Upload, Notification, Event, PasswordResetRequest, Like, UserShopItem, Reward, SupportComment, SupportRequest
+from werkzeug.security import generate_password_hash
+from flask_babel import gettext as _
+import os
+
+__mapper_args__ = {"confirm_deleted_rows": False}
+admin_bp = Blueprint('admin', __name__, url_prefix='/admin')
+
+@admin_bp.route('/reset_requests/delete_all', methods=['GET', 'POST'])
+@login_required
+def admin_delete_all_reset_requests():
+    if not current_user.is_admin:
+        abort(403)
+    db.session.query(PasswordResetRequest).delete()
+    db.session.commit()
+    flash(_('All password reset requests have been deleted.'), 'success')
+    return redirect(url_for('admin.reset_requests'))
+
+@admin_bp.route('/')
+@login_required
+def admin():
+    if not current_user.is_admin:
+        abort(403)
+    users = db.session.query(User).all()
+    posts = db.session.query(Post).all()
+    friendships = db.session.query(Friendship).all()
+    comments = db.session.query(Comment).all()
+    uploads = db.session.query(Upload).all()
+    all_notifications = db.session.query(Notification).order_by(Notification.created_at.desc()).all()
+    events = db.session.query(Event).order_by(Event.timestamp.desc()).limit(50).all()
+    user_shop_items = db.session.query(UserShopItem).all()
+    return render_template(
+        'admin.html',
+        users=users,
+        posts=posts,
+        friendships=friendships,
+        comments=comments,
+        uploads=uploads,
+        all_notifications=all_notifications,
+        events=events,
+        user_shop_items=user_shop_items
+    )
+
+@admin_bp.route('/reset_requests')
+@login_required
+def reset_requests():
+    if not current_user.is_admin:
+        abort(403)
+    requests = db.session.query(PasswordResetRequest).filter_by(status='pending').all()
+    requests_done = db.session.query(PasswordResetRequest).filter_by(status='done').all()
+    requests_rejected = db.session.query(PasswordResetRequest).filter_by(status='rejected').all()
+    return render_template('reset_requests.html', requests=requests, requests_done=requests_done, requests_rejected=requests_rejected)
+
+@admin_bp.route('/reset_requests/<int:req_id>/reject', methods=['POST'])
+@login_required
+def reject_reset_request(req_id):
+    if not current_user.is_admin:
+        abort(403)
+    req = db.session.get(PasswordResetRequest, req_id)
+    if req:
+        req.status = 'rejected'
+        db.session.commit()
+        flash(_('Request rejected.'), 'info')
+    return redirect(url_for('admin.reset_requests'))
+
+@admin_bp.route('/reset_requests/<int:req_id>/reset', methods=['GET', 'POST'])
+@login_required
+def admin_reset_password(req_id):
+    if not current_user.is_admin:
+        abort(403)
+    req = db.session.get(PasswordResetRequest, req_id)
+    if req and req.status == 'pending':
+        user = db.session.get(User, req.user_id)
+        if request.method == 'POST':
+            new_pw = request.form['new_password']
+            if not new_pw or len(new_pw) < 4:
+                flash(_('Password too short.'), 'danger')
+            else:
+                user.password = generate_password_hash(new_pw, method='pbkdf2:sha256')
+                req.status = 'done'
+                db.session.commit()
+                flash(_(f'Password for {user.username} reset.'), 'success')
+                return redirect(url_for('admin.reset_requests'))
+        return render_template('admin_set_password.html', req=req, user=user)
+    return redirect(url_for('admin.reset_requests'))
+
+@admin_bp.route('/delete_post/<int:post_id>', methods=['POST'])
+@login_required
+def admin_delete_post(post_id):
+    if not current_user.is_admin:
+        abort(403)
+    post = db.session.get(Post, post_id)
+    if post:
+        for upload in post.uploads:
+            file_path = os.path.join('static/uploads', upload.filename)
+            try:
+                if os.path.exists(file_path):
+                    os.remove(file_path)
+            except Exception:
+                pass
+        likes = db.session.query(Like).filter_by(post_id=post_id).all()
+        for like in likes:
+            db.session.delete(like)
+        comments = db.session.query(Comment).filter_by(post_id=post_id).all()
+        for comment in comments:
+            db.session.delete(comment)
+        db.session.delete(post)
+        event = Event(message=_(f"Admin {current_user.username} has deleted post {post.id}."))
+        db.session.add(event)
+        notification = Notification(message=_(f"Your post {post.id} has been deleted by an admin."), user_id=post.user_id)
+        db.session.add(notification)
+        db.session.commit()
+        flash(_('Post and associated files deleted.'), 'success')
+    return redirect(url_for('admin.admin'))
+
+@admin_bp.route('/delete_user/<int:user_id>', methods=['POST'])
+@login_required
+def admin_delete_user(user_id):
+    user = db.session.get(User, user_id)
+    if user.is_owner:
+        flash(_('Cannot delete the owner account.'), 'danger')
+        return redirect(url_for('admin.admin')) 
+    if user and not user.is_admin:
+        event = Event(message=f"Admin {current_user.username} hat {user.username} gelöscht.")
+        db.session.add(event)
+        for post in user.posts:
+            db.session.delete(post)
+        for friendship in user.friendships_sent + user.friendships_received:
+            db.session.delete(friendship)
+        for comment in user.comments:
+            db.session.delete(comment)
+        for user_shop_item in user.shop_items:
+            db.session.delete(user_shop_item)
+        for reward in user.rewards:
+            db.session.delete(reward)
+        for like in user.likes:
+            db.session.delete(like)
+        for upload in user.uploads:
+            file_path = os.path.join('static/uploads', upload.filename)
+            try:
+                if os.path.exists(file_path):
+                    os.remove(file_path)
+            except Exception:
+                pass
+            db.session.delete(upload)
+        if user.profile_pic and not user.profile_pic == 'default.png':
+            try:
+                os.remove(os.path.join('static/profile_pics', user.profile_pic))
+            except Exception:
+                pass
+        notifications = db.session.query(Notification).filter_by(user_id=user.id).all()
+        for notif in notifications:
+            db.session.delete(notif)
+        db.session.delete(user)
+        db.session.commit()
+        flash(_('User deleted.'), 'success')
+    else:
+        flash(_('Cannot delete admin or user not found.'), 'danger')
+    return redirect(url_for('admin.admin'))
+
+@admin_bp.route('/delete_pic/<int:user_id>', methods=['POST'])
+@login_required
+def admin_delete_pic(user_id):
+    if not current_user.is_admin:
+        abort(403)
+    user = db.session.get(User, user_id)
+    if user and user.profile_pic and user.profile_pic != 'default.png':
+        try:
+            os.remove(os.path.join('static/profile_pics', user.profile_pic))
+        except Exception:
+            pass
+        user.profile_pic = "default.png"
+        db.session.commit()
+        flash(_(f'Profile picture of {user.username} deleted.'), 'success')
+    return redirect(url_for('admin.admin'))
+
+@admin_bp.route('/delete_all_notifications', methods=['POST'])
+@login_required
+def admin_delete_all_notifications():
+    if not current_user.is_admin:
+        abort(403)
+    db.session.query(Notification).delete()
+    db.session.commit()
+    flash(_('All notifications have been deleted.'), 'success')
+    return redirect(url_for('admin.admin'))
+
+@admin_bp.route('/delete_all_events', methods=['POST'])
+@login_required
+def admin_delete_all_events():
+    if not current_user.is_admin:
+        abort(403)
+    db.session.query(Event).delete()
+    db.session.commit()
+    flash(_('All events have been deleted.'), 'success')
+    return redirect(url_for('admin.admin'))
+
+@admin_bp.route('/delete_upload/<int:upload_id>', methods=['POST'])
+@login_required
+def admin_delete_upload(upload_id):
+    if not current_user.is_admin:
+        abort(403)
+    upload = db.session.get(Upload, upload_id)
+    if upload:
+        file_path = os.path.join('static/uploads', upload.filename)
+        try:
+            if os.path.exists(file_path):
+                os.remove(file_path)
+        except Exception:
+            pass
+        db.session.delete(upload)
+        db.session.commit()
+        flash(_('Upload deleted.'), 'success')
+    return redirect(url_for('admin.admin'))
+
+@admin_bp.route('/delete_all_uploads', methods=['POST'])
+@login_required
+def admin_delete_all_uploads():
+    if not current_user.is_admin:
+        abort(403)
+    db.session.query(Upload).delete()
+    db.session.commit()
+    upload_dir = 'static/uploads'
+    for filename in os.listdir(upload_dir):
+        file_path = os.path.join(upload_dir, filename)
+        try:
+            if os.path.isfile(file_path):
+                os.remove(file_path)
+        except Exception:
+            pass
+    flash(_('All uploads have been deleted.'), 'success')
+    return redirect(url_for('admin.admin'))
+
+@admin_bp.route('/admin/points/<int:user_id>', methods=['POST'])
+@login_required
+def admin_points(user_id):
+    if not current_user.is_admin:
+        abort(403)
+    action = request.form.get('action')
+    try:
+        points = int(request.form['points'])
+    except:
+        flash(_('No Points entered!'))
+        return redirect(url_for('admin.admin'))
+    cuser = db.session.get(User, current_user.id)
+    if not cuser.is_owner:
+        abort(403)
+    if action == 'add':
+        db.session.add(Reward(user_id=user_id, type='admin', points=points))
+        db.session.commit()
+        flash(_('Points added!'), 'success')
+    elif action == 'remove':
+        user = db.session.get(User, user_id)
+        if user.reward_points() >= points:
+            db.session.add(Reward(user_id=user_id, type='admin', points=-points))
+            db.session.commit()
+            flash(_('Points removed!'), 'success')
+        else:
+            flash(_("The user has not enough points to take!"), 'danger')
+    return redirect(url_for('admin.admin'))
+
+@admin_bp.route('/make_admin/<int:user_id>', methods=['POST'])
+@login_required
+def make_admin(user_id):
+    if not current_user.is_admin:
+        abort(403)
+    user = db.session.get(User, user_id)
+    if user and not user.is_admin:
+        user.is_admin = True
+        db.session.commit()
+        flash(_(f"{user.username} is now an admin."), "success")
+    return redirect(url_for('admin.admin'))
+
+@admin_bp.route('/remove_admin/<int:user_id>', methods=['POST'])
+@login_required
+def remove_admin(user_id):
+    if not current_user.is_admin:
+        abort(403)
+    user = db.session.get(User, user_id)
+    if user and user.is_admin and not user.is_owner:
+        user.is_admin = False
+        db.session.commit()
+        flash(_(f"Admin rights of {user.username} removed."), "info")
+    else:
+        flash(_("Owner cannot be removed!"), "danger")
+    return redirect(url_for('admin.admin'))
+
+@admin_bp.route('/wipe_server', methods=['POST'])
+@login_required
+def wipe_server():
+    if not current_user.is_admin and not current_user.is_owner:
+        abort(403)
+
+    db.session.query(Reward).delete()
+    db.session.query(UserShopItem).delete()
+    db.session.query(Like).delete()
+    db.session.query(Comment).delete()
+    db.session.query(Friendship).delete()
+    db.session.query(Post).delete()
+    db.session.query(Upload).delete()
+    db.session.query(Notification).delete()
+    db.session.query(Event).delete()
+    db.session.query(PasswordResetRequest).delete()
+    db.session.query(User).delete()
+    db.session.query(SupportComment).delete()
+    db.session.query(SupportRequest).delete()
+    db.session.commit()
+
+    upload_dir = 'static/uploads'
+    for filename in os.listdir(upload_dir):
+        file_path = os.path.join(upload_dir, filename)
+        try:
+            if os.path.isfile(file_path):
+                os.remove(file_path)
+        except Exception:
+            pass
+
+    profile_dir = 'static/profile_pics'
+    for filename in os.listdir(profile_dir):
+        if filename != 'default.png':
+            file_path = os.path.join(profile_dir, filename)
+            try:
+                if os.path.isfile(file_path):
+                    os.remove(file_path)
+            except Exception:
+                pass
+
+    flash(_('All Data has been deleted.'), 'success')
+    return redirect(url_for('admin.admin'))