183 lines
8.0 KiB
Python
183 lines
8.0 KiB
Python
import httpx
|
|
import os
|
|
from typing import List, Dict, Any
|
|
|
|
class PlaneService:
|
|
def __init__(self):
|
|
self.api_url = os.getenv("PLANE_API_URL", "https://api.plane.so/api/v1")
|
|
self.api_key = os.getenv("PLANE_API_KEY", "")
|
|
self.workspace_slug = os.getenv("PLANE_WORKSPACE_SLUG", "simulation-workspace")
|
|
self.project_id = os.getenv("PLANE_PROJECT_ID", "simulation-project-id")
|
|
|
|
# Enable simulation mode if no API key is provided
|
|
self.is_simulation = not self.api_key or self.api_key == "mock"
|
|
|
|
async def get_label_id(self, label_name: str) -> str:
|
|
"""
|
|
Helper to get Label ID by name, or create if not exists.
|
|
"""
|
|
if self.is_simulation:
|
|
return "simulated-label-id"
|
|
|
|
headers = {
|
|
"X-API-Key": self.api_key,
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
try:
|
|
# 1. List labels (Plane API usually returns all project labels)
|
|
# Adjust URL based on Plane version: /workspaces/{slug}/projects/{id}/labels/ or /issue-labels/
|
|
url = f"{self.api_url}/workspaces/{self.workspace_slug}/projects/{self.project_id}/issue-labels/"
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.get(url, headers=headers)
|
|
|
|
if response.status_code == 200:
|
|
labels = response.json()
|
|
for label in labels:
|
|
if label.get("name") == label_name:
|
|
return label.get("id")
|
|
|
|
# 2. Create if not found
|
|
# Random color generator or fixed
|
|
payload = {"name": label_name, "color": "#ff0000"}
|
|
create_res = await client.post(url, headers=headers, json=payload)
|
|
if create_res.status_code == 201:
|
|
return create_res.json().get("id")
|
|
|
|
except Exception as e:
|
|
print(f"Failed to get/create label {label_name}: {e}")
|
|
|
|
return None
|
|
|
|
async def create_issue(self, title: str, description: str, priority: str = "Medium", estimate: float = None, labels: List[str] = []) -> Dict[str, Any]:
|
|
"""
|
|
Creates an issue in Plane.
|
|
"""
|
|
if self.is_simulation:
|
|
print(f"[SIMULATION] Creating Plane Issue: {title} ({estimate}h) - {priority} - Labels: {labels}")
|
|
return {"id": "simulated-issue-id", "name": title, "status": "created (simulation)"}
|
|
|
|
headers = {
|
|
"X-API-Key": self.api_key,
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
# Map priority to Plane's expected format
|
|
plane_priority = priority.lower()
|
|
if plane_priority not in ["urgent", "high", "medium", "low", "none"]:
|
|
plane_priority = "medium"
|
|
|
|
# Resolve Label IDs
|
|
label_ids = []
|
|
for label_name in labels:
|
|
lid = await self.get_label_id(label_name)
|
|
if lid:
|
|
label_ids.append(lid)
|
|
|
|
payload = {
|
|
"name": title,
|
|
"description_html": f"<p>{description}</p>",
|
|
"priority": plane_priority,
|
|
"label_ids": label_ids
|
|
}
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
try:
|
|
url = f"{self.api_url}/workspaces/{self.workspace_slug}/projects/{self.project_id}/issues/"
|
|
response = await client.post(url, headers=headers, json=payload)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Failed to create Plane issue: {e}")
|
|
return {"error": str(e)}
|
|
|
|
async def update_issue_status(self, issue_identifier: str, status: str = "Done") -> Dict[str, Any]:
|
|
"""
|
|
Updates the status of a Plane issue.
|
|
issue_identifier: Can be issue ID or Sequence ID (e.g., "PROJ-123")
|
|
"""
|
|
if self.is_simulation:
|
|
print(f"[SIMULATION] Updating Plane Issue {issue_identifier} to status: {status}")
|
|
return {"id": issue_identifier, "status": status, "simulated": True}
|
|
|
|
headers = {
|
|
"X-API-Key": self.api_key,
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
# First, we need to find the issue ID if a sequence ID is provided
|
|
# Or if we assume issue_identifier is the UUID
|
|
issue_id = issue_identifier
|
|
|
|
# If it looks like a sequence ID (e.g., PROJ-123), we might need to search for it first
|
|
# For simplicity, let's assume we might need to search or use it directly if Plane supports it.
|
|
# Plane API usually works with UUIDs for updates.
|
|
# Let's try to search if it contains a hyphen and isn't a UUID
|
|
if "-" in issue_identifier and len(issue_identifier) < 30:
|
|
# Rough heuristic for sequence ID vs UUID
|
|
# Search for the issue to get its UUID
|
|
pass # TODO: Implement search logic if needed.
|
|
# For now, let's assume the user provides the UUID or we just try our best.
|
|
|
|
# Plane uses State IDs for status updates, not just string names like "Done".
|
|
# In a real implementation, we would need to fetch the State ID for "Done" from the project states.
|
|
# For this MVP, we will simulate or assume a state_id is passed, or try to map it.
|
|
# But since we can't easily map "Done" to a UUID without fetching states,
|
|
# let's assume for now we are just patching the issue and Plane handles it or we mock it.
|
|
|
|
# Real-world: Fetch states -> Find ID for "Done" -> Update Issue
|
|
# Simplified: Just log the intent or try to update if we had the ID.
|
|
|
|
print(f"Warning: To update status to '{status}' in Plane, we need the specific State UUID.")
|
|
print(f"Skipping actual API call for status update in this MVP version to avoid 400 errors without State IDs.")
|
|
|
|
return {"status": "skipped_implementation_gap", "message": "State ID lookup required for real update"}
|
|
|
|
async def sync_tasks(self, tasks: List[Dict[str, Any]], is_urgent: bool = False) -> Dict[str, Any]:
|
|
"""
|
|
Syncs a list of tasks to Plane.
|
|
Returns details about created tasks including IDs.
|
|
"""
|
|
created_count = 0
|
|
errors = []
|
|
created_tasks_details = []
|
|
|
|
# Determine priority and labels based on urgency
|
|
priority = "Urgent" if is_urgent else "Medium"
|
|
labels = ["Signal-Driven"] if is_urgent else []
|
|
|
|
for task in tasks:
|
|
description = f"Type: {task.get('type', 'General')}<br>Estimated Hours: {task.get('estimated_hours', 0)}"
|
|
|
|
result = await self.create_issue(
|
|
title=task.get("task_name", "Untitled Task"),
|
|
description=description,
|
|
priority=priority,
|
|
estimate=task.get("estimated_hours"),
|
|
labels=labels
|
|
)
|
|
|
|
if "error" in result:
|
|
errors.append(f"Failed to create {task.get('task_name')}: {result['error']}")
|
|
else:
|
|
created_count += 1
|
|
# Append task with its new Plane ID (simulated or real)
|
|
# result contains the issue object from Plane API
|
|
task_with_id = task.copy()
|
|
task_with_id["plane_id"] = result.get("id")
|
|
# Also try to get a sequence ID if available (e.g. "PROJ-123")
|
|
task_with_id["plane_key"] = result.get("sequence_id") or result.get("identifier") or result.get("id")
|
|
task_with_id["status"] = "Todo" # Initial status
|
|
created_tasks_details.append(task_with_id)
|
|
|
|
return {
|
|
"status": "success" if not errors else "partial_success",
|
|
"created_count": created_count,
|
|
"errors": errors,
|
|
"created_tasks": created_tasks_details,
|
|
"simulation_mode": self.is_simulation
|
|
}
|
|
|
|
# Singleton
|
|
plane_service = PlaneService()
|