telegram-apk-downloader/src/download_utils.py

import os
import re

from glob import glob, escape
from contextlib import contextmanager
from subprocess import check_call
from urllib.parse import urlparse, parse_qs

from gplaycli.gplaycli import GPlaycli

class BadPackageNameException(Exception):
    pass

class dotdict(dict):
    """dot.notation access to dictionary attributes"""
    __getattr__ = dict.get
    __setattr__ = dict.__setitem__
    __delattr__ = dict.__delitem__

def validate_package_name(package_name):
    # https://developer.android.com/studio/build/application-id
    """
    - It must have at least two segments (one or more dots).
    - Each segment must start with a letter.
    - All characters must be alphanumeric or an underscore [a-zA-Z0-9_].
    """
    reg = r"^(?:[a-zA-Z]+(?:\d*[a-zA-Z_]*)*)(?:\.[a-zA-Z]+(?:\d*[a-zA-Z_]*)*)+$"
    return re.match(reg, package_name) is not None

class APK:
    def __init__(self, apk, conf={}, conf_file=None):
        self.conf = conf
        self.conf_file = conf_file
        self.package_name = ''
        if 'play.google.com/store/apps/details' in apk:
            url = urlparse(apk)
            if url.netloc == 'play.google.com':
                self.package_name = parse_qs(url.query)['id'][0]
        else:
            self.package_name = apk

    def download(self):
        if not validate_package_name(self.package_name):
            raise BadPackageNameException(f'error downloading {self.package_name}')

        cli = GPlaycli(args=dotdict(self.conf), config_file=self.conf_file)
        cli.download_folder = 'out'
        d = cli.download([self.package_name])  # returns the number of downloaded packages
        if len(d) == 0:
            raise BadPackageNameException(f'error downloading {self.package_name}')

        fname = os.path.join('out', self.package_name + '.apk')
        if os.path.isfile(fname):
            self.file_name = fname

    def check_dimension(self):
        if os.path.getsize(self.file_name) > 50 * 1024 * 1023:
            check_call(['split', '-b', '49M', self.file_name, self.file_name])
            os.remove(self.file_name)
        return glob(escape(self.file_name) + '*')

    @contextmanager
    def send(self):
        files = self.check_dimension()  # split if size >= 50MB
        yield files
        for f in files:  # removing old files
            os.remove(f)