Create app.py

app.py

@@ -0,0 +1,136 @@
+from fastapi import FastAPI, Request
+import requests
+import logging
+import os, json
+from datetime import datetime
+from google.oauth2 import service_account
+from googleapiclient.discovery import build
+
+# Existing service account JSON dictionary
+
+# Retrieve the secret as a string
+google_credentials = os.getenv("GOOGLE_CREDENTIALS")
+
+# Parse the secret if it's stored as a JSON string
+if google_credentials:
+    json_api_dict = json.loads(google_credentials)
+    print("Valid Credentials")
+else:
+    print("No secret found.")
+
+app = FastAPI()
+logging.basicConfig(level=logging.INFO)
+
+# Set your credentials
+CUSTOM_FIELD_ID = os.getenv("CUSTOM_FIELD_ID")
+
+ACCESS_TOKEN = os.getenv("ACCESS_TOKEN")
+
+def update_custom_field(task_id: str, value: str):
+    """Updates the ClickUp task's custom field."""
+    url = f"https://api.clickup.com/api/v2/task/{task_id}/field/{CUSTOM_FIELD_ID}"
+    headers = {
+        "Authorization": ACCESS_TOKEN,
+        "Content-Type": "application/json"
+    }
+    payload = {"value": value}  # Payload should be a dictionary
+    
+    # Corrected to pass the payload correctly
+    response = requests.post(url, json=payload, headers=headers)
+    
+    if response.status_code == 200:
+        print(f"Custom field updated successfully for task {task_id}")
+    else:
+        print(f"Failed to update custom field: {response.text}")
+
+def create_drive_folder(folder_name, month):
+    creds = service_account.Credentials.from_service_account_info(json_api_dict)
+    drive_service = build('drive', 'v3', credentials=creds)
+    parent_folder_id = os.getenv("PARENT_FOLDER_ID")
+    
+    def search_folder(name, parent_id):
+        query = f"name='{name}' and '{parent_id}' in parents and mimeType='application/vnd.google-apps.folder' and trashed=false"
+        results = drive_service.files().list(q=query, fields='files(id, name, webViewLink)').execute()
+        files = results.get('files', [])
+        return files[0] if files else None
+    
+    month_folder = search_folder(month, parent_folder_id)
+    if not month_folder:
+        month_metadata = {'name': month, 'mimeType': 'application/vnd.google-apps.folder', 'parents': [parent_folder_id]}
+        month_folder = drive_service.files().create(body=month_metadata, fields='id, webViewLink').execute()
+    
+    target_folder = search_folder(folder_name, month_folder['id'])
+    if target_folder:
+        return target_folder['webViewLink']
+    
+    folder_metadata = {'name': folder_name, 'mimeType': 'application/vnd.google-apps.folder', 'parents': [month_folder['id']]}
+    folder = drive_service.files().create(body=folder_metadata, fields='id, webViewLink').execute()
+    return folder['webViewLink']
+
+@app.post("/webhook")
+async def receive_webhook(request: Request):
+    """Receives ClickUp webhook events for new tasks."""
+    try:
+        # Parse the JSON payload
+        data = await request.json()
+        logging.info(f"Received webhook: {data}")
+        
+        # Check if the event is 'taskCreated'
+        if data.get("event") == "taskCreated":
+            task_id = data.get("task_id")  # Task ID is in the top level of the data
+            history_items = data.get("history_items", [])
+
+            task_name = "Unnamed Task"  # Default if no name is found
+            creation_timestamp = None  # Placeholder for creation timestamp
+
+            # Loop through history_items to find task creation and task name
+            for item in history_items:
+                if item.get("field") == "task_creation":
+                    # Ensure 'date' is an integer before division
+                    creation_timestamp = int(item.get("date", 0))  # Convert to int if necessary
+
+            # If we have a task_id, make an API call to get task details
+            if task_id:
+                task_name = get_task_details(task_id)
+
+            # If a creation timestamp is found, convert it to a month
+            if creation_timestamp:
+                creation_date = datetime.utcfromtimestamp(creation_timestamp / 1000)  # Convert from milliseconds
+                creation_month = creation_date.strftime("%b %Y")  # Format it as "Mon Year" (e.g., "Feb 2025")
+            else:
+                creation_month = "Unknown Month"
+
+            logging.info(f"New Task Created: {task_name} (ID: {task_id})")
+            logging.info(f"Task Creation Month: {creation_month}")
+        
+        else:
+            logging.info(f"Received a different event: {data.get('event')}")
+    
+    except Exception as e:
+        logging.error(f"Error processing webhook: {e}")
+
+    # Update custom field "1- Drive Link" with a sample value
+    drive_link = create_drive_folder(task_name, creation_month)
+    update_custom_field(task_id, drive_link)
+    
+    return {"status": "success"}
+
+def get_task_details(task_id: str):
+    """Fetch task details from ClickUp API using task_id."""
+    url = f"https://api.clickup.com/api/v2/task/{task_id}"
+    headers = {
+        "Authorization": ACCESS_TOKEN
+    }
+    
+    # Make the API request to ClickUp
+    response = requests.get(url, headers=headers)
+
+    # Check if the request was successful (status code 200)
+    if response.status_code == 200:
+        # Parse and return the task details
+        
+        task_name = response.json()["name"]
+        return task_name
+    else:
+        task_name = "Task name wasn't not found"
+        return task_name