hz4th_coder
2a9f187ce8
- 新增 text 动作：提取页面可读文本（innerText + 空行/行尾清理，剔除脚本样式标签），agent-browser 与 playwright 双后端支持，自动提取页面标题 - 提取历史持久化(SQLite WAL)：每次捕获自动入库，截图文件存持久化目录 - 新增 API：GET /api/history(分页/类型筛选/搜索)、GET /api/history/<id>(详情)、GET /api/history/<id>/file(截图)、DELETE /api/history/<id> - 前端：新增提取文本选项+干净文本展示；新增提取历史卡片（分页/筛选/搜索/查看弹窗/删除/下载/复制）
814 lines
28 KiB
Python
814 lines
28 KiB
Python
#!/home/hz1/miniconda3/envs/openclaw/bin/python3.12
|
||
"""
|
||
Web Capture API - 网页截图与代码提取服务
|
||
使用 agent-browser (Rust + Playwright) 或 Playwright
|
||
"""
|
||
|
||
import os
|
||
import re
|
||
import json
|
||
import asyncio
|
||
import sqlite3
|
||
import shutil
|
||
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)
|
||
|
||
# 持久化存储（历史记录）
|
||
PROJECT_DIR = Path(__file__).parent
|
||
DATA_DIR = PROJECT_DIR / "data"
|
||
CAPTURE_DATA_DIR = DATA_DIR / "captures"
|
||
HISTORY_DB = DATA_DIR / "history.db"
|
||
DATA_DIR.mkdir(exist_ok=True)
|
||
CAPTURE_DATA_DIR.mkdir(exist_ok=True)
|
||
|
||
|
||
def get_db():
|
||
"""获取数据库连接（多进程安全：WAL + busy_timeout）"""
|
||
conn = sqlite3.connect(HISTORY_DB, timeout=15)
|
||
conn.row_factory = sqlite3.Row
|
||
conn.execute("PRAGMA journal_mode=WAL")
|
||
conn.execute("PRAGMA busy_timeout=15000")
|
||
return conn
|
||
|
||
|
||
def init_db():
|
||
"""初始化历史记录表"""
|
||
conn = get_db()
|
||
conn.execute("""
|
||
CREATE TABLE IF NOT EXISTS captures (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
url TEXT NOT NULL,
|
||
title TEXT DEFAULT '',
|
||
action TEXT NOT NULL,
|
||
backend TEXT DEFAULT '',
|
||
status TEXT NOT NULL DEFAULT 'success',
|
||
file_path TEXT DEFAULT '',
|
||
content TEXT DEFAULT '',
|
||
error TEXT DEFAULT '',
|
||
created_at TEXT NOT NULL
|
||
)
|
||
""")
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def save_capture(url, title, action, backend, status, file_path='', content='', error=''):
|
||
"""保存一条提取历史记录"""
|
||
conn = get_db()
|
||
cur = conn.execute(
|
||
"INSERT INTO captures (url, title, action, backend, status, file_path, content, error, created_at) "
|
||
"VALUES (?,?,?,?,?,?,?,?,?)",
|
||
(url, title, action, backend, status, file_path, content, error,
|
||
datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
|
||
)
|
||
conn.commit()
|
||
rid = cur.lastrowid
|
||
conn.close()
|
||
return rid
|
||
|
||
|
||
def persist_screenshot(src_path):
|
||
"""把临时截图复制到持久化目录，供历史记录长期访问"""
|
||
name = f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:8]}.png"
|
||
dest = CAPTURE_DATA_DIR / name
|
||
shutil.copy2(src_path, dest)
|
||
return str(dest)
|
||
|
||
|
||
def clean_text(raw):
|
||
"""清理可读文本：去掉行尾空格、压缩多余空行、去掉首尾空白"""
|
||
if not raw:
|
||
return ""
|
||
lines = [ln.rstrip() for ln in raw.split("\n")]
|
||
text = "\n".join(lines)
|
||
text = re.sub(r"\n{3,}", "\n\n", text)
|
||
return text.strip()
|
||
|
||
|
||
init_db()
|
||
|
||
# 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
|
||
|
||
@staticmethod
|
||
def _decode_eval(output):
|
||
"""agent-browser eval 结果按 JSON 编码返回，这里解码回原始字符串"""
|
||
if output is None:
|
||
return ""
|
||
out = output.strip()
|
||
try:
|
||
return json.loads(out)
|
||
except Exception:
|
||
# 非 JSON（如数字/布尔/纯文本），去掉可能的外层引号
|
||
return out.strip('\"\'')
|
||
|
||
def get_title(self):
|
||
"""获取页面标题"""
|
||
success, output, error = self.run(
|
||
["eval", "document.title"],
|
||
timeout=10000
|
||
)
|
||
if success:
|
||
return self._decode_eval(output)
|
||
return ""
|
||
|
||
def get_text(self):
|
||
"""提取页面可读文本（干净内容，剔除脚本/样式/标签）"""
|
||
success, output, error = self.run(
|
||
["eval", "document.body.innerText"],
|
||
timeout=30000
|
||
)
|
||
if success:
|
||
return True, self._decode_eval(output), ""
|
||
return False, "", 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)
|
||
|
||
# 提取页面标题
|
||
title = session.get_title()
|
||
|
||
# 执行操作
|
||
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, "title": title, "file_path": str(temp_file)}
|
||
else:
|
||
return {"success": False, "title": title, "error": f"Screenshot failed: {error}"}
|
||
|
||
elif action == "html":
|
||
success, html, error = session.get_html()
|
||
if success:
|
||
return {"success": True, "title": title, "html": html}
|
||
else:
|
||
return {"success": False, "title": title, "error": f"Get HTML failed: {error}"}
|
||
|
||
elif action == "text":
|
||
success, raw_text, error = session.get_text()
|
||
if success:
|
||
return {"success": True, "title": title, "text": clean_text(raw_text)}
|
||
else:
|
||
return {"success": False, "title": title, "error": f"Extract text failed: {error}"}
|
||
|
||
else:
|
||
return {"success": False, "title": title, "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)
|
||
|
||
# 提取页面标题
|
||
try:
|
||
title = await page.title()
|
||
except:
|
||
title = ""
|
||
|
||
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, "title": title, "file_path": str(temp_file)}
|
||
|
||
elif action == "html":
|
||
html = await page.content()
|
||
result = {"success": True, "title": title, "html": html}
|
||
|
||
elif action == "text":
|
||
raw_text = await page.evaluate("document.body.innerText")
|
||
result = {"success": True, "title": title, "text": clean_text(raw_text)}
|
||
|
||
else:
|
||
result = {"success": False, "title": title, "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"}
|
||
result = capture_with_agent_browser(
|
||
url, action, scroll_times, scroll_delay, full_page, viewport, wait_time
|
||
)
|
||
result["backend"] = "agent-browser"
|
||
return result
|
||
|
||
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()
|
||
result["backend"] = "playwright"
|
||
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.1.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/html/text) + 自动入库历史",
|
||
"/api/history": "GET - 历史记录分页列表 (?page&page_size&action&search)",
|
||
"/api/history/<id>": "GET - 历史记录详情 / DELETE - 删除记录",
|
||
"/api/history/<id>/file": "GET - 读取历史截图文件",
|
||
"/health": "GET - Health check"
|
||
}
|
||
})
|
||
|
||
|
||
@app.route('/api/history', methods=['GET'])
|
||
def history_list():
|
||
"""
|
||
历史记录分页列表
|
||
参数: page(默认1), page_size(默认15, 最大100), action(筛选: all/screenshot/html/text), search(按url/title模糊搜索)
|
||
"""
|
||
page = max(1, int(request.args.get('page', 1)))
|
||
page_size = min(100, max(1, int(request.args.get('page_size', 15))))
|
||
action = request.args.get('action', 'all')
|
||
search = request.args.get('search', '').strip()
|
||
|
||
where = []
|
||
params = []
|
||
if action and action != 'all':
|
||
where.append("action = ?")
|
||
params.append(action)
|
||
if search:
|
||
where.append("(url LIKE ? OR title LIKE ?)")
|
||
params.append(f"%{search}%")
|
||
params.append(f"%{search}%")
|
||
where_sql = (" WHERE " + " AND ".join(where)) if where else ""
|
||
|
||
conn = get_db()
|
||
total = conn.execute(f"SELECT COUNT(*) AS c FROM captures{where_sql}", params).fetchone()["c"]
|
||
rows = conn.execute(
|
||
f"SELECT id, url, title, action, backend, status, file_path, created_at, "
|
||
f"LENGTH(content) AS content_size, substr(content, 1, 200) AS content_preview, error "
|
||
f"FROM captures{where_sql} ORDER BY id DESC LIMIT ? OFFSET ?",
|
||
params + [page_size, (page - 1) * page_size]
|
||
).fetchall()
|
||
conn.close()
|
||
|
||
records = [dict(r) for r in rows]
|
||
return jsonify({
|
||
"success": True,
|
||
"page": page,
|
||
"page_size": page_size,
|
||
"total": total,
|
||
"total_pages": (total + page_size - 1) // page_size if total else 1,
|
||
"records": records
|
||
})
|
||
|
||
|
||
@app.route('/api/history/<int:rid>', methods=['GET'])
|
||
def history_detail(rid):
|
||
"""历史记录详情（含完整内容）"""
|
||
conn = get_db()
|
||
r = conn.execute("SELECT * FROM captures WHERE id=?", (rid,)).fetchone()
|
||
conn.close()
|
||
if not r:
|
||
return jsonify({"success": False, "error": "记录不存在"}), 404
|
||
return jsonify({"success": True, "record": dict(r)})
|
||
|
||
|
||
@app.route('/api/history/<int:rid>/file', methods=['GET'])
|
||
def history_file(rid):
|
||
"""读取历史截图文件"""
|
||
conn = get_db()
|
||
r = conn.execute("SELECT file_path FROM captures WHERE id=?", (rid,)).fetchone()
|
||
conn.close()
|
||
if not r or not r["file_path"] or not Path(r["file_path"]).exists():
|
||
return jsonify({"success": False, "error": "文件不存在或已删除"}), 404
|
||
return send_file(r["file_path"], mimetype='image/png')
|
||
|
||
|
||
@app.route('/api/history/<int:rid>', methods=['DELETE'])
|
||
def history_delete(rid):
|
||
"""删除历史记录（连带删除截图文件）"""
|
||
conn = get_db()
|
||
r = conn.execute("SELECT file_path FROM captures WHERE id=?", (rid,)).fetchone()
|
||
if r:
|
||
conn.execute("DELETE FROM captures WHERE id=?", (rid,))
|
||
conn.commit()
|
||
conn.close()
|
||
if r and r["file_path"]:
|
||
fp = Path(r["file_path"])
|
||
if fp.exists() and str(fp).startswith(str(CAPTURE_DATA_DIR)):
|
||
try:
|
||
fp.unlink()
|
||
except Exception:
|
||
pass
|
||
return jsonify({"success": True})
|
||
|
||
|
||
@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
|
||
)
|
||
|
||
backend_used = result.get("backend", backend)
|
||
title = result.get("title", "")
|
||
|
||
if not result["success"]:
|
||
save_capture(url, title, action, backend_used, "failed", error=result.get("error", ""))
|
||
return jsonify(result), 400
|
||
|
||
if action == "screenshot":
|
||
persisted = persist_screenshot(result["file_path"])
|
||
save_capture(url, title, action, backend_used, "success", file_path=persisted)
|
||
return send_file(persisted, mimetype='image/png')
|
||
|
||
elif action == "html":
|
||
save_capture(url, title, action, backend_used, "success", content=result["html"])
|
||
return jsonify(result)
|
||
|
||
elif action == "text":
|
||
save_capture(url, title, action, backend_used, "success", content=result["text"])
|
||
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) |