CanvasTodoistConnector

canvas-todoist-link.py

"""
Sync Canvas assignments to Todoist and send alerts to Discord.

"""

import requests
from datetime import datetime
import os
from typing import List, Dict, Tuple

class CanvasToTodoistSync:
    def __init__(self, canvas_api_key: str, canvas_domain: str, todoist_api_key: str, discord_webhook: str, conversion_table: Dict[str, str]):
        self.canvas_api_key = canvas_api_key
        self.canvas_base_url = f"https://{canvas_domain}/api/v1"
        self.todoist_api_key = todoist_api_key
        self.todoist_base_url = "https://api.todoist.com/rest/v2"
        self.discord_webhook = discord_webhook
        self.conversion_table = conversion_table
        
    def get_canvas_headers(self) -> Dict:
        return {
            "Authorization": f"Bearer {self.canvas_api_key}"
        }
    
    def get_todoist_headers(self) -> Dict:
        return {
            "Authorization": f"Bearer {self.todoist_api_key}",
            "Content-Type": "application/json"
        }
    
    def convert_course_name(self, course_name: str) -> str:
        return self.conversion_table.get(course_name) or course_name
    
    def get_active_courses(self) -> List[Dict]:
        """Get favorited active courses from Canvas"""
        response = requests.get(
            f"{self.canvas_base_url}/courses",
            headers=self.get_canvas_headers(),
            params={
                "enrollment_state": "active",
                "include[]": "favorites",
                "favorites": True,
                "per_page": 30

            }
        )
        response.raise_for_status()

        # Filter to only include favorited courses
        courses = response.json()

        return [course for course in courses if course.get('is_favorite', False)]
    
    def get_course_assignments(self, course_id: int) -> List[Dict]:
        """Get all assignments for a specific course"""
        assignments = []

        # Get first page of assignments
        url = f"{self.canvas_base_url}/courses/{course_id}/assignments"
        next_url, curr_url, response = self.get_page_assignments(url)

        # Add assignments to list
        assignments.extend(response)
        print(f"Found {len(assignments)} assignments")

        while(next_url and next_url != curr_url):
            next_url, curr_url, response = self.get_page_assignments(next_url)
            assignments.extend(response)
            print(f"Found {len(assignments)} assignments")

        return assignments
    
    def get_page_assignments(self, url: str) -> Tuple[str, str, List[Dict]]:
        """Get all assignments for a specific course"""
        params = {
            # "bucket": "upcoming",
            "include[]": "submission",
            "per_page": 30,
        }
        response = requests.get(
            url,
            headers=self.get_canvas_headers(),
            params=params
        )
        response.raise_for_status()

        # Get next page URL
        next_url = response.links['next']['url'] if 'next' in response.links else None
        curr_url = response.links['current']['url'] if 'current' in response.links else None

        return (next_url, curr_url, response.json())
    
    def get_or_create_project(self, project_name: str) -> str:
        """Get or create a single Todoist project with specified name"""
        # Get all projects
        response = requests.get(
            f"{self.todoist_base_url}/projects",
            headers=self.get_todoist_headers()
        )
        response.raise_for_status()
        projects = response.json()
        
        # Look for existing project
        for project in projects:
            if project["name"] == project_name:
                return project["id"]
        
        # Create new project if it doesn't exist
        response = requests.post(
            f"{self.todoist_base_url}/projects",
            headers=self.get_todoist_headers(),
            json={"name": project_name}
        )
        response.raise_for_status()
        return response.json()["id"]
    
    def get_or_create_section(self, project_id: str, section_name: str) -> str:
        """Get or create a Todoist section within a specific project"""
        # Get all sections for the project
        response = requests.get(
            f"{self.todoist_base_url}/sections",
            headers=self.get_todoist_headers(),
            params={"project_id": project_id}
        )
        response.raise_for_status()
        sections = response.json()
        
        # Look for existing section
        for section in sections:
            if section["name"] == section_name:
                return section["id"]
        
        # Create new section if it doesn't exist
        response = requests.post(
            f"{self.todoist_base_url}/sections",
            headers=self.get_todoist_headers(),
            json={"project_id": project_id, "name": section_name}
        )
        response.raise_for_status()
        return response.json()["id"]
    
    def get_all_tasks(self, project_id: str) -> List[Dict]:
        """Get all tasks (including completed) in a Todoist project"""
        # Get active tasks
        active_tasks = requests.get(
            f"{self.todoist_base_url}/tasks",
            headers=self.get_todoist_headers(),
            params={"project_id": project_id}
        ).json()
        
        # Get completed tasks
        completed_tasks = requests.get(
            "https://api.todoist.com/sync/v9/completed/get_all",
            headers=self.get_todoist_headers(),
            params={"project_id": project_id}
        ).json()['items']

        # print(active_tasks)
        # print(completed_tasks)
        
        return active_tasks + completed_tasks
    
    def create_task(self, project_id: str, section_id: str, assignment: Dict) -> None:
        """Create a Todoist task for an assignment in a specific section"""
        due_date = assignment.get("due_at")
        if due_date:
            due_date = datetime.strptime(due_date, "%Y-%m-%dT%H:%M:%SZ").strftime("%Y-%m-%d")
        
        task_data = {
            "project_id": project_id,
            "section_id": section_id,
            "content": assignment["name"],
            "description": assignment.get("html_url", ""),
            "due_date": due_date,
            "deadline_date": due_date,
            "priority": 3  # Medium priority
        }
        
        response = requests.post(
            f"{self.todoist_base_url}/tasks",
            headers=self.get_todoist_headers(),
            json=task_data
        )
        response.raise_for_status()

    def alert_header(self, discord_url: str):
        """Send a Discord webhook alert with the current date and time"""
        current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        data = {
            "content": f"Assignments for: {current_time}"
        }
        
        response = requests.post(discord_url, json=data)
        response.raise_for_status()

    def alert_task(self, discord_url: str, assignment: Dict, project_name: str):
        """Send a Discord webhook alert for an assignment"""
        due_date = assignment.get("due_at")
        if due_date:
            due_date = datetime.strptime(due_date, "%Y-%m-%dT%H:%M:%SZ").strftime("%Y-%m-%d")
        
        data = {
            "content": f"{project_name}: {assignment['name']} due {due_date}",
            # "embeds": [
            #     {
            #         "title": assignment["name"],
            #         # "description": assignment.get("description", ""),
            #         "url": assignment.get("html_url", ""),
            #         "color": 0xFF0000,  # Red color
            #         "fields": [
            #             {"name": "Course", "value": project_name},
            #             {"name": "Due Date", "value": due_date}
            #         ]
            #     }
            # ]
        }
        
        response = requests.post(discord_url, json=data)
        response.raise_for_status()

    def is_task_exists(self, task_name: str, all_tasks: List[Dict]) -> bool:
        """Check if a task already exists in Todoist (including completed tasks)"""
        return any(task["content"] == task_name for task in all_tasks)
    
    def sync_assignments(self, project_name: str = "School") -> None:
        """Main function to sync Canvas assignments to Todoist"""
        # Get favorited active courses
        courses = self.get_active_courses()

        print(f"Found {len(courses)} favorited courses")

        # Get or create a single project
        project_id = self.get_or_create_project(project_name)
        
        # Get existing tasks in the project
        existing_tasks = self.get_all_tasks(project_id)
        
        # Track if header is sent
        # Only send header if new assignment found
        sent_header = False

        for course in courses:
            print(f"Processing course: {course['name']}")

            # Convert Canvas course name to Todoist section name
            section_name = self.convert_course_name(course['name'])
            
            # Get or create section for this course
            section_id = self.get_or_create_section(project_id, section_name)
            
            # Get all assignments for the course
            assignments = self.get_course_assignments(course['id'])
            
            # Create Todoist tasks for each assignment
            for assignment in assignments:
                # Skip if assignment is already complete in Canvas

                # print("\tAssignment: ", assignment.get('submission'))

                if assignment.get('submission') and assignment['submission'].get('submitted_at') != None:
                    print(f"\tSkipping submitted task: {assignment['name']}")
                    continue

                if assignment.get('locked_for_user'):
                    print("\tAssignment is locked")
                #     continue

                # Skip if task already exists in Todoist
                if self.is_task_exists(assignment['name'], existing_tasks):
                    print(f"\tSkipping existing task: {assignment['name']}")
                    continue

                # Found new assignment
                if not sent_header:
                    try:
                        self.alert_header(self.discord_webhook)
                        print("\tSent Discord header")
                        sent_header = True
                    except requests.exceptions.RequestException as e:
                        print(f"\tError sending Discord header: {str(e)}")

                # Create task in Todoist within the specific section
                try:
                    self.create_task(project_id, section_id, assignment)
                    print(f"\tCreated task for: {assignment['name']}")
                except requests.exceptions.RequestException as e:
                    print(f"\tError creating task for {assignment['name']}: {str(e)}")

                # Send Discord alert for the task
                try:
                    self.alert_task(self.discord_webhook, assignment, section_name)
                    print(f"\tSent Discord alert for: {assignment['name']}")
                except requests.exceptions.RequestException as e:
                    print(f"\tError sending Discord alert for {assignment['name']}: {str(e)}")

def main():
    # Load environment variables
    # Load environment variables from a .env file
    load_dotenv()

    canvas_api_key = os.getenv("CANVAS_API_KEY")
    canvas_domain = os.getenv("CANVAS_DOMAIN")
    todoist_api_key = os.getenv("TODOIST_API_KEY")
    discord_webhook = os.getenv("DISCORD_WEBHOOK")
    conversion_table = {
            "Web Dev 2": "Web Dev 2 (CS 7830)",
            "Numerical Methods": "Num Methods (CS 7070)",
            "Big Ideas in Science": "Big Ideas (GN_H 2457H)",
            "Capstone 2": "Capstone 2 (CS 4980)",
            "iOS App Dev": "iOS App Dev (CS 4405)",
            "C#/.NET": "C#/.NET (IT 4400)",
            "Graph Theory": "Graph Theory (MATH 4120)",
    }

    if not all([canvas_api_key, canvas_domain, todoist_api_key, discord_webhook]):
        raise ValueError("Missing required environment variables")

    if not conversion_table:
        raise ValueError("Missing required conversion table")
    
    project_name = "School"

    # Create and run sync
    syncer = CanvasToTodoistSync(canvas_api_key, canvas_domain, todoist_api_key, discord_webhook, conversion_table)
    syncer.sync_assignments(project_name)

if __name__ == "__main__":
    main()