pyinstaller 自动更新版本

自动更新版本流程

1. 检测新版本，弹窗提醒更新
2.下载更新文件压缩包
3.将旧的配置文件移到新配置文件夹中
4.关闭当前进程，移除旧文件夹里面的文件，将新文件移到当前位置
5.启动新程序

# -*- coding: utf-8 -*-
import json
import subprocess
import sys
import textwrap
import threading
# import webbrowser
import zipfile
from pathlib import os
# import os
import requests
import shutil
import semver
import tkinter
import tkinter.messagebox
from tqdm.tk import tqdm
from tkinter import Tk
 
"""
pyinstaller -D unit_test.py -n '软件测试'
"""
 
 
class Unit_Test_Version():
    def __init__(self, root=None):
        # 当前版本号
        self.current_version = '1.0.0'
        # 请求软件资源信息
        self.version_url = 'http://127.0.0.1:28082/search_version'
        # 软件名
        self.app_name = '软件测试'
        # 软件类型(同一个表存多个软件的版本)
        self.soft_type = 4
        self.root = root
 
    def upzip_file(self, zip_path=None, unzip_path=None):
        """
        :zip_path 压缩文件路径
        :unzip_path 解压文件路径
        :return 解压 zip 文件,返回所有解压文件夹下的路径
        """
        zip_file = zipfile.ZipFile(zip_path)
        if not os.path.isdir(unzip_path):
            os.mkdir(unzip_path)
        for names in zip_file.namelist():
            zip_file.extract(names, unzip_path)
        zip_file.close()
        return [os.path.join(unzip_path, i).replace('\\', '/') for i in zip_file.namelist()]
 
    def upzip_file_new(self, zip_path=None, unzip_path=None):
        paths = []
        if not os.path.exists(unzip_path):
            os.mkdir(unzip_path)
        with zipfile.ZipFile(file=zip_path, mode='r') as zf:
            for old_name in zf.namelist():
                file_size = zf.getinfo(old_name).file_size
                # 由于源码遇到中文是cp437方式，所以解码成gbk，windows即可正常
                new_name = old_name.encode('cp437').decode('gbk')
                # 拼接文件的保存路径
                new_path = os.path.join(unzip_path, new_name)
                paths.append(new_path)
                # 判断文件是文件夹还是文件
                if file_size > 0:
                    # 是文件，通过open创建文件，写入数据
                    with open(file=new_path, mode='wb') as f:
                        # zf.read 是读取压缩包里的文件内容
                        f.write(zf.read(old_name))
                else:
                    # 是文件夹，就创建
                    os.mkdir(new_path)
        return paths
 
    def get_version_info(self):
        # 获取版本信息
        try:
            data = json.dumps({'soft_type': self.soft_type})
            res = requests.post(self.version_url, data=data, timeout=15)
            response = json.loads(res.text)
        except:
            response = {}
        return response
 
    def del_current_pid(self):
        pid = os.getpid()  # 关闭后台进程
        try:
            os.system('taskkill -f -pid %s' % pid)
        except:
            pass
 
    def check_version(self, version_info):
        self.root = Tk()
        self.root.withdraw()  # 隐藏主窗口
        t = threading.Thread(target=lambda: self.check_for_update(version_info), name='update_thread')
        t.daemon = True  # 守护为True，设置True线程会随着进程一同关闭
        t.start()
        self.root.mainloop()
 
    # def browser_update(self, ver, window):
    #     webbrowser.open(ver.get('updateUrl'))
    #     window.destroy()
 
    def check_for_update(self, version_info):
        version = version_info['version']
        ver = semver.compare(version, self.current_version)
        if ver:
            # 更新的内容
            publish_notes = version_info['remark']
            message = f'当前版本[{self.current_version}], 有新版本[{version}]\n' \
                      f'更新内容:\n{publish_notes}\n请选择立即去下载更新[确定]，暂不更新[取消]？'
            result = tkinter.messagebox.askokcancel(title='更新提示', message=message)
            if result:
                self.is_down_update = True
                self.down_zip(version_info)
                self.root.destroy()
            else:
                self.root.destroy()
 
    def down_zip(self, version_info):
        zip_url, file_name = version_info['zip_url'], version_info['file_name']
        res = requests.get(zip_url)
        # 当前文件夹路径
        # current_path = os.path.dirname(os.path.realpath(sys.argv[0]))
        # 当前文件夹上一级路径
        current_path = os.path.abspath(os.path.dirname(os.path.dirname(__file__)))
        zip_path = os.path.join(current_path, f'{zip_url.split("/")[-1]}')
        unzip_path = os.path.join(current_path, file_name+'_new')
        print('zip_path:', zip_path)
        print('unzip_path:', unzip_path)
        # with open(zip_path, 'wb') as f:
        #     f.write(res.content)
 
        chunk_size = 1024 * 100
        total_size = int(res.headers.get('content-length', 0))
        progress_bar = tqdm(iterable=res.iter_content(chunk_size=chunk_size), tk_parent=None,
                            leave=False, total=total_size, unit='B', unit_scale=True)
        # 设置下载进度条位置
        screenWidth = progress_bar._tk_window.winfo_screenwidth()
        screenHeight = progress_bar._tk_window.winfo_screenheight()
        progress_bar._tk_window.geometry(f"300x160+{screenWidth - 300}+{screenHeight - 400}")
        try:
            with open(zip_path, 'wb') as f:
                for chunk in res.iter_content(chunk_size):
                    if chunk:
                        f.write(chunk)
                        progress_bar.update(len(chunk))
        except Exception as ee:
            progress_bar.close()
            return '下载异常，检查链接'
        progress_bar.close()
        filepath = self.upzip_file_new(zip_path, unzip_path)[0]
        print('filepath:', filepath)
        # 删除压缩包
        if os.path.exists(zip_path):
            os.remove(zip_path)
        # 老版本配置文件
        old_config = os.path.join(current_path, 'config', 'config.ini')
        # 新版本的配置文件路径
        new_config = os.path.join(filepath, '客户端配置文件.ini')
        print('new_config:', new_config)
        print('old_config:', old_config)
        if os.path.exists(new_config) and os.path.exists(old_config):
            os.remove(new_config)
        if os.path.exists(old_config):
            if not os.path.exists(os.path.join(filepath, 'config')):
                os.makedirs(os.path.join(filepath, 'config'))
            shutil.copy(old_config, new_config)
        self.auto_update_setup(filepath)
 
    def make_updater_bat(self, filepath):
        filepath = os.path.abspath(filepath)
        # 当前进程ID
        pid = os.getpid()
        current_path = os.path.dirname(os.path.realpath(sys.argv[0]))
        print('current_path:', current_path)
        # 删除临时文件夹路径
        new_file_dir = os.path.abspath(os.path.join(filepath, ".."))
 
        # old_file = filepath.replace(new_file_dir.replace('\\', '/'), current_path.replace('\\', '/'))  # 旧的文件夹
        # old_file = os.path.abspath(os.path.dirname(os.path.dirname(__file__)))
        old_file = current_path
 
        old_file_dir = os.path.abspath(os.path.dirname(os.path.dirname(__file__)))  # 上一级文件夹
        new_file = filepath      # 新版本文件夹
 
        # 运行exe路径
        # new_file_exe = os.path.join(filepath, f'{self.app_name}.exe').replace('\\', '/')
        # new_file_exe = new_file_exe.replace(new_file_dir.replace('\\', '/'), old_file_dir.replace('\\', '/'))
        new_file_exe = os.path.join(current_path, f'{self.app_name}.exe').replace('\\', '/')
 
        print('filepath:', filepath)
        print('new_file_dir:', new_file_dir.replace('\\', '/'))
        print('current_path:', current_path.replace('\\', '/'))
        # 自动更新脚本
        """
        关闭当前进程，删除所有文件除了updater.bat，将新版本下的所有文件移到当前文件夹下，访问新exe文件
        """
        bat_name = 'updater.bat'
        bat_path = os.path.join(old_file, bat_name)
        print('bat_path:', bat_path)
        
        with open(bat_path, 'w', encoding='gbk') as updater:
            updater.write(textwrap.dedent(f'''\
        @echo off
        echo 正在更新[{self.app_name}]最新版本，请勿关闭窗口...
        ping -n 2 127.0.0.1
        taskkill -f -pid {pid}
        echo 正在复制[{self.app_name}]，请勿关闭窗口...
        for /D %%a in (*) do (
    　　    set str1="%%a"
            if "!str1!" == "!str1:{old_filename}=!" if "!str1!" == "!str1:{old_filename}=!" (
    　　       echo 删除不包含old_file的文件夹： %%a
               move /y "%%a" %{old_path}%
    　　    )
        )
        for %%i in (*) DO (
            if not %%i=={bat_name} if not %%i=={old_filename}(
            echo 删除文件：%%i 
            :: del /s /q "%%i"
            move /y "%%i" %{old_path}%
            )
        )
        for %%a in (*) do (
    　　    set str1=%%a
            if "!str1!" == "!str1:{bat_name}=!" if "!str1!" == "!str1:{bat_name}=!" (
    　　       echo 删除不包含bat的文件： %%a
               move /y "%%a" %{old_path}%
    　　    )
        )
        ping -n 2 127.0.0.1
        xcopy /s /e /y /q "{new_file}\*" "{current_path}"
        ping -n 2 127.0.0.1
        if exist "{new_file_dir}" rd /s /q "{new_file_dir}"
        echo 更新完成，等待自动启动{self.app_name}...
        ping -n 2 127.0.0.1
        "{new_file_exe}"
        exit
        '''))
        updater.flush()
        
        return bat_path
 
    def auto_update_setup(self, filepath):
        # 自动更新
        bat_path = self.make_updater_bat(filepath)
        subprocess.Popen(bat_path, encoding="gbk", shell=True)
 
    def run(self):
        version_info = self.get_version_info()
        if version_info and version_info["code"] == 200:
            if version_info['is_force_update'] == 0:
                # 可选择更新
                self.check_version(version_info)
            else:
                # 强制更新
                self.down_zip(version_info)
 
 
if __name__ == '__main__':
    start = Unit_Test_Version()
    start.run()

# -*- coding: utf-8 -*-
import time
from datetime import datetime
from fastapi.middleware.cors import CORSMiddleware
import uvicorn
from pydantic import BaseModel
import asyncio
from fastapi import FastAPI, File, UploadFile, Form
from link_sql import Link_Sql
from tools import log_count
from starlette.responses import FileResponse
 
"""
http://127.0.0.1:19315/search_version
post请求
参数：soft_type 软件类型
data={
"soft_type":1
}
返回结果：
{
'version': '1.0.2', 
'zip_url': 'https://127.0.0.1:19315/zipfiles/xxxx.zip', 
'file_name': '软件测试', 
'remark': '版本测试',  # 更新内容
'is_force_update': 0 # 0.可选择是否更新 1. 强制更新
}
"""
 
class Comment_Task():
    def __init__(self):
        self.loggings = log_count()
        self.db = Link_Sql(self.loggings)
 
    def search_version(self, item):
        """
        查询版本信息
        """
        soft_type = item.soft_type
        sql = 'select version,zip_url,file_name,remark,is_force_update from plugin_version where soft_type=%s ORDER BY create_time desc'
        result_data = self.db.one_search(sql % soft_type, to_json=True)
        if result_data:
            result_data["code"] = 200
        else:
            result_data["code"] = 500
        print('result_data:', result_data)
        return result_data
 
 
app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 表示允许任何源
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
 
 
class Update_Status(BaseModel):
    soft_type: int
 
 
method_name_dic = {
    # 查询版本信息
    "search_version": Comment_Task().search_version,
}
 
 
@app.post('/search_version')
async def search_version(item: Update_Status):
    method_name = "search_version"
    try:
        loop = asyncio.get_event_loop()
        method_ = method_name_dic[method_name]
        data = await loop.run_in_executor(None, method_, item)
        return data
    except:
        return {"code": 500, "message": "查询失败"}
 
 
@app.get("/zipfiles/{file_path}/{file_name}")
async def download(file_path: str, file_name: str):
    # 下载压缩包 zipfiles为在当前文件 建立的文件夹放置压缩包
    filename = f"zipfiles/{file_path}/{file_name}"
    print('filename:', filename)
    return FileResponse(filename, filename=f"{file_name}")  #展示的下载的文件名
 
 
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=19315)