|
@ -6,90 +6,90 @@ from pywebpush import webpush, WebPushException |
|
|
import json |
|
|
import json |
|
|
import os |
|
|
import os |
|
|
|
|
|
|
|
|
app = Flask(__name__) |
|
|
def create_app(config_name): |
|
|
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///webpush.db' |
|
|
app = Flask(__name__) |
|
|
db.init_app(app) |
|
|
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///webpush.db' |
|
|
|
|
|
db.init_app(app) |
|
|
def generate_and_save_vapid_keys(): |
|
|
|
|
|
vapid = Vapid() |
|
|
|
|
|
vapid.generate_keys() |
|
|
def generate_and_save_vapid_keys(): |
|
|
private_key = vapid.get_private_key().to_pem().decode('utf-8').strip() |
|
|
vapid = Vapid() |
|
|
public_key = vapid.get_public_key().to_pem().decode('utf-8').strip() |
|
|
vapid.generate_keys() |
|
|
|
|
|
private_key = vapid.get_private_key().to_pem().decode('utf-8').strip() |
|
|
key = VAPIDKey(public_key=public_key, private_key=private_key) |
|
|
public_key = vapid.get_public_key().to_pem().decode('utf-8').strip() |
|
|
db.session.add(key) |
|
|
|
|
|
db.session.commit() |
|
|
key = VAPIDKey(public_key=public_key, private_key=private_key) |
|
|
|
|
|
db.session.add(key) |
|
|
|
|
|
db.session.commit() |
|
|
@app.before_first_request |
|
|
|
|
|
def initialize(): |
|
|
|
|
|
if not VAPIDKey.query.first(): |
|
|
def initialize(): |
|
|
generate_and_save_vapid_keys() |
|
|
if not VAPIDKey.query.first(): |
|
|
|
|
|
generate_and_save_vapid_keys() |
|
|
def send_push_notification(subscription_info, message, vapid_key): |
|
|
|
|
|
try: |
|
|
|
|
|
webpush( |
|
|
|
|
|
subscription_info=subscription_info, |
|
|
|
|
|
data=json.dumps(message), |
|
|
|
|
|
vapid_private_key=vapid_key.private_key, |
|
|
|
|
|
vapid_claims={ |
|
|
|
|
|
"sub": "mailto:your-email@example.com" |
|
|
|
|
|
} |
|
|
|
|
|
) |
|
|
|
|
|
except WebPushException as e: |
|
|
|
|
|
print(f"Failed to send notification: {e}") |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@app.route('/web-push/vapid', methods=['GET']) |
|
|
|
|
|
def get_vapid(): |
|
|
|
|
|
key = VAPIDKey.query.first() |
|
|
|
|
|
if key: |
|
|
|
|
|
return jsonify(public_key=key.public_key) |
|
|
|
|
|
return jsonify(error='No VAPID keys found'), 404 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@app.route('/web-push/subscribe', methods=['POST']) |
|
|
|
|
|
def subscribe(): |
|
|
|
|
|
content = request.json |
|
|
|
|
|
vapid_key = VAPIDKey.query.first() |
|
|
|
|
|
|
|
|
|
|
|
if not vapid_key: |
|
|
|
|
|
return jsonify(success=False, error="No VAPID keys available"), 500 |
|
|
|
|
|
|
|
|
|
|
|
subscription = Subscription(endpoint=content['endpoint'], |
|
|
|
|
|
p256dh=content['keys']['p256dh'], |
|
|
|
|
|
auth=content['keys']['auth'], |
|
|
|
|
|
vapid_key_id=vapid_key.id) |
|
|
|
|
|
db.session.add(subscription) |
|
|
|
|
|
db.session.commit() |
|
|
|
|
|
|
|
|
|
|
|
subscription_info = { |
|
|
|
|
|
"endpoint": subscription.endpoint, |
|
|
|
|
|
"keys": { |
|
|
|
|
|
"p256dh": subscription.p256dh, |
|
|
|
|
|
"auth": subscription.auth |
|
|
|
|
|
} |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
message = {"title": "Subscription Successful", "body": "Thank you for subscribing!"} |
|
|
def send_push_notification(subscription_info, message, vapid_key): |
|
|
send_push_notification(subscription_info, message, vapid_key) |
|
|
try: |
|
|
|
|
|
webpush( |
|
|
|
|
|
subscription_info=subscription_info, |
|
|
|
|
|
data=json.dumps(message), |
|
|
|
|
|
vapid_private_key=vapid_key.private_key, |
|
|
|
|
|
vapid_claims={ |
|
|
|
|
|
"sub": "mailto:your-email@example.com" |
|
|
|
|
|
} |
|
|
|
|
|
) |
|
|
|
|
|
except WebPushException as e: |
|
|
|
|
|
print(f"Failed to send notification: {e}") |
|
|
|
|
|
|
|
|
return jsonify(success=True) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@app.route('/web-push/vapid', methods=['GET']) |
|
|
|
|
|
def get_vapid(): |
|
|
|
|
|
key = VAPIDKey.query.first() |
|
|
|
|
|
if key: |
|
|
|
|
|
return jsonify(public_key=key.public_key) |
|
|
|
|
|
return jsonify(error='No VAPID keys found'), 404 |
|
|
|
|
|
|
|
|
@app.route('/web-push/unsubscribe', methods=['DELETE']) |
|
|
|
|
|
def unsubscribe(): |
|
|
|
|
|
content = request.json |
|
|
|
|
|
endpoint = content['endpoint'] |
|
|
|
|
|
subscription = Subscription.query.filter_by(endpoint=endpoint).first() |
|
|
|
|
|
|
|
|
|
|
|
if subscription: |
|
|
@app.route('/web-push/subscribe', methods=['POST']) |
|
|
db.session.delete(subscription) |
|
|
def subscribe(): |
|
|
|
|
|
content = request.json |
|
|
|
|
|
vapid_key = VAPIDKey.query.first() |
|
|
|
|
|
|
|
|
|
|
|
if not vapid_key: |
|
|
|
|
|
return jsonify(success=False, error="No VAPID keys available"), 500 |
|
|
|
|
|
|
|
|
|
|
|
subscription = Subscription(endpoint=content['endpoint'], |
|
|
|
|
|
p256dh=content['keys']['p256dh'], |
|
|
|
|
|
auth=content['keys']['auth'], |
|
|
|
|
|
vapid_key_id=vapid_key.id) |
|
|
|
|
|
db.session.add(subscription) |
|
|
db.session.commit() |
|
|
db.session.commit() |
|
|
return jsonify(success=True, message="Subscription deleted successfully") |
|
|
|
|
|
else: |
|
|
|
|
|
return jsonify(success=False, error="Subscription not found"), 404 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
subscription_info = { |
|
|
|
|
|
"endpoint": subscription.endpoint, |
|
|
|
|
|
"keys": { |
|
|
|
|
|
"p256dh": subscription.p256dh, |
|
|
|
|
|
"auth": subscription.auth |
|
|
|
|
|
} |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
message = {"title": "Subscription Successful", "body": "Thank you for subscribing!"} |
|
|
|
|
|
send_push_notification(subscription_info, message, vapid_key) |
|
|
|
|
|
|
|
|
|
|
|
return jsonify(success=True) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@app.route('/web-push/unsubscribe', methods=['DELETE']) |
|
|
|
|
|
def unsubscribe(): |
|
|
|
|
|
content = request.json |
|
|
|
|
|
endpoint = content['endpoint'] |
|
|
|
|
|
subscription = Subscription.query.filter_by(endpoint=endpoint).first() |
|
|
|
|
|
|
|
|
|
|
|
if subscription: |
|
|
|
|
|
db.session.delete(subscription) |
|
|
|
|
|
db.session.commit() |
|
|
|
|
|
return jsonify(success=True, message="Subscription deleted successfully") |
|
|
|
|
|
else: |
|
|
|
|
|
return jsonify(success=False, error="Subscription not found"), 404 |
|
|
|
|
|
|
|
|
if __name__ == '__main__': |
|
|
return app |
|
|
app.run(debug=True) |
|
|
|
|
|