587 lines
20 KiB
Python
587 lines
20 KiB
Python
#!/home/hz1/miniconda3/envs/openclaw/bin/python3.12
|
||
"""
|
||
Web Capture API - 网页截图与代码提取服务
|
||
使用 agent-browser (Rust + Playwright) 或 Playwright
|
||
"""
|
||
|
||
import os
|
||
import json
|
||
import asyncio
|
||
import subprocess
|
||
import tempfile
|
||
import uuid
|
||
import time
|
||
from pathlib import Path
|
||
from flask import Flask, request, jsonify, send_file, render_template
|
||
from flask_cors import CORS
|
||
from datetime import datetime
|
||
|
||
# 检查 Playwright 是否可用
|
||
try:
|
||
from playwright.async_api import async_playwright
|
||
PLAYWRIGHT_AVAILABLE = True
|
||
# 尝试导入 stealth 反检测
|
||
try:
|
||
from playwright_stealth import stealth_async
|
||
STEALTH_AVAILABLE = True
|
||
except ImportError:
|
||
STEALTH_AVAILABLE = False
|
||
print("⚠️ playwright-stealth 未安装，反爬虫能力降低")
|
||
except ImportError:
|
||
PLAYWRIGHT_AVAILABLE = False
|
||
STEALTH_AVAILABLE = False
|
||
print("⚠️ Playwright 未安装，请运行: pip install playwright && playwright install chromium")
|
||
|
||
# 检查 agent-browser 是否可用
|
||
def check_agent_browser():
|
||
try:
|
||
result = subprocess.run(
|
||
["which", "agent-browser"],
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
return result.returncode == 0
|
||
except:
|
||
return False
|
||
|
||
AGENT_BROWSER_AVAILABLE = check_agent_browser()
|
||
|
||
app = Flask(__name__)
|
||
CORS(app)
|
||
|
||
# 配置
|
||
CAPTURE_DIR = Path(tempfile.gettempdir()) / "web_captures"
|
||
CAPTURE_DIR.mkdir(exist_ok=True)
|
||
|
||
# agent-browser socket 目录
|
||
os.environ.setdefault("AGENT_BROWSER_SOCKET_DIR", "/tmp/agent-browser-sockets")
|
||
Path(os.environ["AGENT_BROWSER_SOCKET_DIR"]).mkdir(exist_ok=True, parents=True)
|
||
|
||
|
||
class AgentBrowserSession:
|
||
"""agent-browser 会话管理器"""
|
||
|
||
def __init__(self, session_id=None):
|
||
self.session_id = session_id or str(uuid.uuid4())[:8]
|
||
self.base_cmd = ["agent-browser", "--session", self.session_id]
|
||
|
||
def run(self, cmd, timeout=60000):
|
||
"""执行 agent-browser 命令"""
|
||
full_cmd = self.base_cmd + cmd
|
||
# 设置必要的环境变量
|
||
env = os.environ.copy()
|
||
env['AGENT_BROWSER_SOCKET_DIR'] = '/tmp/agent-browser-sockets'
|
||
env['AGENT_BROWSER_USER_AGENT'] = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
|
||
try:
|
||
result = subprocess.run(
|
||
full_cmd,
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=timeout / 1000,
|
||
env=env
|
||
)
|
||
return result.returncode == 0, result.stdout, result.stderr
|
||
except subprocess.TimeoutExpired:
|
||
return False, "", "Command timeout"
|
||
except Exception as e:
|
||
return False, "", str(e)
|
||
|
||
def open(self, url):
|
||
"""打开网页"""
|
||
# 添加反爬虫检测的 User-Agent
|
||
env = os.environ.copy()
|
||
env['AGENT_BROWSER_USER_AGENT'] = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
|
||
return self.run(["open", url], timeout=30000)
|
||
|
||
def set_viewport(self, width, height):
|
||
"""设置视口大小"""
|
||
return self.run(["set", "viewport", str(width), str(height)])
|
||
|
||
def screenshot(self, output_path, full_page=False):
|
||
"""截图"""
|
||
cmd = ["screenshot", str(output_path)]
|
||
if full_page:
|
||
cmd.append("--full")
|
||
return self.run(cmd, timeout=30000)
|
||
|
||
def get_html(self):
|
||
"""获取HTML"""
|
||
success, output, error = self.run(
|
||
["eval", "document.documentElement.outerHTML"],
|
||
timeout=10000
|
||
)
|
||
return success, output, error
|
||
|
||
def scroll_down(self, pixels=800):
|
||
"""向下滚动"""
|
||
return self.run(["scroll", "down", str(pixels)])
|
||
|
||
def wait(self, ms):
|
||
"""等待"""
|
||
return self.run(["wait", str(ms)])
|
||
|
||
def close(self):
|
||
"""关闭浏览器"""
|
||
try:
|
||
self.run(["close"])
|
||
except:
|
||
pass
|
||
|
||
|
||
def capture_with_agent_browser(
|
||
url: str,
|
||
action: str = "screenshot",
|
||
scroll_times: int = 0,
|
||
scroll_delay: int = 1000,
|
||
full_page: bool = False,
|
||
viewport: dict = None,
|
||
wait_time: int = 2000
|
||
):
|
||
"""
|
||
使用 agent-browser 捕获网页
|
||
"""
|
||
session = AgentBrowserSession()
|
||
temp_file = None
|
||
|
||
try:
|
||
# 设置视口
|
||
if viewport:
|
||
session.set_viewport(viewport.get("width", 1920), viewport.get("height", 1080))
|
||
|
||
# 打开网页
|
||
success, stdout, stderr = session.open(url)
|
||
if not success:
|
||
# 检查是否是反爬虫拦截
|
||
if "403" in stdout or "Access Denied" in stdout:
|
||
return {"success": False, "error": "网站反爬虫拦截 (403)，建议使用 Playwright 后端或手动添加请求头"}
|
||
return {"success": False, "error": f"Failed to open URL: {stderr}"}
|
||
|
||
# 等待页面加载
|
||
session.wait(wait_time)
|
||
|
||
# 滚动加载动态内容
|
||
if scroll_times > 0:
|
||
for i in range(scroll_times):
|
||
session.scroll_down(800)
|
||
session.wait(scroll_delay)
|
||
|
||
# 执行操作
|
||
if action == "screenshot":
|
||
temp_file = CAPTURE_DIR / f"{session.session_id}.png"
|
||
success, _, error = session.screenshot(temp_file, full_page=full_page)
|
||
|
||
if success and temp_file.exists():
|
||
return {"success": True, "file_path": str(temp_file)}
|
||
else:
|
||
return {"success": False, "error": f"Screenshot failed: {error}"}
|
||
|
||
elif action == "html":
|
||
success, html, error = session.get_html()
|
||
if success:
|
||
return {"success": True, "html": html}
|
||
else:
|
||
return {"success": False, "error": f"Get HTML failed: {error}"}
|
||
|
||
else:
|
||
return {"success": False, "error": f"Unknown action: {action}"}
|
||
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
finally:
|
||
session.close()
|
||
|
||
|
||
async def capture_with_playwright(
|
||
url: str,
|
||
action: str = "screenshot",
|
||
scroll_times: int = 0,
|
||
scroll_delay: int = 1000,
|
||
full_page: bool = False,
|
||
viewport: dict = None,
|
||
wait_time: int = 2000
|
||
):
|
||
"""
|
||
使用 Playwright 捕获网页
|
||
Playwright 有更强的反爬虫能力
|
||
"""
|
||
if not PLAYWRIGHT_AVAILABLE:
|
||
return {"success": False, "error": "Playwright not installed. Run: pip install playwright && playwright install chromium"}
|
||
|
||
session_id = str(uuid.uuid4())[:8]
|
||
|
||
try:
|
||
async with async_playwright() as p:
|
||
# 启动浏览器，添加反检测配置
|
||
# 尝试使用真实 Chrome（如果安装了）
|
||
try:
|
||
# 首先尝试使用真实 Chrome 浏览器
|
||
browser = await p.chromium.launch(
|
||
headless=True,
|
||
channel='chrome', # 使用真实 Chrome 而不是 Chromium
|
||
args=[
|
||
'--disable-blink-features=AutomationControlled',
|
||
'--disable-dev-shm-usage',
|
||
'--no-sandbox',
|
||
'--ignore-certificate-errors',
|
||
]
|
||
)
|
||
except:
|
||
# 如果没有 Chrome，使用 Chromium
|
||
browser = await p.chromium.launch(
|
||
headless=True,
|
||
args=[
|
||
'--disable-blink-features=AutomationControlled',
|
||
'--disable-dev-shm-usage',
|
||
'--no-sandbox',
|
||
'--ignore-certificate-errors',
|
||
]
|
||
)
|
||
|
||
viewport_settings = viewport or {"width": 1920, "height": 1080}
|
||
|
||
# 创建上下文，模拟真实浏览器
|
||
context = await browser.new_context(
|
||
viewport={
|
||
"width": viewport_settings.get("width", 1920),
|
||
"height": viewport_settings.get("height", 1080)
|
||
},
|
||
user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.6099.109 Safari/537.36',
|
||
# 添加更多浏览器特征
|
||
locale='en-US',
|
||
timezone_id='America/New_York',
|
||
color_scheme='light',
|
||
has_touch=False,
|
||
is_mobile=False,
|
||
java_script_enabled=True,
|
||
accept_downloads=True,
|
||
)
|
||
|
||
# 注入反检测脚本
|
||
await context.add_init_script("""
|
||
// 移除 webdriver 标记
|
||
Object.defineProperty(navigator, 'webdriver', {
|
||
get: () => undefined
|
||
});
|
||
|
||
// 添加插件
|
||
Object.defineProperty(navigator, 'plugins', {
|
||
get: () => {
|
||
return [
|
||
{name: 'Chrome PDF Plugin', filename: 'internal-pdf-viewer', description: 'Portable Document Format'},
|
||
{name: 'Chrome PDF Viewer', filename: 'mhjfbmdgncjokjjpdkfnmlglggk', description: ''},
|
||
{name: 'Native Client', filename: 'internal-nacl-plugin', description: ''}
|
||
];
|
||
}
|
||
});
|
||
|
||
// 语言列表
|
||
Object.defineProperty(navigator, 'languages', {
|
||
get: () => ['en-US', 'en', 'zh-CN']
|
||
});
|
||
|
||
// 平台
|
||
Object.defineProperty(navigator, 'platform', {
|
||
get: () => 'Win32'
|
||
});
|
||
|
||
// 硬件并发
|
||
Object.defineProperty(navigator, 'hardwareConcurrency', {
|
||
get: () => 8
|
||
});
|
||
|
||
// 设备内存
|
||
Object.defineProperty(navigator, 'deviceMemory', {
|
||
get: () => 8
|
||
});
|
||
|
||
// Chrome 对象
|
||
window.chrome = {
|
||
runtime: {
|
||
PlatformOs: {MAC: 'mac', WIN: 'win', ANDROID: 'android', CROS: 'cros', LINUX: 'linux', OPENBSD: 'openbsd'},
|
||
PlatformArch: {ARM: 'arm', X86_32: 'x86-32', X86_64: 'x86-64'},
|
||
PlatformNaclArch: {ARM: 'arm', X86_32: 'x86-32', X86_64: 'x86-64'},
|
||
Request: function() {},
|
||
connect: function() {},
|
||
sendMessage: function() {}
|
||
},
|
||
loadTimes: function() {},
|
||
csi: function() {},
|
||
app: {}
|
||
};
|
||
""")
|
||
|
||
page = await context.new_page()
|
||
|
||
# 应用 stealth 反检测（如果可用）
|
||
if STEALTH_AVAILABLE:
|
||
await stealth_async(page)
|
||
|
||
try:
|
||
await page.goto(url, wait_until="domcontentloaded", timeout=30000)
|
||
except Exception as e:
|
||
# 如果 domcontentloaded 超时，尝试 commit
|
||
try:
|
||
await page.goto(url, wait_until="commit", timeout=30000)
|
||
except:
|
||
pass
|
||
|
||
# 固定等待时间（用于验证、加载等过程）
|
||
if wait_time > 0:
|
||
await page.wait_for_timeout(wait_time)
|
||
|
||
# 等待页面基本稳定
|
||
try:
|
||
await page.wait_for_load_state("networkidle", timeout=5000)
|
||
except:
|
||
# networkidle 超时没关系，继续执行
|
||
pass
|
||
|
||
if scroll_times > 0:
|
||
for i in range(scroll_times):
|
||
await page.evaluate("window.scrollBy(0, 800)")
|
||
await page.wait_for_timeout(scroll_delay)
|
||
await page.wait_for_timeout(500)
|
||
|
||
result = {}
|
||
|
||
if action == "screenshot":
|
||
temp_file = CAPTURE_DIR / f"{session_id}.png"
|
||
await page.screenshot(path=str(temp_file), full_page=full_page)
|
||
result = {"success": True, "file_path": str(temp_file)}
|
||
|
||
elif action == "html":
|
||
html = await page.content()
|
||
result = {"success": True, "html": html}
|
||
|
||
else:
|
||
result = {"success": False, "error": f"Unknown action: {action}"}
|
||
|
||
await browser.close()
|
||
return result
|
||
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
|
||
async def capture_with_cdp(
|
||
url_hint: str = "",
|
||
action: str = "screenshot",
|
||
cdp_port: int = 9222
|
||
):
|
||
"""
|
||
使用 CDP 连接到已打开的 Chrome 浏览器
|
||
用于方案三：手动验证后自动截图
|
||
"""
|
||
if not PLAYWRIGHT_AVAILABLE:
|
||
return {"success": False, "error": "Playwright not installed"}
|
||
|
||
try:
|
||
async with async_playwright() as p:
|
||
# 连接到已运行的 Chrome
|
||
browser = await p.chromium.connect_over_cdp(f"http://localhost:{cdp_port}")
|
||
|
||
# 获取所有页面
|
||
contexts = browser.contexts
|
||
result_data = []
|
||
|
||
for context in contexts:
|
||
for page in context.pages:
|
||
current_url = page.url
|
||
title = await page.title()
|
||
|
||
# 如果 URL 包含提示词或者是唯一页面
|
||
if url_hint in current_url or not url_hint:
|
||
if action == "screenshot":
|
||
temp_file = CAPTURE_DIR / f"cdp_capture_{uuid.uuid4()[:8]}.png"
|
||
await page.screenshot(path=str(temp_file), full_page=True)
|
||
result_data.append({
|
||
"url": current_url,
|
||
"title": title,
|
||
"file_path": str(temp_file)
|
||
})
|
||
elif action == "html":
|
||
html = await page.content()
|
||
result_data.append({
|
||
"url": current_url,
|
||
"title": title,
|
||
"html": html
|
||
})
|
||
|
||
if result_data:
|
||
return {"success": True, "pages": result_data}
|
||
else:
|
||
return {"success": False, "error": "No matching page found"}
|
||
|
||
except Exception as e:
|
||
return {"success": False, "error": f"CDP connection failed: {str(e)}"}
|
||
|
||
|
||
def capture_webpage(
|
||
url: str,
|
||
action: str = "screenshot",
|
||
scroll_times: int = 0,
|
||
scroll_delay: int = 1000,
|
||
full_page: bool = False,
|
||
viewport: dict = None,
|
||
wait_time: int = 2000,
|
||
backend: str = "auto"
|
||
):
|
||
"""
|
||
捕获网页
|
||
|
||
Args:
|
||
backend: "auto", "agent-browser", "playwright"
|
||
"""
|
||
# 选择后端
|
||
if backend == "auto":
|
||
if AGENT_BROWSER_AVAILABLE:
|
||
backend = "agent-browser"
|
||
elif PLAYWRIGHT_AVAILABLE:
|
||
backend = "playwright"
|
||
else:
|
||
return {"success": False, "error": "No browser backend available. Install agent-browser or playwright."}
|
||
|
||
# 使用对应后端
|
||
if backend == "agent-browser":
|
||
if not AGENT_BROWSER_AVAILABLE:
|
||
return {"success": False, "error": "agent-browser not available"}
|
||
return capture_with_agent_browser(
|
||
url, action, scroll_times, scroll_delay, full_page, viewport, wait_time
|
||
)
|
||
|
||
elif backend == "playwright":
|
||
if not PLAYWRIGHT_AVAILABLE:
|
||
return {"success": False, "error": "Playwright not available"}
|
||
|
||
try:
|
||
loop = asyncio.new_event_loop()
|
||
asyncio.set_event_loop(loop)
|
||
result = loop.run_until_complete(
|
||
capture_with_playwright(
|
||
url, action, scroll_times, scroll_delay, full_page, viewport, wait_time
|
||
)
|
||
)
|
||
loop.close()
|
||
return result
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
else:
|
||
return {"success": False, "error": f"Unknown backend: {backend}"}
|
||
|
||
|
||
@app.route('/')
|
||
def index():
|
||
"""主页"""
|
||
return render_template('index.html')
|
||
|
||
|
||
@app.route('/api')
|
||
def api_info():
|
||
"""API信息"""
|
||
return jsonify({
|
||
"service": "Web Capture API",
|
||
"version": "2.0.0",
|
||
"backends": {
|
||
"agent-browser": "available" if AGENT_BROWSER_AVAILABLE else "unavailable",
|
||
"playwright": "available" if PLAYWRIGHT_AVAILABLE else "unavailable"
|
||
},
|
||
"endpoints": {
|
||
"/api/capture": "POST - Capture webpage screenshot or HTML",
|
||
"/health": "GET - Health check"
|
||
}
|
||
})
|
||
|
||
|
||
@app.route('/health')
|
||
def health():
|
||
"""健康检查"""
|
||
status = "healthy" if (AGENT_BROWSER_AVAILABLE or PLAYWRIGHT_AVAILABLE) else "degraded"
|
||
return jsonify({
|
||
"status": status,
|
||
"agent-browser": "installed" if AGENT_BROWSER_AVAILABLE else "not found",
|
||
"playwright": "installed" if PLAYWRIGHT_AVAILABLE else "not found",
|
||
"timestamp": datetime.now().isoformat()
|
||
})
|
||
|
||
|
||
@app.route('/api/capture', methods=['POST'])
|
||
def capture():
|
||
"""
|
||
捕获网页接口
|
||
|
||
请求体:
|
||
{
|
||
"url": "https://example.com",
|
||
"action": "screenshot" | "html",
|
||
"scroll_times": 0,
|
||
"scroll_delay": 1000,
|
||
"full_page": false,
|
||
"viewport": {"width": 1920, "height": 1080},
|
||
"wait_time": 2000,
|
||
"backend": "auto" | "agent-browser" | "playwright" | "chrome-cdp",
|
||
"cdp_port": 9222 // 用于 chrome-cdp 后端，连接到已打开的 Chrome
|
||
}
|
||
"""
|
||
data = request.get_json()
|
||
|
||
if not data:
|
||
return jsonify({"success": False, "error": "No JSON data provided"}), 400
|
||
|
||
url = data.get("url")
|
||
if not url:
|
||
return jsonify({"success": False, "error": "URL is required"}), 400
|
||
|
||
# 添加协议前缀
|
||
if not url.startswith(("http://", "https://")):
|
||
url = "https://" + url
|
||
|
||
action = data.get("action", "screenshot")
|
||
scroll_times = int(data.get("scroll_times", 0))
|
||
scroll_delay = int(data.get("scroll_delay", 1000))
|
||
full_page = bool(data.get("full_page", False))
|
||
viewport = data.get("viewport")
|
||
wait_time = int(data.get("wait_time", 2000))
|
||
backend = data.get("backend", "auto")
|
||
cdp_port = int(data.get("cdp_port", 9222))
|
||
url_hint = data.get("url_hint", "")
|
||
|
||
result = capture_webpage(
|
||
url=url,
|
||
action=action,
|
||
scroll_times=scroll_times,
|
||
scroll_delay=scroll_delay,
|
||
full_page=full_page,
|
||
viewport=viewport,
|
||
wait_time=wait_time,
|
||
backend=backend,
|
||
|
||
|
||
)
|
||
|
||
if not result["success"]:
|
||
return jsonify(result), 400
|
||
|
||
if action == "screenshot":
|
||
file_path = result["file_path"]
|
||
return send_file(file_path, mimetype='image/png')
|
||
|
||
elif action == "html":
|
||
return jsonify(result)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
print("🌐 Web Capture API Starting...")
|
||
print(f"📁 Capture directory: {CAPTURE_DIR}")
|
||
print(f"🔧 Backends:")
|
||
print(f" - agent-browser: {'✅' if AGENT_BROWSER_AVAILABLE else '❌'}")
|
||
print(f" - playwright: {'✅' if PLAYWRIGHT_AVAILABLE else '❌'}")
|
||
|
||
if not AGENT_BROWSER_AVAILABLE and not PLAYWRIGHT_AVAILABLE:
|
||
print("\n⚠️ 没有可用的浏览器后端，请安装其中一个:")
|
||
print(" agent-browser: npm install -g agent-browser && agent-browser install")
|
||
print(" playwright: pip install playwright && playwright install chromium")
|
||
|
||
print("\n🚀 Server running on http://0.0.0.0:16025")
|
||
app.run(host='0.0.0.0', port=16025, debug=True) |