496 lines
17 KiB
Python
496 lines
17 KiB
Python
"""
|
|
使用 Camoufox 完成 Firecrawl 注册
|
|
思路：从真实注册页开始，自动处理邮箱验证和 API Key 提取
|
|
"""
|
|
import os
|
|
import re
|
|
import threading
|
|
import time
|
|
import requests as std_requests
|
|
from camoufox.sync_api import Camoufox
|
|
from config import (
|
|
API_KEY_TIMEOUT,
|
|
EMAIL_CODE_TIMEOUT,
|
|
FIRECRAWL_REGISTER_HEADLESS,
|
|
)
|
|
from mail_provider import get_verification_link
|
|
|
|
_HERE = os.path.dirname(os.path.abspath(__file__))
|
|
_SAVE_FILE = os.path.join(_HERE, "firecrawl_accounts.txt")
|
|
_SAVE_LOCK = threading.Lock()
|
|
_FIRECRAWL_SIGNUP_RESULT_TIMEOUT = 15
|
|
|
|
|
|
def attach_signup_feedback_tracker(page):
|
|
"""记录注册请求的关键反馈，方便识别风控或表单错误。"""
|
|
events = []
|
|
|
|
def handle_response(response):
|
|
url = response.url.lower()
|
|
if not any(token in url for token in ("signin", "signup", "auth", "clerk")):
|
|
return
|
|
|
|
try:
|
|
body = response.text()
|
|
except Exception:
|
|
body = ""
|
|
|
|
events.append(
|
|
{
|
|
"url": response.url,
|
|
"status": response.status,
|
|
"body": body[:1500],
|
|
}
|
|
)
|
|
|
|
page.on("response", handle_response)
|
|
return events
|
|
|
|
|
|
def detect_signup_result(page, signup_events):
|
|
"""根据页面内容和网络响应判断注册提交是否真的成功。"""
|
|
snapshots = []
|
|
current_url = page.url.lower()
|
|
|
|
if "confirm-email" in current_url or "confirm_email" in current_url:
|
|
return ("sent", "")
|
|
|
|
try:
|
|
snapshots.append(page.locator("body").inner_text())
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
snapshots.append(page.content())
|
|
except Exception:
|
|
pass
|
|
|
|
snapshots.extend(event.get("body", "") for event in signup_events[-6:])
|
|
combined = "\n".join(snapshots).lower()
|
|
|
|
if "security check failed" in combined or "suspicious activity" in combined:
|
|
return (
|
|
"blocked",
|
|
"Firecrawl 返回了 Security check failed / suspicious activity，当前浏览器指纹或网络被风控拦截。",
|
|
)
|
|
|
|
if "already exists" in combined or "account already exists" in combined:
|
|
return ("exists", "这个邮箱看起来已经注册过了。")
|
|
|
|
if "invalid email" in combined or "email address is invalid" in combined:
|
|
return ("invalid_email", "Firecrawl 认为这个邮箱地址无效。")
|
|
|
|
if "password is not strong enough" in combined or "at least 12 characters" in combined:
|
|
return (
|
|
"weak_password",
|
|
"Firecrawl 拒绝了当前密码强度，至少需要 12 位，并同时包含大小写、数字和特殊字符。",
|
|
)
|
|
|
|
success_markers = (
|
|
"check your email",
|
|
"confirm email",
|
|
"confirmation link",
|
|
"verify your email",
|
|
"verification email",
|
|
"email has been sent",
|
|
"we sent you an email",
|
|
"did not receive the email",
|
|
"once confirmed, you may sign in",
|
|
)
|
|
if any(marker in combined for marker in success_markers):
|
|
return ("sent", "")
|
|
|
|
return ("", "")
|
|
|
|
|
|
def wait_for_signup_result(page, signup_events, timeout=_FIRECRAWL_SIGNUP_RESULT_TIMEOUT):
|
|
"""等待注册提交后的明确结果，避免被风控时继续盲等验证邮件。"""
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
status, message = detect_signup_result(page, signup_events)
|
|
if status:
|
|
return status, message
|
|
time.sleep(1)
|
|
|
|
current_url = page.url.lower()
|
|
if "confirm-email" in current_url or "confirm_email" in current_url:
|
|
return ("sent", "")
|
|
|
|
if "view=signup" in current_url or current_url.rstrip("/").endswith("/signin"):
|
|
return (
|
|
"stalled",
|
|
"提交后页面仍停留在注册页，Firecrawl 没有明确确认已发送验证邮件。",
|
|
)
|
|
|
|
return ("", "")
|
|
|
|
def fill_first_input(page, selectors, value):
|
|
"""填充第一个存在的输入框"""
|
|
for selector in selectors:
|
|
if page.query_selector(selector):
|
|
page.fill(selector, value)
|
|
return selector
|
|
return None
|
|
|
|
def extract_api_key_from_page(page):
|
|
"""从 API Keys 页面提取 API Key"""
|
|
try:
|
|
# 等待 API Keys 页面加载
|
|
time.sleep(3)
|
|
|
|
# 尝试多种选择器查找 API Key
|
|
selectors = [
|
|
'code:has-text("fc-")',
|
|
'[data-testid="api-key"]',
|
|
'.api-key',
|
|
'input[value^="fc-"]',
|
|
'span:has-text("fc-")',
|
|
]
|
|
|
|
for selector in selectors:
|
|
elements = page.query_selector_all(selector)
|
|
for element in elements:
|
|
text = element.inner_text() or element.get_attribute('value') or ''
|
|
match = re.search(r'fc-[a-zA-Z0-9_-]{20,}', text)
|
|
if match:
|
|
return match.group(0)
|
|
|
|
# 从页面 HTML 中提取
|
|
html = page.content()
|
|
matches = re.findall(r'fc-[a-zA-Z0-9_-]{20,}', html)
|
|
if matches:
|
|
return matches[0]
|
|
|
|
return None
|
|
except Exception as e:
|
|
print(f"⚠️ 提取 API Key 失败: {e}")
|
|
return None
|
|
|
|
def create_api_key(page):
|
|
"""在 Dashboard 中创建新的 API Key"""
|
|
try:
|
|
# 查找并点击创建 API Key 按钮
|
|
create_selectors = [
|
|
'button:has-text("Create")',
|
|
'button:has-text("New API Key")',
|
|
'button:has-text("Generate")',
|
|
'[data-testid="create-api-key"]',
|
|
]
|
|
|
|
for selector in create_selectors:
|
|
if page.query_selector(selector):
|
|
page.click(selector)
|
|
time.sleep(2)
|
|
break
|
|
|
|
# 如果有名称输入框，填写名称
|
|
name_input = page.query_selector('input[name="name"], input[placeholder*="name" i]')
|
|
if name_input:
|
|
page.fill('input[name="name"], input[placeholder*="name" i]', 'auto-generated-key')
|
|
time.sleep(1)
|
|
|
|
# 点击确认按钮
|
|
confirm_selectors = [
|
|
'button:has-text("Create")',
|
|
'button:has-text("Generate")',
|
|
'button:has-text("Confirm")',
|
|
'button[type="submit"]',
|
|
]
|
|
|
|
for selector in confirm_selectors:
|
|
if page.query_selector(selector):
|
|
page.click(selector)
|
|
time.sleep(3)
|
|
break
|
|
|
|
return True
|
|
except Exception as e:
|
|
print(f"⚠️ 创建 API Key 失败: {e}")
|
|
return False
|
|
|
|
def save_account(email, password, api_key):
|
|
"""并发注册时串行写入 firecrawl_accounts.txt"""
|
|
with _SAVE_LOCK:
|
|
with open(_SAVE_FILE, 'a', encoding='utf-8') as f:
|
|
f.write(f"{email},{password},{api_key}\n")
|
|
|
|
def verify_api_key(api_key, timeout=30):
|
|
"""真实调用 Firecrawl API，验证新 key 可用"""
|
|
transient_errors = (
|
|
std_requests.exceptions.SSLError,
|
|
std_requests.exceptions.ConnectionError,
|
|
std_requests.exceptions.Timeout,
|
|
)
|
|
last_error = None
|
|
|
|
for attempt in range(1, 4):
|
|
try:
|
|
response = std_requests.post(
|
|
"https://api.firecrawl.dev/v2/scrape",
|
|
json={
|
|
"url": "https://example.com",
|
|
},
|
|
headers={
|
|
"Authorization": f"Bearer {api_key}",
|
|
"Content-Type": "application/json",
|
|
},
|
|
timeout=timeout,
|
|
)
|
|
break
|
|
except transient_errors as exc:
|
|
last_error = exc
|
|
if attempt < 3:
|
|
print(f"⚠️ API Key 调用测试遇到网络/TLS 异常，正在重试 ({attempt}/3): {exc}")
|
|
time.sleep(attempt)
|
|
continue
|
|
print(f"⚠️ API Key 调用测试遇到网络/TLS 异常，暂时无法确认 Key 是否可用: {exc}")
|
|
print(" 这通常是本地代理 / TUN / DNS 劫持链路导致的瞬时握手失败，不一定代表 Key 无效。")
|
|
return None
|
|
except Exception as exc:
|
|
print(f"❌ API Key 调用测试失败: {exc}")
|
|
return False
|
|
else:
|
|
print(f"⚠️ API Key 调用测试未获得有效响应: {last_error}")
|
|
return None
|
|
|
|
if response.status_code == 200:
|
|
print("✅ API Key 调用测试通过")
|
|
return True
|
|
|
|
preview = response.text.strip().replace("\n", " ")[:160]
|
|
print(f"❌ API Key 调用测试失败: HTTP {response.status_code}")
|
|
if preview:
|
|
print(f" 响应: {preview}")
|
|
return False
|
|
|
|
def submit_form(page, input_selector=None):
|
|
"""提交表单"""
|
|
button_selectors = [
|
|
'button[type="submit"]',
|
|
'button:has-text("Sign up")',
|
|
'button:has-text("Continue")',
|
|
'button:has-text("Register")',
|
|
]
|
|
|
|
for selector in button_selectors:
|
|
if page.query_selector(selector):
|
|
try:
|
|
page.click(selector, timeout=3000)
|
|
return True
|
|
except Exception:
|
|
continue
|
|
|
|
if input_selector and page.query_selector(input_selector):
|
|
try:
|
|
page.press(input_selector, 'Enter')
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
return False
|
|
|
|
def register_with_browser(email, password):
|
|
"""使用浏览器自动化注册 Firecrawl"""
|
|
print(f"🌐 使用浏览器模式注册 Firecrawl: {email}")
|
|
|
|
try:
|
|
with Camoufox(headless=FIRECRAWL_REGISTER_HEADLESS) as browser:
|
|
page = browser.new_page()
|
|
signup_events = attach_signup_feedback_tracker(page)
|
|
|
|
# 1. 访问注册页
|
|
print("🧭 进入注册页...")
|
|
page.goto("https://firecrawl.dev/", wait_until="networkidle", timeout=30000)
|
|
time.sleep(2)
|
|
|
|
# 查找并点击 Sign Up 按钮
|
|
signup_selectors = [
|
|
'a:has-text("Sign up")',
|
|
'a:has-text("Sign Up")',
|
|
'button:has-text("Sign up")',
|
|
'a[href*="signup"]',
|
|
'a[href*="register"]',
|
|
]
|
|
|
|
for selector in signup_selectors:
|
|
if page.query_selector(selector):
|
|
page.click(selector)
|
|
time.sleep(3)
|
|
break
|
|
|
|
# 2. 填写注册表单
|
|
print("📝 填写注册信息...")
|
|
|
|
# 填写邮箱
|
|
email_selector = fill_first_input(
|
|
page,
|
|
['input[name="email"]', 'input[type="email"]', 'input[placeholder*="email" i]'],
|
|
email
|
|
)
|
|
if not email_selector:
|
|
print("❌ 未找到邮箱输入框")
|
|
return None
|
|
|
|
time.sleep(1)
|
|
|
|
# 填写密码
|
|
password_selector = fill_first_input(
|
|
page,
|
|
['input[name="password"]', 'input[type="password"]'],
|
|
password
|
|
)
|
|
if not password_selector:
|
|
print("❌ 未找到密码输入框")
|
|
return None
|
|
|
|
time.sleep(1)
|
|
|
|
# 3. 提交注册表单
|
|
print("📤 提交注册...")
|
|
submit_form(page, email_selector)
|
|
status, message = wait_for_signup_result(page, signup_events)
|
|
if status != "sent":
|
|
if message:
|
|
print(f"❌ {message}")
|
|
if status in {"blocked", "stalled"}:
|
|
if FIRECRAWL_REGISTER_HEADLESS:
|
|
print("💡 建议：把 FIRECRAWL_REGISTER_HEADLESS=false，再用可见浏览器重试。")
|
|
else:
|
|
print("💡 建议：切换更干净的网络/代理后重试，或在可见浏览器里手动完成注册动作。")
|
|
return None
|
|
|
|
# 4. 等待邮箱验证链接
|
|
print(f"📧 等待邮箱验证链接（最多 {EMAIL_CODE_TIMEOUT} 秒）...")
|
|
verify_url = get_verification_link(email, timeout=EMAIL_CODE_TIMEOUT)
|
|
if not verify_url:
|
|
print("❌ 未收到验证邮件")
|
|
return None
|
|
|
|
print(f"✅ 收到验证链接: {verify_url[:50]}...")
|
|
|
|
# 5. 访问验证链接
|
|
print("🔗 访问验证链接...")
|
|
page.goto(verify_url, wait_until="networkidle", timeout=60000)
|
|
time.sleep(5)
|
|
|
|
# 6. 检查是否需要登录
|
|
current_url = page.url.lower()
|
|
if 'login' in current_url or 'signin' in current_url:
|
|
print("🔐 需要登录...")
|
|
|
|
# 填写邮箱
|
|
fill_first_input(
|
|
page,
|
|
['input[name="email"]', 'input[type="email"]'],
|
|
email
|
|
)
|
|
time.sleep(1)
|
|
|
|
# 填写密码
|
|
fill_first_input(
|
|
page,
|
|
['input[name="password"]', 'input[type="password"]'],
|
|
password
|
|
)
|
|
time.sleep(1)
|
|
|
|
# 提交登录
|
|
submit_form(page)
|
|
time.sleep(5)
|
|
|
|
# 7. Firecrawl 新版会先进入 onboarding，页面 HTML 里通常已经带有 apiKey。
|
|
print("🔍 先从当前页面提取 API Key...")
|
|
api_key = extract_api_key_from_page(page)
|
|
if api_key:
|
|
print(f"✅ 当前页面已拿到 API Key: {api_key[:20]}...")
|
|
else:
|
|
print("ℹ️ 当前页面未直接拿到 API Key，继续尝试后台入口...")
|
|
|
|
# 8. 当前页面拿不到时，再尝试后台入口。
|
|
if not api_key:
|
|
print("🔑 导航到 API Keys 页面...")
|
|
|
|
# 尝试多种方式找到 API Keys 页面
|
|
api_key_nav_selectors = [
|
|
'a:has-text("API Keys")',
|
|
'a[href*="api-key"]',
|
|
'a[href*="apikey"]',
|
|
'a[href*="keys"]',
|
|
'button:has-text("API Keys")',
|
|
]
|
|
|
|
found_nav = False
|
|
for selector in api_key_nav_selectors:
|
|
if page.query_selector(selector):
|
|
page.click(selector)
|
|
time.sleep(3)
|
|
found_nav = True
|
|
break
|
|
|
|
if not found_nav:
|
|
# 尝试直接访问常见的 API Keys URL
|
|
possible_urls = [
|
|
"https://www.firecrawl.dev/app/api-keys",
|
|
"https://www.firecrawl.dev/app/settings",
|
|
"https://www.firecrawl.dev/app",
|
|
"https://firecrawl.dev/dashboard/api-keys",
|
|
"https://firecrawl.dev/api-keys",
|
|
"https://app.firecrawl.dev/api-keys",
|
|
]
|
|
|
|
for url in possible_urls:
|
|
try:
|
|
page.goto(url, wait_until="networkidle", timeout=15000)
|
|
time.sleep(3)
|
|
if 'api' in page.url.lower() and 'key' in page.url.lower():
|
|
found_nav = True
|
|
break
|
|
except:
|
|
continue
|
|
|
|
if not found_nav:
|
|
print("⚠️ 未找到 API Keys 页面，尝试从当前页面提取...")
|
|
|
|
# 9. 尝试提取现有的 API Key
|
|
print("🔍 查找 API Key...")
|
|
api_key = extract_api_key_from_page(page)
|
|
|
|
# 10. 如果没有找到，尝试创建新的 API Key
|
|
if not api_key:
|
|
print("💡 未找到现有 API Key，尝试创建新的...")
|
|
if create_api_key(page):
|
|
api_key = extract_api_key_from_page(page)
|
|
|
|
if not api_key:
|
|
print("❌ 无法获取 API Key")
|
|
return "SUCCESS_NO_KEY"
|
|
|
|
print(f"✅ 获取到 API Key: {api_key[:20]}...")
|
|
|
|
# 10. 验证 API Key
|
|
print("🧪 验证 API Key 可用性...")
|
|
verify_result = verify_api_key(api_key)
|
|
if verify_result is False:
|
|
print("⚠️ API Key 验证失败，但仍然保存")
|
|
elif verify_result is None:
|
|
print("⚠️ API Key 可用性暂时无法确认，但更像是网络/TLS 问题，仍然保存")
|
|
|
|
save_account(email, password, api_key)
|
|
|
|
print(f"🎉 注册成功")
|
|
print(f" 邮箱: {email}")
|
|
print(f" 密码: {password}")
|
|
print(f" Key : {api_key}")
|
|
return api_key
|
|
|
|
except Exception as e:
|
|
print(f"❌ 注册失败: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return None
|
|
|
|
if __name__ == "__main__":
|
|
from tavily_core import create_email
|
|
email, password = create_email(service="firecrawl")
|
|
register_with_browser(email, password)
|