Feat: create a Flask App for webhook event subscriptions

Author: huaminghuangtw

.gitignore

@@ -1,6 +1,7 @@
 TODOs.txt
+tokens.txt
 __pycache__/
 .idea/
 .vscode/
-configfile/
+.env
 logfile/

Procfile

@@ -0,0 +1 @@
+web: gunicorn --chdir ./src application:app

credentials/config.yaml

@@ -1,6 +1,5 @@
 Flask
 line-bot-sdk
-pyyaml
 stravalib
 requests
 pandas
@@ -10,6 +9,6 @@ selenium
 pyautogui
 pyperclip
 alive-progress
+python-dotenv
 gunicorn
-psycopg2
-
+psycopg2

credentials/tokens.txt

@@ -62,8 +62,8 @@ def upload_Fit_Activity_Files(access_token: str):
             with open(fitfile, 'rb') as fit_file:
                 try:
                     uploads_url = "https://www.strava.com/api/v3/uploads"
-                    payload = { 'client_id': STRAVA_CLIENT_ID,
-                                'data_type': 'fit' }
+                    payload = {'client_id': STRAVA_CLIENT_ID,
+                               'data_type': 'fit'}
                     header = {'Authorization': 'Bearer ' + access_token}
                     f = {'file': fit_file}
                     r = requests.post(uploads_url,
@@ -81,11 +81,11 @@ def upload_Fit_Activity_Files(access_token: str):
                 except (KeyError, TypeError, ValueError):
                     return None
 
-                while (True):
+                while True:
                     # polling the upload status per semaild
                     wait(1)
 
-                    isError, activity_id = check_Upload_Status(access_token, fitfile, upload_ID)
+                    isError, activity_id = check_Upload_Status(access_token, upload_ID)
                     time.sleep(0.05)
 
                     # If there is an error uploading activity file or
@@ -104,7 +104,7 @@ def upload_Fit_Activity_Files(access_token: str):
                 print("")
 
 
-def check_Upload_Status(access_token: str, filename: str, upload_ID: str) -> Tuple[bool, Any]:
+def check_Upload_Status(access_token: str, upload_ID: str) -> Tuple[bool, Any]:
     try:
         uploads_url = "https://www.strava.com/api/v3/uploads/" + upload_ID
         header = {'Authorization': 'Bearer ' + access_token}

requirements.txt

@@ -43,8 +43,7 @@ def main():
     """)
     preprocessing()
     fix_Fit_Activity_Files()
-    TOKEN_FILE_PATH = Path('../credentials/tokens.txt')
-    access_token = get_access_token(TOKEN_FILE_PATH)
+    access_token = get_access_token()
     upload_Fit_Activity_Files(access_token)
 
 

src/StravaUploadTool_Kernel.py

@@ -0,0 +1,58 @@
+from stravalib import Client
+from authentication import *
+from sendLINEMessage import *
+from flask import Flask, request
+
+
+app = Flask(__name__)
+
+
+# Validate webhook subscriptions
+@app.get('/webhook')
+def webhook_get():
+    data = request.args
+    print(data)
+    # Your verify token (should be a random string)
+    VERIFY_TOKEN = "STRAVA"
+    # Parse the query string parameters
+    mode = data['hub.mode']
+    verify_token = data['hub.verify_token']
+    challenge = data['hub.challenge']
+    if (mode != 'subscribe') or (verify_token != VERIFY_TOKEN):
+        print('WEBHOOK_NOT_VERIFIED')
+        return ('INVALID_REQUEST', 401)
+    else:
+        print('WEBHOOK_VERIFIED')
+        return ({'hub.challenge': challenge}, 200)
+
+
+# Receive webhook events
+@app.post('/webhook')
+def webhook_post():
+    print('EVENT_RECEIVED')
+    data = request.json
+    print(data)
+    # You can do whatever you want upon receving a webhook event
+    # Here we send a LINE message with the Strava activity link
+    # when a non-indoor activity is created
+    if (data["aspect_type"] == "create"):
+        client = Client(access_token=get_access_token())
+        activity = client.get_activity(data["object_id"])
+        if (activity.trainer is True): # indoor activity
+            print("Activity " + str(activity.id) + " is an indoor activity, please manually delete it via " +
+                  "https://www.strava.com/activities/" + str(activity.id))
+            return ("INDOOR_ACTIVITY_CREATED", 200)
+        else:
+            if sendLINEMessage("https://www.strava.com/activities/" + str(activity.id)):
+                print("LINE message sent!")
+                return ("ACTIVITY_CREATED", 200)
+            else:
+                return ("SOMETHING_WAS_WRONG", 404)
+    elif (data["aspect_type"] == "update"):
+        return ("ACTIVITY_UPDATED", 200)
+    elif (data["aspect_type"] == "delete"):
+        return ("ACTIVITY_DELETED", 200)
+
+
+if __name__ == "__main__":
+    app.run(debug=True)

src/StravaUploadTool_Main.py

@@ -13,7 +13,7 @@
 client = Client()
 
 
-def get_access_token(file_path: pathlib.Path) -> str:
+def get_access_token(file_path: pathlib.Path = pathlib.Path('../tokens.txt')) -> str:
     """
     Obtain and return an OAuth2 access token for the Strava V3 API
     @params:
@@ -50,14 +50,15 @@ def get_access_token(file_path: pathlib.Path) -> str:
 # ----------------
 def _read_tokens_from_file(file_path: pathlib.Path) -> dict:
     """
-    Read the authentication tokens and expiry time from a text file and return them as a dictionary
+    Read the authentication tokens and expiry time from a text file
+    and return them as a dictionary
     @params:
-        file_path - The path of the file to read the tokens from.
+        file_path - The path of the file to read the tokens from
     @return:
         A dictionary containing the authentication tokens and expiry time
         An empty dictionary if the file cannot be read from successfully
     """
-    print("[Strava]: Reading authentication tokens from '{}'...".format(file_path))
+    print("[Strava]: Reading authentication tokens from '{}'".format(file_path))
 
     tokens = {}
 
@@ -85,8 +86,8 @@ def _write_tokens_to_file(file_path: pathlib.Path, tokens: dict):
         file_path - The path of the file to write the tokens to
         tokens - A dictionary containing the authentication tokens and expiry time to write to the file
     """
-    print("[Strava]: Writing authentication tokens to '{}'...".format(file_path))
-
+    print("[Strava]: Writing authentication tokens to '{}'".format(file_path))
+    
     # Delete the tokens file to remove any expired tokens
     try:
         file_path.unlink()

src/application.py

@@ -162,7 +162,7 @@ def fix_Fit_Activity_Files():
                         print("ERROR - Cannot find the element!")
                         driver.quit()
 
-        # Step 12: finally, close the browser window and change back to the original working directory
+        # Step 12: finally, close the web browser window and change back to the original working directory
         driver.quit()
         os.chdir(owd)
 

src/authentication.py

@@ -1,44 +1,41 @@
-import yaml
-import sys
-from pathlib import Path
+import os
+from dotenv import load_dotenv
 
 
-CONFIG_FILE_PATH = Path('../credentials/config.yaml')
-with CONFIG_FILE_PATH.open(mode='r') as file:
-    config = yaml.safe_load(file)
+load_dotenv("../.env")  # take environment variables from .env
 
 
 try:
-    STRAVA_CLIENT_ID = config["STRAVA_CLIENT_ID"]
+    STRAVA_CLIENT_ID = os.environ["STRAVA_CLIENT_ID"]
 except KeyError:
-    sys.exit("[ERROR]: Please add STRAVA_CLIENT_ID to '{}'".format(CONFIG_FILE_PATH))
+    print("[ERROR]: Please add STRAVA_CLIENT_ID to the list of environment variables!")
 
 try:
-    STRAVA_CLIENT_SECRET = config["STRAVA_CLIENT_SECRET"]
+    STRAVA_CLIENT_SECRET = os.environ["STRAVA_CLIENT_SECRET"]
 except KeyError:
-    sys.exit("[ERROR]: Please add STRAVA_CLIENT_SECRET to '{}'".format(CONFIG_FILE_PATH))
-    
+    print("[ERROR]: Please add STRAVA_CLIENT_SECRET to the list of environment variables")
+
 try:
-    ZWIFT_ACTIVITY_DIR = config["ZWIFT_ACTIVITY_DIR"]
+    GMAIL_USER_ID = os.environ["GMAIL_USER_ID"]
 except KeyError:
-    sys.exit("[ERROR]: Please add ZWIFT_ACTIVITY_DIR to '{}'".format(CONFIG_FILE_PATH))
+    print("[ERROR]: Please add GMAIL_USER_ID to the list of environment variables")
 
 try:
-    GMAIL_USER_ID = config["GMAIL_USER_ID"]
+    GMAIL_PASSWORD = os.environ["GMAIL_PASSWORD"]
 except KeyError:
-    sys.exit("[ERROR]: Please add GMAIL_USER_ID to '{}'".format(CONFIG_FILE_PATH))
+	print("[ERROR]: Please add GMAIL_PASSWORD to the list of environment variables")
 
 try:
-    GMAIL_PASSWORD = config["GMAIL_PASSWORD"]
+    LINE_CHANNEL_ACCESS_TOKEN = os.environ["LINE_CHANNEL_ACCESS_TOKEN"]
 except KeyError:
-    sys.exit("[ERROR]: Please add GMAIL_PASSWORD to '{}'".format(CONFIG_FILE_PATH))
+    print("[ERROR]: Please add LINE_CHANNEL_ACCESS_TOKEN to the list of environment variables")
 
 try:
-    LINE_CHANNEL_ACCESS_TOKEN = config["LINE_CHANNEL_ACCESS_TOKEN"]
+    LINE_CHANNEL_SECRET = os.environ["LINE_CHANNEL_SECRET"]
 except KeyError:
-    sys.exit("[ERROR]: Please add LINE_CHANNEL_ACCESS_TOKEN to '{}'".format(CONFIG_FILE_PATH))
+    print("[ERROR]: Please add LINE_CHANNEL_SECRET to the list of environment variables")
 
 try:
-    LINE_CHANNEL_SECRET = config["LINE_CHANNEL_SECRET"]
+    ZWIFT_ACTIVITY_DIR = os.environ["ZWIFT_ACTIVITY_DIR"]
 except KeyError:
-    sys.exit("[ERROR]: Please add LINE_CHANNEL_SECRET to '{}'".format(CONFIG_FILE_PATH))
+    print("[ERROR]: Please add ZWIFT_ACTIVITY_DIR to the list of environment variables")

src/fitfiletools_automation.py

@@ -0,0 +1,15 @@
+from linebot import LineBotApi
+from linebot.models import TextSendMessage
+from linebot.exceptions import LineBotApiError
+from read_config_file import *
+
+
+line_bot_api = LineBotApi(LINE_CHANNEL_ACCESS_TOKEN)
+
+
+def sendLINEMessage(StrvaActivityURL: str):
+    try:
+        line_bot_api.broadcast(TextSendMessage(text=StrvaActivityURL))
+    except LineBotApiError as e:
+        print("The was an error sending LINE message!")
+    return 200