How to Manage a Team of 10 Engineers Using AI Project Management Tools Like Jira 6.0 and Asana 5.0

import requests
import json
import os
import logging
from datetime import datetime, timedelta
import time

# Configure logging for audit trails
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[logging.FileHandler('jira_asana_sync.log'), logging.StreamHandler()]
)

class JiraAsanaSync:
    def __init__(self):
        # Load API credentials from environment variables (never hardcode!)
        self.jira_url = os.getenv('JIRA_URL', 'https://your-org.atlassian.net')
        self.jira_api_key = os.getenv('JIRA_API_KEY')
        self.jira_project_key = os.getenv('JIRA_PROJECT_KEY', 'ENG')
        self.asana_url = 'https://app.asana.com/api/1.0'
        self.asana_api_key = os.getenv('ASANA_API_KEY')
        self.asana_project_gid = os.getenv('ASANA_PROJECT_GID')

        # Validate required env vars
        if not all([self.jira_api_key, self.asana_api_key, self.asana_project_gid]):
            raise ValueError("Missing required environment variables: JIRA_API_KEY, ASANA_API_KEY, ASANA_PROJECT_GID")

        # Set up headers for both APIs
        self.jira_headers = {
            'Authorization': f'Bearer {self.jira_api_key}',
            'Accept': 'application/json',
            'Content-Type': 'application/json'
        }
        self.asana_headers = {
            'Authorization': f'Bearer {self.asana_api_key}',
            'Accept': 'application/json',
            'Content-Type': 'application/json'
        }

        # Rate limit trackers (Asana 5.0 allows 500 req/min, Jira 6.0 allows 1000 req/min)
        self.jira_last_request = datetime.now()
        self.asana_last_request = datetime.now()
        self.jira_rate_limit = 1000  # req per minute
        self.asana_rate_limit = 500  # req per minute

    def _rate_limit(self, api_type):
        """Enforce rate limits for Jira or Asana APIs"""
        if api_type == 'jira':
            elapsed = (datetime.now() - self.jira_last_request).total_seconds()
            min_interval = 60 / self.jira_rate_limit
            if elapsed < min_interval:
                time.sleep(min_interval - elapsed)
            self.jira_last_request = datetime.now()
        elif api_type == 'asana':
            elapsed = (datetime.now() - self.asana_last_request).total_seconds()
            min_interval = 60 / self.asana_rate_limit
            if elapsed < min_interval:
                time.sleep(min_interval - elapsed)
            self.asana_last_request = datetime.now()

    def sync_jira_to_asana(self):
        """Fetch new Jira issues and create corresponding Asana tasks"""
        try:
            # Query Jira for issues updated in last 15 minutes (adjust based on sync frequency)
            jql = f'project={self.jira_project_key} AND updated >= -15m AND status != Done'
            self._rate_limit('jira')
            response = requests.get(
                f'{self.jira_url}/rest/api/3/search',
                headers=self.jira_headers,
                params={'jql': jql, 'maxResults': 100},
                timeout=10
            )
            response.raise_for_status()
            jira_issues = response.json().get('issues', [])
            logging.info(f"Fetched {len(jira_issues)} Jira issues to sync")

            for issue in jira_issues:
                issue_key = issue['key']
                summary = issue['fields']['summary']
                description = issue['fields'].get('description', '')
                assignee = issue['fields'].get('assignee', {}).get('displayName', 'Unassigned')

                # Check if task already exists in Asana to avoid duplicates
                self._rate_limit('asana')
                existing = requests.get(
                    f'{self.asana_url}/tasks',
                    headers=self.asana_headers,
                    params={'project': self.asana_project_gid, 'name': summary},
                    timeout=10
                )
                existing.raise_for_status()
                if existing.json().get('data', []):
                    logging.info(f"Task {issue_key} already exists in Asana, skipping")
                    continue

                # Create Asana task with Jira context
                self._rate_limit('asana')
                asana_payload = {
                    'data': {
                        'name': f'[{issue_key}] {summary}',
                        'notes': f'Jira Issue: {self.jira_url}/browse/{issue_key}\n\n{description}',
                        'assignee': assignee if assignee != 'Unassigned' else None,
                        'projects': [self.asana_project_gid]
                    }
                }
                create_resp = requests.post(
                    f'{self.asana_url}/tasks',
                    headers=self.asana_headers,
                    json=asana_payload,
                    timeout=10
                )
                create_resp.raise_for_status()
                logging.info(f"Created Asana task for Jira issue {issue_key}")

        except requests.exceptions.RequestException as e:
            logging.error(f"API error during Jira to Asana sync: {e}")
        except Exception as e:
            logging.error(f"Unexpected error during Jira to Asana sync: {e}")

if __name__ == '__main__':
    # Run sync every 15 minutes (use cron or systemd timer in production)
    sync = JiraAsanaSync()
    logging.info("Starting Jira to Asana sync")
    sync.sync_jira_to_asana()
    logging.info("Sync completed")

import requests
import json
import os
import logging
from datetime import datetime, timedelta
import time

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[logging.FileHandler('jira_ai_sprint_planner.log'), logging.StreamHandler()]
)

class JiraAISprintPlanner:
    def __init__(self):
        self.jira_url = os.getenv('JIRA_URL', 'https://your-org.atlassian.net')
        self.jira_api_key = os.getenv('JIRA_API_KEY')
        self.jira_project_key = os.getenv('JIRA_PROJECT_KEY', 'ENG')
        self.board_id = os.getenv('JIRA_BOARD_ID')  # Get from Jira board settings

        if not all([self.jira_api_key, self.board_id]):
            raise ValueError("Missing JIRA_API_KEY or JIRA_BOARD_ID")

        self.headers = {
            'Authorization': f'Bearer {self.jira_api_key}',
            'Accept': 'application/json',
            'Content-Type': 'application/json'
        }
        # Jira 6.0 AI API rate limit: 200 req/hour for predictive endpoints
        self.last_ai_request = datetime.now()
        self.ai_rate_limit = 200  # req per hour

    def _enforce_ai_rate_limit(self):
        """Enforce Jira 6.0 AI API rate limits"""
        elapsed = (datetime.now() - self.last_ai_request).total_seconds()
        min_interval = 3600 / self.ai_rate_limit  # seconds between requests
        if elapsed < min_interval:
            sleep_time = min_interval - elapsed
            logging.info(f"Rate limiting: sleeping {sleep_time:.2f} seconds")
            time.sleep(sleep_time)
        self.last_ai_request = datetime.now()

    def get_backlog_issues(self, max_results=50):
        """Fetch prioritized backlog issues from Jira"""
        try:
            jql = f'project={self.jira_project_key} AND status = Backlog ORDER BY priority DESC'
            response = requests.get(
                f'{self.jira_url}/rest/api/3/search',
                headers=self.headers,
                params={'jql': jql, 'maxResults': max_results},
                timeout=10
            )
            response.raise_for_status()
            issues = response.json().get('issues', [])
            logging.info(f"Fetched {len(issues)} backlog issues for sprint planning")
            return issues
        except requests.exceptions.RequestException as e:
            logging.error(f"Failed to fetch backlog issues: {e}")
            return []

    def generate_ai_sprint_plan(self, sprint_duration_weeks=2, team_capacity_story_points=40):
        """
        Use Jira 6.0 AI Sprint Planner to generate optimal sprint plan
        Args:
            sprint_duration_weeks: Length of sprint in weeks
            team_capacity_story_points: Total team capacity for sprint
        """
        self._enforce_ai_rate_limit()
        try:
            # Prepare payload for Jira 6.0 AI Sprint Planner API
            backlog_issues = self.get_backlog_issues()
            if not backlog_issues:
                logging.warning("No backlog issues to plan")
                return None

            issue_keys = [issue['key'] for issue in backlog_issues]
            payload = {
                'boardId': int(self.board_id),
                'sprintDurationWeeks': sprint_duration_weeks,
                'teamCapacityStoryPoints': team_capacity_story_points,
                'issueKeys': issue_keys,
                'optimizationGoal': 'VELOCITY',  # Options: VELOCITY, RISK_REDUCTION, DEADLINE_COMPLIANCE
                'includeAiRationale': True  # Get explanation for choices
            }

            # Call Jira 6.0 AI Sprint Planner endpoint
            response = requests.post(
                f'{self.jira_url}/rest/api/3/ai/sprint-planner/generate',
                headers=self.headers,
                json=payload,
                timeout=30  # AI endpoints take longer
            )
            response.raise_for_status()
            sprint_plan = response.json()
            logging.info(f"Generated AI sprint plan with {len(sprint_plan.get('plannedIssues', []))} issues")

            # Save plan to file for audit
            with open(f'sprint_plan_{datetime.now().strftime("%Y%m%d_%H%M%S")}.json', 'w') as f:
                json.dump(sprint_plan, f, indent=2)

            return sprint_plan
        except requests.exceptions.RequestException as e:
            logging.error(f"AI Sprint Planner API error: {e}")
            if hasattr(e, 'response') and e.response is not None:
                logging.error(f"API response: {e.response.text}")
            return None
        except Exception as e:
            logging.error(f"Unexpected error generating sprint plan: {e}")
            return None

    def create_sprint_from_ai_plan(self, sprint_plan, sprint_name=None):
        """Create a Jira sprint from the AI-generated plan"""
        if not sprint_plan:
            return None
        try:
            sprint_name = sprint_name or f"Sprint {datetime.now().strftime('%Y-%m-%d')}"
            # Create sprint via Jira API
            self._enforce_ai_rate_limit()
            sprint_payload = {
                'name': sprint_name,
                'boardId': int(self.board_id),
                'startDate': datetime.now().isoformat(),
                'endDate': (datetime.now() + timedelta(weeks=2)).isoformat()
            }
            create_resp = requests.post(
                f'{self.jira_url}/rest/api/3/sprint',
                headers=self.headers,
                json=sprint_payload,
                timeout=10
            )
            create_resp.raise_for_status()
            sprint_id = create_resp.json()['id']
            logging.info(f"Created sprint {sprint_name} with ID {sprint_id}")

            # Add planned issues to sprint
            planned_issues = [issue['key'] for issue in sprint_plan.get('plannedIssues', [])]
            if planned_issues:
                self._enforce_ai_rate_limit()
                move_resp = requests.post(
                    f'{self.jira_url}/rest/api/3/sprint/{sprint_id}/issue',
                    headers=self.headers,
                    json={'issues': planned_issues},
                    timeout=10
                )
                move_resp.raise_for_status()
                logging.info(f"Added {len(planned_issues)} issues to sprint {sprint_id}")

            return sprint_id
        except requests.exceptions.RequestException as e:
            logging.error(f"Failed to create sprint: {e}")
            return None

if __name__ == '__main__':
    planner = JiraAISprintPlanner()
    logging.info("Starting AI sprint planning")
    # Assume 10-person team has 40 story points capacity per 2-week sprint
    sprint_plan = planner.generate_ai_sprint_plan(team_capacity_story_points=40)
    if sprint_plan:
        planner.create_sprint_from_ai_plan(sprint_plan)
        logging.info(f"AI Sprint Plan Rationale: {sprint_plan.get('rationale', 'No rationale provided')}")
    logging.info("Sprint planning completed")

import requests
import json
import os
import logging
from datetime import datetime, timedelta
import time

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[logging.FileHandler('asana_ai_workload.log'), logging.StreamHandler()]
)

class AsanaAIWorkloadConnector:
    def __init__(self):
        self.asana_url = 'https://app.asana.com/api/1.0'
        self.asana_api_key = os.getenv('ASANA_API_KEY')
        self.asana_project_gid = os.getenv('ASANA_PROJECT_GID')
        self.github_api_key = os.getenv('GITHUB_API_KEY')  # To pull commit data for workload
        self.team_gid = os.getenv('ASANA_TEAM_GID')  # Asana team GID

        if not all([self.asana_api_key, self.asana_project_gid, self.team_gid]):
            raise ValueError("Missing ASANA_API_KEY, ASANA_PROJECT_GID, or ASANA_TEAM_GID")

        self.asana_headers = {
            'Authorization': f'Bearer {self.asana_api_key}',
            'Accept': 'application/json',
            'Content-Type': 'application/json'
        }
        # Asana 5.0 AI API rate limit: 100 req/hour for workload endpoints
        self.last_ai_request = datetime.now()
        self.ai_rate_limit = 100

    def _enforce_ai_rate_limit(self):
        """Enforce Asana 5.0 AI API rate limits"""
        elapsed = (datetime.now() - self.last_ai_request).total_seconds()
        min_interval = 3600 / self.ai_rate_limit
        if elapsed < min_interval:
            sleep_time = min_interval - elapsed
            logging.info(f"Asana AI rate limiting: sleeping {sleep_time:.2f} seconds")
            time.sleep(sleep_time)
        self.last_ai_request = datetime.now()

    def get_team_members(self):
        """Fetch team members from Asana"""
        try:
            response = requests.get(
                f'{self.asana_url}/teams/{self.team_gid}/users',
                headers=self.asana_headers,
                timeout=10
            )
            response.raise_for_status()
            members = response.json().get('data', [])
            logging.info(f"Fetched {len(members)} team members")
            return members
        except requests.exceptions.RequestException as e:
            logging.error(f"Failed to fetch team members: {e}")
            return []

    def get_github_commit_data(self, username, days=7):
        """Fetch GitHub commit count for a user over the last N days (to augment workload AI)"""
        if not self.github_api_key or not username:
            return 0
        try:
            headers = {'Authorization': f'token {self.github_api_key}'}
            since = (datetime.now() - timedelta(days=days)).isoformat()
            response = requests.get(
                f'https://api.github.com/search/commits?q=author:{username}+committer-date:>{since}',
                headers=headers,
                timeout=10
            )
            response.raise_for_status()
            return response.json().get('total_count', 0)
        except Exception as e:
            logging.error(f"Failed to fetch GitHub data for {username}: {e}")
            return 0

    def get_ai_workload_predictions(self):
        """Call Asana 5.0 AI Workload Prediction API with custom signals"""
        self._enforce_ai_rate_limit()
        try:
            team_members = self.get_team_members()
            if not team_members:
                return None

            # Prepare custom signals for each team member (GitHub commits, PTO, etc.)
            member_signals = []
            for member in team_members:
                member_gid = member['gid']
                member_name = member['name']
                # Extract GitHub username from Asana user name (assumes format "First Last (github_user)")
                github_user = None
                if '(' in member_name and ')' in member_name:
                    github_user = member_name.split('(')[1].split(')')[0]
                commit_count = self.get_github_commit_data(github_user, days=7) if github_user else 0
                member_signals.append({
                    'memberGid': member_gid,
                    'customSignals': {
                        'githubCommits7d': commit_count,
                        'ptoDays': 0  # Pull from HR API in production
                    }
                })

            # Call Asana 5.0 AI Workload API
            payload = {
                'teamGid': self.team_gid,
                'projectGid': self.asana_project_gid,
                'timeframeWeeks': 2,
                'memberSignals': member_signals,
                'predictBurnout': True,
                'predictCapacity': True
            }

            response = requests.post(
                f'{self.asana_url}/ai/workload-predictions',
                headers=self.asana_headers,
                json=payload,
                timeout=30
            )
            response.raise_for_status()
            predictions = response.json().get('data', {})
            logging.info(f"Generated workload predictions for {len(member_signals)} team members")

            # Save predictions for audit
            with open(f'workload_pred_{datetime.now().strftime("%Y%m%d_%H%M%S")}.json', 'w') as f:
                json.dump(predictions, f, indent=2)

            return predictions
        except requests.exceptions.RequestException as e:
            logging.error(f"Asana AI Workload API error: {e}")
            if hasattr(e, 'response') and e.response is not None:
                logging.error(f"API response: {e.response.text}")
            return None
        except Exception as e:
            logging.error(f"Unexpected error getting workload predictions: {e}")
            return None

    def rebalance_tasks(self, predictions):
        """Auto-reassign tasks based on AI workload predictions (low confidence threshold: 0.7)"""
        if not predictions:
            return
        try:
            overloaded = [m for m in predictions.get('members', []) if m.get('burnoutRisk', 0) > 0.7]
            underloaded = [m for m in predictions.get('members', []) if m.get('capacityRemaining', 0) > 20]
            logging.info(f"Found {len(overloaded)} overloaded, {len(underloaded)} underloaded members")

            for member in overloaded:
                member_gid = member['memberGid']
                # Fetch tasks assigned to overloaded member
                self._enforce_ai_rate_limit()
                tasks_resp = requests.get(
                    f'{self.asana_url}/tasks',
                    headers=self.asana_headers,
                    params={'assignee': member_gid, 'project': self.asana_project_gid, 'completed': False},
                    timeout=10
                )
                tasks_resp.raise_for_status()
                tasks = tasks_resp.json().get('data', [])
                # Reassign 30% of tasks to underloaded members
                reassign_count = int(len(tasks) * 0.3)
                for i in range(min(reassign_count, len(underloaded))):
                    target_member = underloaded[i]
                    task_gid = tasks[i]['gid']
                    self._enforce_ai_rate_limit()
                    update_resp = requests.put(
                        f'{self.asana_url}/tasks/{task_gid}',
                        headers=self.asana_headers,
                        json={'data': {'assignee': target_member['memberGid']}},
                        timeout=10
                    )
                    update_resp.raise_for_status()
                    logging.info(f"Reassigned task {task_gid} from {member_gid} to {target_member['memberGid']}")
        except Exception as e:
            logging.error(f"Failed to rebalance tasks: {e}")

if __name__ == '__main__':
    connector = AsanaAIWorkloadConnector()
    logging.info("Starting Asana AI workload prediction")
    predictions = connector.get_ai_workload_predictions()
    if predictions:
        connector.rebalance_tasks(predictions)
        # Log high-risk members
        high_risk = [m for m in predictions.get('members', []) if m.get('burnoutRisk', 0) > 0.7]
        for m in high_risk:
            logging.warning(f"High burnout risk for {m['name']}: {m['burnoutRisk']*100:.1f}%")
    logging.info("Workload prediction completed")

# Custom Jira 6.0 AI Backlog Groomer Prompt
custom_prompt = {
    "promptContext": "10-person engineering team building B2B SaaS",
    "priorityRules": [
        "P0: Blocks release, customer-facing outage",
        "P1: Required for next sprint release",
        "P2: Nice-to-have, no release impact",
        "P3: Backlog candidate, revisit quarterly"
    ],
    "optimizationGoal": "TIME_TO_MARKET",
    "excludeLabels": ["documentation", "internal-tooling"]
}

# Asana 5.0 Custom Capacity Mapping
role_capacities = {
    "backend": 24,  # hours/week
    "frontend": 26,
    "qa": 30,
    "em": 12
}

def map_asana_capacity(member_gid, role):
    """Update Asana 5.0 workload capacity for a team member"""
    url = f"https://app.asana.com/api/1.0/users/{member_gid}/workload-settings"
    payload = {
        "data": {
            "weeklyCapacityHours": role_capacities.get(role, 32),
            "meetingBufferHours": 10,  # Fixed meeting time per week
            "onCallBufferHours": 4  # Average on-call hours per week
        }
    }
    response = requests.put(url, headers=asana_headers, json=payload)
    return response.json()

# AI Output Audit Script
def audit_sprint_plan(ai_plan, actual_sprint):
    """Compare AI-generated sprint plan to actual sprint outcomes"""
    ai_issues = set([i['key'] for i in ai_plan.get('plannedIssues', [])])
    actual_issues = set([i['key'] for i in actual_sprint.get('issues', [])])
    missing = ai_issues - actual_issues
    extra = actual_issues - ai_issues
    accuracy = len(ai_issues & actual_issues) / len(ai_issues) if ai_issues else 0
    return {
        "accuracy": accuracy * 100,
        "missing_issues": missing,
        "extra_issues": extra
    }