def get_tasks(target_date=None): """ Fetches tasks from Microsoft To Do for a specific date (tasks due on or before that date). Args: target_date: datetime.date object for the target day. If None, uses today. Returns a list of task dictionaries. """ CLIENT_ID = os.getenv('MS_CLIENT_ID', 'CLIENT_ID_KEY') AUTHORITY = "https://login.microsoftonline.com/consumers" SCOPES = ['Tasks.ReadWrite', 'User.Read'] # Setup persistent token cache cache = SerializableTokenCache() cache_file = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), '.token_cache.bin') if os.path.exists(cache_file): with open(cache_file, 'r') as f: cache.deserialize(f.read()) # Authentication with persistent cache app = PublicClientApplication(CLIENT_ID, authority=AUTHORITY, token_cache=cache) accounts = app.get_accounts() result = None if accounts: result = app.acquire_token_silent(SCOPES, account=accounts[0]) if not result or "access_token" not in result: for account in accounts: app.remove_account(account) result = None if not result: flow = app.initiate_device_flow(scopes=SCOPES) if "user_code" not in flow: print(f"[MS To Do] Failed to create device flow: {flow.get('error_description', 'Unknown error')}") return [] if "message" in flow: print(flow['message']) result = app.acquire_token_by_device_flow(flow) if not result or "access_token" not in result: print(f"[MS To Do] Authentication failed: {result.get('error_description', 'No access token') if result else 'No result'}") return [] headers = {"Authorization": f"Bearer {result['access_token']}"} # Get date boundaries for target date # If no target date provided, use today if target_date is None: from datetime import date target_date = date.today() # Convert date to datetime in UTC for comparison now = datetime.now(timezone.utc) target_day_end = datetime.combine(target_date, datetime.max.time()).replace(tzinfo=timezone.utc) # Fetch all lists lists_r = requests.get("https://graph.microsoft.com/v1.0/me/todo/lists", headers=headers) if lists_r.status_code != 200: return [] lists_res = lists_r.json().get("value", []) all_tasks = [] # Fetch tasks from all lists with server-side filtering for task_list in lists_res: list_id = task_list["id"] list_name = task_list["displayName"] # Simple filter - just get incomplete tasks params = {"$filter": "status ne 'completed'"} tasks_r = requests.get( f"https://graph.microsoft.com/v1.0/me/todo/lists/{list_id}/tasks", headers=headers, params=params ) if tasks_r.status_code != 200: continue tasks = tasks_r.json().get("value", []) # Filter and transform tasks for task in tasks: due_date_obj = task.get("dueDateTime") if not due_date_obj: continue due_date_str = due_date_obj.get("dateTime") if not due_date_str: continue try: due_date = datetime.fromisoformat(due_date_str.split('.')[0]) if due_date.tzinfo is None: due_date = due_date.replace(tzinfo=timezone.utc) # Include all tasks due on or before the target date if due_date <= target_day_end: # Determine status based on target date target_day_start = datetime.combine(target_date, datetime.min.time()).replace(tzinfo=timezone.utc) if due_date < target_day_start: status = "OVERDUE" elif due_date <= target_day_end: status = f"DUE {target_date.strftime('%Y-%m-%d')}" else: status = "FUTURE" all_tasks.append({ "list": list_name, "title": task["title"], "due": due_date.strftime("%Y-%m-%d"), "importance": task.get("importance", "normal"), "status": status }) except Exception as e: continue # Print task summary if all_tasks: print(f"[MS To Do] {len(all_tasks)} task(s) for {target_date.strftime('%Y-%m-%d')} or overdue") else: print(f"[MS To Do] No tasks due on {target_date.strftime('%Y-%m-%d')} or overdue") # Save token cache for next run if cache.has_state_changed: with open(cache_file, 'w') as f: f.write(cache.serialize()) return all_tasks