DrissionPage + FastAPI 独立打包成 EXE 方案
一、 项目架构与思路
核心思路是创建一个后台服务型应用:
FastAPI 作为 HTTP 服务器,提供 RESTful API 接口。
DrissionPage 作为核心自动化引擎,在后台运行。
客户端(如 Web 前端、其他程序)通过调用 API 来触发浏览器自动化操作,无需关心底层实现。
使用 pyinstaller 将整个 Python 项目(FastAPI 服务器 + DrissionPage + 所有依赖)打包成一个独立的 exe 文件。
优势:
完全独立:最终用户无需安装 Python、浏览器驱动或任何依赖。
远程调用:可以通过网络 API 控制浏览器,实现分布式部署。
易于集成:任何能发送 HTTP 请求的语言都可以调用其功能。
二、 优化打包方式 (PyInstaller)
打包一个包含浏览器和网络请求的库非常复杂,需要精心配置。
1. 项目结构建议
text
your_project/
├── main.py # FastAPI 应用入口点
├── core/
│ └── automation.py # 封装 DrissionPage 核心操作
├── config.py # 配置文件
├── requirements.txt # 项目依赖
└── build/ # 打包输出目录(自动生成)
2. 关键的 main.py 示例 (FastAPI Server)
python
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
import uvicorn
from core.automation import AutomationManager # 导入封装好的自动化管理器
import asyncio
app = FastAPI(title="DrissionPage Automation Service")
# 解决跨域问题,方便前端调用
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # 生产环境应更严格
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 全局管理自动化实例
automation_manager = AutomationManager()
@app.get("/")
async def root():
return {"message": "DrissionPage Automation Service is Running"}
@app.post("/start-session/")
async def start_session():
"""启动一个浏览器会话"""
try:
session_id = await automation_manager.start_new_session()
return {"status": "success", "session_id": session_id, "message": "Session started"}
except Exception as e:
raise HTTPException(status_code=500, detail=f"Failed to start session: {str(e)}")
@app.post("/run-script/{session_id}")
async def run_script(session_id: str, script_name: str, params: dict = None):
"""在指定会话中运行预定义的脚本"""
try:
result = await automation_manager.run_script(session_id, script_name, params or {})
return {"status": "success", "data": result}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/close-session/{session_id}")
async def close_session(session_id: str):
"""关闭指定浏览器会话"""
try:
await automation_manager.close_session(session_id)
return {"status": "success", "message": f"Session {session_id} closed"}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
if __name__ == "__main__":
# 使用 uvicorn 直接运行,方便调试和打包
uvicorn.run(app, host="0.0.0.0", port=8000)
3. 核心自动化封装 core/automation.py
python
from DrissionPage import ChromiumPage, SessionPage
import asyncio
import uuid
from typing import Dict, Any
class AutomationManager:
def __init__(self):
self.sessions: Dict[str, ChromiumPage] = {}
async def start_new_session(self) -> str:
"""异步方式启动新浏览器,避免阻塞主线程"""
loop = asyncio.get_event_loop()
# 将阻塞的初始化操作放到线程池中执行
page = await loop.run_in_executor(None, self._init_browser)
session_id = str(uuid.uuid4())
self.sessions[session_id] = page
return session_id
def _init_browser(self):
"""同步初始化浏览器"""
# 重要:配置浏览器路径和选项,避免打包后找不到
# 使用 False 防止自动打开浏览器窗口,适合后台运行
page = ChromiumPage(addr_driver_opts=False)
# 或者使用无头模式,不显示图形界面
# page = ChromiumPage(addr_driver_opts=False, headless=True)
return page
async def run_script(self, session_id: str, script_name: str, params: dict) -> Any:
"""运行脚本"""
if session_id not in self.sessions:
raise ValueError(f"Session {session_id} not found")
page = self.sessions[session_id]
# 在这里定义你的各种自动化任务
if script_name == "baidu_search":
return await self._baidu_search(page, params.get('keyword'))
elif script_name == "get_page_title":
return await self._get_page_title(page, params.get('url'))
else:
raise ValueError(f"Unknown script: {script_name}")
async def _baidu_search(self, page: ChromiumPage, keyword: str):
"""示例任务:百度搜索"""
loop = asyncio.get_event_loop()
await loop.run_in_executor(None, self._sync_baidu_search, page, keyword)
return f"Search for '{keyword}' completed."
def _sync_baidu_search(self, page: ChromiumPage, keyword: str):
"""同步的搜索操作"""
page.get('https://www.baidu.com')
page.ele('#kw').input(keyword)
page.ele('#su').click()
page.wait.ele_displayed('#content_left')
async def close_session(self, session_id: str):
"""关闭会话"""
if session_id in self.sessions:
page = self.sessions.pop(session_id)
loop = asyncio.get_event_loop()
await loop.run_in_executor(None, page.quit)
4. 打包配置:pyinstaller.spec 文件 (关键!)
手动创建或通过 pyinstaller main.py 生成后修改 spec 文件。
python
# -*- mode: python ; coding: utf-8 -*-
block_cipher = None
a = Analysis(
['main.py'],
pathex=[],
binaries=[],
# 必须手动添加 DrissionPage 和其他依赖
datas=[],
hiddenimports=[
'DrissionPage',
'fastapi',
'uvicorn',
'uvicorn.lifespan.on',
'uvicorn.lifespan.off',
'asyncio',
# ... 其他可能缺失的库
],
hookspath=[],
hooksconfig={},
runtime_hooks=[],
excludes=[],
win_no_prefer_redirects=False,
win_private_assemblies=False,
cipher=block_cipher,
noarchive=False,
)
# 必须包含 Chromium 驱动文件
import DrissionPage
drission_path = os.path.dirname(DrissionPage.__file__)
driver_files = []
# 尝试收集可能的驱动文件
possible_drivers = [
os.path.join(drission_path, 'chromedriver'),
os.path.join(drission_path, 'geckodriver'),
os.path.join(drission_path, 'msedgedriver'),
]
for driver_path in possible_drivers:
if os.path.exists(driver_path):
driver_files.append((driver_path, '.'))
if driver_files:
a.datas.extend(driver_files)
pyz = PYZ(a.pure, a.zipped_data, cipher=block_cipher)
exe = EXE(
pyz,
a.scripts,
a.binaries,
a.zipfiles,
a.datas,
[],
name='main', # 输出 exe 的名称
debug=False,
bootloader_ignore_signals=False,
strip=False,
upx=True, # 使用 upx 压缩,减小体积
upx_exclude=[],
runtime_tmpdir=None,
console=False, # 设置为 True 可以看到控制台日志,False 则作为后台程序运行
icon='icon.ico', # 可选的图标
)
5. 打包命令
安装依赖:pip install pyinstaller fastapi uvicorn drissionpage
生成初始 spec:pyinstaller main.py
按照上述说明仔细修改生成的 main.spec 文件。
使用 spec 文件打包:pyinstaller main.spec
6. 打包后目录结构
text
dist/
└── main/ # 打包生成的文件夹
├── main.exe # 主可执行文件
├── chromedriver.exe # PyInstaller 复制过来的驱动
├── lib/ # 依赖库
└── ... # 其他文件
三、 调用方式
打包后的 exe 是一个独立的 HTTP 服务器。
1. 启动服务
双击运行 main.exe,它会启动一个本地服务器,默认监听 http://127.0.0.1:8000。
或者在命令行中运行 main.exe,以便查看日志输出。
2. API 调用示例 (使用 Python requests)
任何能发送 HTTP 请求的工具都可以调用,如 Postman、curl、或任何编程语言。
python
import requests
import json
BASE_URL = "http://127.0.0.1:8000"
# 1. 启动一个浏览器会话
response = requests.post(f"{BASE_URL}/start-session/")
session_data = response.json()
session_id = session_data['session_id']
print(f"Session ID: {session_id}")
# 2. 执行一个自动化任务(例如百度搜索)
payload = {
"script_name": "baidu_search",
"params": {
"keyword": "DrissionPage"
}
}
response = requests.post(f"{BASE_URL}/run-script/{session_id}", json=payload)
print(response.json())
# 3. 执行另一个任务(例如获取页面标题)
payload = {
"script_name": "get_page_title",
"params": {
"url": "https://www.example.com"
}
}
response = requests.post(f"{BASE_URL}/run-script/{session_id}", json=payload)
print(response.json())
# 4. 任务完成后,关闭会话,释放资源
response = requests.post(f"{BASE_URL}/close-session/{session_id}")
print(response.json())
3. 查看 API 文档
服务启动后,打开浏览器访问 http://127.0.0.1:8000/docs 即可看到 FastAPI 自动生成的交互式 API 文档(Swagger UI),可以在这里直接测试接口。
四、 重要注意事项与优化提示
防逆向工程:pyinstaller 打包的 exe 容易被反编译。如需商业级保护,考虑使用 pyarmor 等工具进行代码加密。
杀毒软件误报:打包的 Python 程序,尤其是包含浏览器自动化功能的,极易被误报为病毒。需要对用户进行说明或购买商业证书进行签名。
体积优化:最终生成的 exe 会很大(通常 > 100MB),因为包含了 Python 解释器、所有库和浏览器驱动。使用 UPX 压缩可以略微减小体积。
无头模式 (Headless):在服务器部署或不需要图形界面的场景,务必在 _init_browser() 中启用 headless=True,性能更高且更稳定。
会话管理:上述示例使用了简单的内存字典管理会话。生产环境需要增加超时销毁机制,并考虑更持久化的管理方式(如数据库)。
错误日志:确保你的代码中有完善的日志记录(如使用 logging 模块),并将日志写入文件,以便排查打包后程序的运行问题。