Upload app.py

app.py

@@ -1,64 +1,415 @@
 import gradio as gr
-from huggingface_hub import InferenceClient
+from openai import OpenAI
+from google.colab import userdata
+import requests
+import os
+import asana
+from asana.rest import ApiException
+from pprint import pprint
+import openai
+import json
+import datetime
+import dateparser
 
-"""
-For more information on `huggingface_hub` Inference API support, please check the docs: https://huggingface.co/docs/huggingface_hub/v0.22.2/en/guides/inference
-"""
-client = InferenceClient("HuggingFaceH4/zephyr-7b-beta")
+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)
 
-def respond(
-    message,
-    history: list[tuple[str, str]],
-    system_message,
-    max_tokens,
-    temperature,
-    top_p,
-):
-    messages = [{"role": "system", "content": system_message}]
+# Get Asana API key from the environment variable
+access_token = os.environ.get("ASANA_API_KEY")
 
-    for val in history:
-        if val[0]:
-            messages.append({"role": "user", "content": val[0]})
-        if val[1]:
-            messages.append({"role": "assistant", "content": val[1]})
+# Set up the Asana API client with the retrieved access token
+configuration = asana.Configuration()
+configuration.access_token = access_token
+api_client = asana.ApiClient(configuration)
 
-    messages.append({"role": "user", "content": message})
+ASANA_BASE_URL = "https://app.asana.com/api/1.0"
+ASANA_HEADERS = {
+    "Authorization": f"Bearer {access_token}",
+    "Content-Type": "application/json"
+}
 
-    response = ""
+DEFAULT_PROJECT_GID = "1209104858113361"
 
-    for message in client.chat_completion(
-        messages,
-        max_tokens=max_tokens,
-        stream=True,
-        temperature=temperature,
-        top_p=top_p,
-    ):
-        token = message.choices[0].delta.content
+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}
 
-        response += token
-        yield response
+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"
+}
 """
-For information on how to customize the ChatInterface, peruse the gradio docs: https://www.gradio.app/docs/chatinterface
-"""
-demo = gr.ChatInterface(
-    respond,
-    additional_inputs=[
-        gr.Textbox(value="You are a friendly Chatbot.", label="System message"),
-        gr.Slider(minimum=1, maximum=2048, value=512, step=1, label="Max new tokens"),
-        gr.Slider(minimum=0.1, maximum=4.0, value=0.7, step=0.1, label="Temperature"),
-        gr.Slider(
-            minimum=0.1,
-            maximum=1.0,
-            value=0.95,
-            step=0.05,
-            label="Top-p (nucleus sampling)",
-        ),
-    ],
-)
-
-
-if __name__ == "__main__":
-    demo.launch()
+
+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