163 lines
4.8 KiB
Python
163 lines
4.8 KiB
Python
from flask import Flask, render_template, request, redirect, send_from_directory, jsonify
|
|
from config import app, db, bcrypt
|
|
from models import Contact, User, Orders
|
|
from werkzeug.utils import secure_filename
|
|
from werkzeug.exceptions import RequestEntityTooLarge
|
|
import json, os
|
|
import logging
|
|
from pywebpush import webpush, WebPushException
|
|
|
|
|
|
DER_BASE64_ENCODED_PRIVATE_KEY_FILE_PATH = os.path.join(os.getcwd(),"private_key.txt")
|
|
DER_BASE64_ENCODED_PUBLIC_KEY_FILE_PATH = os.path.join(os.getcwd(),"public_key.txt")
|
|
|
|
VAPID_PRIVATE_KEY = open(DER_BASE64_ENCODED_PRIVATE_KEY_FILE_PATH, "r+").readline().strip("\n")
|
|
VAPID_PUBLIC_KEY = open(DER_BASE64_ENCODED_PUBLIC_KEY_FILE_PATH, "r+").read().strip("\n")
|
|
|
|
VAPID_CLAIMS = {
|
|
"sub": "mailto:arcjcb11p@gmail.com"
|
|
}
|
|
|
|
|
|
def send_web_push(subscription_information, message_body):
|
|
return webpush(
|
|
subscription_info=subscription_information,
|
|
data=message_body,
|
|
vapid_private_key=VAPID_PRIVATE_KEY,
|
|
vapid_claims=VAPID_CLAIMS
|
|
)
|
|
|
|
|
|
@app.route("/subscription/", methods=["GET", "POST"])
|
|
def subscription():
|
|
"""
|
|
POST creates a subscription
|
|
GET returns vapid public key which clients uses to send around push notification
|
|
"""
|
|
|
|
if request.method == "GET":
|
|
return Response(response=json.dumps({"public_key": VAPID_PUBLIC_KEY}),
|
|
headers={"Access-Control-Allow-Origin": "*"}, content_type="application/json")
|
|
|
|
subscription_token = request.get_json("subscription_token")
|
|
return Response(status=201, mimetype="application/json")
|
|
|
|
|
|
@app.route("/push_v1/",methods=['POST'])
|
|
def push_v1():
|
|
message = "Push Test v1"
|
|
print("is_json",request.is_json)
|
|
|
|
if not request.json or not request.json.get('sub_token'):
|
|
return jsonify({'failed':1})
|
|
|
|
print("request.json",request.json)
|
|
|
|
token = request.json.get('sub_token')
|
|
try:
|
|
token = json.loads(token)
|
|
send_web_push(token, message)
|
|
return jsonify({'success':1})
|
|
except Exception as e:
|
|
print("error",e)
|
|
return jsonify({'failed':str(e)})
|
|
|
|
|
|
@app.route('/')
|
|
def index():
|
|
files = os.listdir(app.config['UPLOAD_DIRECTORY'])
|
|
images = []
|
|
|
|
for file in files:
|
|
if os.path.splitext(file)[1].lower() in app.config['ALLOWED_EXTENSIONS']:
|
|
images.append(file)
|
|
return render_template('index.html', images=images)
|
|
|
|
@app.route('/list')
|
|
def orderList():
|
|
orders = Orders.query.all()
|
|
all_orders = list(map(lambda x : x.to_json(), orders))
|
|
return render_template('orders.html', orders=all_orders)
|
|
|
|
@app.route('/upload', methods=["POST"])
|
|
def upload():
|
|
try:
|
|
file = request.files['file']
|
|
file_extension = os.path.splitext(file.filename)[1] # [0] is the filename
|
|
|
|
# save the file if it exists
|
|
if file:
|
|
if file_extension not in app.config['ALLOWED_EXTENSIONS']:
|
|
return 'File is not an image'
|
|
file.save(os.path.join(
|
|
app.config['UPLOAD_DIRECTORY'],
|
|
secure_filename(file.filename)
|
|
))
|
|
except RequestEntityTooLarge: # except Exception would catch all errors instead of specified
|
|
return 'File is larger than the limit of 16mb.'
|
|
|
|
return redirect('/')
|
|
|
|
@app.route("/mail", methods=["POST"])
|
|
def mail():
|
|
email = request.json["senderEmail"]
|
|
password = request.json["password"]
|
|
order = request.json["order"]
|
|
|
|
|
|
if email is None:
|
|
return jsonify({"error": "Unauthorized"}), 401
|
|
|
|
if not bcrypt.check_password_hash(user.password, password):
|
|
return jsonify({"error": "Unauthorized"}), 401
|
|
|
|
# Cache session with redis
|
|
session["user_id"] = user.id
|
|
|
|
return jsonify({
|
|
"id": user.id,
|
|
"email": user.email
|
|
})
|
|
|
|
|
|
# Handle Images
|
|
@app.route('/serve-image/<filename>', methods=['GET'])
|
|
def serve_image(filename):
|
|
|
|
return send_from_directory(app.config['UPLOAD_DIRECTORY'], filename)
|
|
|
|
|
|
# New Orders
|
|
@app.route("/new-order", methods=["POST"])
|
|
def register_user():
|
|
email = request.json["senderEmail"]
|
|
order = request.json["order"]
|
|
|
|
new_order= Orders(email=email, order=order)
|
|
|
|
# Write to db
|
|
try:
|
|
db.session.add(new_order) # add to db staging area
|
|
db.session.commit() # add to db and catch any errors that can happen with writing to db
|
|
except Exception as e:
|
|
return (jsonify({"message": str(e)}), 400)
|
|
|
|
|
|
return jsonify({
|
|
"email": email,
|
|
"order confirmed": order,
|
|
})
|
|
# Get orders
|
|
@app.route("/orders", methods=["GET"])
|
|
def orders():
|
|
orders = Orders.query.all()
|
|
json_orders = list(map(lambda x : x.to_json(), orders))
|
|
return jsonify({"orders": json_orders})
|
|
|
|
# Main
|
|
if __name__ == "__main__":
|
|
# Initialize database connection
|
|
with app.app_context():
|
|
db.create_all()
|
|
|
|
app.run(debug=True, port=5080, host="0.0.0.0") |