minitch/lib/handlers.py

import os.path
import hashlib
from typing import Literal
import requests
from lib import itch
from prettytable import PrettyTable
from tqdm import tqdm


class HandlerException(Exception):
    pass

class GameHandler:

    OUTPUT_FORMATS = Literal['table', 'json']
    OS_TYPES = Literal['linux', 'osx', 'windows']

    def __init__(self, api_key, output_format: OUTPUT_FORMATS ='table'):
        self.__itch_client = itch.Client(api_key)
        self.__output_format = output_format

    def __print_table(self, table: PrettyTable):
        if self.__output_format == 'table':
            return print(table)
        if self.__output_format == 'json':
            return print(table.get_json_string())
    
    def find(self, query: str):
        games = self.__itch_client.find_game(query)

        table = PrettyTable(header=True, align='l', autowrap=False)
        table.field_names = ['ID', 'Title', 'Description', 'URL', 'Purchaseable', 'Min price', 'Linux', 'Windows', 'OSX']

        for game in games:
            table.add_row([
                game['id'],
                game['title'] if 'title' in game else '-',
                game['short_text'] if 'short_text' in game else '-',
                game['url'] if 'url' in game else '-',
                'yes' if game['can_be_bought'] == True else 'no',
                float(game['min_price']) / 100,
                'yes' if game['p_linux'] == True else 'no',
                'yes' if game['p_windows'] == True else 'no',
                'yes' if game['p_osx'] == True else 'no'
            ])
        self.__print_table(table)
    
    def uploads(self, game_id: int):
        uploads = self.__itch_client.game_uploads(game_id)

        table = PrettyTable(header=True, align='l', autowrap=False)
        table.field_names = ['ID', 'File Name', 'Last Update', 'Size', 'MD5', 'Linux', 'Windows', 'OSX']

        for upload in uploads:
            table.add_row([
                upload['id'],
                upload['filename'] if 'filename' in upload else '-',
                upload['updated_at'] if 'updated_at' in upload else '-',
                upload['size'] if 'size' in upload else '-',
                upload['md5_hash'] if 'md5_hash' in upload else '-',
                'yes' if upload['p_linux'] == True else 'no',
                'yes' if upload['p_windows'] == True else 'no',
                'yes' if upload['p_osx'] == True else 'no'
            ])
        self.__print_table(table)
    
    def download(self, game_id: int, dst_dir: str, os_type: OS_TYPES = 'linux'):
        meta = self.__itch_client.game_meta(game_id)
        uploads = self.__itch_client.game_uploads(game_id)
        for upload in uploads:
            if upload[f'p_{os_type}'] == True:
                url = self.__itch_client.game_url(upload['id'])
                dst = os.path.join(dst_dir, upload['filename'])
                
                # Download game
                md5 = hashlib.md5()

                with requests.get(url, stream=True) as r:
                    r.raise_for_status()
                    total_size = int(r.headers.get('content-length', 0))
                    with tqdm.wrapattr(open(dst, 'wb'), 'write', miniters=1, desc='Downloading game', total=total_size) as f:
                        for chunk in r.iter_content(chunk_size=8192):
                            f.write(chunk)
                            md5.update(chunk)
                if 'md5_hash' in upload and md5.hexdigest() != upload['md5_hash']:
                    raise HandlerException('Invalid checksum')
                
                return (dst, upload, meta)
        raise HandlerException('No valid upload found')

    def list_local(self, games):
        table = PrettyTable(header=True, align='l', autowrap=False)
        table.field_names = ['ID', 'Title', 'Description', 'URL', 'Last update', 'Installation path']

        for id, game in games.items():
            table.add_row([
                id,
                game['title'],
                game['description'],
                game['url'],
                game['last_update'],
                game['path']
            ])
        self.__print_table(table)
    
    def check_update(self, game_id, last_update, os_type: OS_TYPES = 'linux'):
        uploads = self.__itch_client.game_uploads(game_id)
        for upload in uploads:
            if upload[f'p_{os_type}'] == True:
                return upload['updated_at'] != last_update