| name | web-scraper |
| description | 终极网络数据采集与爬虫工具，Level 12 终极能力：覆盖20大技术模块（HTTP直接请求/浏览器自动化/抓包逆向/移动端APP逆向/JS逆向/协议级采集/AI驱动采集/公开数据源/搜索引擎采集/社交媒体采集/深网暗网采集/网络空间搜索/文档邮件采集/IoT传感器数据/云端浏览器BaaS/验证码完全破解/指纹对抗全方案/代理完整体系/行为模拟/中国平台针对性破解）。支持美团mtgsig/waimai_sign、京东DFPID/XID/sign、淘宝x-sign、拼多多anti-content、抖音_signature/X-Bogus、小红书x-s等全平台破解。WAF绕过：Cloudflare/DataDome/Kasada/Shape/Akamai/PerimeterX。AI驱动：Crawl4AI/ScrapeGraphAI/AgentQL/Browser-Use/Skyvern。指纹对抗：JA4/JA4+/TLS/HTTP2/Canvas/WebGL/AudioContext/WebRTC。代理体系：站大爷/ISP/住宅/移动/Tor。Use when the user asks to "爬取数据", "抓取网页", "web scrape", "crawl", "采集数据", "爬虫", "抓取API", "美团", "京东", "淘宝", "拼多多", "抖音", "小红书", "mtgsig", "waimai_sign", "DFPID", "XID", "x-sign", "anti-content", "X-Bogus", "_signature", "x-s" |
Web Scraper Ultimate - Level 12 终极版本
核心原则: 结果导向，想尽一切办法获取目标数据，不做任何技术妥协或保留。
目录
- 模块1: HTTP直接请求（参数签名逆向）
- 模块2: 浏览器自动化（反检测）
- 模块3: 抓包逆向
- 模块4: 移动端APP逆向
- 模块5: JS逆向
- 模块6: 协议级采集
- 模块7: AI驱动采集
- 模块8: 公开数据源
- 模块9: 搜索引擎采集
- 模块10: 社交媒体采集
- 模块11: 深网/暗网采集
- 模块12: 网络空间搜索
- 模块13: 文档/邮件采集
- 模块14: IoT/传感器数据
- 模块15: 云端浏览器BaaS
- 模块16: 验证码完全破解
- 模块17: 指纹对抗全方案
- 模块18: 代理完整体系
- 模块19: 行为模拟
- 模块20: 中国平台针对性破解
- WAF/Bot检测绕过专项
- 分布式架构
- 自进化引擎
模块1: HTTP直接请求（参数签名逆向）
概述
HTTP直接请求是最基础也是最高效的采集方式。通过完美伪装浏览器指纹（JA3/JA4/TLS/HTTP2），可直接绕过大部分反爬检测。配合参数签名逆向，可直接调用目标API获取结构化数据。
推荐工具
| 工具 | 用途 | 优势 |
|---|
| curl_cffi | TLS/JA3/JA4伪装 | 纯C扩展，完美模拟浏览器TLS握手 |
| httpx | 异步HTTP/2 | 原生HTTP/2支持，性能优异 |
| aiohttp | 高并发异步 | 成熟稳定，生态丰富 |
| requests | 简单同步请求 | 易用，适合快速原型 |
代码示例
1.1 curl_cffi 完整伪装方案
from curl_cffi import requests as curl_requests
from curl_cffi.requests import Session
import json
import time
class UltimateHTTPClient:
"""终极HTTP客户端 - 完全指纹伪装"""
def __init__(self, impersonate="chrome120"):
self.session = Session(impersonate=impersonate)
self.session.timeout = 30
def request_with_full_disguise(self, url, method="GET", **kwargs):
"""
发送完全伪装的HTTP请求
"""
default_headers = {
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
"Accept-Encoding": "gzip, deflate, br",
"Cache-Control": "max-age=0",
"Sec-Ch-Ua": '"Not_A Brand";v="8", "Chromium";v="120", "Google Chrome";v="120"',
"Sec-Ch-Ua-Mobile": "?0",
"Sec-Ch-Ua-Platform": '"Windows"',
"Sec-Fetch-Dest": "document",
"Sec-Fetch-Mode": "navigate",
"Sec-Fetch-Site": "none",
"Sec-Fetch-User": "?1",
"Upgrade-Insecure-Requests": "1"
}
headers = kwargs.pop("headers", {})
default_headers.update(headers)
response = self.session.request(
method=method,
url=url,
headers=default_headers,
**kwargs
)
return response
client = UltimateHTTPClient(impersonate="chrome120")
resp = client.request_with_full_disguise("https://example.com/api/data")
print(resp.text)
1.2 参数签名逆向方法论
import hashlib
import hmac
import base64
import urllib.parse
import json
import time
import re
from typing import Dict, Any
class SignatureCracker:
"""
参数签名逆向通用框架
流程: 抓包→定位→还原→验证
"""
@staticmethod
def analyze_request_pattern(requests_log: list) -> Dict[str, Any]:
"""
分析请求模式，识别签名参数
常见签名参数名: sign, signature, _sign, token, auth, _sig
"""
signature_params = []
for req in requests_log:
params = req.get("params", {})
for key, value in params.items():
if re.match(r'^[a-f0-9]{32}$', str(value)):
signature_params.append({
"param": key,
"type": "MD5",
"value": value,
"other_params": {k: v for k, v in params.items() if k != key}
})
elif re.match(r'^[A-Za-z0-9+/]{43}=?$', str(value)):
signature_params.append({
"param": key,
"type": "HMAC/Base64",
"value": value,
"other_params": {k: v for k, v in params.items() if k != key}
})
return signature_params
@staticmethod
def crack_md5_sign(params: Dict[str, Any], secret_key: str = "",
sort_keys: bool = True,
exclude_keys: list = None) -> str:
"""
破解MD5签名
常见模式: MD5(排序后的参数串 + 密钥)
"""
if exclude_keys is None:
exclude_keys = ["sign", "signature"]
filtered_params = {k: v for k, v in params.items() if k not in exclude_keys}
if sort_keys:
items = sorted(filtered_params.items())
else:
items = filtered_params.items()
sign_str = "&".join([f"{k}={v}" for k, v in items])
if secret_key:
sign_str += secret_key
return hashlib.md5(sign_str.encode()).hexdigest()
@staticmethod
def crack_hmac_sign(params: Dict[str, Any], secret_key: str,
algorithm: str = "sha256") -> str:
"""
破解HMAC签名
"""
message = json.dumps(params, separators=(',', ':'), sort_keys=True)
if algorithm == "sha256":
digest = hmac.new(secret_key.encode(), message.encode(), hashlib.sha256).digest()
elif algorithm == "sha1":
digest = hmac.new(secret_key.encode(), message.encode(), hashlib.sha1).digest()
elif algorithm == "md5":
digest = hmac.new(secret_key.encode(), message.encode(), hashlib.md5).digest()
return base64.b64encode(digest).decode()
params = {
"app_id": "12345",
"timestamp": str(int(time.time())),
"data": "example"
}
sign = SignatureCracker.crack_md5_sign(params, secret_key="secret_key")
print(f"Generated sign: {sign}")
1.3 httpx 异步高并发方案
import httpx
import asyncio
from typing import List, Dict
class AsyncHTTPClient:
"""异步HTTP客户端 - 高并发采集"""
def __init__(self, concurrency: int = 100, http2: bool = True):
limits = httpx.Limits(max_keepalive_connections=50, max_connections=concurrency)
self.client = httpx.AsyncClient(
http2=http2,
limits=limits,
timeout=httpx.Timeout(30.0, connect=5.0)
)
async def fetch_single(self, url: str, headers: Dict = None) -> Dict:
"""获取单个URL"""
try:
response = await self.client.get(url, headers=headers)
return {
"url": url,
"status": response.status_code,
"content": response.text,
"success": response.status_code == 200
}
except Exception as e:
return {
"url": url,
"status": 0,
"content": str(e),
"success": False
}
async def fetch_batch(self, urls: List[str], headers: Dict = None) -> List[Dict]:
"""批量获取URL"""
tasks = [self.fetch_single(url, headers) for url in urls]
return await asyncio.gather(*tasks, return_exceptions=True)
async def close(self):
await self.client.aclose()
async def main():
client = AsyncHTTPClient(concurrency=50)
urls = [f"https://example.com/page/{i}" for i in range(1, 101)]
results = await client.fetch_batch(urls)
await client.close()
print(f"成功: {sum(1 for r in results if r.get('success'))}")
实战技巧
- TLS指纹识别: 使用 https://tls.browserleaks.com/json 检测当前TLS指纹
- JA3指纹: 使用 https://www.ja3er.com/ 查询JA3指纹是否被标记
- 抓包定位签名: 使用Chrome DevTools的"Initiator"功能追踪签名生成位置
- 参数变化分析: 对比多次请求，找出变化的签名参数
- 密钥提取: 在JS代码中搜索关键词:
sign, signature, md5, hmac, secret, key
模块2: 浏览器自动化（反检测）
概述
浏览器自动化是处理JS渲染页面和强反爬站点的核心手段。现代反检测方案从CDP协议层、C++层注入指纹，实现接近真实浏览器的体验。
推荐工具
| 工具 | 核心能力 | 适用场景 |
|---|
| Camoufox | C++层指纹注入，Firefox内核 | 最强反检测方案 |
| Patchright | Playwright反检测Fork | CDP协议层对抗 |
| Nodriver | undetected-chromedriver替代 | 简单反检测 |
| Botright | 反检测+验证码一体化 | 验证码频繁站点 |
| rebrowser-patches | Puppeteer/Playwright补丁 | 轻量级方案 |
代码示例
2.1 Camoufox 终极反检测方案
from camoufox.sync_api import Camoufox
from camoufox.utils import generate_fingerprint
def camoufox_ultimate_scraper(url: str):
"""
Camoufox终极采集方案
特点: C++层指纹注入、内存占用低(200MB)、Firefox内核天然反检测
"""
fingerprint = generate_fingerprint(
os="windows",
browser="firefox",
version="120"
)
with Camoufox(
headless=True,
fingerprint=fingerprint,
geoip=True,
locale="zh-CN",
timezone="Asia/Shanghai"
) as browser:
page = browser.new_page()
page.add_init_script("""
// 覆盖webdriver检测
Object.defineProperty(navigator, 'webdriver', {
get: () => undefined
});
// 覆盖plugins
Object.defineProperty(navigator, 'plugins', {
get: () => [
{name: "Chrome PDF Plugin"},
{name: "Native Client"}
]
});
""")
page.goto(url, wait_until="networkidle")
page.mouse.move(100, 200)
page.wait_for_timeout(500)
page.mouse.move(300, 400)
content = page.content()
browser.close()
return content
2.2 Patchright CDP协议层对抗
from patchright.sync_api import sync_playwright
def patchright_stealth_scraper(url: str):
"""
Patchright反检测方案
特点: 从CDP协议层堵住检测泄露，使用isolated世界隐藏自动化特征
"""
with sync_playwright() as p:
browser = p.chromium.launch(
headless=True,
args=[
'--disable-blink-features=AutomationControlled',
'--disable-web-security',
'--disable-features=IsolateOrigins,site-per-process',
]
)
context = browser.new_context(
viewport={'width': 1920, 'height': 1080},
user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120.0.0.0',
locale='zh-CN',
timezone_id='Asia/Shanghai',
permissions=['geolocation'],
geolocation={'latitude': 39.9, 'longitude': 116.4}
)
page = context.new_page()
page.add_init_script("""
// 删除webdriver
delete Object.getPrototypeOf(navigator).webdriver;
// 模拟chrome.runtime
window.chrome = {
runtime: {
OnInstalledReason: {CHROME_UPDATE: "chrome_update"},
OnRestartRequiredReason: {APP_UPDATE: "app_update"},
PlatformArch: {X86_64: "x86-64"},
PlatformNaclArch: {X86_64: "x86-64"},
PlatformOs: {WIN: "win"},
RequestUpdateCheckStatus: {NO_UPDATE: "no_update"}
}
};
// 覆盖permissions API
const originalQuery = window.navigator.permissions.query;
window.navigator.permissions.query = (parameters) => (
parameters.name === 'notifications' ?
Promise.resolve({state: Notification.permission}) :
originalQuery(parameters)
);
""")
page.goto(url, wait_until="networkidle")
content = page.content()
browser.close()
return content
2.3 Nodriver 快速方案
import nodriver as uc
import asyncio
async def nodriver_scraper(url: str):
"""
Nodriver快速反检测方案
特点: 开箱即用，Cloudflare通过率85%+
"""
browser = await uc.start()
page = await browser.get(url)
await page.sleep(3)
content = await page.get_content()
browser.stop()
return content
2.4 Botright 验证码一体化方案
import botright
async def botright_captcha_scraper(url: str):
"""
Botright验证码一体化方案
特点: 自动检测并解决多种验证码
"""
bot = await botright.Botright()
browser = await botright.launch(
proxy={
"server": "http://proxy.example.com:8080",
"username": "user",
"password": "pass"
},
captcha_solver="capsolver",
captcha_api_key="YOUR_CAPSOLVER_KEY"
)
page = await browser.new_page()
await page.goto(url)
await page.solve_captcha_if_present()
await page.mouse.move_randomly(duration=2.0)
await page.keyboard.type_like_human("search query")
content = await page.content()
await browser.close()
return content
2.5 多浏览器轮换策略
import random
from enum import Enum
class BrowserEngine(Enum):
CAMOUFOX = "camoufox"
PATCHRIGHT = "patchright"
NODRIVER = "nodriver"
PLAYWRIGHT = "playwright"
class BrowserRotator:
"""多浏览器轮换管理器"""
def __init__(self):
self.engines = list(BrowserEngine)
self.current_index = 0
self.failure_counts = {engine: 0 for engine in self.engines}
def get_next_browser(self) -> BrowserEngine:
"""获取下一个浏览器引擎"""
sorted_engines = sorted(self.engines, key=lambda e: self.failure_counts[e])
if random.random() < 0.8:
return sorted_engines[0]
else:
return random.choice(self.engines)
def record_success(self, engine: BrowserEngine):
"""记录成功"""
self.failure_counts[engine] = max(0, self.failure_counts[engine] - 1)
def record_failure(self, engine: BrowserEngine):
"""记录失败"""
self.failure_counts[engine] += 1
rotator = BrowserRotator()
engine = rotator.get_next_browser()
print(f"Using browser: {engine.value}")
实战技巧
- 检测点排查: 使用 https://bot.sannysoft.com/ 测试反检测效果
- WebDriver检测: 确保
navigator.webdriver 为 undefined
- Chrome Runtime: 模拟
window.chrome.runtime 对象
- Permissions API: 覆盖
navigator.permissions.query
- Plugins: 确保
navigator.plugins 非空
- Languages: 设置与IP地理位置匹配的语言
模块3: 抓包逆向
概述
抓包逆向是分析APP和Web应用网络请求的核心技术。通过拦截HTTPS流量，可以分析API参数、签名算法、加密逻辑。
推荐工具
| 工具 | 平台 | 特点 |
|---|
| mitmproxy | 全平台 | 开源，支持自定义脚本 |
| Charles | Win/Mac | 商用，UI友好 |
| HTTP Toolkit | 全平台 | 支持USB直连iPhone |
| Fiddler | Win | 免费，功能全面 |
| Burp Suite | 全平台 | 安全测试标准 |
代码示例
3.1 mitmproxy 自定义脚本
from mitmproxy import http
import json
import base64
class APICapture:
"""API流量捕获与分析"""
def __init__(self):
self.captured_apis = []
def request(self, flow: http.HTTPFlow):
"""拦截请求"""
if "/api/" in flow.request.pretty_url or "/v1/" in flow.request.pretty_url:
capture = {
"type": "request",
"url": flow.request.pretty_url,
"method": flow.request.method,
"headers": dict(flow.request.headers),
"params": dict(flow.request.query),
}
if flow.request.content:
try:
capture["body"] = json.loads(flow.request.content)
except:
capture["body"] = base64.b64encode(flow.request.content).decode()
self.captured_apis.append(capture)
print(f"[API Request] {flow.request.method} {flow.request.pretty_url}")
def response(self, flow: http.HTTPFlow):
"""拦截响应"""
if "/api/" in flow.request.pretty_url:
capture = {
"type": "response",
"url": flow.request.pretty_url,
"status": flow.response.status_code,
"headers": dict(flow.response.headers),
}
if flow.response.content:
try:
capture["body"] = json.loads(flow.response.content)
except:
capture["body"] = base64.b64encode(flow.response.content).decode()[:1000]
self.captured_apis.append(capture)
print(f"[API Response] {flow.response.status_code} {flow.request.pretty_url}")
def done(self):
"""保存捕获的数据"""
with open("captured_apis.json", "w") as f:
json.dump(self.captured_apis, f, indent=2)
addons = [APICapture()]
3.2 SSL Pinning 绕过脚本
Java.perform(function() {
console.log("[*] SSL Pinning Bypass Started");
// 1. OkHttp3 SSL Pinning绕过
try {
var TrustManagerImpl = Java.use('com.android.org.conscrypt.TrustManagerImpl');
TrustManagerImpl.checkTrustedRecursive.implementation = function() {
console.log("[*] OkHttp3 checkTrustedRecursive bypassed");
return [];
};
} catch(e) {
console.log("[-] OkHttp3 not found");
}
// 2. X509TrustManager绕过
try {
var X509TrustManager = Java.use('javax.net.ssl.X509TrustManager');
var SSLContext = Java.use('javax.net.ssl.SSLContext');
var TrustManager = Java.registerClass({
name: 'com.example.TrustManager',
implements: [X509TrustManager],
methods: {
checkClientTrusted: function() {},
checkServerTrusted: function() {},
getAcceptedIssuers: function() { return []; }
}
});
var TrustManagers = [TrustManager.$new()];
var SSLContext_init = SSLContext.init.overload(
'[Ljavax.net.ssl.KeyManager;',
'[Ljavax.net.ssl.TrustManager;',
'java.security.SecureRandom'
);
SSLContext_init.implementation = function(km, tm, random) {
console.log("[*] SSLContext.init() hooked");
SSLContext_init.call(this, km, TrustManagers, random);
};
} catch(e) {
console.log("[-] X509TrustManager hook failed: " + e);
}
// 3. WebView SSL绕过
try {
var WebView = Java.use('android.webkit.WebView');
var SslErrorHandler = Java.use('android.webkit.SslErrorHandler');
var WebViewClient = Java.use('android.webkit.WebViewClient');
WebViewClient.onReceivedSslError.implementation = function(view, handler, error) {
console.log("[*] WebView SSL error bypassed");
handler.proceed();
};
} catch(e) {
console.log("[-] WebView SSL bypass failed");
}
console.log("[*] SSL Pinning Bypass Complete");
});
3.3 流量分析→参数签名复现
import json
from urllib.parse import parse_qs, urlparse
class TrafficAnalyzer:
"""流量分析器 - 从抓包数据中提取签名逻辑"""
def __init__(self, capture_file: str):
with open(capture_file, 'r') as f:
self.captures = json.load(f)
def extract_signature_params(self) -> list:
"""提取包含签名的请求"""
signature_requests = []
for capture in self.captures:
if capture.get("type") == "request":
params = capture.get("params", {})
body = capture.get("body", {})
for key, value in params.items():
if self._is_signature_param(key, value):
signature_requests.append({
"url": capture["url"],
"param_name": key,
"param_value": value,
"other_params": {k: v for k, v in params.items() if k != key},
"source": "url"
})
if isinstance(body, dict):
for key, value in body.items():
if self._is_signature_param(key, value):
signature_requests.append({
"url": capture["url"],
"param_name": key,
"param_value": value,
"other_params": {k: v for k, v in body.items() if k != key},
"source": "body"
})
return signature_requests
def _is_signature_param(self, key: str, value) -> bool:
"""判断是否为签名参数"""
signature_keywords = ['sign', 'signature', '_sign', 'token', 'auth', '_sig', 'hmac']
if any(keyword in key.lower() for keyword in signature_keywords):
return True
if isinstance(value, str):
if len(value) == 32 and all(c in '0123456789abcdef' for c in value.lower()):
return True
if len(value) == 40 and all(c in '0123456789abcdef' for c in value.lower()):
return True
if len(value) > 40 and value.endswith('=') or all(c in 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/=' for c in value):
return True
return False
def analyze_signature_pattern(self, signature_requests: list):
"""分析签名模式"""
if not signature_requests:
print("[-] 未找到签名参数")
return
print(f"[+] 找到 {len(signature_requests)} 个签名请求")
for i, req in enumerate(signature_requests[:5]):
print(f"\n--- 签名请求 {i+1} ---")
print(f"URL: {req['url']}")
print(f"参数名: {req['param_name']}")
print(f"参数值: {req['param_value']}")
print(f"其他参数: {json.dumps(req['other_params'], indent=2)}")
实战技巧
- 证书安装: 确保手机/模拟器信任mitmproxy/Charles的根证书
- Android 7+: 需要Magisk+Move Certificates模块或修改APK
- iOS: 需要越狱或使用HTTP Toolkit的USB直连功能
- SSL Pinning: 使用Frida或Objection绕过
- Protobuf: 使用blackboxprotobuf解码二进制数据
模块4: 移动端APP逆向
概述
移动端APP逆向是获取APP内部API和加密算法的关键技术。通过Frida Hook、APK反编译、so层逆向，可以提取参数签名、设备指纹生成逻辑。
推荐工具
| 工具 | 用途 |
|---|
| Frida | 动态Hook，运行时修改APP行为 |
| jadx | APK反编译，查看Java源码 |
| apktool | APK解包/打包，修改资源 |
| IDA Pro | so层逆向，分析Native代码 |
| Ghidra | 免费逆向工具，替代IDA |
| Objection | Frida的易用封装 |
| Magisk | Root方案，支持LSPosed |
代码示例
4.1 Frida Hook 完整方案
import frida
import sys
hook_script = """
Java.perform(function() {
console.log("[*] Frida Hook Started");
// Hook加密函数示例 - 假设目标使用自定义加密
var CryptoUtils = Java.use("com.example.app.CryptoUtils");
// Hook sign生成函数
CryptoUtils.generateSign.implementation = function(params) {
console.log("[*] generateSign called");
console.log(" params: " + params);
var result = this.generateSign(params);
console.log(" result: " + result);
return result;
};
// Hook MD5函数
CryptoUtils.md5.implementation = function(input) {
console.log("[*] MD5 input: " + input);
var result = this.md5(input);
console.log("[*] MD5 output: " + result);
return result;
};
// Hook设备指纹获取
var DeviceInfo = Java.use("com.example.app.DeviceInfo");
DeviceInfo.getFingerprint.implementation = function() {
var result = this.getFingerprint();
console.log("[*] Device Fingerprint: " + result);
return result;
};
// Hook网络请求
var OkHttpClient = Java.use("okhttp3.OkHttpClient");
var Request = Java.use("okhttp3.Request");
OkHttpClient.newCall.implementation = function(request) {
console.log("[*] HTTP Request:");
console.log(" URL: " + request.url().toString());
console.log(" Method: " + request.method());
var headers = request.headers();
var headersStr = "";
for (var i = 0; i < headers.size(); i++) {
headersStr += headers.name(i) + ": " + headers.value(i) + "\\n";
}
console.log(" Headers: \\n" + headersStr);
return this.newCall(request);
};
});
"""
def on_message(message, data):
if message['type'] == 'send':
print(f"[*] {message['payload']}")
else:
print(f"[-] {message}")
def run_hook(app_package: str):
"""运行Frida Hook"""
device = frida.get_usb_device()
pid = device.spawn([app_package])
session = device.attach(pid)
script = session.create_script(hook_script)
script.on('message', on_message)
script.load()
device.resume(pid)
print(f"[*] Hooking {app_package}, PID: {pid}")
sys.stdin.read()
4.2 APK反编译分析
import subprocess
import os
class APKAnalyzer:
"""APK分析工具"""
def __init__(self, apk_path: str, output_dir: str = "./decompiled"):
self.apk_path = apk_path
self.output_dir = output_dir
os.makedirs(output_dir, exist_ok=True)
def decompile_with_jadx(self):
"""使用jadx反编译APK"""
output_path = os.path.join(self.output_dir, "java_source")
cmd = f"jadx -d {output_path} {self.apk_path}"
subprocess.run(cmd, shell=True)
print(f"[+] 反编译完成: {output_path}")
return output_path
def decode_with_apktool(self):
"""使用apktool解包APK"""
output_path = os.path.join(self.output_dir, "smali_resources")
cmd = f"apktool d -f {self.apk_path} -o {output_path}"
subprocess.run(cmd, shell=True)
print(f"[+] 解包完成: {output_path}")
return output_path
def extract_native_libs(self):
"""提取so库文件"""
import zipfile
libs_dir = os.path.join(self.output_dir, "native_libs")
os.makedirs(libs_dir, exist_ok=True)
with zipfile.ZipFile(self.apk_path, 'r') as zip_ref:
for file in zip_ref.namelist():
if file.endswith('.so'):
zip_ref.extract(file, libs_dir)
print(f"[+] 提取: {file}")
return libs_dir
def search_sign_keywords(self, source_dir: str):
"""搜索签名相关关键词"""
keywords = ['sign', 'signature', 'md5', 'sha1', 'hmac', 'encrypt', 'aes', 'rsa']
results = []
for root, dirs, files in os.walk(source_dir):
for file in files:
if file.endswith('.java'):
file_path = os.path.join(root, file)
with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
content = f.read()
for keyword in keywords:
if keyword in content.lower():
results.append({
"file": file_path,
"keyword": keyword
})
return results
4.3 so层逆向分析
import subprocess
import re
class SOAnalyzer:
"""SO库分析工具"""
def __init__(self, so_path: str):
self.so_path = so_path
def extract_strings(self) -> list:
"""提取字符串"""
result = subprocess.run(
['strings', self.so_path],
capture_output=True,
text=True
)
strings = result.stdout.split('\n')
interesting_patterns = [
r'[a-f0-9]{32}',
r'[a-f0-9]{40}',
r'[a-f0-9]{64}',
r'-----BEGIN (RSA |EC |DSA )?PRIVATE KEY-----',
r'-----BEGIN (RSA |EC |DSA )?PUBLIC KEY-----',
r'AES|DES|RSA|HMAC|SHA',
r'api\.|http|https',
]
interesting_strings = []
for s in strings:
for pattern in interesting_patterns:
if re.search(pattern, s, re.IGNORECASE):
interesting_strings.append(s)
break
return interesting_strings
def list_symbols(self) -> list:
"""列出符号表"""
result = subprocess.run(
['nm', '-D', self.so_path],
capture_output=True,
text=True
)
symbols = []
for line in result.stdout.split('\n'):
if ' T ' in line or ' D ' in line:
parts = line.split()
if len(parts) >= 3:
symbols.append({
"address": parts[0],
"type": parts[1],
"name": parts[2]
})
return symbols
def find_crypto_functions(self) -> list:
"""查找加密相关函数"""
symbols = self.list_symbols()
crypto_keywords = ['encrypt', 'decrypt', 'sign', 'verify', 'hash',
'md5', 'sha', 'aes', 'des', 'rsa', 'hmac']
crypto_funcs = []
for sym in symbols:
name = sym['name'].lower()
if any(kw in name for kw in crypto_keywords):
crypto_funcs.append(sym)
return crypto_funcs
实战技巧
- Frida Server: 确保手机端运行frida-server，版本与PC端一致
- SELinux: Android 5.0+需要关闭SELinux或配置权限
- 反调试绕过: 使用
frida --no-pause绕过启动时检测
- so加密: 使用IDA Pro分析ollvm混淆的so文件
- VMP保护: 使用unicorn模拟执行或寻找逻辑漏洞
模块5: JS逆向
概述
JS逆向是破解Web端参数签名和加密算法的核心技术。通过AST反混淆、JSVMP破解、webpack分析，可以还原加密逻辑并在Python中复现。
推荐工具
| 工具 | 用途 |
|---|
| @babel/core | AST解析与转换 |
| @babel/traverse | AST遍历 |
| @babel/types | AST节点操作 |
| js-beautify | JS代码美化 |
| AST Explorer | 在线AST分析 |
| Node.js | 运行浏览器JS |
代码示例
5.1 AST反混淆
const parser = require("@babel/parser");
const traverse = require("@babel/traverse").default;
const t = require("@babel/types");
const generate = require("@babel/generator").default;
const fs = require("fs");
function deobfuscate(code) {
const ast = parser.parse(code, {
sourceType: "script",
});
traverse(ast, {
StringLiteral(path) {
if (path.node.extra?.raw) {
delete path.node.extra.raw;
}
}
});
traverse(ast, {
StringLiteral(path) {
if (/\\u[0-9a-fA-F]{4}/.test(path.node.value)) {
path.node.value = eval('"' + path.node.value + '"');
}
}
});
traverse(ast, {
MemberExpression(path) {
if (t.isArrayExpression(path.node.object) &&
t.isNumericLiteral(path.node.property)) {
const array = path.node.object.elements;
const index = path.node.property.value;
if (index < array.length) {
path.replaceWith(array[index]);
}
}
}
});
traverse(ast, {
BinaryExpression(path) {
if (t.isNumericLiteral(path.node.left) &&
t.isNumericLiteral(path.node.right)) {
const result = eval(generate(path.node).code);
path.replaceWith(t.numericLiteral(result));
}
}
});
traverse(ast, {
IfStatement(path) {
if (t.isBooleanLiteral(path.node.test)) {
if (path.node.test.value) {
path.replaceWithMultiple(path.node.consequent.body);
} else if (path.node.alternate) {
path.replaceWithMultiple(path.node.alternate.body);
} else {
path.remove();
}
}
}
});
const output = generate(ast, {
compact: false,
quotes: "double",
indent: { style: " " }
}).code;
return output;
}
5.2 Webpack Chunk分析
import re
import json
class WebpackAnalyzer:
"""Webpack打包文件分析器"""
def __init__(self, js_file: str):
with open(js_file, 'r', encoding='utf-8') as f:
self.code = f.read()
def extract_modules(self) -> dict:
"""提取所有模块"""
module_pattern = r'\{(\d+):\s*\[function\([^)]+\)\s*\{([^}]+(?:\{[^}]*\}[^}]*)*)\},\s*\{[^}]*\}\]\}'
modules = {}
for match in re.finditer(module_pattern, self.code):
module_id = match.group(1)
module_code = match.group(2)
modules[module_id] = module_code
return modules
def find_crypto_module(self) -> str:
"""查找加密相关模块"""
modules = self.extract_modules()
crypto_keywords = ['encrypt', 'decrypt', 'sign', 'md5', 'sha', 'aes', 'rsa', 'hmac']
for module_id, code in modules.items():
code_lower = code.lower()
if any(kw in code_lower for kw in crypto_keywords):
print(f"[+] 发现加密模块: {module_id}")
return code
return None
def extract_entry_point(self) -> str:
"""提取入口点"""
entry_pattern = r'\(([function\(require,\s*module,\s*exports\)[\s\S]+?)\)\s*\(\s*\{'
match = re.search(entry_pattern, self.code)
if match:
return match.group(1)
return None
5.3 JSVMP虚拟机破解
import execjs
import json
class JSVMPBreaker:
"""JSVMP破解器"""
def __init__(self, vmp_code: str):
"""
vmp_code: VMP保护的JS代码
"""
self.vmp_code = vmp_code
self.ctx = execjs.compile(vmp_code)
def hook_encrypt_function(self, func_name: str, test_input: str):
"""
Hook加密函数，提取算法
原理: 通过多次调用，观察输入输出关系
"""
hook_code = f"""
var originalFunc = {func_name};
var calls = [];
{func_name} = function(input) {{
var result = originalFunc(input);
calls.push({{input: input, output: result}});
return result;
}};
"""
self.ctx.exec(hook_code)
test_inputs = [
test_input,
"a",
"abc",
"test123",
"hello world"
]
for inp in test_inputs:
try:
result = self.ctx.call(func_name, inp)
print(f"Input: {inp}")
print(f"Output: {result}")
print(f"Length: {len(result)}")
print("---")
except Exception as e:
print(f"Error with input '{inp}': {e}")
def analyze_output_pattern(self, outputs: list):
"""分析输出模式，推断算法"""
for output in outputs:
if len(output) == 32 and all(c in '0123456789abcdef' for c in output.lower()):
return "MD5"
if len(output) == 40 and all(c in '0123456789abcdef' for c in output.lower()):
return "SHA1"
if len(output) == 64 and all(c in '0123456789abcdef' for c in output.lower()):
return "SHA256"
if len(output) % 4 == 0 and all(c in 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/=' for c in output):
return "Base64"
return "Unknown"
5.4 环境补全（Node.js运行浏览器JS）
import execjs
import json
class BrowserEnvironment:
"""浏览器环境补全"""
def __init__(self):
self.env_code = """
// 模拟window对象
var window = {
location: {
href: "https://example.com",
hostname: "example.com",
protocol: "https:",
pathname: "/"
},
navigator: {
userAgent: "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0.0.0",
language: "zh-CN",
languages: ["zh-CN", "zh", "en"],
platform: "Win32",
vendor: "Google Inc.",
webdriver: undefined
},
screen: {
width: 1920,
height: 1080,
colorDepth: 24
},
document: {
cookie: "",
referrer: ""
},
localStorage: {
data: {},
getItem: function(key) { return this.data[key] || null; },
setItem: function(key, value) { this.data[key] = value; },
removeItem: function(key) { delete this.data[key]; }
},
sessionStorage: {
data: {},
getItem: function(key) { return this.data[key] || null; },
setItem: function(key, value) { this.data[key] = value; },
removeItem: function(key) { delete this.data[key]; }
},
addEventListener: function() {},
removeEventListener: function() {}
};
// 模拟document对象
var document = window.document;
// 模拟navigator对象
var navigator = window.navigator;
// 模拟location对象
var location = window.location;
// 模拟localStorage
var localStorage = window.localStorage;
// 模拟sessionStorage
var sessionStorage = window.sessionStorage;
// 模拟console
var console = {
log: function() {},
warn: function() {},
error: function() {},
info: function() {}
};
// 模拟XMLHttpRequest
var XMLHttpRequest = function() {};
// 模拟fetch
var fetch = function() { return Promise.resolve(); };
// 模拟setTimeout/setInterval
var setTimeout = function(fn, delay) { return 0; };
var clearTimeout = function(id) {};
var setInterval = function(fn, delay) { return 0; };
var clearInterval = function(id) {};
"""
def execute_browser_js(self, js_code: str, func_name: str = None, *args):
"""
在补全的浏览器环境中执行JS代码
"""
full_code = self.env_code + "\n" + js_code
ctx = execjs.compile(full_code)
if func_name:
return ctx.call(func_name, *args)
else:
return ctx.eval("this")
实战技巧
- 断点调试: 使用Chrome DevTools的Sources面板设置断点
- Call Stack: 通过调用栈追踪签名生成路径
- Pretty Print: 使用
{}按钮美化压缩代码
- Overrides: 使用Local Overrides修改并持久化JS代码
- XHR/Fetch断点: 在Sources面板设置XHR断点
模块6: 协议级采集
概述
协议级采集是直接对接WebSocket、gRPC、GraphQL、SSE等协议的高级采集技术。相比HTTP，这些协议通常防护较弱，数据更结构化。
推荐工具
| 协议 | 工具 |
|---|
| WebSocket | websockets |
| gRPC | grpcio, grpcurl |
| GraphQL | gql, clairvoyance |
| SSE | aiohttp |
代码示例
6.1 WebSocket 实时数据流采集
import websockets
import asyncio
import json
class WebSocketScraper:
"""WebSocket实时数据流采集器"""
def __init__(self, uri: str, headers: dict = None):
self.uri = uri
self.headers = headers or {}
self.messages = []
async def connect_and_listen(self, subscribe_msg: dict = None, duration: int = 60):
"""
连接并监听WebSocket消息
Args:
subscribe_msg: 订阅消息
duration: 监听时长(秒)
"""
async with websockets.connect(
self.uri,
extra_headers=self.headers,
ping_interval=30,
ping_timeout=10
) as websocket:
print(f"[+] Connected to {self.uri}")
if subscribe_msg:
await websocket.send(json.dumps(subscribe_msg))
print(f"[+] Sent: {subscribe_msg}")
start_time = asyncio.get_event_loop().time()
try:
while asyncio.get_event_loop().time() - start_time < duration:
message = await asyncio.wait_for(
websocket.recv(),
timeout=5.0
)
data = json.loads(message)
self.messages.append(data)
print(f"[+] Received: {data}")
await self.handle_message(data)
except asyncio.TimeoutError:
print("[-] Receive timeout")
except websockets.exceptions.ConnectionClosed:
print("[-] Connection closed")
async def handle_message(self, data: dict):
"""处理接收到的消息 - 子类可重写"""
pass
def save_messages(self, filename: str):
"""保存消息到文件"""
with open(filename, 'w') as f:
json.dump(self.messages, f, indent=2)
print(f"[+] Saved {len(self.messages)} messages to {filename}")
async def crypto_scraper():
scraper = WebSocketScraper(
uri="wss://stream.crypto.com/v2/market",
headers={"Origin": "https://crypto.com"}
)
subscribe_msg = {
"method": "subscribe",
"params": {"channels": ["trade.BTC_USDT"]},
"id": 1
}
await scraper.connect_and_listen(subscribe_msg, duration=60)
scraper.save_messages("crypto_trades.json")
6.2 gRPC Protobuf 逆向与采集
import grpc
import json
from google.protobuf import json_format
class GRPCScraper:
"""gRPC采集器"""
def __init__(self, target: str, use_tls: bool = True):
"""
Args:
target: gRPC服务器地址，如 "api.example.com:50051"
use_tls: 是否使用TLS
"""
self.target = target
if use_tls:
self.channel = grpc.secure_channel(target, grpc.ssl_channel_credentials())
else:
self.channel = grpc.insecure_channel(target)
def list_services(self):
"""列出所有服务 - 需要反射支持"""
try:
from grpc_reflection.v1alpha import reflection_pb2, reflection_pb2_grpc
stub = reflection_pb2_grpc.ServerReflectionStub(self.channel)
request = reflection_pb2.ServerReflectionRequest(list_services="")
response = stub.ServerReflectionInfo(iter([request]))
for resp in response:
if resp.list_services_response:
services = resp.list_services_response.service
return [s.name for s in services]
except Exception as e:
print(f"[-] Reflection not available: {e}")
return []
def describe_service(self, service_name: str):
"""描述服务"""
import subprocess
result = subprocess.run(
['grpcurl', '-plaintext', self.target, 'describe', service_name],
capture_output=True,
text=True
)
return result.stdout
6.3 GraphQL Introspection + Clairvoyance
import requests
import json
class GraphQLScraper:
"""GraphQL采集器"""
def __init__(self, endpoint: str, headers: dict = None):
self.endpoint = endpoint
self.headers = headers or {
"Content-Type": "application/json",
"Accept": "application/json"
}
def introspect(self) -> dict:
"""
执行Introspection查询获取Schema
注意: 部分站点会禁用introspection
"""
introspection_query = """
query IntrospectionQuery {
__schema {
queryType { name }
mutationType { name }
subscriptionType { name }
types {
...FullType
}
directives {
name
description
locations
args {
...InputValue
}
}
}
}
fragment FullType on __Type {
kind
name
description
fields(includeDeprecated: true) {
name
description
args {
...InputValue
}
type {
...TypeRef
}
isDeprecated
deprecationReason
}
inputFields {
...InputValue
}
interfaces {
...TypeRef
}
enumValues(includeDeprecated: true) {
name
description
isDeprecated
deprecationReason
}
possibleTypes {
...TypeRef
}
}
fragment InputValue on __InputValue {
name
description
type { ...TypeRef }
defaultValue
}
fragment TypeRef on __Type {
kind
name
ofType {
kind
name
ofType {
kind
name
ofType {
kind
name
ofType {
kind
name
ofType {
kind
name
ofType {
kind
name
ofType {
kind
name
}
}
}
}
}
}
}
}
"""
response = requests.post(
self.endpoint,
headers=self.headers,
json={"query": introspection_query}
)
if response.status_code == 200:
return response.json()
else:
print(f"[-] Introspection failed: {response.status_code}")
return None
def execute_query(self, query: str, variables: dict = None) -> dict:
"""执行GraphQL查询"""
payload = {"query": query}
if variables:
payload["variables"] = variables
response = requests.post(
self.endpoint,
headers=self.headers,
json=payload
)
return response.json()
def clairvoyance_recon(self):
"""
使用Clairvoyance进行Schema发现
当introspection被禁用时使用
"""
import subprocess
result = subprocess.run(
['python', '-m', 'clairvoyance', self.endpoint, '-o', 'schema.json'],
capture_output=True,
text=True
)
print(result.stdout)
print(result.stderr)
6.4 SSE（Server-Sent Events）流式采集
import aiohttp
import asyncio
import json
class SSEScraper:
"""SSE流式数据采集器"""
def __init__(self, url: str, headers: dict = None):
self.url = url
self.headers = headers or {
"Accept": "text/event-stream",
"Cache-Control": "no-cache"
}
self.events = []
async def listen(self, duration: int = 60):
"""
监听SSE事件流
Args:
duration: 监听时长(秒)
"""
async with aiohttp.ClientSession() as session:
async with session.get(self.url, headers=self.headers) as response:
print(f"[+] Connected to SSE stream: {response.status}")
start_time = asyncio.get_event_loop().time()
current_event = {}
async for line in response.content:
if asyncio.get_event_loop().time() - start_time > duration:
break
line = line.decode('utf-8').strip()
if line.startswith('event:'):
current_event['event'] = line[6:].strip()
elif line.startswith('data:'):
data = line[5:].strip()
if 'data' in current_event:
current_event['data'] += '\n' + data
else:
current_event['data'] = data
elif line.startswith('id:'):
current_event['id'] = line[3:].strip()
elif line.startswith('retry:'):
current_event['retry'] = int(line[6:].strip())
elif line == '':
if current_event:
self.events.append(current_event)
print(f"[+] Event: {current_event.get('event', 'message')}")
try:
data = json.loads(current_event.get('data', '{}'))
await self.handle_event(data)
except json.JSONDecodeError:
pass
current_event = {}
async def handle_event(self, data: dict):
"""处理事件数据 - 子类可重写"""
print(f" Data: {data}")
def save_events(self, filename: str):
"""保存事件到文件"""
with open(filename, 'w') as f:
json.dump(self.events, f, indent=2)
print(f"[+] Saved {len(self.events)} events to {filename}")
async def sse_example():
scraper = SSEScraper("https://stream.example.com/events")
await scraper.listen(duration=60)
scraper.save_events("sse_events.json")
实战技巧
- WebSocket鉴权: 注意在URL或headers中传递token
- gRPC反射: 使用
grpcurl list查看可用服务
- GraphQL深度限制: 避免过深嵌套查询触发限流
- SSE重连: 实现指数退避重连机制
- 协议升级: 关注HTTP/3对WebSocket的影响
模块7: AI驱动采集
概述
AI驱动采集是2025-2026年最前沿的技术方向。通过LLM直接理解页面内容，无需编写XPath/CSS选择器，自适应页面变化。
推荐工具
| 工具 | Stars | 核心能力 |
|---|
| Crawl4AI | 63.5k | LLM原生提取，结构化数据 |
| ScrapeGraphAI | 15k+ | GPT-4V视觉理解 |
| AgentQL | 5k+ | 自然语言查询 |
| Browser-Use | 25k+ | AI控制浏览器 |
| Skyvern | 20.8k | 视觉AI页面理解 |
代码示例
7.1 Crawl4AI 完整方案
from crawl4ai import AsyncWebCrawler
from crawl4ai.extraction_strategy import LLMExtractionStrategy
from crawl4ai.crawler_strategy import LocalSeleniumCrawlerStrategy
import asyncio
import json
class Crawl4AIExtractor:
"""Crawl4AI数据提取器"""
def __init__(self, llm_config: dict = None):
"""
Args:
llm_config: LLM配置，如 {"provider": "openai", "api_key": "xxx"}
"""
self.llm_config = llm_config or {
"provider": "openai",
"api_key": "YOUR_API_KEY",
"model": "gpt-4-vision-preview"
}
async def extract_with_llm(self, url: str, schema: dict, instruction: str = None):
"""
使用LLM提取结构化数据
Args:
url: 目标URL
schema: 数据结构定义
instruction: 额外提取指令
"""
async with AsyncWebCrawler(
verbose=True,
bypass_cloudflare=True
) as crawler:
extraction_strategy = LLMExtractionStrategy(
provider=self.llm_config["provider"],
api_token=self.llm_config["api_key"],
schema=schema,
instruction=instruction or "Extract the data according to the schema",
extraction_type="schema"
)
result = await crawler.arun(
url=url,
extraction_strategy=extraction_strategy,
bypass_cloudflare=True,
magic=True,
wait_for_images=True
)
return {
"success": result.success,
"content": result.extracted_content,
"markdown": result.markdown,
"links": result.links
}
async def extract_products(self, url: str):
"""提取产品信息示例"""
schema = {
"products": [{
"name": "str",
"price": "float",
"currency": "str",
"description": "str",
"image_url": "str",
"in_stock": "boolean"
}]
}
instruction = """
Extract all product information from the page.
For price, extract only the numeric value.
For in_stock, return true if the product is available.
"""
return await self.extract_with_llm(url, schema, instruction)
async def extract_articles(self, url: str):
"""提取文章信息示例"""
schema = {
"articles": [{
"title": "str",
"author": "str",
"publish_date": "str",
"content": "str",
"tags": ["str"]
}]
}
return await self.extract_with_llm(url, schema)
async def crawl4ai_demo():
extractor = Crawl4AIExtractor({
"provider": "openai",
"api_key": "YOUR_API_KEY"
})
result = await extractor.extract_products("https://example.com/products")
print(json.dumps(result, indent=2, ensure_ascii=False))
7.2 ScrapeGraphAI 视觉理解
from scrapegraphai.graphs import SmartScraperGraph, SearchGraph
import os
class ScrapeGraphAIExtractor:
"""ScrapeGraphAI数据提取器"""
def __init__(self, api_key: str = None):
self.api_key = api_key or os.getenv("OPENAI_API_KEY")
self.graph_config = {
"llm": {
"api_key": self.api_key,
"model": "openai/gpt-4-vision-preview",
},
"verbose": True,
"headless": True,
}
def extract(self, prompt: str, url: str):
"""
使用自然语言提示提取数据
Args:
prompt: 自然语言描述要提取的数据
url: 目标URL
"""
smart_scraper = SmartScraperGraph(
prompt=prompt,
source=url,
config=self.graph_config
)
result = smart_scraper.run()
return result
def search_and_extract(self, prompt: str, search_query: str):
"""
搜索并提取数据
Args:
prompt: 提取指令
search_query: 搜索查询
"""
search_graph = SearchGraph(
prompt=prompt,
config=self.graph_config
)
result = search_graph.run(search_query)
return result
7.3 AgentQL 自然语言查询
import agentql
class AgentQLExtractor:
"""AgentQL自然语言查询提取器"""
def __init__(self):
pass
def query_page(self, url: str, query: str):
"""
使用自然语言查询页面
Args:
url: 目标URL
query: AgentQL查询语句
"""
with agentql.wrap() as page:
page.goto(url)
response = page.query(query)
return response
def extract_products(self, url: str):
"""提取产品信息"""
query = """
{
products[] {
name
price (as float)
description
in_stock (as boolean)
image_url
}
}
"""
return self.query_page(url, query)
def extract_search_results(self, url: str):
"""提取搜索结果"""
query = """
{
results[] {
title
url
snippet
}
}
"""
return self.query_page(url, query)
7.4 Browser-Use AI浏览器控制
from browser_use import Agent
from langchain_openai import ChatOpenAI
import asyncio
class BrowserUseAgent:
"""Browser-Use AI浏览器代理"""
def __init__(self, api_key: str = None):
self.llm = ChatOpenAI(
model="gpt-4o",
api_key=api_key
)
async def execute_task(self, task: str, url: str = None):
"""
执行自然语言描述的任务
Args:
task: 任务描述
url: 起始URL（可选）
"""
agent = Agent(
task=task,
llm=self.llm,
use_vision=True
)
result = await agent.run()
return result
async def login_and_extract(self, login_url: str, username: str, password: str, target_data: str):
"""登录并提取数据"""
task = f"""
1. 访问 {login_url}
2. 在用户名输入框输入: {username}
3. 在密码输入框输入: {password}
4. 点击登录按钮
5. 等待页面加载完成
6. 提取以下数据: {target_data}
"""
return await self.execute_task(task)
async def search_and_scrape(self, search_url: str, query: str, num_results: int = 10):
"""搜索并采集结果"""
task = f"""
1. 访问 {search_url}
2. 在搜索框输入: {query}
3. 点击搜索按钮
4. 提取前{num_results}个搜索结果
5. 返回每个结果的标题、链接和描述
"""
return await self.execute_task(task)
async def browser_use_demo():
agent = BrowserUseAgent(api_key="YOUR_API_KEY")
result = await agent.search_and_scrape(
search_url="https://google.com",
query="python web scraping",
num_results=5
)
print(result)
7.5 Skyvern 视觉AI页面理解
from skyvern.agent import SkyvernAgent
import asyncio
class SkyvernExtractor:
"""Skyvern视觉AI提取器"""
def __init__(self, api_key: str = None):
self.agent = SkyvernAgent(api_key=api_key)
async def navigate_and_extract(self, url: str, goal: str):
"""
导航并提取数据
Args:
url: 目标URL
goal: 目标描述
"""
result = await self.agent.run_task(
url=url,
goal=goal
)
return result
async def fill_form(self, url: str, form_data: dict):
"""自动填写表单"""
goal = f"填写表单: {json.dumps(form_data)}"
return await self.navigate_and_extract(url, goal)
async def extract_table(self, url: str, table_description: str):
"""提取表格数据"""
goal = f"找到并提取以下表格: {table_description}"
return await self.navigate_and_extract(url, goal)
async def skyvern_demo():
extractor = SkyvernExtractor(api_key="YOUR_API_KEY")
result = await extractor.navigate_and_extract(
url="https://example.com/data",
goal="提取页面上的所有产品信息表格"
)
print(result)
7.6 LLM直接解析HTML
from openai import AsyncOpenAI
import asyncio
class LLMDirectExtractor:
"""LLM直接解析HTML"""
def __init__(self, api_key: str):
self.client = AsyncOpenAI(api_key=api_key)
async def extract_structured_data(self, html: str, schema: dict, instruction: str = None):
"""
使用LLM从HTML中提取结构化数据
Args:
html: HTML内容
schema: 期望的数据结构
instruction: 额外指令
"""
truncated_html = html[:8000] if len(html) > 8000 else html
system_prompt = """You are a web scraping expert. Extract structured data from the provided HTML according to the given schema. Return only valid JSON."""
user_prompt = f"""
Instruction: {instruction or 'Extract data according to the schema'}
Schema: {json.dumps(schema, indent=2)}
HTML:
{truncated_html}
Extract the data and return as JSON:
"""
response = await self.client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
response_format={"type": "json_object"}
)
result = response.choices[0].message.content
return json.loads(result)
async def llm_extract_demo():
extractor = LLMDirectExtractor(api_key="YOUR_API_KEY")
html = "<html>...</html>"
schema = {
"products": [{
"name": "string",
"price": "number"
}]
}
result = await extractor.extract_structured_data(html, schema)
print(result)
实战技巧
- Token优化: 截断HTML到8k-16k tokens，保留关键内容
- Schema设计: 使用明确的字段名和类型，提高提取准确率
- 多轮提取: 复杂页面分多次提取不同区域
- 结果验证: 使用Pydantic验证提取结果
- 成本优化: 先用传统方法，失败时再用AI方法
模块8: 公开数据源
概述
公开数据源是最合规、最稳定的采集方式。优先使用RSS、Open API、政府开放数据等官方渠道。
推荐工具
| 类型 | 工具/平台 |
|---|
| RSS/Atom | feedparser |
| Open API | requests + 官方SDK |
| 政府数据 | data.gov.cn, data.gov |
| 学术数据 | CrossRef, arXiv API |
代码示例
8.1 RSS/Atom Feed 采集
import feedparser
import requests
from datetime import datetime
class RSSScraper:
"""RSS Feed采集器"""
def __init__(self):
self.feeds = []
def parse_feed(self, url: str):
"""解析RSS Feed"""
feed = feedparser.parse(url)
result = {
"title": feed.feed.get("title", ""),
"link": feed.feed.get("link", ""),
"description": feed.feed.get("description", ""),
"updated": feed.feed.get("updated", ""),
"entries": []
}
for entry in feed.entries:
result["entries"].append({
"title": entry.get("title", ""),
"link": entry.get("link", ""),
"published": entry.get("published", ""),
"summary": entry.get("summary", ""),
"content": entry.get("content", [{}])[0].get("value", "") if entry.get("content") else ""
})
return result
def monitor_feeds(self, urls: list, callback):
"""监控多个Feed"""
for url in urls:
try:
feed = self.parse_feed(url)
callback(feed)
except Exception as e:
print(f"[-] Error parsing {url}: {e}")
8.2 Open API 对接
import requests
from typing import Dict, Any
class OpenAPIClient:
"""Open API客户端"""
def __init__(self, base_url: str, api_key: str = None):
self.base_url = base_url
self.api_key = api_key
self.session = requests.Session()
if api_key:
self.session.headers["Authorization"] = f"Bearer {api_key}"
def get(self, endpoint: str, params: Dict = None) -> Dict:
"""GET请求"""
url = f"{self.base_url}{endpoint}"
response = self.session.get(url, params=params)
response.raise_for_status()
return response.json()
def post(self, endpoint: str, data: Dict = None) -> Dict:
"""POST请求"""
url = f"{self.base_url}{endpoint}"
response = self.session.post(url, json=data)
response.raise_for_status()
return response.json()
8.3 政府开放数据平台
import requests
class GovernmentDataClient:
"""政府开放数据客户端"""
def __init__(self):
self.endpoints = {
"china": "https://data.gov.cn/api",
"usa": "https://api.data.gov",
"eu": "https://data.europa.eu/api"
}
def query_china_data(self, dataset_id: str, params: dict = None):
"""查询中国政府开放数据"""
url = f"{self.endpoints['china']}/datasets/{dataset_id}"
response = requests.get(url, params=params)
return response.json()
def query_usa_data(self, api_key: str, dataset: str, params: dict = None):
"""查询美国政府数据"""
url = f"{self.endpoints['usa']}/{dataset}"
params = params or {}
params["api_key"] = api_key
response = requests.get(url, params=params)
return response.json()
实战技巧
- API限流: 遵守Rate Limit，使用指数退避
- 数据更新: 使用ETag或Last-Modified检测更新
- 增量采集: 只采集新增数据，避免重复
- 数据验证: 验证数据格式和完整性
- 合规使用: 遵守数据使用协议和许可
模块9: 搜索引擎采集
概述
搜索引擎采集是发现目标网站和数据入口的重要手段。通过Google Dorking、搜索引擎API，可以快速定位特定类型的内容。
推荐工具
| 工具 | 用途 |
|---|
| Google Dorking | 高级搜索技巧 |
| SerpAPI | Google搜索结果API |
| ScraperAPI | 通用搜索引擎API |
| Bing API | 官方API |
代码示例
9.1 Google Dorking 高级搜索
class GoogleDorking:
"""Google Dorking高级搜索"""
DORKS = {
"site": "site:{domain}",
"inurl": "inurl:{keyword}",
"intitle": "intitle:{keyword}",
"intext": "intext:{keyword}",
"filetype": "filetype:{ext}",
"ext": "ext:{ext}",
"cache": "cache:{url}",
"related": "related:{domain}",
"link": "link:{url}",
"define": "define:{word}",
"stocks": "stocks:{symbol}",
"book": "book:{title}",
"maps": "maps:{location}",
"movie": "movie:{title}",
"weather": "weather:{location}",
}
@staticmethod
def build_dork(**kwargs) -> str:
"""构建Dork查询"""
parts = []
for key, value in kwargs.items():
if key in GoogleDorking.DORKS:
parts.append(GoogleDorking.DORKS[key].format(**{key: value}))
return " ".join(parts)
@staticmethod
def find_exposed_files(domain: str, extensions: list = None):
"""查找暴露的敏感文件"""
if extensions is None:
extensions = ["pdf", "doc", "docx", "xls", "xlsx", "sql", "env", "config"]
dorks = []
for ext in extensions:
dorks.append(f"site:{domain} filetype:{ext}")
return dorks
@staticmethod
def find_login_pages(domain: str):
"""查找登录页面"""
keywords = ["login", "admin", "signin", "wp-login", "administrator"]
dorks = []
for kw in keywords:
dorks.append(f"site:{domain} inurl:{kw}")
return dorks
@staticmethod
def find_api_endpoints(domain: str):
"""查找API端点"""
patterns = ["/api/", "/v1/", "/v2/", "/graphql", "/rest/"]
dorks = []
for pattern in patterns:
dorks.append(f"site:{domain} inurl:{pattern}")
return dorks
9.2 SerpAPI Google搜索
from serpapi import GoogleSearch
class SerpAPIScraper:
"""SerpAPI Google搜索采集"""
def __init__(self, api_key: str):
self.api_key = api_key
def search(self, query: str, num_results: int = 10, **kwargs):
"""
执行Google搜索
Args:
query: 搜索查询
num_results: 结果数量
**kwargs: 额外参数
"""
params = {
"q": query,
"num": num_results,
"api_key": self.api_key,
"engine": "google",
**kwargs
}
search = GoogleSearch(params)
results = search.get_dict()
return results
def get_organic_results(self, query: str, num_results: int = 10):
"""获取自然搜索结果"""
results = self.search(query, num_results)
organic = []
for result in results.get("organic_results", []):
organic.append({
"title": result.get("title"),
"link": result.get("link"),
"snippet": result.get("snippet"),
"position": result.get("position")
})
return organic
def get_news_results(self, query: str, num_results: int = 10):
"""获取新闻结果"""
results = self.search(query, num_results, tbm="nws")
news = []
for result in results.get("news_results", []):
news.append({
"title": result.get("title"),
"link": result.get("link"),
"source": result.get("source"),
"date": result.get("date"),
"snippet": result.get("snippet")
})
return news
9.3 Bing API 官方接口
import requests
class BingAPIScraper:
"""Bing API搜索采集"""
def __init__(self, api_key: str):
self.api_key = api_key
self.endpoint = "https://api.bing.microsoft.com/v7.0/search"
self.headers = {"Ocp-Apim-Subscription-Key": api_key}
def search(self, query: str, count: int = 10, offset: int = 0):
"""
执行Bing搜索
Args:
query: 搜索查询
count: 结果数量(最大50)
offset: 偏移量
"""
params = {
"q": query,
"count": min(count, 50),
"offset": offset,
"textDecorations": False,
"textFormat": "HTML"
}
response = requests.get(
self.endpoint,
headers=self.headers,
params=params
)
response.raise_for_status()
return response.json()
def get_web_pages(self, query: str, count: int = 50):
"""获取网页结果"""
results = self.search(query, count)
pages = []
for web_page in results.get("webPages", {}).get("value", []):
pages.append({
"name": web_page.get("name"),
"url": web_page.get("url"),
"snippet": web_page.get("snippet"),
"dateLastCrawled": web_page.get("dateLastCrawled")
})
return pages
实战技巧
- Dork组合: 多个dork条件组合使用，精确定位
- 结果去重: 搜索引擎结果可能有重复，需要去重
- 分页采集: 使用offset或start参数获取更多结果
- 限流控制: 搜索引擎API有严格限流，注意控制频率
- 缓存结果: 避免重复搜索相同查询
模块10: 社交媒体采集
概述
社交媒体采集是获取用户生成内容的重要渠道。各平台有不同的防护机制，需要针对性方案。
推荐工具
| 平台 | 工具/方案 |
|---|
| 微博 | Cookie登录+API |
| 小红书 | x-s签名逆向 |
| 抖音 | _signature+X-Bogus |
| Twitter/X | API v2 |
| Facebook | Graph API |
| Instagram | GraphQL API |
代码示例
10.1 微博采集
import requests
import json
import re
class WeiboScraper:
"""微博采集器"""
def __init__(self, cookies: str = None):
self.session = requests.Session()
self.session.headers.update({
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0.0.0",
"Referer": "https://weibo.com/"
})
if cookies:
self.session.headers["Cookie"] = cookies
def get_user_weibo(self, uid: str, page: int = 1):
"""获取用户微博"""
url = "https://weibo.com/ajax/statuses/mymblog"
params = {
"uid": uid,
"page": page,
"feature": 0
}
response = self.session.get(url, params=params)
data = response.json()
weibos = []
for item in data.get("data", {}).get("list", []):
weibos.append({
"id": item.get("id"),
"text": item.get("text_raw", ""),
"created_at": item.get("created_at"),
"reposts_count": item.get("reposts_count"),
"comments_count": item.get("comments_count"),
"attitudes_count": item.get("attitudes_count")
})
return weibos
def search_weibo(self, keyword: str, page: int = 1):
"""搜索微博"""
url = "https://weibo.com/ajax/side/search"
params = {
"q": keyword,
"page": page
}
response = self.session.get(url, params=params)
return response.json()
10.2 小红书 x-s签名破解
import requests
import hashlib
import time
import json
class XiaohongshuScraper:
"""小红书采集器 - x-s签名破解"""
def __init__(self):
self.session = requests.Session()
self.session.headers.update({
"User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15",
"Accept": "application/json, text/plain, */*",
"Accept-Language": "zh-CN,zh;q=0.9",
"X-Sign": "",
"X-Timestamp": "",
"X-S": ""
})
def generate_xs_sign(self, api: str, params: dict = None) -> dict:
"""
生成x-s签名
注意: 这是简化示例，实际算法需要逆向JS获取
"""
timestamp = str(int(time.time() * 1000))
sign_str = f"{api}{timestamp}"
if params:
sign_str += json.dumps(params, separators=(',', ':'), sort_keys=True)
x_s = hashlib.md5(sign_str.encode()).hexdigest()
return {
"X-Timestamp": timestamp,
"X-S": x_s,
"X-Sign": x_s
}
def get_user_notes(self, user_id: str, page: int = 1):
"""获取用户笔记"""
api = "/api/sns/web/v1/user_posted"
params = {
"user_id": user_id,
"page": page,
"page_size": 20
}
headers = self.generate_xs_sign(api, params)
self.session.headers.update(headers)
url = f"https://www.xiaohongshu.com{api}"
response = self.session.get(url, params=params)
return response.json()
10.3 抖音 _signature+X-Bogus
import requests
import hashlib
import time
class DouyinScraper:
"""抖音采集器 - _signature+X-Bogus破解"""
def __init__(self):
self.session = requests.Session()
self.session.headers.update({
"User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15",
"Referer": "https://www.douyin.com/"
})
def generate_xbogus(self, url_params: str, user_agent: str) -> str:
"""
生成X-Bogus参数
注意: 这是简化示例，实际算法需要逆向JS获取
"""
timestamp = int(time.time())
md5_input = f"{url_params}{user_agent}{timestamp}"
xbogus = hashlib.md5(md5_input.encode()).hexdigest()[:21]
return xbogus
def get_user_videos(self, sec_user_id: str, cursor: int = 0):
"""获取用户视频列表"""
base_url = "https://www.douyin.com/aweme/v1/web/aweme/post/"
params = {
"sec_user_id": sec_user_id,
"count": 10,
"cursor": cursor
}
params_str = "&".join([f"{k}={v}" for k, v in params.items()])
xbogus = self.generate_xbogus(params_str, self.session.headers["User-Agent"])
params["X-Bogus"] = xbogus
response = self.session.get(base_url, params=params)
return response.json()
10.4 Twitter/X API v2
import tweepy
class TwitterScraper:
"""Twitter API v2采集器"""
def __init__(self, bearer_token: str):
self.client = tweepy.Client(bearer_token=bearer_token)
def get_user_tweets(self, username: str, max_results: int = 10):
"""获取用户推文"""
user = self.client.get_user(username=username)
user_id = user.data.id
tweets = self.client.get_users_tweets(
id=user_id,
max_results=max_results,
