126 lines
4.2 KiB
Python
126 lines
4.2 KiB
Python
|
from flask import Blueprint, jsonify, session
|
||
|
|
from models.user import User
|
||
|
|
from models.notepad import Notepad
|
||
|
|
from models.note import Note
|
||
|
|
from models.banned_ip import BannedIP
|
||
|
|
from extensions import db
|
||
|
|
|
||
|
|
admin_bp = Blueprint('admin', __name__, url_prefix='/admin')
|
||
|
|
|
||
|
|
def check_admin():
|
||
|
|
user_id = session.get('user_id')
|
||
|
|
if not user_id:
|
||
|
|
return False
|
||
|
|
|
||
|
|
user = User.query.get(user_id)
|
||
|
|
return user and user.is_admin
|
||
|
|
|
||
|
|
@admin_bp.before_request
|
||
|
|
def before_request():
|
||
|
|
if not check_admin():
|
||
|
|
return jsonify({'error': 'Unauthorized'}), 403
|
||
|
|
|
||
|
|
@admin_bp.route('/stats', methods=['GET'])
|
||
|
|
def get_stats():
|
||
|
|
return jsonify({
|
||
|
|
'user_count': User.query.count(),
|
||
|
|
'notepad_count': Notepad.query.count(),
|
||
|
|
'note_count': Note.query.count()
|
||
|
|
})
|
||
|
|
|
||
|
|
@admin_bp.route('/users', methods=['GET'])
|
||
|
|
def get_users():
|
||
|
|
users = User.query.all()
|
||
|
|
user_list = [user.to_dict() for user in users]
|
||
|
|
|
||
|
|
# Add "Unowned" virtual user for managing public/unowned pads
|
||
|
|
unowned_count = Notepad.query.filter_by(user_id=None).count()
|
||
|
|
if unowned_count > 0:
|
||
|
|
user_list.insert(0, {
|
||
|
|
'id': 0,
|
||
|
|
'email': 'Unowned / Public Notepads',
|
||
|
|
'first_name': 'System',
|
||
|
|
'last_name': '',
|
||
|
|
'last_ip': None,
|
||
|
|
'is_admin': False,
|
||
|
|
'is_banned': False
|
||
|
|
})
|
||
|
|
|
||
|
|
return jsonify(user_list)
|
||
|
|
|
||
|
|
@admin_bp.route('/users/<int:user_id>/notepads', methods=['GET'])
|
||
|
|
def get_user_notepads(user_id):
|
||
|
|
# Handle virtual user ID 0 for unowned notepads
|
||
|
|
query_id = None if user_id == 0 else user_id
|
||
|
|
notepads = Notepad.query.filter_by(user_id=query_id).all()
|
||
|
|
return jsonify([notepad.to_dict() for notepad in notepads])
|
||
|
|
|
||
|
|
@admin_bp.route('/notepads/<int:notepad_id>/notes', methods=['GET'])
|
||
|
|
def get_notepad_notes(notepad_id):
|
||
|
|
notepad = Notepad.query.get_or_404(notepad_id)
|
||
|
|
notes = notepad.notes.all()
|
||
|
|
return jsonify([note.to_dict() for note in notes])
|
||
|
|
|
||
|
|
@admin_bp.route('/notes/<int:note_id>', methods=['DELETE'])
|
||
|
|
def delete_note(note_id):
|
||
|
|
note = Note.query.get_or_404(note_id)
|
||
|
|
db.session.delete(note)
|
||
|
|
db.session.commit()
|
||
|
|
return jsonify({'message': 'Note deleted'}), 200
|
||
|
|
|
||
|
|
@admin_bp.route('/notepads/<int:notepad_id>', methods=['DELETE'])
|
||
|
|
def delete_notepad(notepad_id):
|
||
|
|
notepad = Notepad.query.get_or_404(notepad_id)
|
||
|
|
# Clean up notes first to prevent FK errors if cascade is missing
|
||
|
|
for note in notepad.notes:
|
||
|
|
db.session.delete(note)
|
||
|
|
from models.favorite import Favorite
|
||
|
|
Favorite.query.filter_by(notepad_id=notepad.id).delete()
|
||
|
|
db.session.delete(notepad)
|
||
|
|
db.session.commit()
|
||
|
|
return jsonify({'message': 'Notepad deleted'}), 200
|
||
|
|
|
||
|
|
@admin_bp.route('/users/<int:user_id>', methods=['DELETE'])
|
||
|
|
def delete_user(user_id):
|
||
|
|
user = User.query.get_or_404(user_id)
|
||
|
|
|
||
|
|
# Clean up user's notepads and notes
|
||
|
|
notepads = Notepad.query.filter_by(user_id=user.id).all()
|
||
|
|
from models.favorite import Favorite
|
||
|
|
for notepad in notepads:
|
||
|
|
for note in notepad.notes:
|
||
|
|
db.session.delete(note)
|
||
|
|
Favorite.query.filter_by(notepad_id=notepad.id).delete()
|
||
|
|
db.session.delete(notepad)
|
||
|
|
|
||
|
|
Favorite.query.filter_by(user_id=user.id).delete()
|
||
|
|
db.session.delete(user)
|
||
|
|
db.session.commit()
|
||
|
|
return jsonify({'message': 'User deleted'}), 200
|
||
|
|
|
||
|
|
@admin_bp.route('/users/<int:user_id>/ban', methods=['POST'])
|
||
|
|
def ban_user(user_id):
|
||
|
|
user = User.query.get_or_404(user_id)
|
||
|
|
user.is_banned = True
|
||
|
|
db.session.commit()
|
||
|
|
return jsonify({'message': 'User banned'}), 200
|
||
|
|
|
||
|
|
@admin_bp.route('/users/<int:user_id>/unban', methods=['POST'])
|
||
|
|
def unban_user(user_id):
|
||
|
|
user = User.query.get_or_404(user_id)
|
||
|
|
user.is_banned = False
|
||
|
|
db.session.commit()
|
||
|
|
return jsonify({'message': 'User unbanned'}), 200
|
||
|
|
|
||
|
|
@admin_bp.route('/users/<int:user_id>/ban_ip', methods=['POST'])
|
||
|
|
def ban_user_ip(user_id):
|
||
|
|
user = User.query.get_or_404(user_id)
|
||
|
|
if not user.last_ip:
|
||
|
|
return jsonify({'error': 'User has no recorded IP'}), 400
|
||
|
|
|
||
|
|
if not BannedIP.query.filter_by(ip_address=user.last_ip).first():
|
||
|
|
banned = BannedIP(ip_address=user.last_ip)
|
||
|
|
db.session.add(banned)
|
||
|
|
db.session.commit()
|
||
|
|
|
||
|
|
return jsonify({'message': f'IP {user.last_ip} banned'}), 200
|