Integrations

I received a request to email a regular report of call activity from MS Teams for our tenant on the first of the month for the previous month. This is similar to what's available in the Teams admin dashboard, but it doesn't require logging in and running it manually.

The script assumes a config.json file exists in the format shown below. Note that the msg_to value can be a comma-separated list of email addresses. Fill in the details as needed. An app registration with application permissions for Reports.Read.All is required.

{
    "azure_tenant_id":"",
    "azure_client_id":"",
    "azure_client_secret":"",
    "msg_from":"",
    "msg_to":"",
    "smtp_srv":"",
    "smtp_port":25
}

And the script looks like this:

import csv
import json
import msal
import re
import requests
import isodate
import os
from os.path import dirname, abspath
import datetime
from datetime import datetime, timedelta
import sys
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email.mime.text import MIMEText
from email import encoders

# Load config globally
try:
    config_file_path = "config.json"
    config = json.load(open(config_file_path))
except Exception as e:
    print(f"Failed to load config file from {config_file_path}.\r\nException: {e}")
    sys.exit(1)


expiration = datetime.now()
token = {}

# Get access token
def get_token():

    global expiration
    global token

    if datetime.now() > expiration or token == {}:
        tenant = config['azure_tenant_id']
        authority_url = f"https://login.microsoftonline.com/{tenant}"
        client_id = config['azure_client_id']
        client_secret = config['azure_client_secret']
        scope = ['https://graph.microsoft.com/.default']

        client = msal.ConfidentialClientApplication(client_id, authority=authority_url, client_credential=client_secret)
        token_result = client.acquire_token_for_client(scopes=scope)

        if 'access_token' in token_result:
            expiration = datetime.now() + timedelta(seconds=token_result['expires_in']) - timedelta(seconds=60)
            return token_result['access_token']
        else:
            return False
    else:
        return token

def duration_to_minutes(duration_str):
    duration = isodate.parse_duration(duration_str)
    return round(duration.total_seconds()/60)


def send_email(sender_email, recipient_email, subject, body, attachment_filename):
    message = MIMEMultipart()
    message['From'] = sender_email
    message['To'] = recipient_email
    message['Subject'] = subject

    message.attach(MIMEText(body, 'plain'))

    with open(attachment_filename, 'rb') as attachment:
        part = MIMEBase('application', 'octet-stream')
        part.set_payload(attachment.read())

    encoders.encode_base64(part)

    part.add_header(
        "Content-Disposition",
        f"attachment; filename= {os.path.basename(attachment_filename)}",
    )
    message.attach(part)

    with smtplib.SMTP(config['smtp_srv'], config['smtp_port']) as server:
        server.sendmail(sender_email, recipient_email.split(','), message.as_string())


def main():

    current_date = datetime.now()
    first_day_of_current_month = current_date.replace(day=1)
    last_day_of_previous_month = first_day_of_current_month - timedelta(days=1)
    previous_month_name = last_day_of_previous_month.strftime('%B')
    previous_month_year = last_day_of_previous_month.strftime('%Y')
    month_string = f"{previous_month_name}-{previous_month_year}"

    input_file_name = f"/etc/ansible/scripts/reports/teams_usage.csv"
    file_name = f"/etc/ansible/scripts/reports/teams_usage_{month_string}.csv"

    headers = {
        'Authorization' : get_token()
    }
    url = f"https://graph.microsoft.com/v1.0/reports/getTeamsUserActivityCounts(period='D30')"
    response = requests.get(url=url, headers=headers)

    if response.status_code == 200:

        with open(input_file_name, 'w', newline='') as f:
            f.write(response.text)
        f.close()

        with open(input_file_name, 'r') as input_file, open(file_name, 'w', newline='') as output_file:
            data = csv.DictReader(input_file)
            fields = data.fieldnames

            writer = csv.DictWriter(output_file, fieldnames=fields)
            writer.writeheader()

            for row in data:
                row['Audio Duration'] = duration_to_minutes(row['Audio Duration'])
                row['Video Duration'] = duration_to_minutes(row['Video Duration'])
                row['Screen Share Duration'] = duration_to_minutes(row['Screen Share Duration'])

                writer.writerow(row)

        input_file.close()
        output_file.close()


        send_email(sender_email=config['msg_from'], 
                    recipient_email=config['msg_to'], 
                    subject=f"Teams Usage Report for {month_string}",
                    body=f"Teams usage report for {month_string} attached.",
                    attachment_filename=file_name)


        os.remove(file_name)
        os.remove(input_file_name)

    else:

        print("Failed to retrieve Teams Usage report")


if __name__ == '__main__':
    main()