はじめに

直前の投稿から、

ここでPyQt5を利用した水文水質データベースのデータ取得のためのexeプログラムの簡単な説明をした。
この説明のプログラムは

ここで配布しているが、完成ではない状態で配布し始めてため、機能が追加された時、いるかも知らない利用者の方々に通知する必要があると思った。また、既存のコードなどを引用するよりは自分で作ってみた方が勉強になると思った。

機能は、通知するだけじゃ足りないと思い、最新バージョンの自動ダウンロードや、自動更新機能を作ってみることにした。

・最新バージョンの有無を判断する。
・最新バージョンが存在しているのであれば、プログラム上でダウンロード・更新ができるようにする。

主にこの二つのポイントを考慮した。

まずコードから。

コード

# WIS_InfoWindow.py

import sys
import datetime
import json
import re
import requests
import markdown
import zipfile
import shutil
import os
import subprocess

from functools import lru_cache
from PyQt5.QtWidgets import (
                            QApplication, QWidget, QPushButton, QVBoxLayout, QHBoxLayout,
                            QTableWidget, QTableWidgetItem, QTextEdit, QSplitter, QLineEdit,
                            QLabel, QComboBox, QMessageBox, QTabWidget, QTabBar, 
                            QDialog, QGroupBox, QSizePolicy, QProgressBar, QProgressDialog
                            )
from PyQt5.QtCore import Qt, QThread, pyqtSignal, QSize, QUrl, QTimer
from PyQt5.QtGui import QColor, QPixmap, QDesktopServices

# 以下のコードに使用されていないライブラリもあるため注意。
# また、以下のコードのオリジナルコードの一部である。

class UpdateChecker:
    def __init__(self, local_version_file, remote_version_url):
        self.local_version_file = local_version_file
        self.remote_version_url = remote_version_url

    def get_local_version(self):
        with open(self.local_version_file, 'r') as file:
            return file.read().strip()

    def get_remote_version(self):
        response = requests.get(self.remote_version_url)
        if response.status_code == 200:
            return response.text.strip()
        return None

    def check_update(self):
        local_version = self.get_local_version()
        remote_version = self.get_remote_version()

        if local_version and remote_version and local_version != remote_version:
            return True
        return False

class DownloadProgressDialog(QDialog):
    def __init__(self, parent=None):
        super().__init__(parent)
        self.setWindowTitle("ダウンロード")
        self.setFixedSize(400, 120)
        self.setLayout(QVBoxLayout())

        self.label = QLabel("ダウンロード中...")
        self.label.setStyleSheet("font-size: 14px; color: black; font-weight: bold;")
        self.layout().addWidget(self.label)

        self.progressBar = QProgressBar(self)
        self.layout().addWidget(self.progressBar)

    def update_progress(self, percentage):
        self.progressBar.setValue(percentage)

class DownloadThread(QThread):
    update_progress = pyqtSignal(int)

    def __init__(self, url, file_name):
        QThread.__init__(self)
        self.url = url
        self.file_name = file_name

    def run(self):
        response = requests.get(self.url, stream=True)
        total_length = response.headers.get('content-length')

        if total_length is None:
            self.update_progress.emit(100)
        
        else:
            total_length = int(total_length)
            downloaded = 0

            with open(self.file_name, 'wb') as file:
                for data in response.iter_content(chunk_size=4096):
                    downloaded += len(data)
                    file.write(data)
                    done = int(100 * downloaded / total_length)
                    self.update_progress.emit(done)

class App(QWidget):
    def __init__(self):
    .
    .
    .
    
    def check_update(self):
        update_checker = UpdateChecker('VERSION', 'https://raw.githubusercontent.com/refiaa/WIS_Scraper/main/VERSION')
        
        if update_checker.check_update():
            reply = QMessageBox.question(self, "更新情報", "新しいバージョンが利用可能です。最新版にアップデートしてご利用ください。", QMessageBox.Yes | QMessageBox.No)

            if reply == QMessageBox.Yes:
                self.start_update()

    def start_update(self):
        update_checker = UpdateChecker('VERSION', 'https://raw.githubusercontent.com/refiaa/WIS_Scraper/main/VERSION')
        latest_version = update_checker.get_remote_version()

        if latest_version:
            download_url = f"https://github.com/refiaa/WIS_Scraper/releases/download/{latest_version}/{latest_version}.zip"
            self.download_update(download_url, f"{latest_version}.zip")

    def download_update(self, url, file_name):
        self.download_dialog = DownloadProgressDialog(self)
        self.download_dialog.setModal(True)
        self.download_dialog.show()

        self.download_thread = DownloadThread(url, file_name)
        self.download_thread.update_progress.connect(self.download_dialog.update_progress)
        self.download_thread.finished.connect(self.download_thread.deleteLater)
        self.download_thread.finished.connect(self.download_dialog.close)
        self.download_thread.start()
        self.download_thread.finished.connect(lambda: self.ask_update(file_name))

    def ask_update(self, file_name):
        reply = QMessageBox.question(self, "更新情報", "インストールの準備ができました。今すぐインストールしますか？", QMessageBox.Yes | QMessageBox.No)
        if reply == QMessageBox.Yes:
            self.install_update(file_name)

    def install_update(self, file_name):
        try:
            with zipfile.ZipFile(file_name, 'r') as zip_ref:
                zip_ref.extractall("temp_update")

            if self.replace_exe_with_temp(file_name):
                msg_box = QMessageBox(self)
                msg_box.setIcon(QMessageBox.Information)
                msg_box.setWindowTitle("更新情報")
                msg_box.setText("再起動します。")
                msg_box.exec_()

                self.schedule_update()
                QApplication.quit()

        except Exception as e:
            QMessageBox.warning(self, "更新エラー", f"アップデートの中にエラーが起きました。: {e}")

    def replace_exe_with_temp(self, downloaded_file_name):
        try:
            temp_exe_path = os.path.join(os.getcwd(), "temp_update", "WIS_Scraper.exe")
            current_exe_path = sys.executable
            
            bat_path = os.path.join(os.getcwd(), "update_script.bat")

            with open(bat_path, "w") as bat_file:
                bat_file.write(f"""
                    @echo off
                    setlocal enabledelayedexpansion

                    :waitloop
                    timeout /t 1 /nobreak
                    tasklist | find /i "WIS_Scraper.exe" >nul 2>&1
                    if errorlevel 1 (
                        xcopy "temp_update\\*.*" "." /E /H /Y
                        start "" "WIS_Scraper.exe"
                        rmdir /s /q "temp_update"
                        del /f /q "{downloaded_file_name}"
                        del "%~f0"
                    ) else (
                        goto waitloop
                    )
                    """)
            return True
        
        except Exception as e:
            QMessageBox.warning(self, "更新エラー", f"アップデートスクリプトの作成にエラーが起きました。: {e}")
            return False

    def schedule_update(self):
        bat_path = os.path.join(os.getcwd(), "update_script.bat")
        subprocess.Popen(["cmd.exe", "/C" + bat_path], shell=True)

できるだけ簡単に略すると、

github上のVERSIONファイルのRAWに存在するバージョンとlocalに存在するVERSIONファイルのバージョンを比較し、一致しないのであればユーザーに更新版があることを通知し、ダウンロード・更新を行う、

ということである。

適当な説明

class UpdateCheckerは、「github上のVERSIONファイルのRAWに存在するバージョンとlocalに存在するVERSIONファイルのバージョンを比較」を行う。

この結果により、check_updateからはユーザーに新しいバージョンがあることを通知する。

def check_update(self):
    update_checker = UpdateChecker('VERSION', 'https://raw.githubusercontent.com/refiaa/WIS_Scraper/main/VERSION')
        
    if update_checker.check_update():
        reply = QMessageBox.question(self, "更新情報", "新しいバージョンが利用可能です。最新版にアップデートしてご利用ください。", QMessageBox.Yes | QMessageBox.No)

        if reply == QMessageBox.Yes:
            self.start_update()

ユーザーがこの通知に対し、YESをした場合、start_updateに移行する。
筆者は最新リリースのバージョンがgithub上のVERSIONと一致するようにしているため、最新バージョンをgithub上のVERSIONをそのまま使った。

def start_update(self):
    update_checker = UpdateChecker('VERSION', 'https://raw.githubusercontent.com/refiaa/WIS_Scraper/main/VERSION')
    latest_version = update_checker.get_remote_version()

    if latest_version:
        download_url = f"https://github.com/refiaa/WIS_Scraper/releases/download/{latest_version}/{latest_version}.zip"
        self.download_update(download_url, f"{latest_version}.zip")

このあとはdownload_updateに移行し、ダウンロードを実行する。

DownloadThreadにはQThreadを利用した。
また、ダウンロード状況の確認のため、QProgressBarを導入した。


class DownloadProgressDialog(QDialog):
    def __init__(self, parent=None):
        super().__init__(parent)
        self.setWindowTitle("ダウンロード")
        self.setFixedSize(400, 120)
        self.setLayout(QVBoxLayout())

        self.label = QLabel("ダウンロード中...")
        self.label.setStyleSheet("font-size: 14px; color: black; font-weight: bold;")
        self.layout().addWidget(self.label)

        self.progressBar = QProgressBar(self)
        self.layout().addWidget(self.progressBar)

    def update_progress(self, percentage):
        self.progressBar.setValue(percentage)

class DownloadThread(QThread):
    update_progress = pyqtSignal(int)

    def __init__(self, url, file_name):
        QThread.__init__(self)
        self.url = url
        self.file_name = file_name

    def run(self):
        response = requests.get(self.url, stream=True)
        total_length = response.headers.get('content-length')

        if total_length is None:
            self.update_progress.emit(100)
        
        else:
            total_length = int(total_length)
            downloaded = 0

            with open(self.file_name, 'wb') as file:
                for data in response.iter_content(chunk_size=4096):
                    downloaded += len(data)
                    file.write(data)
                    done = int(100 * downloaded / total_length)
                    self.update_progress.emit(done)

このあとはask_updateとinstall_upateに移行する。

ask_updateではインストールの準備ができたことを通知する。
install_updateでは、temp_updateという臨時フォルダを生成し、ダウンロードした最新リリースの圧縮ファイルを解凍する。

しかし、exeプログラムの実行中にデータを更新するのはできないため(win32エラーが発生する)プログラムを一回終了し、Batファイルを利用して、temp_updateのファイルでアップデートを実行することにした。

def replace_exe_with_temp(self, downloaded_file_name):
    try:
        temp_exe_path = os.path.join(os.getcwd(), "temp_update", "WIS_Scraper.exe")
        current_exe_path = sys.executable
            
        bat_path = os.path.join(os.getcwd(), "update_script.bat")

        with open(bat_path, "w") as bat_file:
            bat_file.write(f"""
                @echo off
                setlocal enabledelayedexpansion

                :waitloop
                timeout /t 1 /nobreak
                tasklist | find /i "WIS_Scraper.exe" >nul 2>&1
                if errorlevel 1 (
                    xcopy "temp_update\\*.*" "." /E /H /Y
                        start "" "WIS_Scraper.exe"
                        rmdir /s /q "temp_update"
                        del /f /q "{downloaded_file_name}"
                        del "%~f0"
                    ) else (
                    goto waitloop
                    )
                """)
        return True
        
    except Exception as e:
        QMessageBox.warning(self, "更新エラー", f"アップデートスクリプトの作成にエラーが起きました。: {e}")
        return False

temp_updateや最新リリースの圧縮ファイル、batファイルなどは臨時ファイルであるため、更新（実はただのコピペです）が終わると削除し、プログラムを再実行する。

以上のロジックからgithub上の最新リリースをダウンロード・解凍し、既存のlocalのファイルを最新リリースにアップデートすることに成功した。

しかし、この方法は複数の問題が存在する。

・バージョンをgithub上のVERSIONファイルと一致しているかしてないかで判断しているだけで、localの方が最新だった場合や間違ってgithubのバージョンの入力を間違えた場合、無限にアップデートが実行されることになる。

・一般的とは言えない方法であるため、セキュリティ上で問題がある可能性が高い。

・batファイルで置き換えるだけの方法が一般的・理想的とは言えない。

従って、以上の問題を考慮した改善方法も考えたい。