mattermost-channels/get_custom_emoji.py

#!/usr/bin/env python3

import sys
import os
import re
import json
import urllib.request


destination_dir = "dist/emoji/images"

os.chdir(os.path.dirname(__file__))
os.makedirs(destination_dir, exist_ok=True)

user_agent = "emoji crawler by the MI5 Delegation for Guarding of Access to Resource Development"

MM_SERVER = os.environ["MM_SERVER"]
MM_ACCESSTOKEN = os.environ["MM_ACCESSTOKEN"]


def mm_api_get(path):
	req = urllib.request.Request(f"https://{MM_SERVER}/api{path}")
	req.add_header("Referer", f"https://{MM_SERVER}/")
	req.add_header("User-Agent", user_agent)
	req.add_header("Authorization", "Bearer " + MM_ACCESSTOKEN)
	return urllib.request.urlopen(req)

def mm_api_get_json(*args, **kwargs):
	with mm_api_get(*args, **kwargs) as response:
		return json.load(response)


def get_list_of_custom_emoji():
	page = 0
	per_page = 200 # Maximum allowed in API
	response = []
	while page == 0 or response:
		response = mm_api_get_json(f"/v4/emoji?page={page}&per_page={per_page}&sort=name")
		for emoji in response:
			yield(emoji)
		page += 1


def get_emoji_image(name, emoji_id):
	print(f"Downloading {name}", file=sys.stderr, flush=True)
	with mm_api_get(f"/v4/emoji/{emoji_id}/image") as upstream_response:
		headers = {k.lower(): v for (k, v) in upstream_response.headers.items()}

		filepath = f"{destination_dir}/{name}"
		with open(filepath, "wb") as fh:
			chunk = upstream_response.read1()
			while chunk != b"":
				fh.write(chunk)
				chunk = upstream_response.read1()
		print(f"Saved to {filepath}", file=sys.stderr, flush=True)


def main():
	for emoji in get_list_of_custom_emoji():
		get_emoji_image(emoji["name"], emoji["id"])


if __name__ == "__main__":
	main()