BlinkGTK イベント処理チュートリアル (Python/PyGObject)

#!/usr/bin/env python3
"""読み込みに関わる 4 つのシグナルをコンソールに出す"""
import sys

import gi

gi.require_version("Gtk", "4.0")
gi.require_version("BlinkGTK", "0.1")
from gi.repository import BlinkGTK, Gtk  # noqa: E402

EVENT_NAMES = {
    BlinkGTK.LoadEvent.STARTED: "開始",
    BlinkGTK.LoadEvent.COMMITTED: "コミット",
    BlinkGTK.LoadEvent.FINISHED: "完了",
}


def on_load_changed(view, event):
    print(f"[load-changed] {EVENT_NAMES.get(event, event)}: {view.get_uri()}", flush=True)


def on_load_failed(view, error_code, failing_uri):
    print(f"[load-failed] エラー番号 {error_code}: {failing_uri}", flush=True)


def on_title_changed(view):
    print(f"[title-changed] {view.get_title()}", flush=True)


def on_uri_changed(view):
    print(f"[uri-changed] {view.get_uri()}", flush=True)


def on_activate(app, url):
    window = Gtk.ApplicationWindow(application=app, title="Signal Example")
    window.set_default_size(800, 600)

    view = BlinkGTK.WebView.new()
    view.connect("load-changed", on_load_changed)
    view.connect("load-failed", on_load_failed)
    view.connect("title-changed", on_title_changed)
    view.connect("uri-changed", on_uri_changed)
    window.set_child(view)
    window.present()

    view.load_uri(url)


def main():
    url = sys.argv[1] if len(sys.argv) > 1 else "https://example.com/"
    ok, _argv = BlinkGTK.gtk_init(sys.argv[:1])
    if not ok:
        print("BlinkGTK を初期化できませんでした", file=sys.stderr)
        return 1
    app = Gtk.Application(application_id="org.example.SignalExample")
    app.connect("activate", on_activate, url)
    rc = BlinkGTK.gtk_application_run(app)
    BlinkGTK.gtk_shutdown()
    return rc


if __name__ == "__main__":
    sys.exit(main())

# 簡単なログ出力 (value_nick は "started" などの名前)
web_view.connect("load-changed",
                 lambda w, e: print(f"Load event: {e.value_nick}"))

# タイトル更新 (title-changed には引数が無いので、get_title() で読む)
web_view.connect("title-changed",
                 lambda w: window.set_title(f"{w.get_title()} - Browser"))

def on_load_changed(self, web_view, load_event):
    """読み込みの状態の変化 (状態の表示に使う)"""
    if load_event == BlinkGTK.LoadEvent.COMMITTED:
        self.status_label.set_text("表示する文書が確定しました")
    elif load_event == BlinkGTK.LoadEvent.FINISHED:
        self.status_label.set_text("完了")

def on_load_failed(self, web_view, error_code, failing_uri):
    """読み込みが確定せずに終わった"""
    # iframe など、表示中のページではない失敗は、UI に触る前に除く
    if failing_uri != (web_view.get_uri() or failing_uri):
        return
    if error_code == -3:
        # 中断。停止ボタンや、読み込み中に別のページを開いたときに普通に起きる
        self.status_label.set_text("中断しました")
        return
    self.status_label.set_text(f"読み込めませんでした (エラー番号 {error_code}): {failing_uri}")

def on_title_changed(self, web_view):
    """タイトルの変更"""
    title = web_view.get_title()
    self.set_title(f"{title} - Python Browser" if title else "Python Browser")

def on_uri_changed(self, web_view):
    """URI の変更"""
    uri = web_view.get_uri() or ""
    self.url_entry.set_text(uri)

    # 接続の種類を表示する
    if uri.startswith("https://"):
        self.security_label.set_text("HTTPS")
    elif uri.startswith("http://"):
        self.security_label.set_text("HTTP")
    else:
        self.security_label.set_text("")

def on_is_loading_changed(self, web_view, _pspec):
    """読み込み中の表示と、停止・再読み込みのボタンを切り替える"""
    loading = web_view.is_loading()
    self.progress_bar.set_visible(loading)
    if loading:
        self.progress_bar.pulse()
    self.stop_button.set_sensitive(loading)
    self.reload_button.set_sensitive(not loading)
    self.back_button.set_sensitive(web_view.can_go_back())
    self.forward_button.set_sensitive(web_view.can_go_forward())


web_view.connect("notify::is-loading", self.on_is_loading_changed)

web_view.connect("notify::title", lambda w, pspec: print(f"title = {w.props.title}"))

import sys
from functools import wraps
from typing import Callable


def signal_handler(signal_name: str) -> Callable:
    """ハンドラの中で起きた例外を、シグナル名を付けて表示する"""
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                print(f"{signal_name} のハンドラで例外: {e!r}", file=sys.stderr)
        return wrapper
    return decorator


class BrowserWindow(Gtk.ApplicationWindow):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.web_view = BlinkGTK.WebView.new()
        self.set_child(self.web_view)
        self.web_view.connect("load-changed", self.on_load_changed)
        self.web_view.connect("load-failed", self.on_load_failed)

    @signal_handler("load-changed")
    def on_load_changed(self, web_view, load_event):
        if load_event == BlinkGTK.LoadEvent.FINISHED:
            print("読み込み完了")

    @signal_handler("load-failed")
    def on_load_failed(self, web_view, error_code, failing_uri):
        print(f"確定せずに終了: {failing_uri} (エラー番号 {error_code})")

from typing import Callable, Dict, List


class SignalManager:
    """WebView に接続したハンドラを、シグナル名ごとに覚えておく"""

    def __init__(self, web_view: BlinkGTK.WebView):
        self.web_view = web_view
        self.handlers: Dict[str, List[int]] = {}

    def connect(self, signal_name: str, handler: Callable) -> int:
        handler_id = self.web_view.connect(signal_name, handler)
        self.handlers.setdefault(signal_name, []).append(handler_id)
        return handler_id

    def disconnect(self, signal_name: str) -> None:
        for handler_id in self.handlers.pop(signal_name, []):
            self.web_view.disconnect(handler_id)

    def disconnect_all(self) -> None:
        for signal_name in list(self.handlers):
            self.disconnect(signal_name)


# 使用例
signal_manager = SignalManager(web_view)
signal_manager.connect("load-changed", on_load_changed)
signal_manager.connect("title-changed", on_title_changed)

# 後で外す
signal_manager.disconnect_all()

#!/usr/bin/env python3
"""複数の URL を 1 つずつ順に読み込み、終わったらウィンドウを閉じる"""
import sys

import gi

gi.require_version("Gtk", "4.0")
gi.require_version("BlinkGTK", "0.1")
from gi.repository import BlinkGTK, GLib, Gtk  # noqa: E402

TIMEOUT_SECONDS = 30


class SequentialLoader:
    def __init__(self, window, view, urls):
        self.window = window
        self.view = view
        self.urls = list(urls)
        self.current = None
        self.timeout_id = 0
        self.started = False
        self.error_code = None
        self.timed_out = False
        # 読み込みの終わりは notify::is-loading で受け取る。完了・失敗・中断のどの終わり方でも
        # is-loading は False に戻る。失敗した場合は、その前に load-failed が届く
        view.connect("notify::is-loading", self.on_is_loading_changed)
        view.connect("load-failed", self.on_load_failed)

    def load_next(self):
        if not self.urls:
            print("すべて読み込みました", flush=True)
            self.window.close()  # 最後のウィンドウを閉じると gtk_application_run() が戻る
            return GLib.SOURCE_REMOVE
        self.current = self.urls.pop(0)
        self.started = False
        self.error_code = None
        self.timed_out = False
        self.timeout_id = GLib.timeout_add_seconds(TIMEOUT_SECONDS, self.on_timeout)
        self.view.load_uri(self.current)
        return GLib.SOURCE_REMOVE

    def done(self, message):
        if self.timeout_id:
            GLib.source_remove(self.timeout_id)
            self.timeout_id = 0
        print(f"{message}: {self.current}", flush=True)
        # シグナルの処理の中で次の load_uri() を呼ばず、メインループに戻ってから呼ぶ
        GLib.idle_add(self.load_next)

    def on_is_loading_changed(self, view, _pspec):
        if view.is_loading():
            self.started = True
        elif self.started:
            self.started = False
            if self.timed_out:
                self.done("時間切れで止めました")
            elif self.error_code is not None:
                self.done(f"確定せずに終了 (エラー番号 {self.error_code})")
            else:
                self.done(f"完了 ({view.get_title() or 'タイトルなし'})")

    def on_load_failed(self, view, error_code, failing_uri):
        # iframe などの失敗は除く。Chromium は URL を正規化して通知する (末尾の / など) ので、
        # 入力した文字列ではなく、表示しようとしている URI と比べる
        if failing_uri in (view.get_uri(), self.current) or \
                failing_uri.rstrip("/") == self.current.rstrip("/"):
            self.error_code = error_code

    def on_timeout(self):
        self.timeout_id = 0
        self.timed_out = True
        print(f"{TIMEOUT_SECONDS} 秒で終わらないので止めます: {self.current}", flush=True)
        self.view.stop()  # 止めた後も is-loading が False に戻ってから次へ進む
        return GLib.SOURCE_REMOVE


def on_activate(app, urls):
    window = Gtk.ApplicationWindow(application=app, title="Sequential Loader")
    window.set_default_size(800, 600)
    view = BlinkGTK.WebView.new()
    window.set_child(view)
    window.present()

    loader = SequentialLoader(window, view, urls)
    window.loader = loader  # ウィンドウが生きている間 loader を保持する
    loader.load_next()


def main():
    urls = sys.argv[1:] or ["https://example.com/", "https://www.gtk.org/"]
    ok, _argv = BlinkGTK.gtk_init(sys.argv[:1])
    if not ok:
        print("BlinkGTK を初期化できませんでした", file=sys.stderr)
        return 1
    app = Gtk.Application(application_id="org.example.SequentialLoader")
    app.connect("activate", on_activate, urls)
    rc = BlinkGTK.gtk_application_run(app)
    BlinkGTK.gtk_shutdown()
    return rc


if __name__ == "__main__":
    sys.exit(main())

blinkgtk-run python3 sequential_loader.py https://example.com/ https://www.gtk.org/

#!/usr/bin/env python3
"""4 つのシグナルと設定の切り替えを使うブラウザ"""
import sys

import gi

gi.require_version("Gtk", "4.0")
gi.require_version("BlinkGTK", "0.1")
from gi.repository import BlinkGTK, Gtk  # noqa: E402


class CompleteBrowser(Gtk.ApplicationWindow):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_title("Complete Browser (Python)")
        self.set_default_size(1200, 800)
        self._build_ui()
        self._connect_signals()

    def _build_ui(self):
        vbox = Gtk.Box(orientation=Gtk.Orientation.VERTICAL)

        # ツールバー
        toolbar = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=6)
        for side in ("start", "end", "top", "bottom"):
            getattr(toolbar, f"set_margin_{side}")(6)
        self.back_button = Gtk.Button(label="←", sensitive=False)
        self.forward_button = Gtk.Button(label="→", sensitive=False)
        self.reload_button = Gtk.Button(label="↻")
        self.stop_button = Gtk.Button(label="×", sensitive=False)
        self.security_label = Gtk.Label(label="")
        self.url_entry = Gtk.Entry(hexpand=True)
        for widget in (self.back_button, self.forward_button, self.reload_button,
                       self.stop_button, self.security_label, self.url_entry):
            toolbar.append(widget)

        # 設定のスイッチ
        settings_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=10)
        settings_box.set_margin_start(6)
        settings_box.set_margin_bottom(6)
        self.js_switch = Gtk.Switch(active=True)
        self.images_switch = Gtk.Switch(active=True)
        self.storage_switch = Gtk.Switch(active=True)
        for label, switch in (("JavaScript:", self.js_switch),
                              ("画像:", self.images_switch),
                              ("ローカルストレージ:", self.storage_switch)):
            settings_box.append(Gtk.Label(label=label))
            settings_box.append(switch)

        self.progress_bar = Gtk.ProgressBar(visible=False)

        self.web_view = BlinkGTK.WebView.new()
        self.web_view.set_vexpand(True)

        self.status_label = Gtk.Label(label="準備完了", xalign=0.0)
        self.status_label.set_margin_start(6)
        self.status_label.set_margin_bottom(3)

        for widget in (toolbar, settings_box, self.progress_bar,
                       self.web_view, self.status_label):
            vbox.append(widget)
        self.set_child(vbox)

    def _connect_signals(self):
        self.web_view.connect("notify::is-loading", self.on_is_loading_changed)
        self.web_view.connect("load-changed", self.on_load_changed)
        self.web_view.connect("load-failed", self.on_load_failed)
        self.web_view.connect("title-changed", self.on_title_changed)
        self.web_view.connect("uri-changed", self.on_uri_changed)

        self.back_button.connect("clicked", lambda _b: self.web_view.go_back())
        self.forward_button.connect("clicked", lambda _b: self.web_view.go_forward())
        self.reload_button.connect("clicked", lambda _b: self.web_view.reload())
        self.stop_button.connect("clicked", lambda _b: self.web_view.stop())
        self.url_entry.connect("activate", self.on_url_activate)

        # 設定は次の読み込みから使われるので、変えたら読み込み直す
        self.js_switch.connect("notify::active", self.on_setting_toggled,
                               self.web_view.set_javascript_enabled)
        self.images_switch.connect("notify::active", self.on_setting_toggled,
                                   self.web_view.set_images_enabled)
        self.storage_switch.connect("notify::active", self.on_setting_toggled,
                                    self.web_view.set_local_storage_enabled)

    def on_setting_toggled(self, switch, _pspec, setter):
        setter(switch.get_active())
        if self.web_view.get_uri():
            self.web_view.reload()

    def on_is_loading_changed(self, web_view, _pspec):
        """読み込み中の表示とボタン (リンク・戻る・再読み込みを含め、どの移動でも来る)"""
        loading = web_view.is_loading()
        self.progress_bar.set_visible(loading)
        if loading:
            self.progress_bar.pulse()
            self.status_label.set_text("読み込み中")
        self.stop_button.set_sensitive(loading)
        self.reload_button.set_sensitive(not loading)
        self.back_button.set_sensitive(web_view.can_go_back())
        self.forward_button.set_sensitive(web_view.can_go_forward())

    def on_load_changed(self, web_view, load_event):
        """状態の表示だけに使う"""
        if load_event == BlinkGTK.LoadEvent.FINISHED:
            self.status_label.set_text("完了")

    def on_load_failed(self, web_view, error_code, failing_uri):
        # iframe など、表示中のページではない失敗は、UI に触る前に除く
        if failing_uri != (web_view.get_uri() or failing_uri):
            return
        if error_code == -3:
            self.status_label.set_text("中断しました")
        else:
            self.status_label.set_text(f"読み込めませんでした (エラー番号 {error_code}): {failing_uri}")

    def on_title_changed(self, web_view):
        title = web_view.get_title()
        self.set_title(f"{title} - Complete Browser" if title else "Complete Browser")

    def on_uri_changed(self, web_view):
        uri = web_view.get_uri() or ""
        self.url_entry.set_text(uri)
        if uri.startswith("https://"):
            self.security_label.set_text("HTTPS")
        elif uri.startswith("http://"):
            self.security_label.set_text("HTTP")
        else:
            self.security_label.set_text("")

    def on_url_activate(self, entry):
        url = entry.get_text().strip()
        if url and "://" not in url and not url.startswith(("about:", "data:")):
            url = "https://" + url
        if url:
            self.web_view.load_uri(url)


def on_activate(app, url):
    window = CompleteBrowser(application=app)
    window.present()
    window.web_view.load_uri(url)


def main():
    url = sys.argv[1] if len(sys.argv) > 1 else "https://example.com/"
    ok, _argv = BlinkGTK.gtk_init(sys.argv[:1])
    if not ok:
        print("BlinkGTK を初期化できませんでした", file=sys.stderr)
        return 1
    app = Gtk.Application(application_id="org.example.CompleteBrowser")
    app.connect("activate", on_activate, url)
    rc = BlinkGTK.gtk_application_run(app)
    BlinkGTK.gtk_shutdown()
    return rc


if __name__ == "__main__":
    sys.exit(main())