Script to send Discord version and module updates to webhooks

notifier.py

# MIT License

# Copyright(c) 2021 Austin Whyte

# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files(the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:

# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.

# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

import aiohttp
import asyncio
import re
import os
import time
import json
import functools

# Put your webhooks here
# Format: ('url', ['events']) Legacy behavior is ['canary', 'modules_canary']
# Available subscriptions: canary, ptb, stable, modules_canary, modules_ptb, modules_stable
WEBHOOK_URLS = [
    ('https://canary.discord.com/api/webhooks/XXXXXXXXXXXXXXXXXX/YYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYY',
    ['canary', 'modules_canary']),
    ('https://canary.discord.com/api/webhooks/XXXXXXXXXXXXXXXXXX/YYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYY',
    ['ptb', 'modules_ptb']),
    ('https://canary.discord.com/api/webhooks/XXXXXXXXXXXXXXXXXX/YYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYY',
    ['stable', 'modules_stable']),
]

raw_json = {}
async def send_webhook_message(url, message):
    async with aiohttp.ClientSession() as session:
        current_time = time.strftime('%H:%M')
        to_send = f'[`{current_time}`] {message}'
        if len(to_send) > 1990:
            print("Sending in chunks")
            chunk = ''
            for line in to_send.splitlines():
                if len(chunk + line) > 1990:
                    await session.post(url, data={
                        'content': chunk
                    })
                    chunk = ''
                else:
                    chunk += f'{line}\n'
            if chunk != '':
                await session.post(url, data={
                    'content': chunk
                })
        else:
            await session.post(url, data={
                'content': to_send
            })

async def get_app_version(branch, platform):
    url = f'https://canary.discord.com/api/updates/{branch}?platform={platform}'
    print(f'Getting app version for {branch}/{platform} via {url}')

    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            return (await resp.json())['name']

async def get_app_build_id(branch):
    if branch == 'stable':
        branch = ''
    else:
        branch = f'{branch}.'
    
    async with aiohttp.ClientSession() as session:
        async with session.head(f'https://{branch}discord.com/app') as resp:
            return resp.headers['X-Build-Id']

async def get_module_versions(branch, platform, version):
    url = f'https://canary.discord.com/api/modules/{branch}/versions.json?host_version={version}&platform={platform}'

    print(f'Getting module versions via {url}')
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            return await resp.json()

async def get_build_number(branch='canary'):
    if branch == 'stable':
        branch = ''
    else:
        branch = f'{branch}.'
    print(f'Determining build number for branch {branch}')
    async with aiohttp.ClientSession() as session:
        async with session.get(f'https://{branch}discord.com/app') as resp:
            body = await resp.read()
            decoded = body.decode('utf-8')
            regex = re.compile(r'Build Number: (?:\"\).concat\(\")?(\d+)')
            assets_regex = re.compile(r'<script src="(\/assets/[A-Za-z0-9]+.js)"')
            for match in assets_regex.findall(decoded):
                url = f'https://{branch}discord.com{match}'
                print(f'Trying {url}')
                async with session.get(url) as r1:
                    build = (await r1.read()).decode('utf-8')
                    matches = regex.findall(build)
                    if matches:
                        print(f'Found build number {matches[0]}')
                        return matches[0]
            print('Exhausted all assets files and didn\'t find a match')
            return 'Unknown'

branch_translations = {
    'canary': 'Canary',
    'stable': 'Stable',
    'ptb': 'PTB'
}
async def app_version(branch):
    branch_json = raw_json.get(branch, {})

    current_build_id = await get_app_build_id(branch)
    stored_build_id = branch_json.get('build_id', None)
    stored_build_number = branch_json.get('build_number', 0)

    msg = ''
    if current_build_id != stored_build_id:
        current_build_number = await get_build_number(branch)
        msg += f'New {branch_translations.get(branch, branch)} Build: {current_build_id} ({current_build_number})'
        if current_build_number != 'Unknown' and stored_build_number != 'Unknown' and int(current_build_number) < int(stored_build_number):
            msg += f'\n\t:warning: BUILD NUMBER WENT BACKWARDS ({stored_build_number} -> {current_build_number}) :warning:'
    else:
        current_build_number = stored_build_number
    return (msg, current_build_number, current_build_id)

async def module_versions(branch, platforms = ['osx', 'win', 'linux']):
    branch_json = raw_json.get(branch, {})

    msg = ''
    all_modules = {}
    branch_modules = branch_json.get('modules', {})
    for platform in platforms:
        platform_msg = ''
        current_appver = await get_app_version(branch, platform)
        modules = branch_modules.get(platform, {})
        stored_appver = modules.get('_version', None)
        print(f'Stored {branch}/{platform}: {stored_appver}')

        if stored_appver != current_appver:
            platform_msg += f'New app version: {current_appver}\n\n'
            modules = {} # Reset modules so everything is  new
        
        module_versions = await get_module_versions(branch, platform, current_appver)
        module_names = list(modules.keys())
        for module, version in module_versions.items():
            stored_module_version = modules.get(module, None)
            if stored_module_version is None:
                platform_msg += f'New module: **{module}**: `{version}`\n'
            elif version != stored_module_version:
                platform_msg += f'Module update **{module}**: `{version}`\n'
                if int(version) < int(stored_module_version):
                    platform_msg += f'\t:warning: Module version went backwards ({stored_module_version} -> {version}) :warning:\n'
            if module in module_names: module_names.remove(module)
        
        for removed_module in module_names:
            if removed_module == '_version':
                continue
            platform_msg += f'Module removed **{removed_module}**\n'
            if removed_module in modules: del modules[removed_module]
        module_versions['_version'] = current_appver
        all_modules[platform] = module_versions

        if platform_msg != '':
            msg += f'\n{platform} {branch}\n----------\n{platform_msg}'
    return (msg, all_modules)

function_map = {
    'canary': functools.partial(app_version, 'canary'),
    'ptb': functools.partial(app_version, 'ptb'),
    'stable': functools.partial(app_version, 'stable'),
    'modules_canary': functools.partial(module_versions, 'canary'),
    'modules_ptb': functools.partial(module_versions, 'ptb'),
    'modules_stable': functools.partial(module_versions, 'stable')
}

def save_build_info(branch, build_number, build_id):
    platform_json = raw_json.get(branch, {})
    platform_json['build_number'] = build_number
    platform_json['build_id'] = build_id
    raw_json[branch] = platform_json

def save_module_info(branch, modules):
    print(f"Saving module info {modules}")
    branch_json = raw_json.get(branch, {})
    branch_json['modules'] = modules
    raw_json[branch] = branch_json

async def main():
    with open('data.json', 'r') as f:
        global raw_json
        raw_json = json.loads(''.join(f.readlines()))

    results = {}
    for key, function in function_map.items():
        results[key] = await function()

    print(results)
    
    for webhook, subscriptions in WEBHOOK_URLS:
        webhook_msg = ''
        for sub in subscriptions:
            msg = results.get(sub, None)
            to_append = msg[0] if msg is not None else f"Unknown sub {sub}"
            if to_append != '':
                webhook_msg += f'{to_append}\n'
        print(f"Sending: {webhook_msg}")

        if webhook_msg != '':
            await send_webhook_message(webhook, webhook_msg)
    
    save_build_info('canary', results['canary'][1], results['canary'][2])
    save_build_info('ptb', results['ptb'][1], results['ptb'][2])
    save_build_info('stable', results['stable'][1], results['stable'][2])

    save_module_info('canary', results['modules_canary'][1])
    save_module_info('ptb', results['modules_ptb'][1])
    save_module_info('stable', results['modules_stable'][1])

    with open('data.json', 'w') as f:
        f.write(json.dumps(raw_json))

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())