Spaces:
Sleeping
Sleeping
File size: 5,362 Bytes
a051e25 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 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 |