如何在同一程序中协同运行Playwright与Pycord版Discord Bot

核心问题分析 #

你的代码存在两个致命问题：

1. sync_playwright()的上下文管理器会阻塞主线程，导致后续的bot.run()根本无法执行；
2. Pycord基于asyncio异步事件循环，同步版Playwright会直接阻塞事件循环，导致Bot无法响应任何命令。

下面提供两种可直接落地的整合方案：

方案一：使用Playwright异步版本（推荐） #

直接切换到Playwright的异步API，和Pycord共用同一个asyncio事件循环，彻底避免线程切换的复杂度。

修改后的完整代码示例 #

import asyncio
from playwright.async_api import async_playwright, Page, BrowserContext
import discord
from discord.ext import commands

# 全局变量存储Playwright实例（也可以用类封装更规范）
browser_ctx: BrowserContext = None
page: Page = None

# 异步版本的Playwright操作函数
async def get_round_image(token: str, outfile: str=None):
    global page
    # 替换为你的实际业务逻辑
    await page.goto(f"https://example.com/round?token={token}")
    if outfile:
        await page.screenshot(path=outfile)

async def guess(lat: float, lng: float, token: str):
    global page
    # 替换为你的实际业务逻辑
    await page.fill("#lat-input", str(lat))
    await page.fill("#lng-input", str(lng))
    await page.click("#submit-guess")

async def get_game_info(token: str):
    global page
    await page.goto(f"https://example.com/game/{token}")
    return await page.inner_text("#game-info")

async def create_game():
    global page
    await page.goto("https://example.com/create-game")
    await page.click("#create-btn")
    return await page.locator("#game-token").inner_text()

# Discord Bot部分
intents = discord.Intents.all()
bot = commands.Bot(command_prefix="!", intents=intents)

@bot.command()
async def guess_cmd(ctx: commands.Context, lat: float, lng: float, token: str):
    try:
        await guess(lat, lng, token)
        await ctx.send("猜测已提交！")
    except Exception as e:
        await ctx.send(f"操作失败：{str(e)}")

@bot.command()
async def create_game_cmd(ctx: commands.Context):
    try:
        token = await create_game()
        await ctx.send(f"游戏创建成功，Token：{token}")
    except Exception as e:
        await ctx.send(f"创建失败：{str(e)}")

# 统一初始化逻辑
async def main():
    global browser_ctx, page
    async with async_playwright() as p:
        browser_ctx = await p.webkit.launch_persistent_context(
            "./userdata/",
            headless=True,
            viewport={"width": 1920, "height": 1080}
        )
        page = await browser_ctx.new_page()
        # 启动Bot，直到Bot关闭才退出Playwright上下文
        await bot.start(DISCORD_TOKEN)

if __name__ == "__main__":
    asyncio.run(main())

关键改动说明 #

• 替换sync_playwright为async_playwright，所有Playwright操作函数改为异步（添加async关键字，调用Playwright方法时加await）；
• 用asyncio.run(main())统一管理事件循环，先初始化Playwright上下文和页面，再启动Bot；
• 改用await bot.start()替代bot.run()，前者是异步方法，不会阻塞事件循环，能和Playwright的异步逻辑完美兼容。

方案二：将同步Playwright放到后台线程 #

如果不想修改已有同步Playwright代码，可以把Playwright的运行放到单独线程，通过线程安全队列处理Bot命令与Playwright操作的交互。

代码示例 #

import threading
import queue
from playwright.sync_api import sync_playwright, Page, BrowserContext
import discord
from discord.ext import commands

# 线程安全队列用于传递任务，全局变量存储Playwright实例
task_queue = queue.Queue()
browser_ctx: BrowserContext = None
page: Page = None

# 原有同步Playwright函数保持不变
def get_round_image(token: str, outfile: str=None):
    global page
    # 你的原有业务逻辑
    page.goto(f"https://example.com/round?token={token}")
    if outfile:
        page.screenshot(path=outfile)

def guess(lat: float, lng: float, token: str):
    global page
    page.fill("#lat-input", str(lat))
    page.fill("#lng-input", str(lng))
    page.click("#submit-guess")

def get_game_info(token: str):
    global page
    page.goto(f"https://example.com/game/{token}")
    return page.inner_text("#game-info")

def create_game():
    global page
    page.goto("https://example.com/create-game")
    page.click("#create-btn")
    return page.locator("#game-token").inner_text()

# Playwright后台线程处理函数
def playwright_worker():
    global browser_ctx, page
    with sync_playwright() as p:
        browser_ctx = p.webkit.launch_persistent_context(
            "./userdata/",
            headless=True,
            viewport={"width": 1920, "height": 1080}
        )
        page = browser_ctx.new_page()
        # 持续处理队列中的任务
        while True:
            task = task_queue.get()
            if task is None:  # 终止信号
                break
            func, args, callback = task
            try:
                result = func(*args)
                callback(True, result)
            except Exception as e:
                callback(False, str(e))
            task_queue.task_done()

# Discord Bot部分
intents = discord.Intents.all()
bot = commands.Bot(command_prefix="!", intents=intents)

def task_callback(success, result, ctx):
    # 通过Bot的事件循环异步发送消息
    async def send_msg():
        if success:
            await ctx.send(f"操作成功：{result}" if result else "操作成功！")
        else:
            await ctx.send(f"操作失败：{result}")
    bot.loop.create_task(send_msg())

@bot.command()
async def guess_cmd(ctx: commands.Context, lat: float, lng: float, token: str):
    task_queue.put((guess, (lat, lng, token), lambda s, r: task_callback(s, r, ctx)))

@bot.command()
async def create_game_cmd(ctx: commands.Context):
    task_queue.put((create_game, (), lambda s, r: task_callback(s, r, ctx)))

# 启动线程和Bot
if __name__ == "__main__":
    # 启动Playwright后台线程
    pw_thread = threading.Thread(target=playwright_worker, daemon=True)
    pw_thread.start()
    # 启动Bot（同步阻塞直到Bot关闭）
    bot.run(DISCORD_TOKEN)
    # Bot关闭后发送终止信号给线程
    task_queue.put(None)
    pw_thread.join()

关键改动说明 #

• 把Playwright的初始化和任务处理放到后台线程，通过queue.Queue实现Bot命令与Playwright操作的通信；
• 每个Bot命令将任务放入队列，后台线程执行完成后通过回调函数通知Bot发送结果；
• 标记线程为daemon=True，确保程序退出时线程能自动终止，避免资源泄漏。

注意事项 #

• 方案一更简洁，无线程安全问题，优先推荐使用；
• 方案二中，队列会串行处理所有任务，确保Playwright页面实例不会被多线程同时访问，保证线程安全；
• 使用launch_persistent_context时，注意./userdata/目录的读写权限，避免无法保存用户数据。

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