hz4th_coder
82ff8a2610
- 新增 GET /api/backends 返回可用后端/默认顺序/说明，供前端动态渲染 - 前端表单新增「捕获后端」下拉框: 自动(默认,优先agent-browser超时自动回退playwright)/agent-browser/playwright/chrome-cdp，不可用的置灰 - 接通 chrome-cdp 后端: 之前函数已实现但 capture_webpage 未接线(传 backend=chrome-cdp 会报 Unknown backend)，现支持 screenshot/html/text，连接已开的 Chrome(--remote-debugging-port) - 前端 currentData 携带 backend 字段
1436 lines
55 KiB
Python
1436 lines
55 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 base64
|
||
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"
|
||
HTML_DATA_DIR = DATA_DIR / "html" # 原始HTML按月归档目录
|
||
HISTORY_DB = DATA_DIR / "history.db"
|
||
DATA_DIR.mkdir(exist_ok=True)
|
||
CAPTURE_DATA_DIR.mkdir(exist_ok=True)
|
||
HTML_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 '',
|
||
caller TEXT DEFAULT '游客',
|
||
call_method TEXT DEFAULT 'api',
|
||
html_path TEXT DEFAULT '',
|
||
created_at TEXT NOT NULL
|
||
)
|
||
""")
|
||
# 兼容旧库：缺列则 ALTER TABLE 补充
|
||
cols = [row[1] for row in conn.execute("PRAGMA table_info(captures)").fetchall()]
|
||
for col, ddl in (("caller", "TEXT DEFAULT '游客'"),
|
||
("call_method", "TEXT DEFAULT 'api'"),
|
||
("html_path", "TEXT DEFAULT ''")):
|
||
if col not in cols:
|
||
conn.execute(f"ALTER TABLE captures ADD COLUMN {col} {ddl}")
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def save_capture(url, title, action, backend, status, file_path='', content='', error='',
|
||
caller='游客', call_method='api', html_path=''):
|
||
"""保存一条提取历史记录（含调用者/调用方式/原始HTML路径）"""
|
||
conn = get_db()
|
||
cur = conn.execute(
|
||
"INSERT INTO captures (url, title, action, backend, status, file_path, content, error, caller, call_method, html_path, created_at) "
|
||
"VALUES (?,?,?,?,?,?,?,?,?,?,?,?)",
|
||
(url, title, action, backend, status, file_path, content, error, caller, call_method, html_path,
|
||
datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
|
||
)
|
||
conn.commit()
|
||
rid = cur.lastrowid
|
||
conn.close()
|
||
return rid
|
||
|
||
|
||
def save_raw_html(html, url=''):
|
||
"""把最原始 HTML 保存到 data/html/<YYYY-MM>/ 按月目录归档；失败返回 ''"""
|
||
if not html:
|
||
return ''
|
||
try:
|
||
month = datetime.now().strftime("%Y-%m")
|
||
month_dir = HTML_DATA_DIR / month
|
||
month_dir.mkdir(parents=True, exist_ok=True)
|
||
name = f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:8]}.html"
|
||
dest = month_dir / name
|
||
dest.write_text(html if isinstance(html, str) else str(html), encoding="utf-8")
|
||
return str(dest)
|
||
except Exception:
|
||
return ''
|
||
|
||
|
||
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()
|
||
|
||
# ===== 按需截图（AI智能判断滚动）配置 =====
|
||
CONFIG_FILE = DATA_DIR / "config.json"
|
||
|
||
DEFAULT_SMART_CONFIG = {
|
||
"base_url": "https://www.autodl.art/api/v1",
|
||
"api_key": "F9MBfolzuapqTsD4KmUf9qen720rXvUZ3Sp3IrWiCTukqonx",
|
||
"model": "qwen3.6-plus",
|
||
"prompt": (
|
||
"你是网页内容完整性判断助手。下面是一张网页滚动截图的当前视口画面。"
|
||
"请判断：这个网页的主题内容（正文/主要内容）是否已经完整截取完成，是否还需要继续向下滚动？\n"
|
||
"规则：\n"
|
||
"1. 如果当前视口底部已经是页面底部（如页脚、版权信息、导航链接列表、没有更多正文），说明已完整 → complete=true\n"
|
||
"2. 如果底部还有正文内容被截断、或还有更多正文需要滚动才能看到 → complete=false\n"
|
||
"只输出JSON：{\"complete\": true/false, \"reason\": \"简短理由\"}"
|
||
),
|
||
"max_scrolls": 20,
|
||
"scroll_ratio": 0.85, # 每次滚动视口高度的比例（重叠 15% 保证拼接连续）
|
||
"timeout": 120
|
||
}
|
||
|
||
|
||
def load_smart_config():
|
||
"""读取按需截图 LLM 配置（缺省回默认）"""
|
||
cfg = json.loads(json.dumps(DEFAULT_SMART_CONFIG))
|
||
if CONFIG_FILE.exists():
|
||
try:
|
||
saved = json.loads(CONFIG_FILE.read_text(encoding="utf-8"))
|
||
if isinstance(saved, dict):
|
||
for k in DEFAULT_SMART_CONFIG:
|
||
if k in saved:
|
||
cfg[k] = saved[k]
|
||
except Exception:
|
||
pass
|
||
return cfg
|
||
|
||
|
||
def save_smart_config(cfg):
|
||
"""保存按需截图 LLM 配置"""
|
||
merged = json.loads(json.dumps(DEFAULT_SMART_CONFIG))
|
||
if isinstance(cfg, dict):
|
||
for k in DEFAULT_SMART_CONFIG:
|
||
if k in cfg and cfg[k] not in (None, ""):
|
||
merged[k] = cfg[k]
|
||
CONFIG_FILE.parent.mkdir(exist_ok=True)
|
||
CONFIG_FILE.write_text(json.dumps(merged, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
return merged
|
||
|
||
|
||
def extract_json_from_content(content):
|
||
"""从大模型输出中稳健提取 JSON 对象（兼容 ```json 代码块包裹、前后废话）"""
|
||
if not content:
|
||
return None
|
||
text = str(content).strip()
|
||
if text.startswith("```"):
|
||
text = re.sub(r"^```[a-zA-Z]*\s*", "", text)
|
||
text = re.sub(r"\s*```$", "", text)
|
||
s, e = text.find("{"), text.rfind("}")
|
||
if s != -1 and e > s:
|
||
text = text[s:e + 1]
|
||
try:
|
||
return json.loads(text)
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def llm_judge_screenshot(image_path, cfg):
|
||
"""
|
||
调用视觉大模型判断当前截图是否已覆盖主题内容、是否还需滚动
|
||
返回: {"complete": bool, "reason": str}
|
||
异常时抛 ValueError（调用方决定如何处置）
|
||
"""
|
||
import requests
|
||
img_b64 = base64.b64encode(Path(image_path).read_bytes()).decode()
|
||
url = cfg["base_url"].rstrip("/") + "/chat/completions"
|
||
payload = {
|
||
"model": cfg["model"],
|
||
"messages": [{
|
||
"role": "user",
|
||
"content": [
|
||
{"type": "text", "text": cfg["prompt"]},
|
||
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{img_b64}"}}
|
||
]
|
||
}],
|
||
"max_tokens": 300
|
||
}
|
||
headers = {"Authorization": f"Bearer {cfg['api_key']}", "Content-Type": "application/json"}
|
||
resp = requests.post(url, headers=headers, json=payload, timeout=int(cfg.get("timeout", 120)))
|
||
if resp.status_code != 200:
|
||
raise ValueError(f"LLM 接口返回 {resp.status_code}: {resp.text[:200]}")
|
||
try:
|
||
content = resp.json()["choices"][0]["message"]["content"]
|
||
except Exception:
|
||
raise ValueError(f"LLM 响应格式异常: {resp.text[:200]}")
|
||
parsed = extract_json_from_content(content)
|
||
if not parsed or "complete" not in parsed:
|
||
raise ValueError(f"LLM 判断结果解析失败，原始输出: {content[:200]}")
|
||
return {"complete": bool(parsed["complete"]), "reason": str(parsed.get("reason", ""))}
|
||
|
||
|
||
def stitch_images(image_paths, out_path):
|
||
"""把多张视口截图纵向拼接成一张长图（所有图对齐到最宽宽度）"""
|
||
from PIL import Image
|
||
images = [Image.open(p).convert("RGB") for p in image_paths]
|
||
if not images:
|
||
return None
|
||
width = max(im.width for im in images)
|
||
height = sum(im.height for im in images)
|
||
canvas = Image.new("RGB", (width, height), "white")
|
||
y = 0
|
||
for im in images:
|
||
canvas.paste(im, (0, y))
|
||
y += im.height
|
||
out_path.parent.mkdir(exist_ok=True, parents=True)
|
||
canvas.save(out_path)
|
||
return str(out_path)
|
||
|
||
|
||
# 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'
|
||
# agent-browser 内部导航等待约 30s 超时，这里留足 45s 让它返回自己的超时报错（避免被 subprocess 硬杀）
|
||
return self.run(["open", url], timeout=45000)
|
||
|
||
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 后端或手动添加请求头"}
|
||
if "timed out" in stderr.lower() or "timeout" in stderr.lower():
|
||
return {"success": False, "error": "页面加载超时（可能是慢加载/持续加载页面），已自动切换 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():
|
||
res = {"success": True, "title": title, "file_path": str(temp_file)}
|
||
# 顺带抓取最原始 HTML 存档
|
||
ok, html_out, _ = session.get_html()
|
||
if ok and html_out:
|
||
res["html"] = session._decode_eval(html_out)
|
||
return res
|
||
else:
|
||
return {"success": False, "title": title, "error": f"Screenshot failed: {error}"}
|
||
|
||
elif action == "html":
|
||
success, html, error = session.get_html()
|
||
if success:
|
||
# 修复：eval 结果按 JSON 编码返回，需解码成最原始 HTML
|
||
return {"success": True, "title": title, "html": session._decode_eval(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:
|
||
res = {"success": True, "title": title, "text": clean_text(raw_text)}
|
||
# 顺带抓取最原始 HTML 存档
|
||
ok, html_out, _ = session.get_html()
|
||
if ok and html_out:
|
||
res["html"] = session._decode_eval(html_out)
|
||
return res
|
||
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=60000)
|
||
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)}
|
||
# 顺带抓取最原始 HTML 存档
|
||
try:
|
||
result["html"] = await page.content()
|
||
except Exception:
|
||
pass
|
||
|
||
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)}
|
||
# 顺带抓取最原始 HTML 存档
|
||
try:
|
||
result["html"] = await page.content()
|
||
except Exception:
|
||
pass
|
||
|
||
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
|
||
})
|
||
|
||
elif action == "text":
|
||
text = await page.evaluate("document.body.innerText")
|
||
result_data.append({
|
||
"url": current_url,
|
||
"title": title,
|
||
"text": clean_text(text)
|
||
})
|
||
|
||
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 _scroll_pixels(view_h, ratio):
|
||
"""按视口高度与比例计算单次滚动像素（下限 50）"""
|
||
return max(50, int(view_h * float(ratio)))
|
||
|
||
|
||
def smart_capture_agent_browser(url, cfg, wait_time, viewport):
|
||
"""agent-browser 后端：滚动截图 + 视觉大模型实时判断"""
|
||
session = AgentBrowserSession()
|
||
vw = viewport.get("width", 1280) if viewport else 1280
|
||
vh = viewport.get("height", 700) if viewport else 700
|
||
steps = []
|
||
shot_paths = []
|
||
max_scrolls = int(cfg.get("max_scrolls", 20))
|
||
stop_reason = ""
|
||
|
||
try:
|
||
session.set_viewport(vw, vh)
|
||
success, _, err = session.open(url)
|
||
if not success:
|
||
if "timed out" in err.lower() or "timeout" in err.lower():
|
||
return {"success": False, "error": "页面加载超时（可能是慢加载/持续加载页面），已自动切换 Playwright 后端重试；若仍失败请增大等待时间或换其他后端"}
|
||
return {"success": False, "error": f"打开网页失败: {err}"}
|
||
session.wait(wait_time)
|
||
|
||
# 先回到顶部
|
||
session.run(["eval", "window.scrollTo(0, 0)"], timeout=5000)
|
||
session.wait(300)
|
||
|
||
for i in range(max_scrolls):
|
||
step_no = i + 1
|
||
shot = CAPTURE_DIR / f"smart_{session.session_id}_{step_no}.png"
|
||
ok, _, serr = session.screenshot(shot)
|
||
if not ok or not shot.exists():
|
||
return {"success": False, "error": f"第 {step_no} 步截图失败: {serr}"}
|
||
shot_paths.append(shot)
|
||
|
||
# 视觉大模型实时判断
|
||
try:
|
||
judge = llm_judge_screenshot(shot, cfg)
|
||
steps.append({"step": step_no, "complete": judge["complete"], "reason": judge["reason"]})
|
||
except ValueError as ve:
|
||
# 判断失败：保守停止，保留已截内容
|
||
steps.append({"step": step_no, "complete": True, "reason": f"⚠️ 大模型判断失败，停止滚动: {ve}"})
|
||
stop_reason = str(ve)
|
||
break
|
||
|
||
if judge["complete"]:
|
||
break
|
||
|
||
# 记录滚动前位置，防止页面无法再滚动
|
||
try:
|
||
_, before, _ = session.run(["eval", "window.scrollY"], timeout=5000)
|
||
before = float(session._decode_eval(before) or 0)
|
||
except Exception:
|
||
before = -1
|
||
|
||
session.scroll_down(_scroll_pixels(vh, cfg.get("scroll_ratio", 0.85)))
|
||
session.wait(int(wait_time) if wait_time else 500)
|
||
|
||
if before >= 0:
|
||
try:
|
||
_, after_raw, _ = session.run(["eval", "window.scrollY"], timeout=5000)
|
||
after = float(session._decode_eval(after_raw) or before)
|
||
if after <= before:
|
||
steps.append({"step": step_no, "complete": True, "reason": "已滚动到页面底部（滚动位置不再变化）"})
|
||
break
|
||
except Exception:
|
||
pass
|
||
else:
|
||
steps.append({"step": max_scrolls, "complete": True, "reason": "达到最大滚动次数上限，停止"})
|
||
|
||
# 拼接长图
|
||
out = CAPTURE_DATA_DIR / f"smart_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:8]}.png"
|
||
stitched = stitch_images(shot_paths, out)
|
||
title = session.get_title()
|
||
|
||
res = {
|
||
"success": True,
|
||
"title": title,
|
||
"file_path": stitched,
|
||
"steps": steps,
|
||
"total_steps": len(shot_paths),
|
||
"stop_reason": stop_reason,
|
||
"backend": "agent-browser"
|
||
}
|
||
# 顺带抓取最原始 HTML 存档
|
||
ok, html_out, _ = session.get_html()
|
||
if ok and html_out:
|
||
res["html"] = session._decode_eval(html_out)
|
||
return res
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
finally:
|
||
session.close()
|
||
|
||
|
||
async def smart_capture_playwright(url, cfg, wait_time, viewport):
|
||
"""playwright 后端：滚动截图 + 视觉大模型实时判断"""
|
||
if not PLAYWRIGHT_AVAILABLE:
|
||
return {"success": False, "error": "Playwright not installed"}
|
||
vw = viewport.get("width", 1280) if viewport else 1280
|
||
vh = viewport.get("height", 700) if viewport else 700
|
||
steps = []
|
||
shot_paths = []
|
||
max_scrolls = int(cfg.get("max_scrolls", 20))
|
||
stop_reason = ""
|
||
|
||
try:
|
||
async with async_playwright() as p:
|
||
try:
|
||
browser = await p.chromium.launch(
|
||
headless=True, channel='chrome',
|
||
args=['--disable-blink-features=AutomationControlled', '--disable-dev-shm-usage',
|
||
'--no-sandbox', '--ignore-certificate-errors']
|
||
)
|
||
except Exception:
|
||
browser = await p.chromium.launch(
|
||
headless=True,
|
||
args=['--disable-blink-features=AutomationControlled', '--disable-dev-shm-usage',
|
||
'--no-sandbox', '--ignore-certificate-errors']
|
||
)
|
||
context = await browser.new_context(
|
||
viewport={"width": vw, "height": vh},
|
||
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'
|
||
)
|
||
page = await context.new_page()
|
||
if STEALTH_AVAILABLE:
|
||
try:
|
||
from playwright_stealth import stealth_async
|
||
await stealth_async(page)
|
||
except Exception:
|
||
pass
|
||
try:
|
||
await page.goto(url, wait_until="domcontentloaded", timeout=60000)
|
||
except Exception:
|
||
try:
|
||
await page.goto(url, wait_until="commit", timeout=30000)
|
||
except Exception:
|
||
pass
|
||
if wait_time > 0:
|
||
await page.wait_for_timeout(wait_time)
|
||
try:
|
||
await page.wait_for_load_state("networkidle", timeout=5000)
|
||
except Exception:
|
||
pass
|
||
try:
|
||
await page.evaluate("window.scrollTo(0, 0)")
|
||
await page.wait_for_timeout(300)
|
||
except Exception:
|
||
# 页面导航导致上下文销毁，等稳定后再试
|
||
try:
|
||
await page.wait_for_timeout(800)
|
||
await page.evaluate("window.scrollTo(0, 0)")
|
||
except Exception:
|
||
pass
|
||
|
||
async def _safe_scroll_y():
|
||
try:
|
||
return await page.evaluate("window.scrollY")
|
||
except Exception:
|
||
return None
|
||
|
||
for i in range(max_scrolls):
|
||
step_no = i + 1
|
||
shot = CAPTURE_DIR / f"smart_pw_{uuid.uuid4().hex[:8]}_{step_no}.png"
|
||
try:
|
||
await page.screenshot(path=str(shot))
|
||
except Exception:
|
||
try:
|
||
await page.wait_for_timeout(800)
|
||
await page.screenshot(path=str(shot))
|
||
except Exception as e:
|
||
return {"success": False, "error": f"第 {step_no} 步截图失败: {e}"}
|
||
shot_paths.append(shot)
|
||
|
||
try:
|
||
judge = llm_judge_screenshot(shot, cfg)
|
||
steps.append({"step": step_no, "complete": judge["complete"], "reason": judge["reason"]})
|
||
except ValueError as ve:
|
||
steps.append({"step": step_no, "complete": True, "reason": f"⚠️ 大模型判断失败，停止滚动: {ve}"})
|
||
stop_reason = str(ve)
|
||
break
|
||
|
||
if judge["complete"]:
|
||
break
|
||
|
||
before = await _safe_scroll_y()
|
||
try:
|
||
await page.evaluate(f"window.scrollBy(0, {_scroll_pixels(vh, cfg.get('scroll_ratio', 0.85))})")
|
||
await page.wait_for_timeout(wait_time if wait_time else 500)
|
||
except Exception:
|
||
# 滚动失败视为到页面底部
|
||
steps.append({"step": step_no, "complete": True, "reason": "页面导航/滚动异常，视为已到末尾"})
|
||
break
|
||
after = await _safe_scroll_y()
|
||
if before is not None and after is not None and after <= before:
|
||
steps.append({"step": step_no, "complete": True, "reason": "已滚动到页面底部（滚动位置不再变化）"})
|
||
break
|
||
else:
|
||
steps.append({"step": max_scrolls, "complete": True, "reason": "达到最大滚动次数上限，停止"})
|
||
|
||
out = CAPTURE_DATA_DIR / f"smart_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:8]}.png"
|
||
stitched = stitch_images(shot_paths, out)
|
||
try:
|
||
title = await page.title()
|
||
except Exception:
|
||
title = ""
|
||
res = {
|
||
"success": True,
|
||
"title": title,
|
||
"file_path": stitched,
|
||
"steps": steps,
|
||
"total_steps": len(shot_paths),
|
||
"stop_reason": stop_reason,
|
||
"backend": "playwright"
|
||
}
|
||
# 顺带抓取最原始 HTML 存档
|
||
try:
|
||
res["html"] = await page.content()
|
||
except Exception:
|
||
pass
|
||
await browser.close()
|
||
return res
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
|
||
def smart_capture(url, cfg, wait_time, viewport, backend="auto"):
|
||
"""按需截图总入口：滚动截图 + 视觉大模型实时判断，返回拼接长图"""
|
||
if backend == "auto":
|
||
backend = "agent-browser" if AGENT_BROWSER_AVAILABLE else "playwright"
|
||
|
||
if backend == "agent-browser":
|
||
result = smart_capture_agent_browser(url, cfg, wait_time, viewport)
|
||
# 打开页面超时 → 自动切 Playwright 重试（domcontentloaded 更宽容）
|
||
if (not result.get("success")
|
||
and "自动切换 Playwright" in (result.get("error") or "")
|
||
and PLAYWRIGHT_AVAILABLE):
|
||
try:
|
||
loop = asyncio.new_event_loop()
|
||
asyncio.set_event_loop(loop)
|
||
result = loop.run_until_complete(smart_capture_playwright(url, cfg, wait_time, viewport))
|
||
loop.close()
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
return result
|
||
elif backend == "playwright":
|
||
try:
|
||
loop = asyncio.new_event_loop()
|
||
asyncio.set_event_loop(loop)
|
||
result = loop.run_until_complete(smart_capture_playwright(url, cfg, wait_time, viewport))
|
||
loop.close()
|
||
return result
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
return {"success": False, "error": f"Unknown backend: {backend}"}
|
||
|
||
|
||
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",
|
||
cdp_port: int = 9222
|
||
):
|
||
"""
|
||
捕获网页
|
||
|
||
Args:
|
||
backend: "auto", "agent-browser", "playwright", "chrome-cdp"
|
||
"""
|
||
# 选择后端
|
||
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"
|
||
# agent-browser 打开页面超时（慢加载/持续加载页）→ 自动切 Playwright 重试（domcontentloaded 更宽容）
|
||
if (not result.get("success")
|
||
and "自动切换 Playwright" in (result.get("error") or "")
|
||
and PLAYWRIGHT_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"
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
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)}
|
||
|
||
elif backend == "chrome-cdp":
|
||
# 连接已打开的 Chrome（需 --remote-debugging-port=<cdp_port>），归一化为单结果
|
||
if not PLAYWRIGHT_AVAILABLE:
|
||
return {"success": False, "error": "Playwright not installed (chrome-cdp 依赖 playwright)"}
|
||
try:
|
||
loop = asyncio.new_event_loop()
|
||
asyncio.set_event_loop(loop)
|
||
r = loop.run_until_complete(
|
||
capture_with_cdp(url_hint=url, action=action, cdp_port=cdp_port)
|
||
)
|
||
loop.close()
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
if not r.get("success"):
|
||
return r
|
||
if not r.get("pages"):
|
||
return {"success": False, "error": "chrome-cdp 未找到匹配页面（请确认 Chrome 已开 --remote-debugging-port）"}
|
||
p0 = r["pages"][0]
|
||
res = {"success": True, "title": p0.get("title", ""), "backend": "chrome-cdp"}
|
||
if action == "screenshot":
|
||
res["file_path"] = p0["file_path"]
|
||
elif action == "html":
|
||
res["html"] = p0["html"]
|
||
elif action == "text":
|
||
res["text"] = p0["text"]
|
||
return res
|
||
|
||
else:
|
||
return {"success": False, "error": f"Unknown backend: {backend}"}
|
||
|
||
|
||
@app.route('/')
|
||
def index():
|
||
"""主页"""
|
||
return render_template('index.html')
|
||
|
||
|
||
@app.route('/api/backends')
|
||
def api_backends():
|
||
"""返回可用后端及默认选择顺序，供前端渲染选择器"""
|
||
available = []
|
||
if AGENT_BROWSER_AVAILABLE:
|
||
available.append("agent-browser")
|
||
if PLAYWRIGHT_AVAILABLE:
|
||
available.append("playwright")
|
||
# chrome-cdp 依赖 playwright 的 connect_over_cdp
|
||
available.append("chrome-cdp")
|
||
unavailable = []
|
||
if not AGENT_BROWSER_AVAILABLE:
|
||
unavailable.append("agent-browser")
|
||
if not PLAYWRIGHT_AVAILABLE:
|
||
unavailable.append("playwright")
|
||
unavailable.append("chrome-cdp")
|
||
return jsonify({
|
||
"success": True,
|
||
"default": "auto",
|
||
"available": available,
|
||
"unavailable": unavailable,
|
||
"priority": available + unavailable,
|
||
"labels": {
|
||
"auto": "自动（推荐：优先 agent-browser，打开超时自动回退 Playwright）",
|
||
"agent-browser": "agent-browser（Rust 版，反爬强、快，默认首选）",
|
||
"playwright": "Playwright（Python 版，domcontentloaded 更宽容，慢页面更稳）",
|
||
"chrome-cdp": "Chrome CDP（连接已打开的 Chrome，需 --remote-debugging-port 启动）"
|
||
}
|
||
})
|
||
|
||
|
||
@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/smart) + 自动入库历史(记录调用者/方式/原始HTML)",
|
||
"/api/backends": "GET - 可用后端列表与默认顺序(供前端渲染后端选择器)",
|
||
"/api/history": "GET - 历史记录分页列表 (?page&page_size&action&search)",
|
||
"/api/history/<id>": "GET - 历史记录详情 / DELETE - 删除记录",
|
||
"/api/history/<id>/file": "GET - 读取历史截图文件",
|
||
"/api/history/<id>/html": "GET - 读取保存的原始HTML文件",
|
||
"/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, caller, call_method, html_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>/html', methods=['GET'])
|
||
def history_html_file(rid):
|
||
"""读取保存的原始 HTML 文件（data/html/<月份>/xxx.html）"""
|
||
conn = get_db()
|
||
r = conn.execute("SELECT html_path FROM captures WHERE id=?", (rid,)).fetchone()
|
||
conn.close()
|
||
if not r or not r["html_path"] or not Path(r["html_path"]).exists():
|
||
return jsonify({"success": False, "error": "原始HTML文件不存在或已删除"}), 404
|
||
return send_file(r["html_path"], mimetype='text/html; charset=utf-8')
|
||
|
||
|
||
@app.route('/api/history/<int:rid>', methods=['DELETE'])
|
||
def history_delete(rid):
|
||
"""删除历史记录（连带删除截图文件 / 原始HTML文件）"""
|
||
conn = get_db()
|
||
r = conn.execute("SELECT file_path, html_path FROM captures WHERE id=?", (rid,)).fetchone()
|
||
if r:
|
||
conn.execute("DELETE FROM captures WHERE id=?", (rid,))
|
||
conn.commit()
|
||
conn.close()
|
||
if r:
|
||
for fp_s in (r["file_path"], r["html_path"]):
|
||
if not fp_s:
|
||
continue
|
||
fp = Path(fp_s)
|
||
if fp.exists() and str(fp).startswith(str(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/smart/config', methods=['GET', 'POST'])
|
||
def smart_config_endpoint():
|
||
"""获取 / 保存按需截图 LLM 配置"""
|
||
if request.method == 'GET':
|
||
cfg = load_smart_config()
|
||
return jsonify({"success": True, "config": cfg})
|
||
data = request.get_json() or {}
|
||
cfg = save_smart_config(data.get("config") or data)
|
||
return jsonify({"success": True, "config": cfg})
|
||
|
||
|
||
@app.route('/api/smart/test', methods=['POST'])
|
||
def smart_test_endpoint():
|
||
"""测试按需截图 LLM 接口连通性（发一条小文本消息验证 base_url/api_key/model）"""
|
||
import requests
|
||
data = request.get_json() or {}
|
||
cfg = dict(load_smart_config())
|
||
if data.get("config"):
|
||
for k in DEFAULT_SMART_CONFIG:
|
||
if k in data["config"] and data["config"][k] not in (None, ""):
|
||
cfg[k] = data["config"][k]
|
||
elif data.get("base_url") or data.get("api_key") or data.get("model"):
|
||
for k in ("base_url", "api_key", "model"):
|
||
if data.get(k):
|
||
cfg[k] = data[k]
|
||
|
||
url = cfg["base_url"].rstrip("/") + "/chat/completions"
|
||
payload = {"model": cfg["model"], "messages": [{"role": "user", "content": "ping"}], "max_tokens": 5}
|
||
headers = {"Authorization": f"Bearer {cfg['api_key']}", "Content-Type": "application/json"}
|
||
t0 = time.time()
|
||
try:
|
||
resp = requests.post(url, headers=headers, json=payload, timeout=int(cfg.get("timeout", 120)))
|
||
cost = round(time.time() - t0, 2)
|
||
if resp.status_code != 200:
|
||
return jsonify({"success": False, "error": f"HTTP {resp.status_code}: {resp.text[:200]}", "latency": cost}), 400
|
||
d = resp.json()
|
||
try:
|
||
reply = d["choices"][0]["message"]["content"][:80]
|
||
except Exception:
|
||
reply = "(无内容)"
|
||
return jsonify({"success": True, "latency": cost, "model": cfg["model"], "reply": reply})
|
||
except Exception as e:
|
||
return jsonify({"success": False, "error": str(e)}), 400
|
||
|
||
|
||
@app.route('/api/capture', methods=['POST'])
|
||
def capture():
|
||
"""
|
||
捕获网页接口
|
||
|
||
请求体:
|
||
{
|
||
"url": "https://example.com",
|
||
"action": "screenshot" | "html" | "text" | "smart",
|
||
"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
|
||
"smart_config": {...} // 可选，按需截图时临时覆盖 LLM 配置
|
||
}
|
||
"""
|
||
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")
|
||
if action not in ("screenshot", "html", "text", "smart"):
|
||
return jsonify({"success": False, "error": f"Unknown action: {action}（可选 screenshot/html/text/smart）"}), 400
|
||
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", "")
|
||
|
||
# ---- 调用者与调用方式识别 ----
|
||
# 调用者：优先请求体 caller，其次请求头 X-Caller/X-Project/X-App，默认游客
|
||
caller = (data.get("caller") or request.headers.get("X-Caller")
|
||
or request.headers.get("X-Project") or request.headers.get("X-App")
|
||
or "游客")
|
||
# 调用方式：请求体 call_method 或请求头 X-From；缺省按 Referer 判断（本前端=web，否则 api）
|
||
call_method = data.get("call_method") or request.headers.get("X-From") or ""
|
||
if not call_method:
|
||
ref = request.headers.get("Referer") or ""
|
||
call_method = "web" if (ref and request.host in ref) else "api"
|
||
elif call_method not in ("web", "api"):
|
||
call_method = "api"
|
||
|
||
# 按需截图：滚动截图 + 视觉大模型实时判断
|
||
if action == "smart":
|
||
smart_cfg = load_smart_config()
|
||
override = data.get("smart_config") or {}
|
||
if isinstance(override, dict):
|
||
for k in DEFAULT_SMART_CONFIG:
|
||
if k in override and override[k] not in (None, ""):
|
||
smart_cfg[k] = override[k]
|
||
result = smart_capture(url, smart_cfg, wait_time, viewport, backend)
|
||
backend_used = result.get("backend", backend)
|
||
title = result.get("title", "")
|
||
if not result["success"]:
|
||
save_capture(url, title, "smart", backend_used, "failed", error=result.get("error", ""),
|
||
caller=caller, call_method=call_method)
|
||
return jsonify(result), 400
|
||
# 拼接长图已存持久化目录，直接入库；原始HTML按月归档
|
||
steps_log = json.dumps(result.get("steps", []), ensure_ascii=False)
|
||
html_path = save_raw_html(result.get("html", ""))
|
||
rid = save_capture(url, title, "smart", backend_used, "success",
|
||
file_path=result["file_path"], content=steps_log,
|
||
html_path=html_path, caller=caller, call_method=call_method)
|
||
result["history_id"] = rid
|
||
result["raw_html_path"] = html_path
|
||
return jsonify(result)
|
||
|
||
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,
|
||
cdp_port=cdp_port
|
||
)
|
||
|
||
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", ""),
|
||
caller=caller, call_method=call_method)
|
||
return jsonify(result), 400
|
||
|
||
# 每次提取后都把最原始 HTML 按月归档到本地
|
||
html_path = save_raw_html(result.get("html", ""))
|
||
|
||
if action == "screenshot":
|
||
persisted = persist_screenshot(result["file_path"])
|
||
save_capture(url, title, action, backend_used, "success", file_path=persisted,
|
||
html_path=html_path, caller=caller, call_method=call_method)
|
||
return send_file(persisted, mimetype='image/png')
|
||
|
||
elif action == "html":
|
||
save_capture(url, title, action, backend_used, "success", content=result["html"],
|
||
html_path=html_path, caller=caller, call_method=call_method)
|
||
return jsonify(result)
|
||
|
||
elif action == "text":
|
||
save_capture(url, title, action, backend_used, "success", content=result["text"],
|
||
html_path=html_path, caller=caller, call_method=call_method)
|
||
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) |