Untitled

import json
import logging
import os
import pathlib
import tarfile
import urllib.parse
import plistlib
 
from xml.etree import ElementTree
from argparse import ArgumentParser
 
 
def _update_url(url, url_prefix):
    parsed = urllib.parse.urlparse(url.strip())
    url_postfix = parsed._replace(netloc="", scheme="").geturl()
 
    d = "downloads"
    if not url_prefix.endswith("/"):
        d = "/" + d
    if not url_postfix.startswith("/"):
        d = d + "/"
 
    return url_prefix + d + url_postfix
 
def _update_url_mac(url, url_prefix, download_url):
    parsed = urllib.parse.urlparse(url.strip())
    url_postfix = parsed._replace(netloc="", scheme="").geturl()
 
    d = "downloads"
    if not url_prefix.endswith("/"):
        d = "/" + d
    if not url_postfix.startswith("/"):
        d = d + "/"
 
    #return url_prefix + d + url_postfix
    return url_prefix + download_url + '/myteam.dmg'
 
def _update_url_in_plist(url, url_prefix, full_version):
    parsed = urllib.parse.urlparse(url.strip())
    url_postfix = parsed._replace(netloc="", scheme="").geturl()
    url_postfix = os.path.join(*url_postfix.split(full_version, 1)[1:])
 
    d = "downloads/ios/" + full_version
    if not url_prefix.endswith("/"):
        d = "/" + d
    if not url_postfix.startswith("/"):
        d = d + "/"
 
    return url_prefix + d + url_postfix
 
 
def update_mac_dist(myteam_app_config_path, url_prefix, download_url):
    version_file_name = "version.xml"
    version_file_path = os.path.join(os.path.dirname(myteam_app_config_path), version_file_name)
 
    namespaces = dict([
        node for _, node in ElementTree.iterparse(version_file_path, events=["start-ns"])
    ])
 
    et = ElementTree.parse(version_file_path)
 
    for k, v in namespaces.items():
        ElementTree.register_namespace(k, v)
 
    for link_obj in et.findall("channel/item/sparkle:releaseNotesLink", namespaces):
        link_obj.text = _update_url_mac(link_obj.text, url_prefix, download_url)
 
    enclosure_tag = et.find("channel/item/enclosure")
    enclosure_tag.set("url", _update_url_mac(enclosure_tag.get("url"), url_prefix, download_url))
    et.write(version_file_path, "UTF-8", True)
 
 
def update_ios_dist(myteam_app_config_path, url_prefix, full_version):
    plist_file_name = "ios_manifest.plist"
    plist_file_path = os.path.join(os.path.dirname(myteam_app_config_path), plist_file_name)
 
    with open(plist_file_path, "rb+") as plist_handle:
        plist_data = plistlib.load(plist_handle)
 
        if "items" not in plist_data:
            return
 
        for item in plist_data["items"]:
            if "assets" not in item:
                continue
 
            for asset in item["assets"]:
                if "url" not in asset:
                    continue
 
                asset["url"] = _update_url_in_plist(asset["url"], url_prefix, full_version)
 
        plist_handle.seek(os.SEEK_SET)
        plist_handle.write(plistlib.dumps(plist_data))
        plist_handle.truncate()
 
 
def create_symlink(app_path):
    symlink_tmp_postfix = ".tmp"
    symlink_path_tmp = os.path.join(
        *("/",) + tuple(app_path.rsplit("/")[:-1]) + ("latest" + symlink_tmp_postfix,)
    )
    symlink_path = symlink_path_tmp.replace(symlink_tmp_postfix, "")
 
    if os.path.exists(symlink_path_tmp):
        log.warning(f"Found previously created temporary symlink: {symlink_path_tmp}. Try to remove it.")
        os.remove(symlink_path_tmp)
 
    os.symlink(app_path, symlink_path_tmp)
    os.rename(symlink_path_tmp, symlink_path)
 
    if os.path.realpath(symlink_path) == app_path:
        log.info(f"OK: {symlink_path} -> {app_path}")
    else:
        log.error(f"Fail: {symlink_path} -> os.path.realpath(symlink_path) (expected: {app_path})")
 
 
def deploy(src_apps_dir, dst_apps_dir, url_prefix, client_config_path, external_log = None):
    global log
    if external_log is not None:
        log = external_log
    if not (src_apps_dir and dst_apps_dir):
        return
 
    if not os.path.exists(src_apps_dir):
        log.error(f"{src_apps_dir} does not exists")
        return
 
    pathlib.Path(dst_apps_dir).mkdir(parents=True, exist_ok=True)
 
    myteam_apps_config_path_list = []
    for archive in os.scandir(src_apps_dir):
        if not (archive.is_file() and archive.name.endswith(".tgz")):
            log.warning(f"Unexpected file/directory {archive.path} in source apps directory")
            continue
 
        new_latest_path = ""
        with tarfile.open(archive.path, "r:gz") as tar_handle:
            tar_handle.extractall(path=dst_apps_dir)
 
            myteam_apps_config_path = None
            for tarinfo in tar_handle:
                if not new_latest_path:
                    new_latest_path = os.path.join(dst_apps_dir, tarinfo.name)
 
                if tarinfo.name.endswith("myteam-config.json"):
                    myteam_apps_config_path = os.path.join(dst_apps_dir, tarinfo.name)
 
            if not new_latest_path:
                log.error(f"Can't create new symlink, because {archive.name} has unexpected structure; skipped")
                continue
 
            if myteam_apps_config_path:
                myteam_apps_config_path_list.append(myteam_apps_config_path)
 
            create_symlink(new_latest_path)
 
    apps_config = {}
    for app_config_path in myteam_apps_config_path_list:
        with open(app_config_path, "rb") as config_handle:
            myteam_app_download_config = json.load(config_handle)
            for platform, app_config in myteam_app_download_config.items():
 
                if platform.startswith("mac_"):
                    update_mac_dist(app_config_path, url_prefix, myteam_app_download_config[platform]['url'])
 
                if platform == "ios":
                    full_version = myteam_app_download_config[platform]["full_version"]
                    update_ios_dist(app_config_path, url_prefix, full_version)
 
                if "url" not in app_config:
                    log.error(f"Bad app config format, url not found: {app_config_path}")
                    return
 
                myteam_app_download_config[platform]["url"] = url_prefix + myteam_app_download_config[platform]["url"]
            apps_config.update(myteam_app_download_config)
 
    with open(client_config_path, 'r+') as client_config_handle:
        client_config = json.load(client_config_handle)
        client_config["apps"] = apps_config
        client_config_handle.seek(os.SEEK_SET)
        json.dump(client_config, client_config_handle, indent=2)
        client_config_handle.truncate()
 
 
if __name__ == '__main__':
    parser = ArgumentParser()
    parser.add_argument('--src-apps-dir', dest="src_apps_dir", default=None, help="")
    parser.add_argument('--dst-apps-dir', dest="dst_apps_dir", default=None, help="")
    parser.add_argument("--client-config-path", dest="client_config_path", default=None, help="")
    parser.add_argument("--url-prefix", dest="url_prefix", default=None, help="")
 
    args = parser.parse_args()
    log = logging.getLogger()
    log.setLevel(logging.DEBUG)
    logging.basicConfig(format='%(levelname)-8s [%(filename)s:%(lineno)-5d] %(message)s',
                        datefmt='%Y-%m-%d:%H:%M:%S',
                        level=logging.INFO)
 
    deploy(args.src_apps_dir, args.dst_apps_dir, args.url_prefix, args.client_config_path)