PythonのWebアプリのデスクトップアプリ化（改良版） - 闘うITエンジニアの覚え書き

streamlit==1.50.0
pywebview==6.0

plotly
pyvista[all]
stpyvista
pydeck

scipy
rfc3987

pyinstaller

# 署名検証用
pefile
cryptography

import multiprocessing
import os
import queue
import shutil
import signal
import socket
import subprocess
import sys
import tempfile
import threading
import time
import traceback

from common import check_edge_installed
from common import get_initial_content
from common import init_platform
from common import install_webview2
from common import load_server_env
from common import get_server_port
from common import print_log


# アプリケーション名
APP_NAME = "Sample App"

# アプリケーションサーバ実行ファイル名(拡張子は自動補完)
APP_SERVER_EXE = "myapp_server"

# streamlitの設定ファイル
APP_CONFIG_FILE = os.path.join(".streamlit", "config.toml")

# 許可するサーバー実行ファイルの証明書サムプリント(空の時はチェックしない)
# TODO: 外部化
ALLOWED_THUMB_PRINTS = "81CD829A58A13480C5......"


class MyParameters():
    def __init__(self):
        self.server_proc = None
        self.child_pid_value = None
        self.log_thread = None
        self.stop_event = None
        self.server_port = None
        self.server_exe_path = None
        self.server_tmp_base = None
        self.server_tmp_dir = None

    def set_value(self, server_proc, child_pid_value, log_thread, stop_event):
        self.server_proc = server_proc
        self.child_pid_value = child_pid_value
        self.log_thread = log_thread
        self.stop_event = stop_event


def start_server_process(app_path, log_queue, pid_value, server_tmp_base, errs, force):
    """アプリケーションサーバを起動する."""

    try:
        if not app_path:
            raise ValueError("app_path is empty")

        exe_path = os.path.abspath(app_path)
        if not os.path.exists(exe_path):
            raise FileNotFoundError(f"Executable not found: {exe_path}")

        # 証明書チェック
        if not force:
            is_certified = check_server_certificate(app_path)
            if not is_certified:
                print_log("[ERROR] Server certificate check failed.")
                errs["CERT_ERROR"] = "証明書エラー"
                print_log("cert_error(p): " + errs["CERT_ERROR"])
                return
                # raise RuntimeError("Server certificate check failed.")

        pid_value.value = os.getpid()

        if sys.platform == "win32":
            # 子プロセスの展開先ベースを指定（PyInstaller は TEMP/TMP 配下に _MEIxxxxx を作る）
            env = os.environ.copy()
            if server_tmp_base:
                try:
                    os.makedirs(server_tmp_base, exist_ok=True)
                except Exception:
                    pass
                env["TEMP"] = server_tmp_base
                env["TMP"] = server_tmp_base
                env["TMPDIR"] = server_tmp_base
            popen_kwargs = {
                "stdout": subprocess.PIPE,
                "stderr": subprocess.STDOUT,
                "encoding": "utf-8",                                         # 標準入出力のエンコーディング
                "errors": "namereplace",                                     # 標準入出力の文字化け部分はUnicode名に変換
                "creationflags": getattr(subprocess, "CREATE_NO_WINDOW", 0),  # コンソールなしで実行(環境によっては未定義の可能性がある為、getattr使用)
                "universal_newlines": True,                                   # 標準入出力は文字列
                "env": env,
            }

            print_log(f"[INFO] start server process: {exe_path}")
            proc = subprocess.Popen([exe_path], **popen_kwargs)
            pid_value.value = proc.pid

            try:
                if proc.stdout is not None:
                    for line in iter(proc.stdout.readline, ""):
                        if not line:
                            break
                        log_queue.put(line)
            finally:
                if proc.stdout is not None:
                    proc.stdout.close()
                proc.wait()
        else:
            # Unix系: TMPDIR を指定して execv
            if server_tmp_base:
                try:
                    os.makedirs(server_tmp_base, exist_ok=True)
                except Exception:
                    pass
                os.environ["TEMP"] = server_tmp_base
                os.environ["TMP"] = server_tmp_base
                os.environ["TMPDIR"] = server_tmp_base
            os.execv(exe_path, [exe_path])

    except SystemExit:
        pass
    except Exception:
        log_queue.put("[STREAMLIT ERROR] Failed to launch executable:\\n" + traceback.format_exc())


def check_server_certificate(exe_path: str) -> bool:
    """サーバー実行ファイルの署名をチェックする."""
    if sys.platform != "win32":
        # Windows以外はチェックしない
        return True

    if not ALLOWED_THUMB_PRINTS:
        # 許可する証明書サムプリントが空の時はチェックしない
        print_log("[INFO] SKIP check certificate")
        return True

    try:
        from check_sign_win import get_exe_thumbprint
    except ImportError:
        print_log("[WARN] Unable to import check_sign module; skipping certificate check.")
        return False

    try:
        print_log("[INFO] START check certificate")
        thumbprint = get_exe_thumbprint(exe_path)
        print_log(f"[INFO] Executable thumbprint: {thumbprint}")
        if thumbprint.upper() == ALLOWED_THUMB_PRINTS.upper():
            print_log("[INFO] Server certificate OK.")
            return True
        else:
            print_log("[ERROR] Server certificate thumbprint does not match expected value.")
            return False
    except Exception as e:
        print_log(f"[ERROR] Failed to verify server certificate: {e}")
        return False
    finally:
        print_log("[INFO] END check certificate")


def start_app_server(params, errs, force=False):
    """Streamlitサーバ起動."""
    child_pid_value = multiprocessing.Value("i", -1)
    stop_event = threading.Event()
    server_proc = multiprocessing.Process(
        target=start_server_process,
        args=(params.server_exe_path, params.log_queue, child_pid_value, params.server_tmp_dir, errs, force),
    )
    server_proc.start()

    def print_logs(q):
        while True:
            if stop_event.is_set():
                break
            try:
                msg = q.get(timeout=0.5)
            except queue.Empty:
                if stop_event.is_set() or not server_proc.is_alive():
                    break
                continue
            except Exception:
                if stop_event.is_set() or not server_proc.is_alive():
                    break
                continue

            if msg:
                print_log(f"[STREAMLIT] {msg}", end="")

    log_thread = threading.Thread(target=print_logs, args=(params.log_queue,), daemon=False)
    log_thread.start()

    params.set_value(server_proc, child_pid_value, log_thread, stop_event)


def wait_server_starting(window, params, force=False):
    """アプリケーションサーバが起動するまで待機する."""
    print_log("[INFO] Waiting for app server port...")

    # プロセス跨ぎでエラー情報を共有する為、Manager を使用
    with multiprocessing.Manager() as manager:

        # サーバ起動
        errs = manager.dict()
        start_app_server(params, errs, force)

        def force_start(e):
            """証明書エラー時の強制起動ボタン押下処理."""
            print_log("[INFO] clicked force start")
            window.load_html(get_initial_content())
            wait_server_starting(window, params, force=True)

        def wait_for_port(host, port, timeout=60):
            """指定ポートが開くまで待機する."""
            start = time.time()
            while time.time() - start < timeout:
                if "CERT_ERROR" in errs:
                    print_log("cert_error(w): " + errs["CERT_ERROR"])
                    window.load_html("<h1 style='color: red'>アプリケーションの起動に失敗しました</h1><div><span>内部サーバ証明書が不正です。このまま起動しますか？</span><button id='force_start'>起動する</button></div>")
                    button = window.dom.get_element('#force_start')
                    button.events.click += force_start
                    return False
                try:
                    with socket.create_connection((host, port), timeout=1):
                        return True
                except OSError:
                    time.sleep(0.5)
            return False

        if wait_for_port("127.0.0.1", params.server_port, timeout=120):
            print_log("[INFO] app server is ready.")
            window.load_url(f"http://localhost:{params.server_port}")
        else:
            if "CERT_ERROR" not in errs:
                server_proc = params.server_proc
                errors = ["[ERROR] app server did not start within timeout."]
                if server_proc is not None and server_proc.exitcode is not None:
                    errors.append(f"[ERROR] Streamlit process exit code: {server_proc.exitcode}")
                while params.log_queue is not None and not params.log_queue.empty():
                    errors.append(f"[STREAMLIT] {params.log_queue.get()}")
                for err in errors:
                    print_log(err)
                window.load_html("<h1 style='color: red'>アプリケーションの起動に失敗しました</h1><pre>" + "\r\n".join(errors) + "</pre>")


def get_server_exe_path(filename):
    """実行ファイルのパスを取得する"""
    ext = ".exe" if sys.platform == "win32" else ""
    candidates = []
    try:
        # この exe と同じディレクトリにある場合
        exec_dir = os.path.dirname(sys.executable)
        candidates.append(os.path.join(exec_dir, f"{filename}{ext}"))
    except Exception:
        pass
    # ./dist/{filename}[.exe] にある場合(開発環境)
    candidates.append(os.path.abspath(os.path.join("dist", f"{filename}{ext}")))
    # ./{filename}[.exe] にある場合(開発環境)
    candidates.append(os.path.abspath(f"{filename}{ext}"))

    exe_path = ""
    for path in candidates:
        if os.path.exists(path) and os.access(path, os.X_OK):
            exe_path = path
            print_log(f"[INFO] found server exe: {path}")
            break
        else:
            print_log(f"[INFO] Checked and not found: {path}")

    return exe_path


def terminate_server_process(params, timeout=5):
    """サーバプロセスを停止する."""
    print_log("[INFO] Shutting down Streamlit...")
    log_queue = params.log_queue
    proc = params.server_proc
    child_pid_value = params.child_pid_value
    log_thread = params.log_thread
    stop_event = params.stop_event

    if stop_event is not None:
        stop_event.set()

    if log_queue is not None:
        try:
            log_queue.put_nowait("")
        except Exception:
            pass

    launcher_pid = getattr(proc, "pid", None)
    child_pid = -1
    if child_pid_value is not None:
        try:
            child_pid = child_pid_value.value
        except Exception as exc:
            print_log(f"[WARN] Unable to read child process pid: {exc}")

    if proc is not None:
        try:
            proc.terminate()
            proc.join(timeout)
        except Exception as exc:
            print_log(f"[WARN] Failed to terminate Streamlit launcher gracefully: {exc}")

        if proc.is_alive():
            print_log("[WARN] Streamlit launcher still running; forcing termination.")
            if sys.platform == "win32":
                try:
                    subprocess.run(
                        ["taskkill", "/PID", str(proc.pid), "/T", "/F"],
                        check=False,
                        stdout=subprocess.DEVNULL,
                        stderr=subprocess.DEVNULL,
                    )
                except Exception as exc:
                    print(f"[ERROR] taskkill failed: {exc}")
            else:
                try:
                    proc.kill()
                except Exception as exc:
                    print(f"[ERROR] Failed to force kill Streamlit launcher: {exc}")
            try:
                proc.join(timeout)
            except Exception:
                pass

    target_pid = child_pid if child_pid > 0 else launcher_pid
    if target_pid and (launcher_pid is None or target_pid != launcher_pid or (proc is not None and not proc.is_alive())):
        if sys.platform == "win32":
            try:
                subprocess.run(
                    ["taskkill", "/PID", str(target_pid), "/T", "/F"],
                    check=False,
                    stdout=subprocess.DEVNULL,
                    stderr=subprocess.DEVNULL,
                )
            except Exception as exc:
                print_log(f"[ERROR] taskkill (child) failed: {exc}")
        else:
            try:
                os.kill(target_pid, signal.SIGTERM)
            except ProcessLookupError:
                pass
            except Exception as exc:
                print(f"[ERROR] Failed to signal Streamlit process: {exc}")

    if log_thread is not None and log_thread.is_alive():
        try:
            log_thread.join(timeout)
        except Exception:
            pass

    if log_queue is not None:
        try:
            log_queue.cancel_join_thread()
        except Exception:
            pass
        try:
            log_queue.close()
        except Exception:
            pass

    if proc is not None and proc.is_alive():
        print_log(f"[ERROR] Streamlit launcher is still alive (pid={proc.pid}).")

    try:
        if proc is not None:
            proc.close()
    except Exception:
        pass


def cleanup_server_tmp(params):
    """サーバー用一時ディレクトリを削除する."""

    parent_dir = params.server_tmp_base
    tmp_dir = params.server_tmp_dir

    try:
        print_log(f"[INFO] Cleaning up server temp directory: {tmp_dir}, (_MEIPASS parent: {parent_dir})")
        if not tmp_dir or not os.path.isdir(tmp_dir):
            return
        # 親の _MEIPASS そのものは消さない(pyinstaller実装に影響するため)
        if parent_dir and os.path.abspath(tmp_dir) == os.path.abspath(parent_dir):
            return
        # ハンドル解放のタイムラグがある場合に備えてリトライ
        retries = 5
        for i in range(retries):
            try:
                shutil.rmtree(tmp_dir)
                break
            except Exception:
                if i == retries - 1:
                    raise
                time.sleep(0.6)
    except Exception as e:
        print_log(f"[WARN] Failed to cleanup server temp '{tmp_dir}': {e}")


def create_tmp_dir():
    """サーバー用一時ディレクトリを作成する."""
    parent_mei = getattr(sys, "_MEIPASS", None)
    if parent_mei:
        tmp_dir = parent_mei + "_s"
    else:
        tmp_dir = os.path.join(tempfile.gettempdir(), "myapp_server_s")
    try:
        os.makedirs(tmp_dir, exist_ok=True)
    except Exception:
        pass
    return parent_mei, tmp_dir


# -----------------------
# メイン処理
# -----------------------
def main():
    # 初期設定
    init_platform()

    # 環境確認など
    if not check_edge_installed() and sys.platform == "win32":
        install_webview2()
        sys.exit(1)

    # パラメータ初期化
    params = MyParameters()
    params.log_queue = multiprocessing.Queue()

    # アプリケーションサーバの path、ポート取得
    envs = load_server_env(APP_CONFIG_FILE)
    params.server_port = get_server_port(envs)
    params.server_exe_path = get_server_exe_path(APP_SERVER_EXE)

    # サーバー用一時ベースディレクトリを決定
    tmp_base_dir, tmp_dir = create_tmp_dir()
    params.server_tmp_base = tmp_base_dir
    params.server_tmp_dir = tmp_dir

    # Webviewの表示
    try:
        import webview
        content = get_initial_content()
        window = webview.create_window(APP_NAME, html=content, width=1000, height=700, text_select=True)
        if params.server_exe_path:
            webview.start(wait_server_starting, (window, params), gui=os.environ["PYWEBVIEW_GUI"], debug=False)
        else:
            window.load_html("<h1 style='color: red'>アプリケーションの起動に失敗しました</h1><div>(内部サーバが存在しません)</div>")
    except Exception as e:
        print_log("[ERROR] WebView failed to start:")
        print_log(f"Exception: {e}")
        traceback.print_exc()
    finally:
        # サーバ停止
        terminate_server_process(params)
        # 一時ディレクトリを削除
        cleanup_server_tmp(params)


if __name__ == "__main__":
    main()

from streamlit.web import cli as stcli
import os
import sys

from common import get_server_port, load_server_env, resource_path


# streamlitの設定ファイル
APP_CONFIG_FILE = os.path.join(".streamlit", "config.toml")

# TOPページのPATH
TOP_PAGE = os.path.join("pages", "top.py")


if __name__ == "__main__":

    # アプリケーションサーバのpath、ポート取得
    host = os.environ.get("STREAMLIT_SERVER_ADDRESS") or os.environ.get("STREAMLIT_HOST") or "localhost"
    streamlit_env = load_server_env(APP_CONFIG_FILE)
    APP_PORT = get_server_port(streamlit_env)

    # Streamlitを起動
    sys.argv = [
        "streamlit",
        "run",
        resource_path(TOP_PAGE),
        "--server.headless", "true",
        "--server.address", host,
        "--server.port", str(APP_PORT),
        "--logger.level", "error",
        "--global.developmentMode", "false",
    ]
    sys.exit(stcli.main())

import datetime
import multiprocessing
import os
import re
import sys
import pathlib
import tomllib


def print_log(*args, **kwargs):
    t = datetime.datetime.now()
    p_args = [t.strftime("%Y-%m-%d %H:%M:%S"), *args]
    print(*p_args, **kwargs)


def init_platform():
    """multiprocessing 初期化"""
    if sys.platform == "darwin":
        # Mac なら fork 使用
        multiprocessing.set_start_method("fork")
        os.environ["PYWEBVIEW_GUI"] = "cocoa"
    elif sys.platform == "win32":
        # exe化した時に親子プロセスの区別をつける為に必要
        multiprocessing.freeze_support()
        os.environ["PYWEBVIEW_GUI"] = "edgechromium"
    else:
        multiprocessing.set_start_method("fork")
        os.environ["PYWEBVIEW_GUI"] = "gtk"


def check_edge_installed():
    """Edge Webview2 インストール確認"""
    if sys.platform != "win32":
        print("[INFO] Non-Windows OS, skipping Edge check")
        return True

    try:
        import winreg
    except ImportError:
        print("[WARN] winreg module not available")
        return False

    keys_to_check = [
        r"SOFTWARE\Microsoft\EdgeUpdate",
        r"SOFTWARE\Microsoft\Edge",
    ]

    for key_path in keys_to_check:
        try:
            with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, key_path):
                print(f"[INFO] Found registry key: {key_path}")
                return True
        except FileNotFoundError:
            continue

    print("[WARN] No Edge registry key found. WebView2 may not work.")
    return False


def install_webview2():
    """WebView2 ランタイムのインストール案内（GUI + ブラウザ起動）"""
    if sys.platform != "win32":
        return

    import webbrowser
    try:
        import ctypes
        MB_OK = 0x00000000
        MB_ICONINFORMATION = 0x00000040
        ctypes.windll.user32.MessageBoxW(
            None,
            "WebView2 ランタイムが見つかりません。\nOK を押すとダウンロードページを開きます。",
            "WebView のインストールが必要です",
            MB_OK | MB_ICONINFORMATION,
        )
    except Exception:
        # メッセージボックスに失敗しても続行（ブラウザ起動）
        pass

    url = "https://developer.microsoft.com/ja-jp/microsoft-edge/webview2/"
    try:
        webbrowser.open(url)
    except Exception:
        pass

    # インストール後の再起動を促すため、ここで異常終了させる
    raise RuntimeError("WebView2 runtime not installed. Prompted user to install.")


def to_streamlit_envs(env_dict):
    """
    streamlitの設定ファイル情報を環境変数として取得する.

    Returns:
        {"STREAMLIT_XXXX": 値, "STREAMLIT_XXXX": 値, ...}
    """
    results = {}
    keys = env_dict.copy().keys()
    for camel_key1 in keys:
        values = env_dict[camel_key1]
        snake_key1 = re.sub("([A-Z])", lambda x: "_" + x.group(1).lower(), camel_key1).upper()
        child_keys = values.keys()
        for camel_key2 in child_keys:
            val = values[camel_key2]
            snake_key2 = re.sub("([A-Z])", lambda x: "_" + x.group(1).lower(), camel_key2).upper()
            env_name = f"STREAMLIT_{snake_key1}_{snake_key2}"
            results[env_name] = val
    return results


def resource_path(relative_path):
    """リソースパス解決"""
    if hasattr(sys, "_MEIPASS"):
        print("sys._MEIPASS: ", sys._MEIPASS)
        return os.path.join(sys._MEIPASS, relative_path)
    return os.path.join(os.path.abspath("."), relative_path)


def load_server_env(config_path):
    """サーバ設定ファイル読み込み"""
    # 設定ファイルの内容を取得
    internal_config_path = resource_path(config_path)
    print(f"internal_config_path: {internal_config_path}")
    data = {}
    if os.path.isfile(internal_config_path):
        with open(internal_config_path, mode="rb") as f:
            data = to_streamlit_envs(tomllib.load(f))

    overwrite_config_path = os.path.join(str(pathlib.Path(sys.executable).parent), config_path)
    print(f"overwrite_config_path: {overwrite_config_path}")

    # 上書き設定ファイルがある場合は取得 及び 上書き
    if os.path.isfile(overwrite_config_path) and internal_config_path != overwrite_config_path:
        with open(overwrite_config_path, mode="rb") as f:
            data2 = to_streamlit_envs(tomllib.load(f))
            keys = data2.keys()
            for key in keys:
                data[key] = data2[key]
    # 環境変数を設定
    for key in data.keys():
        print(f"{key} = {data[key]}")
        os.environ[key] = f"{data[key]}"

    return data


def get_server_port(streamlit_env):
    """streamlitのポート番号取得"""
    if "STREAMLIT_SERVER_PORT" in streamlit_env:
        return streamlit_env["STREAMLIT_SERVER_PORT"]
    return 8501


def get_initial_content():
    """初期ページ取得"""
    return """<!DOCTYPE html>
<html lang="ja">
<head>
<meta charset="UTF-8" />
<style>
body {
  text-align: center;
  margin-top: 20%;
}
.loading {
  position: relative;
}
.loading::after {
  content: "";
  display: inline-block;
  width: 30px;
  text-align: left;
  animation: dots 1.5s steps(3, end) infinite;
}
@keyframes dots {
  0%  { content: ""; }
  25% { content: "."; }
  50% { content: ".."; }
  75% { content: "..."; }
}
</style>
</head>
<body>

<h1 class="loading">アプリケーションを起動しています</h1>

</body>
</html>
"""

import subprocess
import json


def get_exe_thumbprint(exe_path):
    info = get_signature_info(exe_path)
    if "Thumbprint" not in info:
        raise ValueError("証明書を取得できませんでした。")
    return info["Thumbprint"]


def get_signature_info(exe_path: str):
    ps_script = f"""
    $sig = Get-AuthenticodeSignature '{exe_path}';
    $obj = [PSCustomObject]@{{
        Status = $sig.Status.ToString();
        Signer = $sig.SignerCertificate.Subject;
        Issuer = $sig.SignerCertificate.Issuer;
        Thumbprint = $sig.SignerCertificate.Thumbprint;
    } };
    $obj | ConvertTo-Json -Compress
    """
    result = subprocess.run(["powershell", "-Command", ps_script], capture_output=True, text=True)
    if result.returncode != 0:
        raise RuntimeError("署名情報の取得に失敗しました")

    info = json.loads(result.stdout)
    return info


if __name__ == "__main__":
    # 確認用
    exe = "dist/myapp_server.exe"
    info = get_signature_info(exe)
    print("署名検証結果:")
    print(json.dumps(info, indent=2, ensure_ascii=False))

    # 署名者(Signer)、証明書ハッシュ(Thumbprint) などを照合
    if "Your Company Name" in info["Signer"]:
        print("OK")
    else:
        print("NG")

x_build.sh

#!/bin/bash
#====================================================
# build app
#====================================================

venv_dir=".venv"
stime=`date "+%Y-%m-%d %H:%M:%S"`

# Python仮想環境作成
if [ ! -d "${venv_dir}" ]; then
    echo "Creating virtual environment..."
    python -m venv ${venv_dir}
fi
source ${venv_dir}/bin/activate

# 必要パッケージインストール
pip install --upgrade pip
pip install -r requirements.txt

# ビルド形式
#BUILD_MODE="--onedir"
BUILD_MODE=--onefile
#NOCONSOLE=--noconsole
NOCONSOLE=

# webviewビルド
echo "Building webview exe..."
pyinstaller ${BUILD_MODE} ${NOCONSOLE} \
    --clean \
    --noconfirm \
    --add-data ".streamlit:.streamlit" \
    myapp_webview.py


# webアプリ(streamlit)ビルド
echo "Building Streamlit exe..."
pyinstaller ${BUILD_MODE} ${NOCONSOLE} \
    --clean \
    --noconfirm \
    --add-data "pages:pages" \
    --add-data ".streamlit:.streamlit" \
    --collect-all streamlit \
    --collect-all pyvista \
    --collect-all stpyvista \
    myapp_server.py

deactivate

etime=`date "+%Y-%m-%d %H:%M:%S"`
echo ""
echo "process times ... started at: ${stime} , finished at: ${etime}"
echo ""

win_build_pyinstaller.bat

@echo off
REM ###############################################
REM pyinstaller でビルド
REM ###############################################

set STIME=%DATE% %TIME%

set venv_dir=.venv

REM 仮想環境の有効化
IF NOT EXIST %venv_dir%\Scripts\activate.bat python -m venv %venv_dir%
call %venv_dir%\Scripts\activate

REM 必要パッケージインストール
set PYTHONUTF8=1
REM pip install --upgrade pip
pip install -r requirements.txt

REM set BUILD_MODE=--onedir
set BUILD_MODE=--onefile
REM set NOCONSOLE=--noconsole
set NOCONSOLE=

echo "webview build"
pyinstaller ^
    %BUILD_MODE% %NOCONSOLE% ^
    --clean ^
    --noconfirm ^
    --add-data ".streamlit:.streamlit" ^
    myapp_webview.py

echo "streamlit build"
pyinstaller ^
    %BUILD_MODE% %NOCONSOLE% ^
    --clean ^
    --noconfirm ^
    --add-data "pages:pages" ^
    --add-data ".streamlit:.streamlit" ^
    --collect-all streamlit ^
    --collect-all pyvista ^
    --collect-all stpyvista ^
    myapp_server.py

call %venv_dir%\Scripts\deactivate

set ETIME=%DATE% %TIME%
echo.
echo process times ... started at: %STIME% , finished at: %ETIME%
echo.

win_build_nuitka.bat

@echo off
REM ====================================================
REM Nuitka でビルド（Windows）
REM ====================================================

set STIME=%DATE% %TIME%

set venv_dir=.venv_webview

REM 仮想環境の有効化
IF NOT EXIST %venv_dir%\Scripts\activate.bat python -m venv %venv_dir%
call %venv_dir%\Scripts\activate

REM 必要パッケージインストール
set PYTHONUTF8=1
REM pip install --upgrade pip
pip install -r requirements_webview.txt
pip install Nuitka zstandard

REM 出力ディレクトリ作成
IF NOT EXIST dist_nuitka_webview mkdir dist_nuitka_webview
IF NOT EXIST dist_nuitka_server mkdir dist_nuitka_server

REM Nuitka のキャッシュをワークスペース内に制限（任意）
set NUITKA_CACHE_DIR=%CD%\.nuitka-cache
IF NOT EXIST "%NUITKA_CACHE_DIR%" mkdir "%NUITKA_CACHE_DIR%"

REM 追加データ
set INCLUDE_STREAMLIT_DIR=
IF EXIST .streamlit (
  set "INCLUDE_STREAMLIT_DIR=--include-data-dir=.streamlit=.streamlit"
)
set INCLUDE_PAGES=
IF EXIST pages (
  set "INCLUDE_PAGES=--include-data-dir=pages=pages"
)

REM デバッグ用に一時ディレクトリを残す場合は "cached" を指定
set ONEFILE_CACHE_MODE="--onefile-cache-mode=auto"
REM set ONEFILE_CACHE_MODE="--onefile-cache-mode=cached"

REM webviewビルド実行
REM 一部設定を追加（pywebview の動作に必要なdllがコピーされない為）
python -m nuitka ^
  --onefile ^
  --remove-output ^
  --output-dir=dist_nuitka_webview ^
  --disable-plugin=anti-bloat ^
  --nofollow-import-to=tkinter,_curses,curses ^
  --enable-plugin=multiprocessing ^
  --no-deployment-flag=self-execution ^
  %ONEFILE_CACHE_MODE% ^
  %INCLUDE_STREAMLIT_DIR% ^
  --include-data-files=%venv_dir%\Lib\site-packages\webview\lib\WebBrowserInterop.x86.dll=webview/lib/WebBrowserInterop.x86.dll ^
  --include-data-files=%venv_dir%\Lib\site-packages\webview\lib\runtimes\win-arm64\native\WebView2Loader.dll=webview/lib/runtimes/win-arm64/native/WebView2Loader.dll ^
  --include-data-files=%venv_dir%\Lib\site-packages\webview\lib\runtimes\win-x86\native\WebView2Loader.dll=webview/lib/runtimes/win-x86/native/WebView2Loader.dll ^
  myapp_webview.py

REM serverビルド実行
python -m nuitka ^
  --onefile ^
  --remove-output ^
  --output-dir=dist_nuitka_server ^
  --disable-plugin=anti-bloat ^
  --nofollow-import-to=tkinter,_curses,curses ^
  --enable-plugin=multiprocessing ^
  --no-deployment-flag=self-execution ^
  %ONEFILE_CACHE_MODE% ^
  %INCLUDE_STREAMLIT_DIR% ^
  %INCLUDE_PAGES% ^
  myapp_server.py

call %venv_dir%\Scripts\deactivate

set ETIME=%DATE% %TIME%
echo.
echo process times ... started at: %STIME% , finished at: %ETIME%
echo.