launcher/launcher.py

import io
import json
import mimetypes
import os
import shutil
import subprocess
import threading
from pathlib import Path
from shutil import rmtree
from subprocess import CREATE_NO_WINDOW, Popen

import requests
import webview

from launcher import config
from log import logger

mimetypes.add_type("text/html", ".html")
mimetypes.add_type("text/css", ".css")
mimetypes.add_type("application/javascript", ".js")

version = "v0.3"

get_new_version_url = (
    "https://git.zhaozuohong.vip/api/v1/repos/mower-ng/launcher/releases/latest"
)

upgrade_script_name = "upgrade.bat"


def custom_event(data):
    data = json.dumps({"log": data})
    js = f"var event = new CustomEvent('log', {{detail: {data}}}); window.dispatchEvent(event);"
    window.evaluate_js(js)


mirror_list = {
    "pypi": "https://pypi.org/simple",

    "aliyun": "https://mirrors.aliyun.com/pypi/simple/",
    "tuna": "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple",
    "sjtu": "https://mirror.sjtu.edu.cn/pypi/web/simple",
}

command_list = {
    "lfs": "git\\bin\\git lfs install",
    "ensurepip": "python\\python -m ensurepip --default-pip",
    "clone": "git\\bin\\git -c lfs.concurrenttransfers=100 clone https://git.zhaozuohong.vip/mower-ng/mower-ng.git --branch slow",
    "fetch": lambda: f"..\\git\\bin\\git fetch origin {config.conf.branch} --progress",
    "switch": lambda: f"..\\git\\bin\\git -c lfs.concurrenttransfers=100 switch -f {config.conf.branch} --progress",
    "reset": lambda: f"..\\git\\bin\\git -c lfs.concurrenttransfers=200 reset --hard origin/{config.conf.branch}",
    "pip_install": lambda: f"..\\python\\Scripts\\pip install --no-cache-dir -i {mirror_list[config.conf.mirror]} -r requirements.txt --no-warn-script-location",
    "webview": "start ..\\python\\pythonw webview_ui.py",
    "manager": "start ..\\python\\pythonw manager.py",
}


def read_stream(stream, log_func):
    text_io = io.TextIOWrapper(stream, encoding='utf-8', errors='replace')
    try:
        for line in iter(text_io.readline, ''):
            text = line.rstrip('\n')
            log_func(f"log: {text}")
            custom_event(text.strip() + "\n")
    finally:
        text_io.close()


class Api:

    def load_config(self):
        logger.info("读取配置文件")
        return config.conf.model_dump()

    def save_config(self, conf):
        logger.info(f"更新配置文件{conf}")
        config.conf = config.Conf(**conf)
        config.save_conf()

    def get_version(self):
        return version

    def get_new_version(self):
        logger.info("获取最新版本号")
        response = requests.get(get_new_version_url)
        return response.json()

    # 更新启动器本身
    def update_self(self, download_url):
        # 下载压缩包的全路径
        download_path = os.path.join(os.getcwd(), os.path.basename(download_url))
        logger.info(f"下载新版本: {download_url}到{download_path}")
        response = requests.get(download_url, stream=True)
        if response.status_code == 200:
            with open(download_path, "wb") as file:
                shutil.copyfileobj(response.raw, file)
                logger.info("下载完成")
        else:
            logger.error(f"下载新版本失败: {response.status_code}")
            return f"下载新版本失败: {response.status_code}"

        script_path = os.path.join(os.getcwd(), upgrade_script_name)
        folder_path = os.path.join(os.getcwd(), "_internal")
        exe_path = os.path.join(os.getcwd(), "launcher.exe")
        with open(script_path, "w") as b:
            TempList = f"@echo off\n"
            TempList += f"timeout /t 3 /nobreak\n"  # 等待进程退出
            TempList += f"rmdir {folder_path}\n"  # 删除_internal
            TempList += f"del {exe_path}\n"  # 删除exe
            TempList += f"tar -xf {download_path} -C ..\n"  # 解压压缩包
            TempList += f"timeout /t 1 /nobreak\n"  # 等待解压
            TempList += f"start {exe_path}\n"  # 启动新程序
            TempList += f"del {download_path}\n"  # 删除压缩包
            TempList += f"exit"
            b.write(TempList)
        # 不显示cmd窗口
        Popen([script_path], creationflags=CREATE_NO_WINDOW)
        os._exit(0)

    def rm_site_packages(self):
        site_packages_path = Path("./python/Lib/site-packages")
        if site_packages_path.exists():
            rmtree(site_packages_path)
            return "site-packages目录移除成功"
        return "python\\Lib\\site-packages目录不存在"

    def rm_python_scripts(self):
        python_scripts_path = Path("./python/Scripts")
        if python_scripts_path.exists():
            rmtree(python_scripts_path)
            return "Scripts目录移除成功"
        return "python\\Scripts目录不存在"

    def run(self, command, cwd=None):
        command = command_list[command]
        if callable(command):
            command = command()
        custom_event(command + "\n")
        logger.info(f"cwd {cwd}")
        logger.info(f"command {command}")
        try:
            with subprocess.Popen(
                    command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True, cwd=cwd, bufsize=0,
                    universal_newlines=False
            ) as p:
                stdout_thread = threading.Thread(target=read_stream, args=(p.stdout, logger.info))
                stderr_thread = threading.Thread(target=read_stream, args=(p.stderr, logger.info))

                stdout_thread.start()
                stderr_thread.start()

                stdout_thread.join()
                stderr_thread.join()
            if p.returncode == 0:
                return "success"
        except Exception as e:
            logger.error(f"command {command} 执行异常{e}")
            logger.exception(e)
            custom_event(str(e))
        return "failed"


# 如果当前路径存在更新脚本，则删除
if Path(upgrade_script_name).exists():
    os.remove(upgrade_script_name)

# url = "ui/dist/index.html"
url = "http://localhost:5173/"
window = webview.create_window(
    f"mower-ng launcher {version}", url, js_api=Api()
)
webview.start()