Windows Store Analytics (Paid-, Prepaid code data) to Plausible Analytics

windows_store_plausible.py

##############################################
# Windows Store Analytics to Plausible Event #
#               Author: Bostrot              #
##############################################

from typing import Text
import requests
import json

urlRequestToken = 'https://login.microsoftonline.com/TENANT_ID/oauth2/token'
urlAnalytics = 'https://manage.devcenter.microsoft.com/v1.0/my/analytics/appacquisitions?applicationId=APP_ID&startDate=2021-08-08&groupby=acquisitionType'
urlAnalyticsPage = 'https://manage.devcenter.microsoft.com/v1.0/my/analytics/appchannelconversions?applicationId=APP_ID&startDate=2021-08-08'
urlPostAnalytics = 'https://plausible.io/api/event'
fdata = ''
headers = ''
analyticsData = {
    'total': 0.0,
    'totalCustomers': 0,
    'totalCodes': 0,
    'clickCount': 0,
}
headersAnalytics = {
    'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/85.0.4183.121 Safari/537.36 OPR/71.0.3770.284',
    'Content-Type': 'application/json',
    'X-Forwarded-For': '127.0.0.1',
}
bodyAnalytics = {
    'name': 'paid',
    'url': 'app://localhost/installs',
    'domain': 'YOURDOMAIN',
}

def getToken() -> object:
    body = {
        # AZURE AD API KEY
        'grant_type': 'client_credentials',
        'client_id': 'CLIENT_ID',
        'client_secret': 'CLIENT_SECRET',
        'resource': 'AZURE_APP_ID',
    }

    tokenRequest = requests.post(urlRequestToken, data=body)

    if tokenRequest.status_code == 200:
        token = json.loads(tokenRequest.text)['access_token']

    headers = {'Authorization': 'Bearer: ' + token}
    return headers

def getStoreData(headers):
    analyticsRequest = requests.get(urlAnalytics, headers=headers)

    if analyticsRequest.status_code == 200:
        data = json.loads(analyticsRequest.text)
        for item in data['Value']:
            if item['acquisitionType'] == 'Paid':
                analyticsData['totalCustomers'] += item['acquisitionQuantity']
            if item['acquisitionType'] == 'Prepaid Code':
                analyticsData['totalCodes'] += item['acquisitionQuantity']
            analyticsData['total'] += item['purchasePriceUSDAmount']

def postEvent(name):
    bodyAnalytics['name'] = name
    requestPostAnalytics = requests.post(
        urlPostAnalytics, json=bodyAnalytics, headers=headersAnalytics)

def openCache() -> Text:
    f = open('cache.json', 'r')
    fdata = f.read()
    f.close
    return fdata

def sendStoreEvents(fdata):
    try:
        fdata = json.loads(fdata)
        diffCustomers = analyticsData['totalCustomers'] - \
            fdata['totalCustomers']
        diffCodes = analyticsData['totalCodes'] - fdata['totalCodes']
        if diffCustomers > 0:
            # write analytics
            for i in range(diffCustomers):
                postEvent('paid')
        if diffCodes > 0:
            # write analytics
            for i in range(diffCodes):
                postEvent('code')
    except:
        for i in range(analyticsData['totalCustomers']):
            postEvent('paid')
        for i in range(analyticsData['totalCodes']):
            postEvent('code')
        pass


def getAnalyticsStorefront(headers) -> bool:
    analyticsRequest = requests.get(urlAnalyticsPage, headers=headers)

    if analyticsRequest.status_code == 200:
        dataClicks = json.loads(analyticsRequest.text)
        for item in dataClicks['Value']:
            analyticsData['clickCount'] += item['clickCount']
        return True
    return False


def postEventStorefront(fdata):
    bodyAnalytics['domain'] = 'app://localhost/storefront'
    try:
        fdata = json.loads(fdata)
        diffClicks = analyticsData['clickCount'] - fdata['clickCount']
        if diffClicks > 0:
            # write analytics
            for i in range(diffClicks):
                postEvent('pageview')
    except:
        for i in range(analyticsData['clickCount']):
            postEvent('pageview')
        pass


def cache():
    f = open('cache.json', 'w')
    f.write(json.dumps(analyticsData))
    f.close

def main() -> int:
    headers = getToken()
    getStoreData(headers)
    data = openCache()
    sendStoreEvents(data)
    gotStoreData = getAnalyticsStorefront(headers)
    if (gotStoreData):
        postEventStorefront(data)
    cache()
    return 0

if __name__ == "__main__":
    main()