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

x_init.sh

#!/bin/bash
#!/bin/bash
#====================================================
# setup python virtual env
#====================================================

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

python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
deactivate

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

win_init.bat

@echo off
REM #####################################################
REM setup python virtual env
REM #####################################################

for /f "usebackq" %%d in (`date /t`) do for /f "usebackq" %%t in (`time /t`) do set stime=%%d %%t

python -m venv .venv
call .venv\Scripts\activate

set PYTHONUTF8=1
pip install -r requirements.txt

call .venv\Scripts\deactivate

for /f "usebackq" %%d in (`date /t`) do for /f "usebackq" %%t in (`time /t`) do set etime=%%d %%t

echo.
echo process times ... started at: %stime% , finished at: %etime%
echo.

import multiprocessing
import os
import socket
import sys
import time
import threading
import traceback
from streamlit.web import cli as stcli

from common import (
    check_edge_installed,
    get_initial_content,
    init_platform,
    install_webview2,
    get_server_port,
    load_server_env,
    resource_path
)

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

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

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


def start_server_process(app_path, log_queue):
    """アプリケーションサーバ(streamlit)を起動する"""
    args = [
        "streamlit", "run", app_path,
        "--global.developmentMode", "false",  # config.toml による定義が許可されていない為、これだけ起動時引数でセット
    ]
    sys.argv = args

    # stdout/err を Queue経由でログ出力
    class QueueWriter:
        def __init__(self, q):
            self.q = q
            self.encoding = 'utf-8'

        def write(self, msg):
            if msg:
                self.put_queue(msg)

        def flush(self):
            pass

        def put_queue(self, message):
            message = (message.decode() if type(message) is bytes else message).replace("\r\n", "").replace("\n", "")
            message = message.lstrip()
            if not message:
                return
            self.q.put(message + "\r\n")

    sys.stdout = QueueWriter(log_queue)
    sys.stderr = QueueWriter(log_queue)

    try:
        stcli.main()
    except SystemExit:
        pass
    except Exception:
        log_queue.put("[STREAMLIT ERROR] Exception in Streamlit process:\n" + traceback.format_exc())


def start_app_server(app_path, log_queue):
    """アプリケーションサーバを起動する"""
    # pyinstaller で windows exe 化する場合に subprocess が不安定な為、multiprocessing を使用する.
    # 事前に multiprocessing.set_start_method('spawn')(※1) しておけば subprocess でも起動は出来るが不安定。
    # ※1 ... windowsでは fork は出来ない為 spawn を使用する必要がある
    server_proc = multiprocessing.Process(
        target=start_server_process,
        args=(app_path, log_queue)
    )
    server_proc.start()

    # ログを別スレッドで表示
    def print_logs(q):
        while True:
            try:
                msg = q.get(timeout=1)
                print(f"[STREAMLIT] {msg}", end="")
            except Exception:
                if not server_proc.is_alive():
                    break

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

    return server_proc


def wait_server_starting(window, app_port, server_proc, log_queue):
    """アプリケーションサーバが起動するまで待機する"""
    print("[INFO] Waiting for Streamlit port...")

    def wait_for_port(host, port, timeout=60):
        """ポート待機"""
        start = time.time()
        while time.time() - start < timeout:
            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", app_port, timeout=30):
        print("[INFO] Streamlit server is ready.")
        window.load_url(f"http://localhost:{app_port}")
    else:
        # タイムアウト時に Streamlit プロセスの exitcode と残りログを出力
        errors = ["[ERROR] Streamlit server did not start within timeout."]
        if server_proc.exitcode is not None:
            errors.append(f"[ERROR] Streamlit process exit code: {server_proc.exitcode}")
        while not log_queue.empty():
            errors.append(f"[STREAMLIT] {log_queue.get()}")
        for err in errors:
            print(err)
        window.load_html("<h1 style='color: red'>アプリケーションの起動に失敗しました</h1><pre>" + "\r\n".join(errors) + "</pre>")


# -----------------------
# メイン処理
# -----------------------
if __name__ == "__main__":
    # 初期設定
    init_platform()

    # 環境確認など
    if not check_edge_installed() and sys.platform == "win32":
        try:
            install_webview2()
        except Exception as e:
            print(f"[ERROR] WebView2 auto-install failed: {e}")
            print("[INFO] Please install WebView2 manually from https://developer.microsoft.com/en-us/microsoft-edge/webview2/")
        sys.exit(1)

    # 設定ファイル読み込み
    streamlit_env = load_server_env(APP_CONFIG_FILE)
    APP_PORT = get_server_port(streamlit_env)
    APP_PATH = resource_path(APP_FILE)

    # streamlitの起動
    print(f"[INFO] server port {APP_PORT}")
    print(f"[INFO] app path: {APP_PATH}")
    log_queue = multiprocessing.Queue()
    server_proc = start_app_server(APP_PATH, log_queue)

    # Webviewの表示
    try:
        import webview
        content = get_initial_content()
        window = webview.create_window(APP_NAME, html=content, width=1000, height=700, text_select=True)
        webview.start(wait_server_starting, (window, APP_PORT, server_proc, log_queue), gui=os.environ["PYWEBVIEW_GUI"], debug=False)
    except Exception as e:
        print("[ERROR] WebView failed to start:")
        print(f"Exception: {e}")
        traceback.print_exc()
    finally:
        print("[INFO] Shutting down Streamlit...")
        server_proc.terminate()

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


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"):
        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 streamlit as st


def root():
    st.title("TOPページ")


def page1():
    st.title("ページ1")


def page2():
    st.title("ページ2")


# サイドバー(ナビゲーションメニュー)
pg = st.navigation([
    st.Page(root, title="TOP page"),
    st.Page(page1, title="First page"),
    st.Page(page2, title="Second page"),
    st.Page("./sample_plotly.py", title="sample plotly"),
    st.Page("./sample_pydeck.py", title="sample pydeck"),
    st.Page("./sample_pyvista.py", title="sample pyvista"),
])
pg.run()

import plotly.graph_objects as go
import plotly.figure_factory as ff
import streamlit as st
from numpy.random import default_rng as rng


st.title("Plotlyサンプル")

fig = go.Figure()
fig.add_trace(
    go.Scatter(
        x=[1, 2, 3, 4, 5],
        y=[1, 3, 2, 5, 4]
    )
)

st.plotly_chart(fig, config={'scrollZoom': False})

hist_data = [
    rng(0).standard_normal(200) - 2,
    rng(1).standard_normal(200),
    rng(2).standard_normal(200) + 2,
]
group_labels = ["Group 1", "Group 2", "Group 3"]

fig = ff.create_distplot(
    hist_data, group_labels, bin_size=[0.1, 0.25, 0.5]
)

st.plotly_chart(fig)

import pandas as pd
import pydeck as pdk
import streamlit as st
from numpy.random import default_rng as rng

st.title("PyDeckサンプル")

df = pd.DataFrame(
    rng(0).standard_normal((1000, 2)) / [50, 50] + [37.76, -122.4],
    columns=["lat", "lon"],
)

st.pydeck_chart(
    pdk.Deck(
        map_style=None,  # Use Streamlit theme to pick map style
        initial_view_state=pdk.ViewState(
            latitude=37.76,
            longitude=-122.4,
            zoom=11,
            pitch=50,
        ),
        layers=[
            pdk.Layer(
                "HexagonLayer",
                data=df,
                get_position="[lon, lat]",
                radius=200,
                elevation_scale=4,
                elevation_range=[0, 1000],
                pickable=True,
                extruded=True,
            ),
            pdk.Layer(
                "ScatterplotLayer",
                data=df,
                get_position="[lon, lat]",
                get_color="[200, 30, 0, 160]",
                get_radius=200,
            ),
        ],
    )
)

import pyvista as pv
import streamlit as st
import sys


if sys.platform == "darwin":
    # macOS用の動作不良対応
    # https://github.com/edsaac/stpyvista/issues/14
    # https://github.com/edsaac/stpyvista/issues/34
    from stpyvista.trame_backend import stpyvista
    import multiprocessing
    multiprocessing.set_start_method("fork", force=True)
else:
    from stpyvista import stpyvista

st.title("A cube")
plotter = pv.Plotter(window_size=[400, 400])
mesh = pv.Cube(center=(0, 0, 0))
mesh['myscalar'] = mesh.points[:, 2] * mesh.points[:, 0]
plotter.add_mesh(mesh, scalars='myscalar', cmap='bwr')
plotter.view_isometric()
plotter.background_color = 'white'
stpyvista(plotter, key="pv_cube")

[server]
port = 8501
headless = true
fileWatcherType = "none"

[client]
toolbarMode = "viewer"

[browser]
gatherUsageStats = false

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=

# 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.py

deactivate

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

win_build.bat

@echo off
REM ###############################################
REM build app
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=

pyinstaller ^
    %BUILD_MODE% %NOCONSOLE% ^
    --clean ^
    --noconfirm ^
    --add-data "pages:pages" ^
    --add-data ".streamlit:.streamlit" ^
    --collect-all streamlit ^
    --collect-all pyvista ^
    --collect-all stpyvista ^
    myapp.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

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
pip install Nuitka zstandard

REM 出力ディレクトリ作成と古い生成物のクリーン
IF NOT EXIST dist mkdir dist
IF EXIST dist\myapp.build rmdir /s /q dist\myapp.build
IF EXIST dist\myapp.exe del /q dist\myapp.exe

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

REM 追加データ（存在チェックしてから付与）
set INCLUDE_PAGES=
IF EXIST pages (
  set "INCLUDE_PAGES=--include-data-dir=pages=pages"
)

set INCLUDE_STREAMLIT_DIR=
IF EXIST .streamlit (
  set "INCLUDE_STREAMLIT_DIR=--include-data-dir=.streamlit=.streamlit"
)

REM ビルド実行
python -m nuitka ^
  --onefile ^
  --remove-output ^
  --output-dir=dist ^
  --disable-plugin=anti-bloat ^
  --nofollow-import-to=tkinter,_curses,curses ^
  %INCLUDE_PAGES% ^
  %INCLUDE_STREAMLIT_DIR% ^
  --include-package-data=streamlit ^
  --include-package-data=plotly ^
  --include-package-data=pyvista ^
  --include-package-data=stpyvista ^
  --include-package-data=pydeck ^
  myapp.py

call %venv_dir%\Scripts\deactivate

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