通过steel-browser搭建可以通过web查看内容的浏览器
https://github.com/steel-dev/steel-browser
使用docker镜像搭建
通过构建xvfb服务允许使用有头浏览器
docker-compose.yml配置文件如下
services:
# 添加xvfb服务
xvfb:
image: alpine:latest
restart: unless-stopped
command: >
sh -c "apk add --no-cache xvfb dbus &&
rm -f /tmp/.X10-lock /run/dbus/dbus.pid /run/dbus/system_bus_socket &&
mkdir -p /run/dbus /tmp/.X11-unix &&
chmod 1777 /tmp/.X11-unix &&
dbus-daemon --system --fork &&
Xvfb :10 -screen 0 2560x1440x24 -listen tcp -ac &
sleep 1 && tail -f /dev/null"
environment:
# 由于steel-browser虚拟显示默认在:10，因此设置为:10，command中的设置同理
- DISPLAY=:10
volumes:
- x11-unix:/tmp/.X11-unix # 关键：共享 X11 套接字
# 健康检查
healthcheck:
test: ["CMD", "pgrep", "Xvfb"]
interval: 5s
timeout: 3s
retries: 30
start_period: 15s
networks:
- steel-browser-network
steel-browser:
image: ghcr.io/steel-dev/steel-browser
restart: unless-stopped
# 使用 entrypoint 先清理锁文件，再执行原始命令，通过该命令可以查看镜像的入口点docker inspect ghcr.io/steel-dev/steel-browser
entrypoint: >
/bin/sh -c "rm -f /tmp/steel-chrome/SingletonLock /tmp/steel-chrome/SingletonSocket &&
/app/api/entrypoint.sh"
ports:
# web的端口，内部默认3000
- "3001:3000"
# CDP的端口，内部通过9223端口的nginx转发，如果想要修改CDP中/json接口返回信息中的端口只需修改nginx.conf中的Host添加端口
- "9221:9223"
volumes:
- ./logs:/data/logs
- ./exports:/tmp/steel-browser-exports
- ./steel-chrome:/tmp/steel-chrome
- ./.cache:/app/.cache
- ./nginx.conf:/app/api/nginx.conf # 关键：内部nginx转发配置，Host可以修改CDP中/json接口返回信息中的IP+端口
- x11-unix:/tmp/.X11-unix # 关键：共享 X11 套接字
environment:
# 关闭无头浏览器
- CHROME_HEADLESS=false
# 声明网页端后台地址
- DOMAIN=${DOMAIN:-10.126.126.2:3001}
# 声明CDP地址，暂未测试出效果
- CDP_DOMAIN=${CDP_DOMAIN:-10.126.126.2:9221}
- LOG_STORAGE_ENABLED=true
- LOG_STORAGE_PATH=/data/logs/browser-logs.duckdb
# 添加虚拟显示和相关配置
- DISPLAY=:10
# 等待 xvfb 启动
depends_on:
xvfb:
condition: service_healthy # 等待健康检查通过
networks:
- steel-browser-network
networks:
steel-browser-network:
driver: bridge
volumes:
x11-unix:nginx.conf配置如下
events {
worker_connections 1024;
}
http {
map $http_upgrade $connection_upgrade {
default upgrade;
'' keep-alive;
}
server {
listen 9223;
location / {
proxy_pass http://127.0.0.1:9222;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection $connection_upgrade;
proxy_set_header Host $host:9221;
gzip off;
proxy_set_header Accept-Encoding "";
proxy_read_timeout 86400;
proxy_send_timeout 86400;
proxy_buffering off;
proxy_request_buffering off;
chunked_transfer_encoding on;
}
}
} 通过预构建xvfb避免每次都要安装xvfb
Dockerfile文件
FROM alpine:latest
# 安装 xvfb 和相关依赖
RUN apk add --no-cache xvfb dbus font-alias mkfontscale fontconfig && \
rm -f /tmp/.X10-lock /run/dbus/dbus.pid /run/dbus/system_bus_socket && \
mkdir -p /run/dbus /tmp/.X11-unix && \
chmod 1777 /tmp/.X11-unix
# 初始化 dbus
RUN dbus-daemon --system --fork || true
# 清理包缓存减小镜像体积
RUN apk cache clean
# 使用 tini 作为 init 系统
RUN apk add --no-cache tini && \
chmod +s /sbin/tini || true
ENTRYPOINT ["tini", "--"]
CMD ["sh", "-c", "Xvfb :10 -screen 0 2560x1440x24 -listen tcp -ac & sleep infinity"]通过以下命令构建镜像
docker build -t my-xvfb:latest -f Dockerfile .调整docker-compose.yml配置文件中的xvfb部分内容
services:
# 添加xvfb服务
xvfb:
image: my-xvfb:latest # 使用预构建的镜像
restart: unless-stopped
command: >
sh -c "rm -f /tmp/.X10-lock /run/dbus/dbus.pid /run/dbus/system_bus_socket &&
mkdir -p /run/dbus /tmp/.X11-unix &&
chmod 1777 /tmp/.X11-unix &&
dbus-daemon --system --fork &&
Xvfb :10 -screen 0 2560x1440x24 -listen tcp -ac &
sleep 1 && tail -f /dev/null"
environment:
# 由于steel-browser虚拟显示默认在:10，因此设置为:10，command中的设置同理
- DISPLAY=:10
volumes:
- x11-unix:/tmp/.X11-unix # 关键：共享 X11 套接字
# 健康检查
healthcheck:
test: ["CMD", "pgrep", "Xvfb"]
interval: 5s
timeout: 3s
retries: 30
start_period: 15s
networks:
- steel-browser-network通过python长期开启本地浏览器CDP
定时自动重启浏览器，当浏览器关闭时自动开启
import logging
from logging.handlers import TimedRotatingFileHandler
import subprocess
import psutil
import shutil
import schedule
import time
import os
import sys
# 配置日志基本设置
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
# 日志文件路径
log_file_path = 'log/output.log'
# 确保日志目录存在
log_directory = os.path.dirname(log_file_path)
if not os.path.exists(log_directory):
os.makedirs(log_directory)
# 创建TimedRotatingFileHandler，每天轮换日志文件
handler = TimedRotatingFileHandler(log_file_path, encoding='utf-8', when="midnight", interval=1, backupCount=30)
# 使用日志格式器
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
# 为日志记录系统添加handler
logging.getLogger().addHandler(handler)
# 定义 Chrome 路径（常见安装位置，如果不对请手动修改）
# "C:\Program Files\Google\Chrome\Application\chrome.exe"
chromePath = r"C:\Program Files\Google\Chrome\Application\chrome.exe"
# 定义调试端口
# 由于chrome默认只能127.0.0.1地址访问因此需要使用 netsh 端口转发 netsh interface portproxy add v4tov4 listenport=9222 listenaddress=你的局域网IP connectport=9222 connectaddress=10.126.126.2
debugPort = 9222
# 定义用户数据目录，用于保存你的登录信息、插件等，实现持久化
userDataDir = "D:\openclaw\Launch-ChromeDebug\chrome-debug-profile"
process_name = "chrome.exe"
process_search = f"--remote-debugging-port={debugPort}"
program_start_command = [chromePath, f"--remote-debugging-port={debugPort}", "--remote-debugging-address=0.0.0.0", f"--user-data-dir={userDataDir}", '--remote-allow-origins=*']
program_directory = "."
# 自动执行端口转发
subprocess.Popen(["cmd", "/c", f"netsh interface portproxy add v4tov4 listenport={debugPort} listenaddress=0.0.0.0 connectport={debugPort} connectaddress=127.0.0.1"], shell=True)
# 查询程序的pid
def get_process_id(process_name):
for proc in psutil.process_iter(['pid', 'name', 'cmdline']):
if process_name == proc.info['name']:
# logging.info(f"cmdline：{proc.info['cmdline']}")
for cmd_arg in proc.info['cmdline']:
if process_search in cmd_arg:
# logging.info(f"程序：{process_name}，pid：{proc.info['pid']}")
return proc.info['pid']
return None
# 检测程序是否在执行
def check_process_running(process_name):
# for process in psutil.process_iter(['name', 'cmdline']):
# if process.info['name'] and process.info['name'] == process_name:
# return True
# elif process.info['cmdline'] and process_name in process.info['cmdline']:
# return True
# return False
pid = get_process_id(process_name)
if pid:
return True
else:
return False
# 关闭服务器
def stop_program():
logging.info("服务器关闭")
if sys.platform.startswith('win'):
# subprocess.Popen('taskkill /F /IM ' + process_name, shell=True)
pid = get_process_id(process_name)
logging.info(f"程序：{process_name}，pid：{pid}")
if pid:
subprocess.Popen(f"taskkill /F /PID {pid}", shell=True)
# elif sys.platform.startswith('linux') or sys.platform.startswith('darwin'):
# subprocess.Popen(['pkill', '-f', 'your_program.py'])
# 启动服务器
def start_program():
logging.info("服务器启动")
if sys.platform.startswith('win'):
# subprocess.Popen(program_path, shell=True)
# subprocess.run([program_start_command], input=b'\n', shell=True, cwd=program_directory)
process = subprocess.Popen(program_start_command,
shell=True,
cwd=program_directory,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
universal_newlines=True,
encoding='gbk')
# # 实时获取控制台输出内容
# for line in process.stdout:
# out = process.stdout.readline()
# logging.info(out, end='')
# if out == '' and process.poll() is not None:
# break
# elif '请按任意键继续. . .' in out:
# # 当输出中出现特定内容时
# process.stdin.write('\n')
# process.stdin.flush() # 确保输入被发送
# #
time.sleep(10) # 等待10秒
# 强制终止子进程
process.kill()
# 等待进程结束
process.wait()
# 检查是否有错误发生
if process.returncode != 0:
logging.info(f"Command execution failed with exit code {process.returncode}")
# elif sys.platform.startswith('linux') or sys.platform.startswith('darwin'):
# subprocess.Popen(['python', 'your_program.py'])
# 重启任务
restarting = False
def stop_and_restart_job():
# 在更改全局变量前先声明
global restarting # 声明restarting为全局变量
restarting = True
while check_process_running(process_name): # 等待程序结束完成
stop_program()
time.sleep(10) # 等待10秒，确保程序完全关闭
logging.info(f"启动{process_name}程序")
while not check_process_running(process_name): # 查询程序是否启动，如果未启动则再次启动
start_program()
time.sleep(60)
restarting = False
# 当程序崩溃时自动启动
def auto_start_job():
# 在更改全局变量前先声明
global start_time # 声明start_time为全局变量
if not restarting and not check_process_running(process_name):
logging.info("检测到服务器未启动，开始自启动")
start_program()
# 定义任务调度函数
def schedule_job():
schedule.every(1).minutes.do(auto_start_job) # 每分钟检测一次程序是否崩溃
schedule.every().day.at("23:53").do(stop_and_restart_job) # 每天23：53重启服务器
# 启动任务调度器
schedule_job()
logging.info(f"检测{process_name}程序是否在执行：{check_process_running(process_name)}")
# if not check_process_running(process_name):
# logging.info("程序未执行，启动程序更新")
# run_steamcmd_update_command_realtime()
# start_program()
while True:
schedule.run_pending() # 检查是否有任务需要执行
time.sleep(1) # 避免过于频繁的检查