184 lines
7.1 KiB
Python
184 lines
7.1 KiB
Python
import json
|
|
from datetime import datetime, timedelta
|
|
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_patch_request(token, endpoint, data):
|
|
return requests.patch(
|
|
endpoint,
|
|
json=data,
|
|
#data=data,
|
|
headers={'Authorization': 'Bearer ' + token['access_token']},).json()
|
|
|
|
def update_calendar_event(token, event_id, data, user_id=USER_ID):
|
|
endpoint= "https://graph.microsoft.com/v1.0/users/" + user_id + f"/calendar/events/{event_id}"
|
|
return execute_patch_request(token, endpoint, data)
|
|
|
|
def execute_get_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_get_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():
|
|
|
|
t1=datetime.now()
|
|
t2=t1 + timedelta(days=365)
|
|
t1 = t1.strftime("%Y-%m-%d")
|
|
t2 = t2.strftime("%Y-%m-%d")
|
|
|
|
filter = f"?StartDateTime={t1}&EndDateTime={t2}"
|
|
|
|
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')
|
|
parser.add_argument('--update_test_event', help='Execute a patch request for testing', 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"], e["attendees"], e["id"]) for e in events], indent=2))
|
|
|
|
if args.update_test_event:
|
|
|
|
data = {
|
|
"attendees" :
|
|
[
|
|
{
|
|
"type": "required",
|
|
"status": {
|
|
"response": "none",
|
|
"time": "0001-01-01T00:00:00Z"
|
|
},
|
|
"emailAddress": {
|
|
"name": "Simone Profus",
|
|
"address": "simone.profus@propedal.at"
|
|
}
|
|
}
|
|
],
|
|
}
|
|
|
|
data["attendees"][0]["emailAddress"]["name"] = "Patrick S."
|
|
data["attendees"][0]["emailAddress"]["address"] = "struebin.patrick@gmail.com"
|
|
|
|
#Test2
|
|
event_id = "AAMkADY0MDg1MTVjLTg5ZjItNGQxYS04MGQ3LWY2NjJmYjM0YmZhOQBGAAAAAADXD7SdVoWYQI4RYXbBumMEBwAf_ngZxs71RonY3GuLL8TVAADHFw_OAAAf_ngZxs71RonY3GuLL8TVAABLH9QmAAA="
|
|
ret = update_calendar_event(token, event_id, data)
|
|
|
|
print(json.dumps(ret, indent=2))
|
|
|
|
print("Done.") |