Spaces:
Sleeping
Sleeping
File size: 7,130 Bytes
f0ba3c6 | 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 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 | import asyncio
from datetime import datetime
import logging
import os
from typing import Any, Dict, List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel
from integrations.asana_service import asana_service
from integrations.jira_service import get_jira_service
from integrations.microsoft365_service import microsoft365_service
from integrations.zoho_projects_service import ZohoProjectsService
router = APIRouter(prefix="/api/atom/projects/live", tags=["projects-live"])
logger = logging.getLogger(__name__)
# --- Data Models ---
class UnifiedTask(BaseModel):
name: str
platform: str # 'asana', 'jira', 'zoho', 'planner'
status: str
priority: Optional[str] = "normal"
assignee: Optional[str] = None
due_date: Optional[str] = None
project_name: Optional[str] = None
url: Optional[str] = None
class ProjectStats(BaseModel):
total_active_tasks: int
completed_today: int
overdue_count: int
tasks_by_platform: Dict[str, int]
class LiveProjectsResponse(BaseModel):
ok: bool = True
stats: ProjectStats
tasks: List[UnifiedTask]
providers: Dict[str, bool]
# --- Helper Functions ---
def map_asana_task(task: Dict[str, Any]) -> UnifiedTask:
return UnifiedTask(
id=task.get("gid"),
name=task.get("name") or "Untitled Task",
platform="asana",
status="completed" if task.get("completed") else "active",
assignee=task.get("assignee_name"),
due_date=task.get("due_on"),
project_name=None, # Expensive to fetch per task, skipped for live view speed
url=task.get("url")
)
def map_jira_issue(issue: Dict[str, Any], base_url: str) -> UnifiedTask:
fields = issue.get("fields", {})
return UnifiedTask(
id=issue.get("key"),
name=fields.get("summary") or "Untitled Issue",
platform="jira",
status=fields.get("status", {}).get("name", "Unknown"),
priority=fields.get("priority", {}).get("name", "normal"),
assignee=fields.get("assignee", {}).get("displayName"),
due_date=fields.get("duedate"),
project_name=fields.get("project", {}).get("name"),
url=f"{base_url}/browse/{issue.get('key')}"
)
def map_zoho_task(task: Dict[str, Any]) -> UnifiedTask:
return UnifiedTask(
id=task.get("id_string"),
name=task.get("name") or "Untitled Zoho Task",
platform="zoho",
status="completed" if task.get("status", {}).get("type") == "completed" else "active",
priority=task.get("priority", "normal"),
assignee=task.get("created_person"),
due_date=task.get("end_date"),
project_name=task.get("project_name")
)
def map_planner_task(task: Dict[str, Any]) -> UnifiedTask:
return UnifiedTask(
id=task.get("id", "planner_task"),
name=task.get("title") or "Untitled MS Task",
platform="planner",
status="completed" if task.get("completedDateTime") else "active",
priority="normal", # Planner has complexity in priority mapping
due_date=task.get("dueDateTime"),
project_name="MS Planner"
)
# --- Endpoints ---
@router.get("/board", response_model=LiveProjectsResponse)
async def get_live_project_board(
limit: int = 50,
# User ID dependency would ideally be here
):
"""
Fetch live tasks from connected Project Management tools (Asana, Jira)
and aggregate them into a unified board view.
"""
tasks = []
providers_status = {"asana": False, "jira": False, "zoho": False, "planner": False}
# 1. Fetch Asana Tasks
try:
# Get user's Asana access token from environment or use service account
asana_token = os.getenv("ASANA_ACCESS_TOKEN")
if not asana_token:
logger.warning("ASANA_ACCESS_TOKEN not configured, skipping Asana fetch")
else:
# Use asana_service to fetch tasks
asana_tasks = asana_service.get_user_tasks(user_id=user_id, limit=limit)
tasks.extend([map_asana_task(t) for t in asana_tasks])
providers_status["asana"] = True
except Exception as e:
logger.warning(f"Failed to fetch live Asana tasks: {e}")
# 2. Fetch Jira Issues
try:
jira = get_jira_service()
if jira:
# Verify if environmental config is present (Mock/Dev mode)
test_conn = jira.test_connection()
if test_conn.get("authenticated"):
# JQL for open issues assigned to current user fallback
# "assignee = currentUser() AND status != Done"
jql = "order by created DESC"
raw_data = jira.search_issues(jql=jql, max_results=limit)
raw_issues = raw_data.get("issues", [])
base_url = jira.base_url
tasks.extend([map_jira_issue(i, base_url) for i in raw_issues])
providers_status["jira"] = True
else:
logger.info("Jira service not available (credentials missing)")
except Exception as e:
logger.warning(f"Failed to fetch live Jira issues: {e}")
# 3. Fetch Zoho Projects Tasks
try:
zoho_token = os.getenv("ZOHO_CRM_ACCESS_TOKEN") # Reusing same base secret if applicable or ZOHO_PROJECTS_TOKEN
portal_id = os.getenv("ZOHO_PROJECTS_PORTAL_ID")
if zoho_token and portal_id:
zoho = ZohoProjectsService()
raw_tasks = await zoho.get_all_active_tasks(zoho_token, portal_id, limit=limit)
tasks.extend([map_zoho_task(t) for t in raw_tasks])
providers_status["zoho"] = True
except Exception as e:
logger.warning(f"Failed to fetch live Zoho Projects tasks: {e}")
# 4. Fetch Microsoft Planner Tasks
try:
ms_token = os.getenv("MICROSOFT_365_ACCESS_TOKEN")
if ms_token:
res = await microsoft365_service.get_planner_tasks(access_token=ms_token, top=limit)
if res.get("status") == "success":
raw_tasks = res.get("data", {}).get("value", [])
tasks.extend([map_planner_task(t) for t in raw_tasks])
providers_status["planner"] = True
except Exception as e:
logger.warning(f"Failed to fetch live MS Planner tasks: {e}")
# Calculate Stats
total_active = len(tasks)
# Simple logic for overdue - robust logic would parse dates
overdue = 0
platform_counts = {
"asana": len([t for t in tasks if t.platform == 'asana']),
"jira": len([t for t in tasks if t.platform == 'jira']),
"zoho": len([t for t in tasks if t.platform == 'zoho']),
"planner": len([t for t in tasks if t.platform == 'planner'])
}
return LiveProjectsResponse(
ok=True,
stats=ProjectStats(
total_active_tasks=total_active,
completed_today=0, # Need more logic/history for this
overdue_count=overdue,
tasks_by_platform=platform_counts
),
tasks=tasks,
providers=providers_status
)
|