Super simple and small Web Push and Notification service
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 

94 lines
2.1 KiB

  1. # (A) INIT
  2. # (A1) LOAD MODULES
  3. from flask import Flask, render_template, request, make_response, send_from_directory
  4. from pywebpush import webpush, WebPushException
  5. import json
  6. import time
  7. import threading
  8. # (A2) FLASK SETTINGS + INIT - CHANGE TO YOUR OWN!
  9. HOST_NAME = "localhost"
  10. HOST_NAME = "0.0.0.0"
  11. HOST_PORT = 80
  12. VAPID_SUBJECT = "mailto:your@email.com"
  13. VAPID_PRIVATE = "YOUR-PRIVATE-KEY"
  14. app = Flask(__name__,
  15. static_folder='../static',
  16. template_folder='.',
  17. )
  18. app.debug = True
  19. #app.config['EXPLAIN_TEMPLATE_LOADING'] = True
  20. # (B) VIEWS
  21. # (B1) "LANDING PAGE"
  22. @app.route("/")
  23. def index():
  24. return render_template("S2_perm_sw.html")
  25. # (B2) SERVICE WORKER
  26. @app.route("/S3_sw.js")
  27. def sw():
  28. import sys
  29. print(repr(app.static_folder), file=sys.stderr)
  30. response = make_response(send_from_directory(app.static_folder, "S3_sw.js"))
  31. return response
  32. threadlist = []
  33. def donotify(sub, sleep=10):
  34. global threadlist
  35. for t in threadlist:
  36. t.join(0)
  37. threadlist = [ t for t in threadlist if t.is_alive() ]
  38. if sleep:
  39. time.sleep(sleep)
  40. print('sending notification...')
  41. try:
  42. # Mozilla appears to not require the vapid key.
  43. # Chrome requires the vapid key.
  44. if True:
  45. kwargs = dict(vapid_private_key=None,
  46. #vapid_claims=None,
  47. )
  48. else:
  49. kwargs = dict(vapid_private_key=VAPID_PRIVATE,
  50. vapid_claims={ "sub": VAPID_SUBJECT },
  51. )
  52. webpush(
  53. subscription_info = sub,
  54. data = json.dumps({
  55. "title" : "Welcome!",
  56. "body" : "Yes, it works!",
  57. "icon" : "static/i-ico.png",
  58. "image" : "static/i-banner.png"
  59. }), **kwargs
  60. )
  61. print('done')
  62. except WebPushException as ex:
  63. print(ex)
  64. t = threading.Thread(target=donotify, args=(sub,))
  65. t.run()
  66. threadlist.append(t)
  67. # (B3) PUSH DEMO
  68. @app.route("/push", methods=["POST"])
  69. def push():
  70. # (B3-1) GET SUBSCRIBER
  71. sub = json.loads(request.form["sub"])
  72. import sys
  73. print('sub:', repr(json.dumps(sub)), file=sys.stderr)
  74. # (B3-2) TEST PUSH NOTIFICATION
  75. result = "OK"
  76. donotify(sub, 1)
  77. return result
  78. # (C) START
  79. if __name__ == "__main__":
  80. app.run(HOST_NAME, HOST_PORT)