140 lines
5.5 KiB
Python
140 lines
5.5 KiB
Python
import json
|
|
from datetime import datetime
|
|
import logging
|
|
import argparse
|
|
|
|
import requests
|
|
import msal
|
|
|
|
try:
|
|
from zoneinfo import ZoneInfo
|
|
except ImportError:
|
|
from backports.zoneinfo import ZoneInfo
|
|
|
|
config = {
|
|
"authority": "https://login.microsoftonline.com/propedal.at",
|
|
"client_id": "52f192c4-875d-44a2-b28a-575e920225e5", # client public id (from azure web interface)#"da3fc28c-5fcf-4884-9477-903a4420cc3d",
|
|
"scope": ["https://graph.microsoft.com/.default"], # scopes from api
|
|
"secret": "irj8Q~PliZzSe7JnXEaiWKQ6v0CAg1DTZOO~Ccsf" # api secret key (from azure web interface)#"bqP8Q~SEp_AmYYDZoEykXxZdADoCOhzOOhbO3c3T"
|
|
}
|
|
USER_ID = "simone.profus@propedal.at" # user with calendar #"2af02ca1-77fc-46fd-90af-c754306081cb" #
|
|
CALENDAR_ID = "AAMkADY0MDg1MTVjLTg5ZjItNGQxYS04MGQ3LWY2NjJmYjM0YmZhOQBGAAAAAADXD7SdVoWYQI4RYXbBumMEBwAf_ngZxs71RonY3GuLL8TVAAAAAAEGAAAf_ngZxs71RonY3GuLL8TVAADHFxN2AAA=" # calendar id - determined by /users/id/calendars
|
|
|
|
WEEKDAYS= {0:"Mo", 1:"Di", 2:"Mi", 3:"Do", 4: "Fr", 5:"Sa", 6: "So"}
|
|
# Optional logging
|
|
logging.basicConfig(format='%(asctime)s,%(msecs)d %(levelname)-4s [%(filename)s:%(lineno)d] %(message)s',
|
|
datefmt='%Y-%m-%d:%H:%M:%S',
|
|
level=logging.INFO)
|
|
# logging.getLogger("msal").setLevel(logging.INFO) # Optionally disable MSAL DEBUG logs
|
|
|
|
|
|
def get_access_token():
|
|
#with open("auth_config.json") as f:
|
|
# config = json.load(f)
|
|
|
|
# Create a preferably long-lived app instance which maintains a token cache.
|
|
app = msal.ConfidentialClientApplication(
|
|
config["client_id"], authority=config["authority"],
|
|
client_credential=config["secret"],
|
|
# token_cache=... # Default cache is in memory only.
|
|
# You can learn how to use SerializableTokenCache from
|
|
# https:#msal-python.readthedocs.io/en/latest/#msal.SerializableTokenCache
|
|
)
|
|
|
|
# The pattern to acquire a token looks like this.
|
|
result = None
|
|
|
|
# Firstly, looks up a token from cache
|
|
# Since we are looking for token for the current app, NOT for an end user,
|
|
# notice we give account parameter as None. # TODO: token never exists in cache; make app long living
|
|
result = app.acquire_token_silent(config["scope"], account=None)
|
|
|
|
if result is None:
|
|
logging.info("No suitable token exists in cache. Let's get a new one from AAD.")
|
|
return app.acquire_token_for_client(scopes=config["scope"])
|
|
else:
|
|
logging.info("Token was found in cache.")
|
|
|
|
if not "access_token" in result: # a final check
|
|
logging.error(result.get("error"))
|
|
logging.error(result.get("error_description"))
|
|
logging.error(result.get("correlation_id")) # You may need this when reporting a bug
|
|
raise AssertionError("No access token present")
|
|
|
|
return result
|
|
|
|
|
|
def execute_request(token: dict, endpoint:str):
|
|
return requests.get( # Use token to call downstream service
|
|
endpoint,
|
|
headers={'Authorization': 'Bearer ' + token['access_token']},).json()
|
|
|
|
def execute_user_request(token, endpoint, user_id=USER_ID):
|
|
return execute_request(token, "https://graph.microsoft.com/v1.0/users/" + user_id + f"/{endpoint}")
|
|
|
|
def get_all_calendar_events():
|
|
token = get_access_token()
|
|
|
|
return execute_user_request(token, f"calendars/{CALENDAR_ID}/events").get("value")
|
|
|
|
def get_future_calendar_events():
|
|
|
|
filter = "?StartDateTime=2022-06-12&EndDateTime=2023-06-30" # TODO: Add today dynamically + 1 from time.time in right format
|
|
|
|
token = get_access_token()
|
|
|
|
return execute_user_request(token, f"calendars/{CALENDAR_ID}/calendarview{filter}")
|
|
|
|
def convert_datetimes(events):
|
|
for event in events:
|
|
|
|
start_date_time = datetime.fromisoformat(event["start"]["dateTime"][:-1]) \
|
|
.replace(tzinfo=ZoneInfo('UTC')).astimezone(ZoneInfo('localtime'))
|
|
|
|
stop_date_time = datetime.fromisoformat(event["end"]["dateTime"][:-1]) \
|
|
.replace(tzinfo=ZoneInfo('UTC')).astimezone(ZoneInfo('localtime'))
|
|
|
|
event["start"]["date"] = start_date_time.strftime("%d.%m.%Y")
|
|
event["start"]["time"] = start_date_time.strftime("%H:%M:%S")
|
|
event["end"]["date"] = stop_date_time.strftime("%d.%m.%Y")
|
|
event["end"]["time"] = stop_date_time.strftime("%H:%M:%S")
|
|
event["duration"] = str(stop_date_time - start_date_time)[:-3]
|
|
event["weekday"] = WEEKDAYS[start_date_time.weekday()]
|
|
|
|
return events
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
parser = argparse.ArgumentParser(description='Process some integers.')
|
|
parser.add_argument('--show_calendars', help='Show available calendars', action='store_true')
|
|
parser.add_argument('--show_event_entries', help='Show available event fields', action='store_true')
|
|
parser.add_argument('--show_events', help='Show available events for selected calendar', action='store_true')
|
|
|
|
args = parser.parse_args()
|
|
|
|
# Calling graph using the access token
|
|
token = get_access_token()
|
|
|
|
if args.show_calendars:
|
|
calendars = execute_user_request(token, "calendars")
|
|
cal_name_id = [(c["name"], c["id"]) for c in calendars["value"]]
|
|
|
|
print("Available calendars are:")
|
|
print(json.dumps(cal_name_id, indent=2))
|
|
|
|
if args.show_event_entries or args.show_events:
|
|
events= get_future_calendar_events().get("value")
|
|
|
|
if args.show_event_entries:
|
|
print(json.dumps(list(events)[0].keys(), indent=2))
|
|
|
|
if args.show_events:
|
|
print(json.dumps(events, indent=2))
|
|
|
|
convert_datetimes(events)
|
|
|
|
print(json.dumps([(e["subject"], e["start"], e["duration"]) for e in events], indent=2))
|
|
|
|
|
|
print("Done.") |