66 lines
2.7 KiB
Python
66 lines
2.7 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""Resolve ClickUp UserConnection and configure ClickupService."""
|
|
|
|
import logging
|
|
import re
|
|
from typing import Any, Dict, Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_UUID_RE = re.compile(
|
|
r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
|
|
class ClickupConnectionHelper:
|
|
def __init__(self, method_instance: Any):
|
|
self.method = method_instance
|
|
self.services = method_instance.services
|
|
|
|
def _normalize_connection_reference(self, ref: str) -> str:
|
|
"""Match listConnections / getUserConnectionFromConnectionReference formats."""
|
|
if ref.startswith("connection:"):
|
|
return ref
|
|
if _UUID_RE.match(ref):
|
|
return ref
|
|
# LLM often copies "clickup:username" without the connection: prefix
|
|
if ":" in ref:
|
|
return f"connection:{ref}"
|
|
return ref
|
|
|
|
def get_clickup_connection(self, connection_reference: str) -> Optional[Dict[str, Any]]:
|
|
try:
|
|
ref = (connection_reference or "").split(" [")[0].strip()
|
|
if not ref:
|
|
return None
|
|
ref = self._normalize_connection_reference(ref)
|
|
chat = getattr(self.services, "chat", None)
|
|
if not chat or not hasattr(chat, "getUserConnectionFromConnectionReference"):
|
|
logger.warning("Chat service missing; cannot resolve ClickUp connection")
|
|
return None
|
|
user_connection = chat.getUserConnectionFromConnectionReference(ref)
|
|
if not user_connection:
|
|
logger.warning("No user connection for reference/id %s", connection_reference)
|
|
return None
|
|
authority = getattr(user_connection.authority, "value", None) or str(
|
|
user_connection.authority
|
|
)
|
|
if authority != "clickup":
|
|
logger.warning("Connection %s is not ClickUp (authority=%s)", user_connection.id, authority)
|
|
return None
|
|
status = getattr(user_connection.status, "value", None) or str(user_connection.status)
|
|
if status not in ("active", "pending"):
|
|
logger.warning("Connection %s status not active: %s", user_connection.id, status)
|
|
|
|
cu = getattr(self.services, "clickup", None)
|
|
if not cu:
|
|
return None
|
|
if not cu.setAccessTokenFromConnection(user_connection):
|
|
logger.warning("Failed to set ClickUp token for connection %s", user_connection.id)
|
|
return None
|
|
return {"id": user_connection.id, "userConnection": user_connection}
|
|
except Exception as e:
|
|
logger.error("get_clickup_connection error: %s", e)
|
|
return None
|