import os import io import base64 from PIL import Image from flask import send_file, url_for, render_template from flask_restx import Namespace, Resource, fields, reqparse from flask_jwt_extended import jwt_required, get_jwt_identity from flask import jsonify from app.models import Profile from app.services.user import UserService from app.utils import admin_required import json from app.config.extensions import bcrypt from flask_jwt_extended import create_access_token from app.services.mail import MailService from werkzeug.datastructures import FileStorage UPLOAD_FOLDER = 'uploads' # Create an instance of the Namespace for account-related endpoints account_namespace = Namespace('account', description='Account operations') email_model = account_namespace.model('Email', { 'email': fields.String(required=True) }) account_model = account_namespace.model('Email',{"email":fields.String(required=True)}) upload_parser = reqparse.RequestParser() upload_parser.add_argument('image', location='files', type=FileStorage, required=True) user_model = account_namespace.model('Email', { 'email': fields.String(required=True, description='Username'), 'password': fields.String(required=True, description='Password'), }) @account_namespace.route('/me') class AccountMe(Resource): @jwt_required() # Ensure the request is protected by a valid JWT access token using the 'jwt_required' decorator @account_namespace.doc(security='Bearer_Token') def get(self): """ Endpoint that retreive single user based on User Token Args: Resource (strin): User Token Returns: json: json Data of user information """ # Get the current user's identity from the JWT token current_user_id = get_jwt_identity() # print("Current User ID: ",current_user_id) # For this example, we'll just return a mock user dictionary. user_info = UserService.get_user_by_email(current_user_id) # print("user dict", user_info) user_info.ensure_client_role() user_info.save() # if not user_info.profile or user_info.profile is None: user_info.profile = Profile() user_info.save() print(user_info.to_dict()) return jsonify(user_info.to_dict(), 200) # Return user information as JSON with HTTP status code 200 @jwt_required() # Ensure the request is protected by a valid JWT access token using the 'jwt_required' decorator # @account_namespace.doc(security='Bearer_Token') def patch(self): # os.makedirs(UPLOAD_FOLDER, exist_ok=True) # Get the current user's identity from the JWT token. current_user_id = get_jwt_identity() # Get the request JSON data from the payload. user_update_json = account_namespace.payload # user_info = UserService.get_user_by_email(current_user_id) # Check for username, first_name and last_name for field in ['first_name', 'last_name', 'username']: if field in user_update_json: setattr(user_info, field, user_update_json.pop(field)) for field in ['updated_at', 'tags', 'roles']: user_update_json.pop(field, None) # Handle profile photo upload if 'avatar' in user_update_json: # avatar_file = user_update_json.pop('avatar') # Write bytes to file and save in upload folder if 'name' in avatar_file: file_path = os.path.join(UPLOAD_FOLDER, avatar_file["name"]) else : import uuid file_path = os.path.join(UPLOAD_FOLDER, f"{uuid.uuid4()}.png") if 'base64file' in avatar_file: image = Image.open(io.BytesIO(base64.b64decode(avatar_file["base64file"].split(',')[1] ))) image.save(file_path) user_update_json["avatar"] = f"http://173.249.58.66:8082/account/avatar/{avatar_file['name']}" # url_for('account_image_resource', file_name= avatar_file["name"], _external=True, _scheme='http', _server='173.249.58.66:8082') # user_info.profile = Profile(**user_update_json) for key, value in user_update_json.items(): if hasattr(user_info.profile, key): setattr(user_info.profile, key, value) user_info.save() return user_info.to_dict(), 200 # Return user information as JSON with HTTP status code 200 @account_namespace.route('/exist') class AccountExist(Resource): @account_namespace.expect(email_model) def post(self): """ Endpoint that check user if exists or not based on user email Returns: json: json format of user boolean if exist or not """ # Get the request JSON data from the payload. data = account_namespace.payload # For this example, we'll just return a mock user dictionary. user_info = UserService.get_user_by_email(data["email"]) if user_info and user_info is not None: return {'exists': True}, 200 else: return {'exists': False}, 200 @account_namespace.route('/deactivate') class AccountDeactivateResource(Resource): @jwt_required() def delete(self): try: # Access the user identity (user ID) from the JWT token current_user_id = get_jwt_identity() # Retrieve user information based on the user's identity user_info = UserService.get_user_by_email(current_user_id) # Deactivate the user's account by toggling the active status if user_info.is_active: user_info.toggle_active_status() # Archive the user's account if not user_info.deleted: user_info.archive() return { 'msg': 'Your account has been deactivated. Please contact our administrator within 72 hours to discuss account reactivation. Failure to do so may result in permanent account deletion.', 'data': None, 'status': 200 }, 200 except Exception as e: if hasattr(e, 'code') and e.code is not None: # If the exception has a status code attribute, return it return { 'msg': str(e), 'data': None, 'status': e.code }, e.code # Other status code # If no specific status code is provided, return a 500 Internal Server Error return { 'msg': 'An error occurred while processing your request: ' + str(e), 'data': None, 'status': 500 }, 500 # Internal Server Error @account_namespace.route('/stats') class AccountStatsResource(Resource): def get(self): stats = [ { 'stats': '21,459', 'title': 'Session', 'trendDiff': '+29', 'icon': 'tabler:user', 'subtitle': 'Total Users' }, { 'stats': '4,567', 'trendDiff': '+18', 'title': 'Paid Users', 'avatarColor': 'error', 'icon': 'tabler:user-plus', 'subtitle': 'Last week analytics' }, { 'stats': '19,860', 'trendDiff': '-14', 'trend': 'negative', 'title': 'Active Users', 'avatarColor': 'success', 'icon': 'tabler:user-check', 'subtitle': 'Last week analytics' }, { 'stats': '237', 'trendDiff': '+42', 'title': 'Pending Users', 'avatarColor': 'warning', 'icon': 'tabler:user-exclamation', 'subtitle': 'Last week analytics' } ] return {'statsHorizontalWithDetails': stats} @account_namespace.route('/avatar/') class ImageResource(Resource): def get(self, file_name): # Specify the directory where the images are stored image_directory = "/app/uploads" # Replace with the actual directory path # Serve the requested image file from the specified directory return send_file(os.path.join(image_directory, file_name), mimetype='image/png') @account_namespace.route('/upload_image') class ImageResource(Resource): @account_namespace.expect(upload_parser) def post(self,username): args = upload_parser.parse_args() uploaded_file = args['image'] user = UserService.get_user_by_email_or_username(username) if user: file_path = f"{UPLOAD_FOLDER}/{user['username']}_{uploaded_file.filename}" uploaded_file.save(file_path) user = UserService.get_user_by_email_or_username(username) # if user has a profile already update the avatar pic if user['profile']: user['profile'].avatar = file_path user['profile'].save() return {'message': 'Image uploaded successfully'}, 201 else: # create new one user['profile'] = Profile(avatar=file_path) user.save() else: return {'message': 'User not found'}, 404 @account_namespace.route("/users") class UsersList(Resource): @jwt_required() # Ensure the request is protected by a valid JWT access token using the 'jwt_required' decorator @admin_required() @account_namespace.doc(security='Bearer_Token', description="Return all users") def get(self): # Get the current user's identity from the JWT token users = UserService.all() formated_data = [] for user in users: formated_data.append(user) return jsonify(formated_data) @account_namespace.route("/users/") class UserInfo(Resource): @jwt_required() # Ensure the request is protected by a valid JWT access token using the 'jwt_required' decorator @admin_required() @account_namespace.doc(security='Bearer_Token', description="Return all users") def get(self,public_id): user = UserService.get_user_by_id(public_id) if user: return jsonify(user) return {"message":"User Not Found!"},404 @account_namespace.expect(account_model) def put(self,public_id): print(account_namespace.payload) def delete(self,public_id): user = UserService.get_user_by_id(public_id) user.delete() return {"message":"User Are Deleted Success!"},200 @account_namespace.route('/change-password/') class AccountChangePasswordResource(Resource): @jwt_required() @account_namespace.expect(user_model) def post(self,token): data = account_namespace.payload email = data.get("email") new_password = data.get('password') # Assuming you have a user database or some mechanism to get user information # Here, we are using the current_user for demonstration purposes current_user = UserService.get_user_by_email(email) if current_user: # Update the password new_pass = UserService.change_password(user_id=current_user['public_id'],new_password=new_password) return {'message': 'Password updated successfully'}, 200 else: return {'message': 'Invalid username or password'}, 401 @account_namespace.route('/forgot-password') class ForgotPasswordResource(Resource): @account_namespace.expect(user_model) def post(self): data = account_namespace.payload username = data.get('username') email = data.get('email') current_user = UserService.get_user_by_email(email) print("current user: ",current_user) # Assuming you have a user database or some mechanism to get user information # Here, we are using the example_user for demonstration purposes if email == current_user.email: additional_claims = { "is_superuser": current_user.is_superuser, "is_active": current_user.is_active, "is_staff": current_user.is_staff, "is_admin": current_user.is_admin, } # Generate a token and send a password reset link to the user's email token = create_access_token(identity=current_user.email, additional_claims=additional_claims) reset_link = f'http://0.0.0.0:8002/api/account/change_password/{token}' # send email MailService.send_email( subject="Account Activation: Email Verification", recipient = current_user.email, message = render_template('reset_password.html', password_reset_link=reset_link), html=True ) return {'message': 'Password reset link sent to your email', 'reset_link': reset_link}, 200 else: return {'message': 'Invalid username or email'}, 401