70 lines
1.9 KiB
Python
70 lines
1.9 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
||
|
import sys
|
||
|
import os
|
||
|
import re
|
||
|
import json
|
||
|
import mimetypes
|
||
|
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()}
|
||
|
mime = headers.get("content-type")
|
||
|
if not mime.startswith("image/"):
|
||
|
raise Exception(f"Got response with MIME type {mime}, not an image")
|
||
|
extension = mimetypes.guess_extension(mime)
|
||
|
|
||
|
filepath = f"{destination_dir}/{name}{extension}"
|
||
|
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()
|