team_updates_clickup.py

#!/usr/bin/env python3
"""
Script to fetch ClickUp tasks, list teams, or list spaces within a team.

Usage Examples:

1. List all available Teams/Workspaces:
   python team_updates_clickup.py --list-teams

2. List all Spaces within a specific Team/Workspace (replace YOUR_TEAM_ID):
   python team_updates_clickup.py --list-spaces-in-team YOUR_TEAM_ID

3. Generate a task update report for an entire Workspace for the last 24 hours:
   python team_updates_clickup.py --team-id YOUR_TEAM_ID --hours_duration 24

4. Generate a task update report for specific Space IDs within a Workspace for a date range:
   python team_updates_clickup.py --team-id YOUR_TEAM_ID --space-ids "SPACE_ID_1,SPACE_ID_2" --date-range "YYYY-MM-DD to YYYY-MM-DD"

5. Generate a report with a specific output path and debug logging:
   python team_updates_clickup.py --team-id YOUR_TEAM_ID --output-path ./reports --debug --hours_duration 48

Prerequisites:
- CLICKUP_PERSONAL_TOKEN environment variable must be set.
- Required Python packages (see requirements.txt, typically aiohttp, pytz, python-dotenv).
"""
import os
import json
import datetime
import asyncio
import aiohttp
import argparse
from datetime import datetime, timedelta, timezone
import pytz
import sys
import logging
import csv
import os.path
from dotenv import load_dotenv

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# Load .env file
load_dotenv()

# Configuration
CLICKUP_TOKEN = os.environ.get('CLICKUP_PERSONAL_TOKEN')
if not CLICKUP_TOKEN:
    logger.error("ClickUp token not found. Please set the CLICKUP_PERSONAL_TOKEN environment variable.")
    sys.exit(1)

CLICKUP_API_BASE_URL = "https://api.clickup.com/api/v2"

# Command line arguments
parser = argparse.ArgumentParser(description='Fetch ClickUp tasks, list teams, or list spaces within a team.')
parser.add_argument('--list-teams', action='store_true', help='List all available Team IDs (Workspaces) and their names.')
parser.add_argument('--list-spaces-in-team', type=str, metavar='TEAM_ID', help='List all Spaces within the specified Team ID (Workspace ID).')
parser.add_argument('--team-id', type=str, help='ClickUp Team ID (Workspace ID) for generating the task report. If not provided and no other action is specified, lists available Team IDs.')
parser.add_argument('--space-ids', type=str, help='Comma-separated list of Space IDs to filter tasks for the report. Requires --team-id to be set for report generation.')
parser.add_argument('--date-range', type=str, help='Date range for task report, e.g., "YYYY-MM-DD to YYYY-MM-DD". Overrides --hours_duration.')
parser.add_argument('--hours_duration', type=int, help='Hours to look back for task report. Used if --date-range is not set.')
parser.add_argument('--output-path', type=str, help='Directory to save the output CSV file. Defaults to ./output.')
parser.add_argument('--debug', action='store_true', help='Enable debug logging')
args = parser.parse_args()

HEADERS = {
    "Authorization": CLICKUP_TOKEN,
    "Content-Type": "application/json"
}

class RateLimitError(Exception):
    """Custom exception for rate limit errors"""
    pass

def clean_csv_string(text):
    """Cleans a string for CSV output by removing newlines and escaping quotes."""
    if text is None:
        return ""
    cleaned_text = str(text).replace('\r\n', ' ').replace('\n', ' ').replace('\r', ' ')
    return cleaned_text.replace('"', '""')

def get_pst_time():
    """Get current time in PST/PDT"""
    ptz_timezone = pytz.timezone('America/Los_Angeles')
    return datetime.now(ptz_timezone)

def get_default_time_range():
    """Get default time range: 6am PT yesterday to 6am PT today"""
    now = get_pst_time()
    today_6am = now.replace(hour=6, minute=0, second=0, microsecond=0)
    
    if now.hour < 6: # If current time is before 6 AM PT
        yesterday_6am = today_6am - timedelta(days=1)
        day_before_6am = today_6am - timedelta(days=2) # e.g. Monday 3AM, range is Sat 6AM to Sun 6AM
        return day_before_6am, yesterday_6am
    else: # Current time is 6 AM PT or later
        yesterday_6am = today_6am - timedelta(days=1) # e.g. Monday 10AM, range is Sun 6AM to Mon 6AM
        return yesterday_6am, today_6am

def parse_date_range(date_range_str):
    """Parse date range string into start and end datetime objects in UTC"""
    try:
        if date_range_str:
            logger.info(f"Using provided --date-range: {date_range_str}")
            parts = date_range_str.split(" to ")
            if len(parts) != 2:
                raise ValueError("Invalid date range format. Expected 'YYYY-MM-DD HH:MM to YYYY-MM-DD HH:MM' or 'YYYY-MM-DD to YYYY-MM-DD'")
            
            start_str, end_str = parts
            try:
                start_time_pt = datetime.strptime(start_str.strip(), "%Y-%m-%d %H:%M")
                end_time_pt = datetime.strptime(end_str.strip(), "%Y-%m-%d %H:%M")
            except ValueError: # Try parsing as YYYY-MM-DD
                start_time_pt = datetime.strptime(start_str.strip(), "%Y-%m-%d")
                # For end_time, if only date is given, set to end of that day
                end_time_pt = datetime.strptime(end_str.strip(), "%Y-%m-%d").replace(hour=23, minute=59, second=59)

            ptz_timezone = pytz.timezone('America/Los_Angeles')
            start_time_pt = ptz_timezone.localize(start_time_pt)
            end_time_pt = ptz_timezone.localize(end_time_pt)

        elif args.hours_duration is not None:
            logger.info(f"Using --hours_duration: {args.hours_duration} hours back from current PT time.")
            end_time_pt = get_pst_time()
            start_time_pt = end_time_pt - timedelta(hours=args.hours_duration)
        else:
            logger.info("No --date-range or --hours_duration provided, using default time range (yesterday 6am PT to today 6am PT)")
            start_time_pt, end_time_pt = get_default_time_range()

        logger.info(f"Using PT date range: {start_time_pt.strftime('%Y-%m-%d %H:%M')} to {end_time_pt.strftime('%Y-%m-%d %H:%M')}")
        # Convert to UTC for API
        start_time_utc = start_time_pt.astimezone(timezone.utc)
        end_time_utc = end_time_pt.astimezone(timezone.utc)
        logger.info(f"UTC date range for API: {start_time_utc.isoformat()} to {end_time_utc.isoformat()}")
        return start_time_utc, end_time_utc, end_time_pt # Return end_time_pt for filename
    except Exception as e:
        logger.error(f"Error parsing date range: {e}. Using default.")
        start_time_pt_def, end_time_pt_def = get_default_time_range()
        return start_time_pt_def.astimezone(timezone.utc), end_time_pt_def.astimezone(timezone.utc), end_time_pt_def

async def make_request(session, url, method="GET", params=None, retry_count=0, max_retries=5, initial_backoff=1, max_backoff=60):
    """Make an HTTP request with exponential backoff."""
    logger.debug(f"Making {method} request to {url} with params {params}")
    try:
        async with session.request(method, url, headers=HEADERS, params=params) as resp:
            if resp.status == 429: # ClickUp uses 429 for rate limits
                backoff = min(initial_backoff * (2 ** retry_count), max_backoff)
                retry_after = resp.headers.get("Retry-After", backoff) # ClickUp might send Retry-After
                try:
                    wait_time = int(retry_after)
                except ValueError:
                    wait_time = backoff
                
                logger.warning(f"Rate limit hit (Status 429). Waiting {wait_time:.2f} seconds before retry {retry_count + 1}/{max_retries}")
                await asyncio.sleep(wait_time)
                if retry_count < max_retries:
                    return await make_request(session, url, method, params, retry_count + 1)
                else:
                    raise RateLimitError("Max retries exceeded for rate limit")
            
            resp.raise_for_status() 
            return await resp.json()
    except aiohttp.ClientResponseError as e: # Catch specific client errors for better logging
        logger.error(f"HTTP error: {e.status} {e.message} for URL {url}")
        # Log response body if available and helpful for debugging
        try:
            error_body = await e.text()
            logger.debug(f"Error response body: {error_body[:500]}") # Log first 500 chars
        except Exception:
            pass # Ignore if can't read body
        
        if retry_count < max_retries and e.status not in [401, 403, 404]: # Don't retry auth/permission issues
            backoff = min(initial_backoff * (2 ** retry_count), max_backoff)
            await asyncio.sleep(backoff)
            return await make_request(session, url, method, params, retry_count + 1)
        raise
    except aiohttp.ClientError as e: # Catch other client errors (network, etc.)
        logger.error(f"Network or client error: {e} for URL {url}")
        if retry_count < max_retries:
            backoff = min(initial_backoff * (2 ** retry_count), max_backoff)
            await asyncio.sleep(backoff)
            return await make_request(session, url, method, params, retry_count + 1)
        raise

async def fetch_and_print_available_teams(session):
    """Fetches and prints available ClickUp teams/workspaces."""
    logger.info("Fetching available Team IDs...")
    url = f"{CLICKUP_API_BASE_URL}/team"
    try:
        response_data = await make_request(session, url)
        teams = response_data.get("teams")
        if teams:
            logger.info("Available Teams/Workspaces:")
            for team in teams:
                print(f"  Name: {team.get('name')}, ID: {team.get('id')}")
            logger.info("\nPlease re-run the script with the desired --team-id.")
        else:
            logger.warning("No teams found or unexpected API response format when fetching teams.")
            logger.debug(f"Full response for /team: {response_data}")
    except RateLimitError as e:
        logger.error(f"Rate limit error while fetching teams: {e}")
    except Exception as e:
        logger.error(f"Error fetching or parsing teams: {e}", exc_info=args.debug)

async def fetch_and_print_spaces(session, team_id_for_spaces):
    """Fetches and prints available Spaces within a given ClickUp Team/Workspace."""
    logger.info(f"Fetching Spaces for Team ID: {team_id_for_spaces}...")
    url = f"{CLICKUP_API_BASE_URL}/team/{team_id_for_spaces}/space"
    params = {"archived": "false"} # Get non-archived spaces
    try:
        response_data = await make_request(session, url, params=params)
        spaces = response_data.get("spaces")
        if spaces:
            logger.info(f"Available Spaces in Team ID {team_id_for_spaces}:")
            for space in spaces:
                print(f"  Name: {space.get('name')}, ID: {space.get('id')}")
        elif spaces == []: # Explicitly check for an empty list vs. None
            logger.info(f"No Spaces found in Team ID {team_id_for_spaces}.")
        else:
            logger.warning(f"Could not retrieve spaces or unexpected API response format for Team ID {team_id_for_spaces}.")
            logger.debug(f"Full response for /team/{team_id_for_spaces}/space: {response_data}")
    except RateLimitError as e:
        logger.error(f"Rate limit error while fetching spaces for Team ID {team_id_for_spaces}: {e}")
    except Exception as e:
        logger.error(f"Error fetching or parsing spaces for Team ID {team_id_for_spaces}: {e}", exc_info=args.debug)

async def generate_markdown_report(fetched_tasks, output_dir, filename_prefix, end_time_pt_for_filename, duration_hours):
    """Generates a Markdown report summarizing team activity."""
    logger.info("Generating Markdown team report...")
    md_report_path = os.path.join(output_dir, f"{filename_prefix}team_report.md")

    report_type_str = ""
    if duration_hours == 24:
        report_type_str = "Daily"
    elif duration_hours == (24 * 7):
        report_type_str = "Weekly"
    else:
        report_type_str = f"{duration_hours} Hour"
    
    end_date_title_str = end_time_pt_for_filename.strftime('%Y-%m-%d')
    md_report_title = f"# {end_date_title_str} {report_type_str} ClickUp Team Update\n\n"

    user_activities = {}
    user_emails_map = {} # To store display_name -> email mapping

    for task in fetched_tasks:
        assignees_list = task.get('assignees', [])
        if not assignees_list: 
            continue

        task_title = task.get('name', 'No Title')
        task_link = task.get('url', '#')
        task_custom_id = task.get('custom_id')
        task_status_obj = task.get('status', {})
        task_status_type = task_status_obj.get('type', 'unknown').lower()

        item_details_md = f"{task_title} " 
        if task_custom_id: 
            item_details_md += f"[{task_custom_id}]({task_link})"
        else: 
            item_details_md += f"[Link]({task_link})"

        for assignee in assignees_list:
            # Determine the display name for the user
            assignee_display_name = assignee.get('username')
            if not assignee_display_name:
                 assignee_display_name = assignee.get('email', f"user_id_{assignee.get('id', 'unknown')}")
            
            # Store email if available and not already stored for this display name
            assignee_email = assignee.get('email')
            if assignee_email and assignee_display_name not in user_emails_map:
                user_emails_map[assignee_display_name] = assignee_email
            
            if assignee_display_name not in user_activities:
                user_activities[assignee_display_name] = {'done': [], 'wip': []}

            if task_status_type == 'closed':
                user_activities[assignee_display_name]['done'].append(item_details_md)
            else: 
                user_activities[assignee_display_name]['wip'].append(item_details_md)

    # Build Markdown string
    md_content_parts = [md_report_title]
    for user_display_name in sorted(user_activities.keys()):
        user_email = user_emails_map.get(user_display_name)
        if user_email:
            md_content_parts.append(f"### @[{user_display_name}](mailto:{user_email})\n\n")
        else:
            md_content_parts.append(f"### @{user_display_name}\n\n")
        
        if user_activities[user_display_name]['done']:
            md_content_parts.append("done:\n")
            for item_md in user_activities[user_display_name]['done']:
                md_content_parts.append(f"- {item_md}\n")
            md_content_parts.append("\n")
        
        if user_activities[user_display_name]['wip']:
            md_content_parts.append("wip:\n")
            for item_md in user_activities[user_display_name]['wip']:
                md_content_parts.append(f"- {item_md}\n")
            md_content_parts.append("\n")
        
        if not user_activities[user_display_name]['done'] and not user_activities[user_display_name]['wip']:
            # Check against the actual last appended header before popping
            expected_header_no_email = f"### @{user_display_name}\n\n"
            expected_header_with_email = f"### @[{user_display_name}](mailto:{user_email})\n\n" if user_email else ""
            if md_content_parts[-1] == expected_header_no_email or (user_email and md_content_parts[-1] == expected_header_with_email):
                 md_content_parts.pop()

    try:
        with open(md_report_path, "w", encoding='utf-8') as f_md:
            f_md.write("".join(md_content_parts))
        logger.info(f"Markdown team report saved to {os.path.abspath(md_report_path)}")
    except IOError as e:
        logger.error(f"Failed to write Markdown report file: {e}")

async def fetch_all_updated_tasks(session, team_id, start_time_ms, end_time_ms, space_ids_list=None):
    """Fetch all tasks (including subtasks) updated within the time range for a team, optionally filtered by space_ids."""
    all_tasks = []
    page = 0 # ClickUp API uses 0-indexed pages for this endpoint it seems
    
    while True:
        # https://clickup.com/api/clickup-api-docs/reference/get-tasks
        url = f"{CLICKUP_API_BASE_URL}/team/{team_id}/task"
        params = {
            "subtasks": "true",
            "include_closed": "true", # To get tasks that were closed/done in the period
            "date_updated_gt": int(start_time_ms),
            "date_updated_lt": int(end_time_ms),
            "order_by": "updated", # Order by updated date
            "reverse": "true",     # Get most recently updated first
            "page": page
            # "space_ids[]", "project_ids[]", "list_ids[]" can also be used for more specific targeting
        }
        if space_ids_list:
            params["space_ids[]"] = space_ids_list
            logger.debug(f"Fetching tasks page {page} for team {team_id} (spaces: {space_ids_list}) with params: {params}")
        else:
            logger.debug(f"Fetching tasks page {page} for team {team_id} with params: {params}")
        
        try:
            response_data = await make_request(session, url, params=params)
            tasks_page = response_data.get("tasks", [])
            
            if not tasks_page:
                logger.info(f"No more tasks found on page {page} for team {team_id}.")
                break
            
            all_tasks.extend(tasks_page)
            logger.info(f"Fetched {len(tasks_page)} tasks from page {page}. Total fetched so far: {len(all_tasks)}")
            
            # ClickUp's 'Get Tasks' endpoint response includes a 'last_page': boolean field.
            # If it's true, no need to fetch next page.
            # However, the public documentation for /team/{team_id}/task does not explicitly state 'last_page'.
            # It's safer to rely on an empty 'tasks' array to stop.
            # If response_data.get("last_page") is True:
            #    logger.info("Last page of tasks reached.")
            #    break

            page += 1
            # Safety break for very large number of pages, adjust if necessary
            if page > 100: # Assuming 100 tasks per page, this is 10,000 tasks.
                logger.warning("Reached maximum page limit (100) for fetching tasks. Some tasks might be missed.")
                break
                
        except RateLimitError as e:
            logger.error(f"Rate limit error while fetching tasks: {e}")
            raise # Propagate to stop further processing
        except Exception as e:
            logger.error(f"Error fetching tasks page {page}: {e}")
            break # Stop if there's an error on a page
            
    logger.info(f"Total tasks fetched for team {team_id} in range: {len(all_tasks)}")
    return all_tasks

async def main():
    if args.debug:
        logger.setLevel(logging.DEBUG)
        logger.debug("Debug logging enabled")

    # Action handling
    async with aiohttp.ClientSession() as session:
        if args.list_teams:
            await fetch_and_print_available_teams(session)
            sys.exit(0)

        if args.list_spaces_in_team:
            await fetch_and_print_spaces(session, args.list_spaces_in_team)
            sys.exit(0)

        # If neither list_teams nor list_spaces_in_team is used, proceed to report generation or default team listing.
        if not args.team_id:
            logger.info("No --team-id provided for report generation, and no other action specified (e.g., --list-teams or --list-spaces-in-team). Listing available teams:")
            await fetch_and_print_available_teams(session)
            logger.info("\nTo generate a report, provide --team-id and optionally --space-ids, along with date range/duration.")
            logger.info("To list spaces within a team, use: --list-spaces-in-team <TEAM_ID>")
            sys.exit(0)

        # Proceed with task report generation
        team_id_for_report = args.team_id

        space_ids_for_report = []
        if args.space_ids:
            if not args.team_id:
                logger.error("--space-ids argument requires --team-id to be specified for report generation.")
                sys.exit(1)
            space_ids_for_report = [s_id.strip() for s_id in args.space_ids.split(',') if s_id.strip()]
            if not space_ids_for_report:
                logger.warning("--space-ids was provided but contained no valid IDs after parsing. Report will include all spaces in the team.")
            else:
                logger.info(f"Report will be filtered for Space IDs: {space_ids_for_report} within Team ID: {team_id_for_report}")

        logger.info(f"Proceeding with task report generation for Team ID: {team_id_for_report}" + (f" (filtered by Space IDs: {space_ids_for_report})" if space_ids_for_report else ""))

        start_time_utc, end_time_utc, end_time_pt_for_filename = parse_date_range(args.date_range)
        start_time_ms = int(start_time_utc.timestamp() * 1000)
        end_time_ms = int(end_time_utc.timestamp() * 1000)

        logger.info(f"Fetching tasks for Team ID: {team_id_for_report}")
        logger.info(f"Time range (ms): {start_time_ms} to {end_time_ms}")

        output_dir = args.output_path if args.output_path else "./output"
        if args.output_path or not args.output_path: # Ensure directory exists if we are generating a report
            os.makedirs(output_dir, exist_ok=True)
            logger.info(f"Output directory: {os.path.abspath(output_dir)}")
        
        duration_seconds = (end_time_utc - start_time_utc).total_seconds()
        duration_hours = round(duration_seconds / 3600)
        end_date_str_for_prefix = end_time_pt_for_filename.strftime('%Y-%m-%d-%H-%M-%S')
        
        prefix_type_str = ""
        if duration_hours == 24:
            prefix_type_str = "Daily"
        elif duration_hours == (24 * 7):
            prefix_type_str = "Weekly"
        else:
            prefix_type_str = f"{duration_hours}h"
        filename_prefix = f"{prefix_type_str}-ClickUp-{end_date_str_for_prefix}-"
        logger.info(f"Using filename prefix: {filename_prefix}")
        
        processed_tasks_for_csv = []
        raw_tasks_data_for_json = [] # For storing raw task data

        try:
            # Fetch tasks first
            fetched_tasks = await fetch_all_updated_tasks(session, team_id_for_report, start_time_ms, end_time_ms, space_ids_list=space_ids_for_report)
            
            if fetched_tasks:
                # Store raw data for JSON output
                raw_tasks_data_for_json.extend(fetched_tasks)

                # --- Output Raw Task Data (JSON) ---
                if raw_tasks_data_for_json:
                    raw_json_path = os.path.join(output_dir, f"{filename_prefix}tasks_raw_data.json")
                    try:
                        with open(raw_json_path, "w", encoding='utf-8') as f_json:
                            json.dump(raw_tasks_data_for_json, f_json, indent=2)
                        logger.info(f"Raw task data saved to {os.path.abspath(raw_json_path)}")
                    except IOError as e:
                        logger.error(f"Failed to write raw task JSON file: {e}")
                # --- End Output Raw Task Data (JSON) ---

                logger.info(f"Processing {len(fetched_tasks)} tasks for CSV output...")
                for task in fetched_tasks:
                    title = clean_csv_string(task.get('name', 'N/A'))
                    link = task.get('url', 'N/A')
                    custom_id_val = task.get('custom_id', 'N/A') # Get custom ID
                    if custom_id_val is None: # API might return null for no custom ID
                        custom_id_val = 'N/A'
                    
                    assignees_list = task.get('assignees', [])
                    assignee_names = []
                    if assignees_list:
                        for assignee in assignees_list:
                            # Prefer username, fallback to email if username is missing
                            if assignee.get('username'):
                                assignee_names.append(assignee['username'])
                            elif assignee.get('email'):
                                 assignee_names.append(assignee['email'])
                    assignees_str = ", ".join(assignee_names) if assignee_names else "Unassigned"
                    
                    date_updated_ms_str = task.get('date_updated')
                    last_updated_iso = 'N/A'
                    if date_updated_ms_str:
                        try:
                            date_updated_s = int(date_updated_ms_str) / 1000
                            last_updated_iso = datetime.fromtimestamp(date_updated_s, tz=timezone.utc).isoformat()
                        except ValueError:
                            logger.warning(f"Could not parse date_updated '{date_updated_ms_str}' for task ID {task.get('id')}")
                    
                    processed_tasks_for_csv.append({
                        "title": title,
                        "link": link,
                        "custom_id": custom_id_val, # Add custom_id to dict
                        "assignee": assignees_str,
                        "last_updated": last_updated_iso
                    })
            else:
                logger.info("No tasks found in the specified range to process.")

        except RateLimitError as e:
            logger.error(f"Critical rate limit error during operation: {e}")
            sys.exit(1)
        except Exception as e:
            logger.error(f"An unexpected error occurred during API interaction: {e}", exc_info=args.debug)
            sys.exit(1)

    if processed_tasks_for_csv:
        csv_file_path = os.path.join(output_dir, f"{filename_prefix}tasks_updates.csv")
        csv_headers = ["title", "link", "custom_id", "assignee", "last_updated"]
        
        try:
            with open(csv_file_path, "w", newline='', encoding='utf-8') as f:
                writer = csv.DictWriter(f, fieldnames=csv_headers)
                writer.writeheader()
                writer.writerows(processed_tasks_for_csv)
            logger.info(f"Successfully wrote {len(processed_tasks_for_csv)} tasks to {os.path.abspath(csv_file_path)}")
        except IOError as e:
            logger.error(f"Failed to write CSV file: {e}")
            sys.exit(1)
    else:
        logger.info("No processed tasks to write to CSV.")

    # --- Generate Markdown Report ---
    if fetched_tasks: # Use fetched_tasks which contains the raw task data
        await generate_markdown_report(fetched_tasks, output_dir, filename_prefix, end_time_pt_for_filename, duration_hours)
    else:
        logger.info("No tasks fetched, skipping Markdown report generation.")

if __name__ == "__main__":
    asyncio.run(main())