app.py

import gradio as gr
from openai import OpenAI
import requests
import os
import asana
from asana.rest import ApiException
from pprint import pprint
import openai
import json
import datetime
import dateparser
import multiprocessing.pool

# Monkey patch ApiClient.__del__
def noop_del(self):
    print("Monkey patched ApiClient.__del__ called. Doing nothing.")
    pass

asana.api_client.ApiClient.__del__ = noop_del

# Monkey patch multiprocessing.pool.Pool.__del__
def noop_pool_del(self):
    print("Monkey patched multiprocessing.pool.Pool.__del__ called. Doing nothing.")
    pass

multiprocessing.pool.Pool.__del__ = noop_pool_del

settings={
    "PREFER_DATES_FROM": "future",
    "RELATIVE_BASE": datetime.datetime.now()
}
# Get the OpenAI API key from the environment variable
open_ai_key = os.environ.get("OPENAI_API_KEY")

# Initialize the OpenAI client with the API key
client = OpenAI(api_key=open_ai_key)

# Get Asana API key from the environment variable
access_token = os.environ.get("ASANA_API_KEY")

# Set up the Asana API client with the retrieved access token
configuration = asana.Configuration()
configuration.access_token = access_token
api_client = asana.ApiClient(configuration)

ASANA_BASE_URL = "https://app.asana.com/api/1.0"
ASANA_HEADERS = {
    "Authorization": f"Bearer {access_token}",
    "Content-Type": "application/json"
}

DEFAULT_PROJECT_GID = "1209104858113361"

def create_asana_task(name, due_date=None):
    """Create a task in Asana."""
    url = f"{ASANA_BASE_URL}/tasks"
    data = {
        "data": {
            "name": name,
            "projects": [DEFAULT_PROJECT_GID]
        }
    }
    if due_date:
        data["data"]["due_on"] = due_date  # Asana uses 'due_on' in YYYY-MM-DD format
    resp = requests.post(url, json=data, headers=ASANA_HEADERS)
    if resp.status_code == 201:
        return resp.json()["data"]  # returns the newly created task object
    else:
        return {"error": resp.text}

def list_asana_tasks(only_open=True):
    """List tasks in the default project, optionally filtering for only open tasks."""
    url = f"{ASANA_BASE_URL}/projects/{DEFAULT_PROJECT_GID}/tasks"
    params = {
        "opt_fields": "name,completed"  # Include the "completed" field to verify task status
    }
    if only_open:
        params["completed_since"] = "now"  # Fetch only incomplete or recently updated tasks

    resp = requests.get(url, headers=ASANA_HEADERS, params=params)
    if resp.status_code == 200:
        tasks = resp.json()["data"]
        if only_open:
            # Filter out completed tasks if only_open is True
            tasks = [task for task in tasks if not task.get("completed", False)]
        return tasks
    else:
        return {"error": resp.text}

def complete_asana_task(task_gid):
    """Mark a task as complete."""
    url = f"{ASANA_BASE_URL}/tasks/{task_gid}"
    data = {
        "data": {
            "completed": True
        }
    }
    resp = requests.put(url, json=data, headers=ASANA_HEADERS)
    if resp.status_code == 200:
        return resp.json()["data"]
    else:
        return {"error": resp.text}         

def call_llm(user_message, conversation_history=None):
    today_date = datetime.date.today().strftime("%Y-%m-%d")
    messages = [{"role": "system", "content": system_prompt}]
    messages.append({"role": "user", "content": user_message})

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        temperature=0.2,
        max_tokens=200
    )
    # Access 'content' using dot notation instead of indexing
    llm_content = response.choices[0].message.content
    print(f"Debug: Raw LLM Content: {llm_content}")
    return llm_content

# Global variables
last_task_list = []

def execute_turn(user_message, test_feed_iter=None):
    global last_task_list  # Declare global variable at the top of the function
    llm_output = call_llm(user_message)
    parsed = parse_llm_response(llm_output)
    action = parsed.get("action")

    if action == "CREATE_TASK":
        task_name = parsed.get("name", "").strip()
        due_date = parsed.get("due")  # Could be a natural language string like "tomorrow"

        # --- 1) OVERRIDE "NEW TASK" NAME IF POSSIBLE ---
        if (not task_name or task_name.lower() == "new task"):
            if test_feed_iter is not None:
                # If we have more lines in TEST_FEED, use them
                try:
                    override_name = next(test_feed_iter)  # get the next line from the feed
                    task_name = override_name.strip()
                    print(f"Debug: Overriding 'New Task' with '{task_name}' from TEST_FEED")
                except StopIteration:
                    # If there's nothing left in the feed, fallback to user prompt
                    if not task_name:
                        print("Bot: Task name cannot be empty. Please try again.")
                        return
            else:
                # If we don't have a test_feed_iter, just prompt the user
                print("Bot: What would you like the name of the task to be?")
                task_name = input(USER_PROMPT).strip()
                if not task_name:
                    print("Bot: Task name cannot be empty. Please try again.")
                    return

        # --- 2) OVERRIDE DUE DATE IF POSSIBLE ---
        if not due_date:
            if test_feed_iter is not None:
                # Try to parse the next line as a date
                try:
                    maybe_date = next(test_feed_iter)
                    parsed_date = dateparser.parse(maybe_date)
                    if parsed_date:
                        due_date = parsed_date.strftime('%Y-%m-%d')
                        print(f"Debug: Overriding due date with '{due_date}' from TEST_FEED")
                    else:
                        # Not recognized as a date; do nothing
                        print(f"Debug: '{maybe_date}' did not parse as a date; skipping due.")
                except StopIteration:
                    pass

            # If we still have no due_date after that, fallback to user prompt
            if not due_date:
                user_due_date = input(USER_PROMPT).strip()
                if user_due_date:
                    parsed_date = dateparser.parse(
                        user_due_date,
                        settings={
                            "PREFER_DATES_FROM": "future",
                            # Optionally: "RELATIVE_BASE": datetime.datetime.now()
                        }
                    )
                    if parsed_date:
                        due_date = parsed_date.strftime('%Y-%m-%d')
                    else:
                        print("Bot: I couldn’t understand the due date. Skipping it.")
                        due_date = None

        # --- 3) ATTEMPT TO CREATE THE TASK ---
        print(f"Debug: Attempting to create task with name '{task_name}' and due date '{due_date}'")
        result = create_asana_task(task_name, due_date)
        # Provide a confirmation message if successful
        if "error" in result:
            print("Bot: Sorry, I had trouble creating the task:", result["error"])
        else:
            message = f"Bot: I've created your task '{result['name']}' (ID: {result['gid']})."
            if due_date:
                message += f" It's due on {due_date}."
            print(message)

    elif action == "LIST_TASKS":
        # (Unmodified code for listing tasks)
        filter_type = parsed.get("filter", "open")  # Default to "open" if no filter is provided
        only_open = filter_type == "open"
        tasks = list_asana_tasks(only_open=only_open)
        if "error" in tasks:
            print("Bot: Sorry, I had trouble listing tasks:", tasks["error"])
        elif not tasks:
            if only_open:
                print("Bot: You have no open tasks!")
            else:
                print("Bot: You have no tasks!")
        else:
            task_type = "open" if only_open else "all"
            print(f"Here are your {task_type} tasks:")
            last_task_list.clear()  # Clear previous tasks
            for t in tasks:
                task_info = {'name': t['name'], 'gid': t['gid']}
                last_task_list.append(task_info)  # Store task info
                print(f"- {t['name']} (ID: {t['gid']})")

    elif action == "COMPLETE_TASK":
        # (Unmodified code for completing tasks)
        task_gid = parsed.get("task_gid")
        task_name = parsed.get("name")  # Capture task name for fuzzy matching

        if task_gid:
            result = complete_asana_task(task_gid)
            if "error" in result:
                print("Bot: Sorry, I couldn’t complete the task:", result["error"])
            else:
                print(f"Task '{result['name']}' marked as complete.")
        elif task_name:
            tasks = list_asana_tasks()
            if "error" in tasks:
                print("Bot: Sorry, I had trouble fetching tasks to find a match.")
                return

            matches = [t for t in tasks if task_name.lower() in t['name'].lower()]

            if len(matches) == 1:
                task_to_close = matches[0]
                result = complete_asana_task(task_to_close["gid"])
                if "error" in result:
                    print(f"Bot: Sorry, I couldn’t complete the task: {result['error']}")
                else:
                    print(f"Task '{task_to_close['name']}' marked as complete.")
            elif len(matches) > 1:
                print("Bot: I found multiple tasks matching that name. "
                      "Please provide the ID of the task you'd like to close:")
                for task in matches:
                    print(f"- {task['name']} (ID: {task['gid']})")
            else:
                print(f"Bot: I couldn’t find any tasks matching '{task_name}'.")
        else:
            # Attempt to extract ordinal-based task position.
            ordinal_map = {
                'first': 1,
                'second': 2,
                'third': 3,
                'fourth': 4,
                'fifth': 5,
                'sixth': 6,
                'seventh': 7,
                'eighth': 8,
                'ninth': 9,
                'tenth': 10
            }
            words = user_message.lower().split()
            ordinal_position = None
            for word in words:
                if word in ordinal_map:
                    ordinal_position = ordinal_map[word] - 1  # zero-based index
                    break
                elif word.isdigit():
                    ordinal_position = int(word) - 1
                    break

            if ordinal_position is not None and last_task_list:
                if 0 <= ordinal_position < len(last_task_list):
                    task_to_close = last_task_list[ordinal_position]
                    result = complete_asana_task(task_to_close["gid"])
                    if "error" in result:
                        print(f"Bot: Sorry, I couldn’t complete the task: {result['error']}")
                    else:
                        print(f"Task '{task_to_close['name']}' marked as complete.")
                else:
                    print("Bot: The task number you specified is out of range.")
            else:
                print("Bot: Please provide a valid task name, ID, or position to close.")
    else:
        # No recognized action, or just normal text
        print(llm_output)

def extract_task_id_from_message(message):
    """
    Extract task ID (task_gid) from the user message.
    Example input: "Can we close task 1234567890?"
    Example output: "1234567890"
    """
    import re
    # Use a regular expression to find a numeric sequence in the message
    match = re.search(r'\b\d{10,}\b', message)  # Look for 10+ digit numbers
    if match:
        return match.group(0)  # Return the first match
    return None  # If no match found, return None

def parse_llm_response(llm_output):
    try:
        print(f"Debug: Raw LLM Content: {llm_output}")  # Debug raw output

        # Strip the backticks and "json" tag
        if llm_output.startswith("```json") and llm_output.endswith("```"):
            llm_output = llm_output.strip("```").strip("json").strip()

        print(f"Debug: Cleaned LLM Output: {llm_output}")  # Debug cleaned output

        # Parse the cleaned JSON
        parsed_response = json.loads(llm_output)
        print(f"Debug: Parsed Response: {parsed_response}")  # Debug parsed JSON
        return parsed_response
    except json.JSONDecodeError as e:
        print(f"Error: Failed to parse LLM response: {e}")
        return {"action": "NONE"}  # Fallback
    except Exception as e:
        print(f"Error: Unexpected issue in parse_llm_response: {e}")
        return {"action": "NONE"}  # Fallback

def run_manual_chat():
    print("Hello! I'm your Asana Copilot!")
    print("I can help you create new tasks, list your tasks, and mark tasks as completed.")
    print("Let me know how I can help.")
    print("Want to end our chat? Just type 'quit' to exit.\n")

    USER_PROMPT = "[USER]\n>>> "
    TURN_BREAK = "-------------------\n"

    while True:
        user_input = input(USER_PROMPT).strip()
        if user_input.lower() == "quit":
            print("\nExiting the chat. Goodbye!")
            break

        print(TURN_BREAK + "[COPILOT]")
        execute_turn(user_input)
        print(TURN_BREAK)

system_prompt = """
You are a friendly AI Copilot that helps users interface with Asana -- namely creating new tasks, listing tasks, and marking tasks as complete.
You will interpret the user's request and respond with structured JSON.
Today's date is {today_date}.

Rules:
1. If the user asks to create a task, respond with:
   { "action": "CREATE_TASK", "name": "<TASK NAME>", "due": "<YYYY-MM-DD>" }
   If they gave a date in any format. For words like 'tomorrow', interpret it as {today_date} + 1 day, etc.
   If no date is given or you cannot parse it, omit the 'due' field.
2. If the user asks to list tasks, respond with:
   {"action": "LIST_TASKS", "filter": "open"}  # For "list my open tasks" or similar
   {"action": "LIST_TASKS", "filter": "all"}  # For "list all my tasks" or similar
   If the user specifies "open tasks" or similar, return only incomplete tasks. If the user specifies "all tasks," return all tasks (completed and incomplete).
   If the intent is unclear, default to showing only open tasks.
3. If the user asks to complete a task, respond with:
   { "action": "COMPLETE_TASK", "task_gid": "<ID>" }
   OR
   { "action": "COMPLETE_TASK", "name": "<TASK NAME>" }
   OR
   { "action": "COMPLETE_TASK", "position": <NUMBER> }
   Use 'position' if the user refers to a task by its position in the list (e.g., "third one").
4. If no action is needed, respond with:
   { "action": "NONE" }

Examples:
- User: "Close task 1209105096577103"
  Response: { "action": "COMPLETE_TASK", "task_gid": "1209105096577103" }

- User: "Can you close rub jason's feet?"
  Response: { "action": "COMPLETE_TASK", "name": "rub jason's feet" }

- User: "List all my tasks"
  Response: { "action": "LIST_TASKS" }

- User: "Create a task called 'Finish report' due tomorrow"
  Response: { "action": "CREATE_TASK", "name": "Finish report", "due": "2025-01-08" }

- User: "Close the third one"
  Response: { "action": "COMPLETE_TASK", "position": 3 }

- User: "Complete task number 5"
  Response: { "action": "COMPLETE_TASK", "position": 5 }

- {"action": "LIST_TASKS", "filter": "open"}  # For "list my open tasks"
- {"action": "LIST_TASKS", "filter": "all"}  # For "list all my tasks"
- {"action": "CREATE_TASK", "name": "Task Name", "due": "2025-01-15"}
- {"action": "COMPLETE_TASK", "task_gid": "1209105096577103"}

Again, always respond in JSON format. Example:
{
  "action": "CREATE_TASK",
  "name": "Submit Assignment",
  "due": "2023-12-31"
}

If no action is required, respond with:
{
  "action": "NONE"
}
"""

def process_input(user_input):
    """
    This function takes the user's input, processes it using the Asana Copilot logic,
    and returns the formatted output for display.
    """
    global last_task_list  # Make sure to handle global variables

    # Simulate the conversation turn
    response = ""  # Initialize an empty string to accumulate the response

    # Process the user input using the execute_turn function
    # Capture the output from execute_turn (e.g., print statements) and append it to 'response'
    import io
    from contextlib import redirect_stdout

    with io.StringIO() as buf, redirect_stdout(buf):
        execute_turn(user_input)
        output = buf.getvalue()

    response += output  # Append the captured output to the response

    return response