Upload New File server.py

server.py

+from flask import Flask, request, send_from_directory
+import random
+import json
+import database_helper as database_helper
+import re
+from flask_sock import Sock
+import time
+app = Flask(__name__)
+sock = Sock(app)
+
+# function to signin the user with post method
+# We will expect 2 string values email address password.
+# if the authentication is successful we will return access token
+@app.route("/sign_in", methods=["POST"])
+def sign_in():
+    email_address = request.form.get("email", "")
+    password = request.form.get("password", "")
+    print("Email address: " + email_address, "password: " + password)
+
+    if len(email_address) == 0 or len(password) == 0:
+        return json.dumps(
+            {"success": False, "message": "Form data missing or incorrect type."}
+        ), 403
+
+    result = database_helper.find_user_by_email_password(
+        email_address, password)
+
+    if result["status"] == True:
+        if len(result["data"]) == 0:
+            return json.dumps(
+                {"success": False, "message": "Wrong username or password."}
+            ), 401
+        else:
+            token = random.getrandbits(30)
+            database_helper.remove_logged_in_session(email_address)
+            database_helper.add_user_email_token(email_address, token)
+            return json.dumps(
+                {
+                    "success": True,
+                    "message": "Successfully sign in.",
+                    "data": str(token),
+                }
+            ), 200
+
+    return json.dumps({"success": False, "message": "error in sign in"}), 500
+
+
+# function to signup the user with post method
+# We will expect 7 string values following: email, password, firstname, familyname, gender, city and country.
+# if the sign_up is successful we will return access token
+@app.route("/sign_up", methods=["POST"])
+def sign_up():
+    regex = r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,7}\b"
+    email_address = request.form.get("email", "")
+    password = request.form.get("password", "")
+    firstname = request.form.get("firstname", "")
+    familyname = request.form.get("familyname", "")
+    gender = request.form.get("gender", "")
+    city = request.form.get("city", "")
+    country = request.form.get("country", "")
+
+    if (
+        len(email_address) == 0
+        or len(password) == 0
+        or len(firstname) == 0
+        or len(familyname) == 0
+        or len(gender) == 0
+        or len(city) == 0
+        or len(country) == 0
+    ):
+        return json.dumps(
+            {"success": False, "message": "Form data missing or incorrect type."}
+        ), 403
+
+    if len(password) < 8:
+        return json.dumps(
+            {
+                "success": False,
+                "message": "Password should be at least 8 characters long.",
+            }
+        ), 401
+
+    if re.fullmatch(regex, email_address) is None:
+        return json.dumps(
+            {
+                "success": False,
+                "message": "Invalid email, please provide valid email address",
+            }
+        ), 401
+
+    if database_helper.user_exists_email(email_address):
+        return json.dumps({"success": False, "message": "User already exists."}), 403
+
+    result = database_helper.create_user(
+        email_address, password, firstname, familyname, gender, city, country
+    )
+
+    if result["status"] == True:
+        return json.dumps(
+            {"success": True, "message": "Successfully created a new user."}
+        ), 200
+
+    return json.dumps({"success": False, "message": "error in sign up"}), 500
+
+
+# function to signout the user with post method
+# We will expect 1 string values token.
+# if the token is valid we will signout the user
+@app.route("/sign_out", methods=["DELETE"])
+def sign_out():
+    token = request.headers.get("token", "")
+
+    if len(token) == 0:
+        return json.dumps({"success": False, "message": "Token missing or incorrect."}), 401
+
+    result = database_helper.find_token(token)
+
+    if (len(result["data"])) == 0:
+        return json.dumps({"success": False, "message": "You are not signed in."}), 403
+
+    result = database_helper.remove_token(token)
+
+    if result["status"] == True:
+        return json.dumps({"success": True, "message": "Successfully signed out."}), 200
+    elif result["status"] == False:
+        return json.dumps({"success": False, "message": result["data"]}), 200
+
+    return json.dumps({"success": False, "message": "error in sign out"}), 500
+
+
+# function to change password of the user with put method
+# We will expect 3 string values token, oldPassword, newPassword: .
+# if the data is valid we will change password of the user
+@app.route("/change_password", methods=["PUT"])
+def change_password():
+    token = request.headers.get("token", "")
+    oldPassword = request.form.get("oldPassword", "")
+    newPassword = request.form.get("newPassword", "")
+
+    if len(token) == 0:
+        return json.dumps({"success": False, "message": "Token missing or incorrect."}), 401
+
+    if len(oldPassword) == 0 or len(newPassword) == 0:
+        return json.dumps(
+            {"success": False, "message": "Form data missing or incorrect type."}
+        ), 403
+
+    if len(oldPassword) < 8 or len(newPassword) < 8:
+        return json.dumps(
+            {
+                "success": False,
+                "message": "old password and new password length should be grater then 8.",
+            }
+        ), 400
+
+    if oldPassword == newPassword:
+        return json.dumps(
+            {
+                "success": False,
+                "message": "old password and new password should be different.",
+            }
+        ), 406
+
+    # validate token
+    result = database_helper.find_token(token)
+    if (len(result["data"])) == 0:
+        return json.dumps({"success": False, "message": "You are not logged in."}), 401
+
+    email = result["data"][0][1]
+
+    # find_user_by_email_password
+    result = database_helper.find_user_by_email_password(email, oldPassword)
+
+    if result["status"] == True and (len(result["data"])) == 0:
+        return json.dumps({"success": False, "message": "Wrong password."}), 401
+
+    result = database_helper.change_password(email, newPassword)
+    if result["status"] == True:
+        return json.dumps({"success": True, "message": "Password changed."}), 200
+
+    return json.dumps({"success": False, "message": "error in change password"}), 500
+
+
+# function to get the user with get method.
+# We will expect 2 string values email and token: .
+# if the data is valid we return the user data
+@app.route("/get_user_data_by_email", methods=["POST"])
+def get_user_data_by_email():
+    token = request.headers.get("token", "")
+    emailAddress = request.form.get("email", "")
+    if len(token) == 0:
+        return json.dumps({"success": False, "message": "Token missing or incorrect."}), 401
+
+    if len(emailAddress) == 0:
+        return json.dumps(
+            {"success": False, "message": "Form data missing or incorrect type."}
+        ), 403
+
+    # validate token
+    result = database_helper.find_token(token)
+    if (len(result["data"])) == 0:
+        return json.dumps({"success": False, "message": "You are not logged in."}), 401
+
+    # find_user_by_email_password
+    result = database_helper.find_user_by_email(emailAddress)
+
+    if result["status"] == True and (len(result["data"])) != 0:
+        temp_user = {
+            "email": result["data"][0][0],
+            "firstname": result["data"][0][2],
+            "familyname": result["data"][0][3],
+            "gender": result["data"][0][4],
+            "city": result["data"][0][5],
+            "country": result["data"][0][6],
+        }
+        return json.dumps(
+            {"success": True, "message": "user details", "data": temp_user}, 200
+        )
+    elif result["status"] == True and (len(result["data"])) == 0:
+        return json.dumps({"success": False, "message": "Invalid user email"}), 401
+
+    return json.dumps({"success": False, "message": "error in get_user_data_by_email"}), 500
+
+
+# function to get the user with post method
+# We will expect 1 string values token: .
+# if the data is valid we return the user data
+@app.route("/get_user_data_by_token", methods=["GET"])
+def get_user_data_by_token():
+    token = request.headers.get("token", "")
+
+    if len(token) == 0:
+        return json.dumps({"success": False, "message": "Token missing or incorrect."}), 401
+
+    # validate token
+    result = database_helper.find_token(token)
+    if (len(result["data"])) == 0:
+        return json.dumps({"success": False, "message": "You are not logged in."}), 401
+
+    email = result["data"][0][1]
+
+    # find_user_by_email_password
+    result = database_helper.find_user_by_email(email)
+
+    if result["status"] == True and (len(result["data"])) != 0:
+        temp_user = {
+            "email": result["data"][0][0],
+            "firstname": result["data"][0][2],
+            "familyname": result["data"][0][3],
+            "gender": result["data"][0][4],
+            "city": result["data"][0][5],
+            "country": result["data"][0][6],
+        }
+        return json.dumps(
+            {"success": True, "message": "user details", "data": temp_user}
+        ), 200
+
+    return json.dumps({"success": False, "message": "error in get_user_data_by_token"}), 500
+
+
+# function to get the user messages post method
+# We will expect 1 string values token: .
+# if the data is valid we return the user data
+@app.route("/get_user_messages_by_token", methods=["GET"])
+def get_user_messages_by_token():
+    token = request.headers.get("token", "")
+
+    if len(token) == 0:
+        return json.dumps({"success": False, "message": "Token missing or incorrect."}), 401
+
+    # validate token
+    result = database_helper.find_token(token)
+    if (len(result["data"])) == 0:
+        return json.dumps({"success": False, "message": "You are not logged in."}), 401
+
+    email = result["data"][0][1]
+
+    # find_user_by_email_password
+    result = database_helper.get_message_from_email(email)
+
+    if result["status"] == True and (len(result["data"])) == 0:
+        return json.dumps({"success": True, "message": "No messages available"}), 200
+    elif result["status"] == True and (len(result["data"])) != 0:
+        temp_data = []
+        for message in result["data"]:
+            temp_data.append({"writer": message[2], "content": message[1]})
+        return json.dumps({"success": True, "data": temp_data}), 200
+
+    return json.dumps({"success": False, "message": "error in get user data by token"}), 500
+
+
+# function to get the user messages post method
+# We will expect 1 string values email: .
+# if the data is valid we return the user data
+@app.route("/get_user_messages_by_email", methods=["POST"])
+def get_user_messages_by_email():
+    token = request.headers.get("token", "")
+    email = request.form.get("email", "")
+    
+    if len(token) == 0:
+        return json.dumps({"success": False, "message": "Token missing or incorrect."}), 401
+
+    if len(email) == 0:
+        return json.dumps(
+            {"success": False, "message": "Form data missing or incorrect type."}
+        ), 403
+
+    # validate token
+    result = database_helper.find_token(token)
+    if (len(result["data"])) == 0:
+        return json.dumps({"success": False, "message": "You are not logged in."}), 401
+
+    # validate email
+    result = database_helper.find_user_by_email(email)
+    if result["status"] == True and (len(result["data"])) == 0:
+        return json.dumps({"success": False, "message": "Invalid user email"}), 401
+
+    # find_user_by_email_password
+    result = database_helper.get_message_from_email(email)
+
+    if result["status"] == True and (len(result["data"])) == 0:
+        return json.dumps({"success": True, "message": "No messages available"}), 200
+    elif result["status"] == True and (len(result["data"])) != 0:
+        temp_data = []
+        for message in result["data"]:
+            temp_data.append({"writer": message[2], "content": message[1]})
+        return json.dumps({"success": True, "data": temp_data}), 200
+
+    return json.dumps({"success": False, "message": "error in get user data by token"}), 500
+
+
+# function to get the post messages
+# We will expect 3 string values token, message, email.
+# if the data is valid we return the user data
+@app.route("/post_message", methods=["POST"])
+def post_message():
+    recipient_email = request.form.get("email", "")
+    token = request.headers.get("token", "")
+    message = request.form.get("message", "")
+    
+    print("recipient_email" + recipient_email + " message" + message + " token" + token)
+
+    if len(token) == 0:
+        return json.dumps({"success": False, "message": "Token missing or incorrect."}), 401
+
+    if len(message) == 0:
+        return json.dumps(
+            {"success": False, "message": "Form data missing or incorrect type."}
+        ), 403
+  
+    # validate token
+    result = database_helper.find_token(token)
+    if (len(result["data"])) == 0:
+        return json.dumps({"success": False, "message": "You are not logged in."}), 401
+    
+    curr_user_email = result["data"][0][1]
+
+    # validate email
+    if recipient_email:        
+        result = database_helper.find_user_by_email(recipient_email)
+        if result["status"] == True and (len(result["data"])) == 0:
+            return json.dumps({"success": False, "message": "Invalid user email"}), 401
+    else:
+         recipient_email = curr_user_email
+
+    result = database_helper.add_message_and_email(
+        curr_user_email, message, recipient_email
+    )
+
+    if result["status"] == True:
+        return json.dumps({"success": True, "message": "Message posted"}), 200
+
+    return json.dumps({"success": False, "message": "error in get_user_data_by_token"}), 500
+
+# function to Validate token
+# We will expect 1 string values token.
+@app.route("/validate_token", methods=["GET"])
+def validate_token():
+    token = request.headers.get("token", "")
+    if database_helper.token_exist(token):
+        return json.dumps({"success": True, "message": "Valid Token"}), 201
+    else:
+        return json.dumps({"success": False, "message": "Invalid Token"}), 401
+    
+
+@sock.route('/api')
+def handle_websocket(ws):
+    if not ws:
+        return json.dumps({'Expected WebSocket request.'}), 401
+
+    token  = ws.receive()
+    while True:
+        time.sleep(2)
+        print("1 = ", token)
+        # validate token
+        result = database_helper.find_token(token)
+        print("len = ", len(result["data"]))
+        if (len(result["data"])) == 0:
+            ws.send(json.dumps({"success": True, "message": "User signed out"}))
+
+# function to load template
+@app.route('/')
+def index():
+    return send_from_directory('static', 'client.html'), 200