如何将Discum的WebSocket网关事件与Textual TUI模块对接？

你的核心问题是Discum的网关事件运行在后台线程，而Textual的UI更新必须在主线程执行，直接在网关回调里操作UI会触发线程安全错误，导致界面无法正常更新。以下是具体的修正方案：

关键修正点 #

• 所有UI更新操作必须通过Textual提供的call_from_thread方法调度到主线程执行
• 修正代码中变量名不一致（比如res和resp混用）、方法调用错误的问题
• 确保Discum网关在Textual启动后再启动，避免提前触发事件时App还未初始化

完整可运行示例代码 #

from textual.app import App, ComposeResult
from textual.widgets import Label, Footer, Container, Static
import discum

# 自定义消息部件
class MessageWidget(Static):
    def __init__(self, content: str):
        super().__init__(content)
        self.styles.padding = (1, 2)
        self.styles.margin = (0, 0, 1, 0)
        self.styles.background = "#2b2d31"

class CordApp(App):
    BINDINGS = []
    CSS = """
    #login-status {
        width: 100%;
        height: 3;
        background: #1e1f22;
        padding: 1;
    }
    #message-cont {
        width: 100%;
        height: 100%;
        overflow-y: auto;
        background: #36393f;
        padding: 2;
    }
    """

    def compose(self) -> ComposeResult:
        yield Label("Logging in...", id="login-status")
        yield Footer()
        yield Container(id="message-cont")

    def update_login_status(self, status_text: str):
        """主线程中更新登录状态"""
        login_status = self.query_one("#login-status", Label)
        login_status.update(status_text)

    def append_message(self, content: str):
        """主线程中添加消息部件"""
        msg_cont = self.query_one("#message-cont", Container)
        msg_cont.mount(MessageWidget(content))
        # 自动滚动到底部
        msg_cont.scroll_end(animate=False)

# 全局变量
currentChannel = "922797200814264361"
bot = discum.Client(token="YOUR_DISCORD_TOKEN", log=False)

# 网关事件处理
@bot.gateway.command
def handle_discord_events(resp):
    if resp.event.ready_supplemental:
        user = bot.gateway.session.user
        # 调度UI更新到主线程
        app.call_from_thread(app.update_login_status, f"Logged in as {user['username']}#{user['discriminator']}")
    if resp.event.message:
        msg = resp.parsed.auto()
        if msg["channel_id"] == currentChannel:
            # 调度消息添加到主线程
            app.call_from_thread(app.append_message, msg["content"])

if __name__ == "__main__":
    app = CordApp()
    # 启动Discum网关后台线程
    bot.gateway.run_background()
    # 启动Textual前台UI
    app.run()

方案说明 #

1. 线程安全调度：使用app.call_from_thread()将UI操作从Discum的后台线程提交到Textual的主线程执行，这是Textual官方推荐的跨线程UI更新方式。
2. 部件定义：补充了MessageWidget的完整实现，确保消息能正常显示。
3. 启动顺序：先初始化App，再启动Discum网关后台线程，避免网关事件触发时App还未完成初始化。

替代方案参考 #

如果你遇到Discum的局限性，可以考虑：

• 使用discord.py（异步Discord库）结合Textual的异步支持，通过asyncio实现更顺畅的事件同步
• 改用Textual的on_message自定义事件系统，将Discum的网关事件转换为Textual内部事件处理

内容的提问来源于stack exchange，提问作者PlanetBluto