Microsoft TODO

mstodo.py

#!/usr/bin/python
import subprocess, os, json, ast, os.path
from random import randint
from urllib.parse import urlparse
from urllib.parse import parse_qsl
from http.server import BaseHTTPRequestHandler, HTTPServer
import threading

from requests_oauthlib import OAuth2Session, TokenUpdated
from oauthlib.oauth2 import BackendApplicationClient
from oauthlib.oauth2.rfc6749.errors import TokenExpiredError



client_id = "<ID>"
client_secret = "<SECRET>"

#   https://login.microsoftonline.com/common/oauth2/v2.0/authorize
#   https://login.microsoftonline.com/common/oauth2/v2.0/token

# Constant strings for OAuth2 flow
# The OAuth authority
authority = 'https://login.microsoftonline.com'

# The authorize URL that initiates the OAuth2 client credential flow for admin consent
authorize_url = '{0}{1}'.format(authority, '/common/oauth2/v2.0/authorize')

# The token issuing endpoint
token_url = '{0}{1}'.format(authority, '/common/oauth2/v2.0/token')

#redirect_url = 'https://login.microsoftonline.com/common/oauth2/nativeclient'
redirect_url = 'http://localhost:8383'

scopes = ['offline_access', 'https://outlook.office.com/Tasks.ReadWrite']

token_store = '.config/mstodo.token'


class Handler(BaseHTTPRequestHandler):
    auth_callback = ''
    server = None
    def do_GET(self):
        #url = urlparse(self.path)
        #dict(parse_qsl(url.query)).get('code')
        Handler.auth_callback = "{0}{1}".format(redirect_url, self.path)
        #kill server
        assassin1 = threading.Thread(target=Handler.server.shutdown)
        assassin1.daemon = True
        
        assassin0 = threading.Thread(target=Handler.server.socket.close)
        assassin0.daemon = True
        
        assassin1.start()
        assassin0.start()
        

class MSTask:
    def __init__(self, token_file = None):
        self.token_file = token_file or os.path.join(os.environ['HOME'], token_store)
        self.oauth = None
        self.token = None
        self.authorizing = False
        if os.path.isfile(self.token_file):
            with open(self.token_file, 'rt') as f:
                token_str = f.read()
                self.token = ast.literal_eval(token_str)

        os.environ['OAUTHLIB_INSECURE_TRANSPORT'] = '1'
        os.environ['OAUTHLIB_RELAX_TOKEN_SCOPE'] = '1'
        self.oauth = OAuth2Session(client_id,
                                   token=self.token,
                                   redirect_uri=redirect_url,
                                   auto_refresh_url=token_url,
                                   auto_refresh_kwargs=
                                   {
                                       'client_id':client_id,
                                       'client_secret':client_secret
                                   },
                                   scope=scopes)

    def authorize(self):
        self.authorizing = True
        authorization_url, state = self.oauth.authorization_url(authorize_url,
                                                                access_type='offline',
                                                                prompt='consent')

        #response = oauth.get(authorization_url)

        subprocess.call(["xdg-open", authorization_url])

        server = HTTPServer(('localhost', 8383), Handler)
        Handler.server = server
        server.serve_forever()

        self.token = self.get_token(Handler.auth_callback)
        with open(self.token_file, 'wt') as f:
            f.write(json.dumps(self.token))
        
        return self.token

    def get_token(self, auth_response):        
        token = self.oauth.fetch_token(token_url, client_secret=client_secret,
                              authorization_response=auth_response)

        self.authorizing = False
        return token

    def get_json(self, url):
        resp = self.oauth.get(url)
        if resp:
            body_s = resp.content.decode('utf-8')
            body = json.loads(body_s)
            return body
    
    def get_top_tasks(self):
        if self.authorizing:
            return
        elif not self.oauth.authorized:
            self.authorize()
        try:
            tasks = self.get_json("https://outlook.office.com/api/v2.0/me/tasks?$filter=Status ne 'Completed'")
            folders = self.get_json('https://outlook.office.com/api/v2.0/me/taskfolders')
            if tasks and folders:
                fs = folders['value']
                ts = tasks['value']
                return ['{0} @{1}'.format(t['Subject'], f['Name']) for t in ts for f in fs if f['Id'] == t['ParentFolderId'] and t['Status'] != 'Completed']
                
        except TokenExpiredError as te:
            self.authorize()
            return self.get_top_tasks()
        except TokenUpdated as tu:
            with open(self.token_file, 'wt') as f:
                f.write(json.dumps(tu.token))
            return self.get_top_tasks()


if __name__ == '__main__':
    task = MSTask()
    tasks = task.get_top_tasks()
    print(tasks[randint(0, len(tasks) - 1)])
    print('---')

    for t in tasks:
        print('{0} | trim=true '.format(t))

    print('---')
    print('Refresh | refresh=true')

https://msdn.microsoft.com/en-us/office/office365/api/complex-types-for-mail-contacts-calendar#TaskStatusV2

http://docs.oasis-open.org/odata/odata/v4.0/errata03/os/complete/part1-protocol/odata-v4.0-errata03-os-part1-protocol-complete.html#_The_$filter_System