script.py

"""
Batch2TickTick

This script automates the process of importing hierarchical tasks from a text file
into TickTick via their OAuth 2.0 API. It supports up to 5 levels of task nesting
and handles authentication, project selection, and batch task creation.

File Format (tasks.txt):
    - Plain text: Root level task (LV0)
    - LV1-Title: First level subtask
    - LV2-Title: Second level subtask
    - ... up to LV4

Example:
    Project Planning
    LV1-Research Phase
    LV2-Market Analysis
    LV2-Competitor Research
    LV1-Development Phase
    LV2-Backend Development
    LV3-API Design

Requirements:
    - .env file with TICKTICK_CLIENT_ID and TICKTICK_CLIENT_SECRET
    - tasks.txt file with properly formatted tasks
"""

import requests
from datetime import datetime, timedelta
from requests.auth import HTTPBasicAuth
import webbrowser
import json
import os
import inquirer
import sys
import re
from tqdm import tqdm
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

# ============================================================================
# CONFIGURATION
# ============================================================================

# TickTick OAuth credentials loaded from environment
TICKTICK_CLIENT_ID = os.getenv("TICKTICK_CLIENT_ID")
TICKTICK_CLIENT_SECRET = os.getenv("TICKTICK_CLIENT_SECRET")

# OAuth scope defines API permissions (read and write tasks)
SCOPE = "tasks:write tasks:read"

# OAuth redirect URI (must match TickTick app configuration)
REDIRECT_URI = "http://localhost:8088"

# ============================================================================
# CONSTANTS
# ============================================================================

# Valid next levels for each task level in the hierarchy
# Index represents current level, values are allowed next levels
# Example: After LV0, you can have LV0 or LV1
# After LV2, you can have LV0, LV1, LV2, or LV3
NEIGHBORS = [[0, 1], [0, 1, 2], [0, 1, 2, 3], [0, 1, 2, 3, 4], [0, 1, 2, 3, 4]]


# ============================================================================
# AUTHENTICATION & TOKEN MANAGEMENT
# ============================================================================


def generate_token_info() -> bool:
    """
    Initiates OAuth 2.0 flow to obtain TickTick access token.

    Checks if a valid token already exists before initiating the flow.
    If not, opens browser for user authentication and exchanges the
    authorization code for an access token.

    Returns:
        bool: True if new token was generated, False if valid token exists
    """
    # Check for existing valid token to avoid unnecessary authentication
    token_info = get_stored_token_info()
    if token_info and not token_is_expired(token_info):
        print("\n✅ Access token already obtained.")
        return False

    # Construct OAuth authorization URL
    auth_url = f"https://ticktick.com/oauth/authorize?scope={SCOPE}&client_id={TICKTICK_CLIENT_ID}&redirect_uri={REDIRECT_URI}&response_type=code"
    print(
        f"👉 Please visit the following URL to authenticate with TickTick:\n {auth_url}"
    )

    # Open browser for user authentication
    webbrowser.open(auth_url)

    # User manually enters authorization code from redirect URL
    authorization_code = input("\n👉 Enter the authorization code: ")

    # Exchange authorization code for access token
    token_url = "https://ticktick.com/oauth/token"
    data = {
        "client_id": TICKTICK_CLIENT_ID,
        "client_secret": TICKTICK_CLIENT_SECRET,
        "code": authorization_code,
        "grant_type": "authorization_code",
        "scope": SCOPE,
        "redirect_uri": REDIRECT_URI,
    }

    # Use HTTP Basic Auth as required by TickTick OAuth
    auth = HTTPBasicAuth(TICKTICK_CLIENT_ID, TICKTICK_CLIENT_SECRET)
    response = requests.post(token_url, data=data, auth=auth)

    if response.status_code == 200:
        token_info = response.json()
        print("\n✅ Access token obtained successfully : ")
        store_token_info(token_info)
        return True
    else:
        print(f"\n❌ Error while obtaining access token : {response.text} ")
        return False


def store_token_info(token_info: dict) -> bool:
    """
    Persists OAuth token information to disk with expiration metadata.

    Enriches token data with creation timestamp and calculates absolute
    expiration time for easier validation. Stores as JSON for human readability.

    Args:
        token_info: Dictionary containing OAuth token data from TickTick

    Returns:
        bool: True if storage succeeded, False otherwise
    """
    try:
        # Add creation timestamp for audit trail
        token_info["created_at"] = datetime.now().isoformat()

        # Convert relative expiration (seconds) to absolute timestamp
        if "expires_in" in token_info:
            expires_at = datetime.now() + timedelta(seconds=token_info["expires_in"])
            token_info["expires_at"] = expires_at.isoformat()

        # Persist to JSON file with UTF-8 encoding for international characters
        with open("token_info.json", "w", encoding="utf-8") as file:
            json.dump(token_info, file, indent=4, ensure_ascii=False)
        return True
    except Exception as e:
        print(f"\n❌ Error while storing token info : {e}")
        return False


def token_is_expired(token_info: dict) -> bool:
    """
    Checks if the stored OAuth token has expired.

    Args:
        token_info: Dictionary containing token data with 'expires_at' field

    Returns:
        bool: True if token expired, False if still valid or no expiration set
    """
    if "expires_at" in token_info:
        return datetime.now() > datetime.fromisoformat(token_info["expires_at"])
    else:
        # If no expiration data, assume token is valid (shouldn't happen normally)
        return False


def get_stored_token_info() -> dict:
    """
    Retrieves OAuth token information from persistent storage.

    Returns:
        dict: Token information dictionary, or empty dict if file doesn't exist
              or an error occurred
    """
    if not os.path.exists("token_info.json"):
        return {}

    try:
        with open("token_info.json", "r", encoding="utf-8") as file:
            return json.load(file)
    except Exception as e:
        print(f"\n❌ Error while getting stored token info : {e}")
        return {}


def set_headers(access_token: str) -> dict:
    """
    Constructs HTTP headers for TickTick API requests.

    Args:
        access_token: OAuth access token for authentication

    Returns:
        dict: Headers dictionary with Authorization and Content-Type,
              or empty dict if no token provided
    """
    if access_token:
        return {
            "Authorization": f"Bearer {access_token}",
            "Content-Type": "application/json",
        }
    else:
        return {}


# ============================================================================
# PROJECT MANAGEMENT
# ============================================================================


def getProjects(headers: dict) -> list:
    """
    Fetches all projects from TickTick account.

    Args:
        headers: HTTP headers including authentication token

    Returns:
        list: List of project dictionaries, or empty list on error
    """
    try:
        response = requests.get(
            "https://api.ticktick.com/open/v1/project",
            headers=headers,
        )

        # Raise exception for HTTP errors (4xx, 5xx)
        response.raise_for_status()
        return response.json()
    except Exception as e:
        print(f"\n❌ Error while getting projects : {e} ")
        return []


def selectProject(projects: list) -> tuple:
    """
    Presents interactive project selection menu to user.

    Uses inquirer library for arrow-key navigation. Hides cursor during
    selection for cleaner UI experience.

    Args:
        projects: List of project dictionaries from TickTick

    Returns:
        tuple: (project_name: str, project_id: str) or (None, None) if no projects
    """
    hide_cursor()

    # Create mapping for easy lookup after selection
    choices_map = {project["name"]: project["id"] for project in projects}

    if projects:
        question = inquirer.prompt(
            [
                inquirer.List(
                    "project",
                    message="Select a project",
                    choices=list(choices_map.keys()),
                )
            ]
        )
        show_cursor()
        return question["project"], choices_map[question["project"]]
    else:
        show_cursor()
        return None, None


# ============================================================================
# TASK CREATION
# ============================================================================


def add_tasks_batch(headers: dict, project_id: str = None) -> bool:
    """
    Reads tasks from file and creates them in TickTick with proper hierarchy.

    Maintains parent-child relationships by tracking the most recent task ID
    at each level. When creating a subtask (LV1-LV4), uses the parent ID from
    the previous level.

    Task Format:
        - No prefix: Root level task (LV0)
        - LV1-Title: Child of most recent LV0 task
        - LV2-Title: Child of most recent LV1 task
        - etc.

    Args:
        headers: HTTP headers with authentication
        project_id: Optional TickTick project ID (None = Inbox)

    Returns:
        bool: True if all tasks created successfully, False on error
    """
    try:
        with open("tasks.txt", "r", encoding="utf-8") as file:
            lines = file.readlines()

        # Track most recent task ID at each level for parent-child relationships
        # Example: When creating LV2 task, use parent_ids["LV1"] as parent
        parent_ids = {
            "LV0": None,
            "LV1": None,
            "LV2": None,
            "LV3": None,
            "LV4": None,
        }

        # Initialize sortOrder tracking for each level
        # TickTick sorts by ascending sortOrder (lower = top)
        # We use spacing of 1,000,000 per task starting from 0
        # This allows inserting ~999,999 tasks between any two existing tasks
        SORT_ORDER_SPACING = 1000000

        # Track sortOrder counters for each level independently
        # Each level maintains its own ascending sequence starting at 0
        sort_order_counters = {
            "LV0": 0,
            "LV1": 0,
            "LV2": 0,
            "LV3": 0,
            "LV4": 0,
        }

        hide_cursor()

        # Process each line with progress bar
        for line in tqdm(lines, desc="📝 Adding tasks", unit="task"):
            # Skip empty lines
            if not line.strip():
                continue

            # Root level task (no LV prefix)
            if not line.startswith("LV"):
                current_lv = "LV0"
                current_task = {
                    "projectId": project_id if project_id else None,
                    "title": line.strip(),
                    "sortOrder": sort_order_counters[current_lv],
                }
                # Increment sortOrder for next task at this level
                sort_order_counters[current_lv] += SORT_ORDER_SPACING
            else:
                # Parse hierarchical task (LVn-Title format)
                match = re.match(r"LV(\d+)-", line)
                if match:
                    depth = int(match.group(1))
                    current_lv = f"LV{depth}"
                    # Extract title after "LVn-" prefix
                    title_formatted = line.strip().split("-", 1)[1]
                    # Link to parent at previous level (depth - 1)
                    parent_id = parent_ids[f"LV{depth - 1}"]
                    current_task = {
                        "projectId": project_id if project_id else None,
                        "title": title_formatted,
                        "parentId": parent_id,
                        "sortOrder": sort_order_counters[current_lv],
                    }
                    # Increment sortOrder for next task at this level
                    sort_order_counters[current_lv] += SORT_ORDER_SPACING

            # Create task via TickTick API
            response = requests.post(
                "https://api.ticktick.com/open/v1/task",
                headers=headers,
                json=current_task,
            )
            response.raise_for_status()

            # Store created task ID for potential child tasks
            response_json: dict = response.json()
            parent_ids[current_lv] = response_json.get("id")

        print("\n✅ All tasks have been created successfully!")
        show_cursor()
        return True
    except Exception as e:
        print(f"\n❌ Error while creating tasks : {e} ")
        return False


# ============================================================================
# FILE VALIDATION
# ============================================================================


def verify_txt_file() -> bool:
    """
    Validates the hierarchical structure of tasks.txt file.

    Uses a state machine approach to ensure task levels follow valid
    transitions. For example, after a LV0 task, only LV0 or LV1 are valid.
    After LV2, you can have LV0-LV3. This prevents orphaned subtasks.

    Validation Rules:
        - LV1 must follow LV0
        - LV2 must follow LV1 or lower
        - Cannot jump levels (e.g., LV0 -> LV3 directly)
        - Empty lines are ignored

    Returns:
        bool: True if file structure is valid, False otherwise
    """
    if not os.path.exists("tasks.txt"):
        print("\n❌ tasks.txt file not found.")
        return False

    with open("tasks.txt", "r", encoding="utf-8") as file:
        # State machine: tracks which levels are currently valid
        # None = initial state, True = valid, False = invalid
        possible_lv = [None] * 5

        for i, line in enumerate(file):
            # Skip empty lines
            if not line.strip():
                continue

            # Root level task validation
            if not line.startswith("LV") and (
                all(lv is None for lv in possible_lv) or is_valid(0, possible_lv)
            ):
                update_possible_neighbors(0, possible_lv)
                continue

            # Hierarchical task validation (LVn-Title)
            if line.startswith("LV"):
                match = re.match(r"LV(\d+)-", line)
                if match:
                    current_lv = int(match.group(1))
                    if is_valid(current_lv, possible_lv):
                        update_possible_neighbors(current_lv, possible_lv)
                        continue

            # Invalid structure detected
            print(f"\n❌ Error at line {i + 1}:  {line}")
            print("❌ tasks.txt file is not valid.")
            return False

    print("\n✅ tasks.txt file is valid.")
    return True


def update_possible_neighbors(current_lv: int, possible_lv: list) -> None:
    """
    Updates state machine with valid next levels after current level.

    After processing a task at current_lv, determines which levels can
    legally follow based on NEIGHBORS constant.

    Args:
        current_lv: Current task level (0-4)
        possible_lv: State array tracking valid levels (modified in-place)
    """
    # Reset all levels to invalid
    for i in range(len(possible_lv)):
        possible_lv[i] = False

    # Mark allowed next levels as valid
    for i in NEIGHBORS[current_lv]:
        possible_lv[i] = True


def is_valid(current_lv: int, possible_lv: list) -> bool:
    """
    Checks if current level is valid given the current state.

    Args:
        current_lv: Level to validate (0-4)
        possible_lv: Current state of valid levels

    Returns:
        bool: True if current_lv is allowed in current state
    """
    return possible_lv[current_lv]


# ============================================================================
# UI UTILITIES
# ============================================================================


def hide_cursor() -> None:
    """
    Hides terminal cursor using ANSI escape sequence.

    Improves visual experience during interactive prompts and progress bars
    by preventing cursor blinking.
    """
    sys.stdout.write("\x1b[?25l")
    sys.stdout.flush()


def show_cursor() -> None:
    """
    Restores terminal cursor visibility using ANSI escape sequence.

    Should always be called after hide_cursor() to ensure cursor returns
    to normal state, even after errors.
    """
    sys.stdout.write("\x1b[?25h")
    sys.stdout.flush()


# ============================================================================
# MAIN PROGRAM FLOW
# ============================================================================


def main() -> None:
    """
    Main entry point for the TickTick task batch importer.

    Orchestrates the complete workflow:
    1. Display welcome banner
    2. Validate tasks.txt file structure
    3. Handle OAuth authentication
    4. Optionally select destination project
    5. Batch import tasks with hierarchy
    6. Report success/failure status
    """
    # Display application banner
    ascii_art = r"""
 ____        _       _     ___ _______ _      _ _______ _      _
|  _ \      | |     | |   |__ \__   __(_)    | |__   __(_)    | |
| |_) | __ _| |_ ___| |__    ) | | |   _  ___| | _| |   _  ___| | __
|  _ < / _` | __/ __| '_ \  / /  | |  | |/ __| |/ / |  | |/ __| |/ /
| |_) | (_| | || (__| | | |/ /_  | |  | | (__|   <| |  | | (__|   <
|____/ \__,_|\__\___|_| |_|____| |_|  |_|\___|_|\_\_|  |_|\___|_|\_\
                                                              
    """
    print(ascii_art)
    print("========================================================================")

    # Step 1: Validate input file before proceeding
    if not verify_txt_file():
        return

    # Step 2: Ensure valid OAuth token exists
    generate_token_info()
    token_info = get_stored_token_info()

    if token_info:
        print("✅ Access token obtained successfully.\n")

        # Step 3: Ask user whether to select a specific project or use Inbox
        question = inquirer.prompt(
            [
                inquirer.Confirm(
                    "project", message="Do you want to select a project ? (n => Inbox)"
                )
            ]
        )

        # Prepare authentication headers for API requests
        headers = set_headers(token_info["access_token"])

        # Step 4: Handle project selection or default to Inbox
        if question["project"]:
            projects = getProjects(headers)
            if projects:
                project_name, project_id = selectProject(projects)
                if project_name and project_id:
                    print(f"\n✅ Project selected : {project_name}")
                    print("\n⏳ Adding tasks to selected project...")
                    hasbeenAdded = add_tasks_batch(headers, project_id)
                else:
                    print("\n❌ Error while selecting project.")
            else:
                print("\n❌ No projects found.")
        else:
            # No project selected - tasks will be added to default Inbox
            print("\n✅ No project selected. Adding tasks to Inbox...")
            hasbeenAdded = add_tasks_batch(headers)

        # Step 5: Report final status
        if hasbeenAdded:
            print("\n✅ Tasks added successfully to TickTick !")
        else:
            print("\n❌ Error while adding tasks to TickTick.")
    else:
        print("\n❌ Error while obtaining access token.")

    print("\n\nProgram finished successfully.")


# Entry point - executes main() when script is run directly
if __name__ == "__main__":
    main()