调度和启动core/manager.py:
py
from __future__ import annotations # 启用未来注解特性，允许在类型提示中使用前向引用
import asyncio # 引入异步IO基础库，用于调度协程
import json # 处理JSON序列化，用于写入代理配置
import logging # 提供日志记录功能，方便排查运行状态
import random # 随机数工具，用于调度间隔抖动
import shutil # 文件与目录复制操作，用于准备浏览器数据目录
import socket # 网络套接字库，用于探测可用端口
import time # 时间工具，用于超时控制
from asyncio.subprocess import Process # 引入异步子进程类型，描述外部浏览器进程
from contextlib import suppress # 安全忽略特定异常，防止无谓报错
from dataclasses import dataclass # 数据类装饰器，简化会话对象定义
from pathlib import Path # 路径处理库，统一跨平台路径操作
from typing import Awaitable, Callable, List # 类型提示集合，包含回调类型定义
from playwright.async_api import Browser, BrowserContext, Page, Playwright, async_playwright # 异步版Playwright接口，用于连接已启动的浏览器
import env # 加载环境变量配置，获取运行参数
from database.db import fetch_next_signup_x, update_signup_x_status # 数据库操作函数，负责获取与更新signup_x数据
from database.model import SignupXModel # 数据模型，用于类型提示与字段访问
logger = logging.getLogger(__name__) # 创建模块级日志记录器，按模块名称输出日志
def _ensure_logging_configured() -> None:
"""初始化日志配置，保证INFO级别消息也能输出到控制台。
参数:
无
返回:
None
"""
root_logger = logging.getLogger() # 获取根日志记录器
if root_logger.handlers: # 若已经存在处理器则认为外部已配置
return
logging.basicConfig( # 设置基础日志配置，输出到标准输出
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
root_logger.setLevel(logging.INFO) # 进一步确保根日志级别为INFO
@dataclass(slots=True) # 使用数据类并启用slots以减少内存开销
class BrowserSession: # 封装单个浏览器会话相关状态
"""指纹浏览器会话信息
参数:
signup (SignupXModel): 对应的signup_x数据库记录
port (int): 浏览器远程调试端口
process (Process): 浏览器外部进程对象
playwright (Playwright): Playwright实例，用于后续自动化控制
browser (Browser): 通过CDP连接得到的浏览器对象
context (BrowserContext): 浏览器上下文对象
page (Page): 启动后默认的标签页，用于后续操作
profile_dir (Path): 用户的数据目录
返回:
None
""" # 数据类文档注释，说明结构与字段含义
signup: SignupXModel # 对应的signup_x记录，包含用户注册资料
port: int # 当前浏览器实例占用的远程调试端口
process: Process # 指纹浏览器的异步子进程句柄
playwright: Playwright # Playwright主对象，用于后续操作
browser: Browser # 通过CDP连接得到的浏览器句柄
context: BrowserContext # 浏览器上下文对象，管理页面集合
page: Page # 默认打开的标签页对象
profile_dir: Path # 浏览器用户数据目录路径
async def close(self) -> None: # 定义异步关闭方法，释放所有资源
"""关闭Playwright连接并终止浏览器进程
参数:
无
返回:
None
""" # 方法文档注释，描述清理流程
if self.browser.is_connected(): # 判断浏览器是否仍然连接，避免重复关闭
await self.browser.close() # 异步关闭Playwright持有的浏览器句柄
await self.playwright.stop() # 停止Playwright实例，释放底层资源
if self.process.returncode is None: # 检查子进程是否仍在运行
self.process.terminate() # 请求指纹浏览器进程终止
with suppress(ProcessLookupError): # 忽略可能出现的进程查找异常
await self.process.wait() # 等待进程真正结束，清理系统资源
logger.info("已关闭浏览器端口 %s", self.port) # 记录关闭操作的端口信息
class FingerprintBrowserManager: # 管理指纹浏览器实例的核心类
"""fingerprint-chromium浏览器管理器
参数:
project_root (Path | None): 项目根目录，不提供时默认取当前模块的上级目录
返回:
None
""" # 管理器文档注释，描述职责与初始化参数
# 构造函数，完成路径与依赖检查
def __init__(self, project_root: Path | None = None):
_ensure_logging_configured() # 确保日志系统已就绪
self.project_root = Path(project_root or Path(__file__).resolve().parent.parent) # 计算项目根目录，默认取上级路径
if not env.BROWSER_PATH: # 校验浏览器路径配置是否存在
raise ValueError("BROWSER_PATH环境变量未配置") # 缺少配置时抛出错误提醒
self.browser_path = Path(env.BROWSER_PATH) # 将浏览器路径转换为Path对象便于操作
if not self.browser_path.exists(): # 检查实际文件是否存在
raise FileNotFoundError(f"浏览器路径不存在: {self.browser_path}") # 文件不存在时抛出异常
self.chromium_data_dir = self.project_root / "chromium_data" # 定义指纹浏览器数据根目录
self.default_template_dir = self.project_root / "ext" / "Default" # 指向Default扩展模板目录
self.proxy_template_dir = self.project_root / "ext" / "proxy" # 指向代理扩展模板目录
self.captcha_dir = self.project_root / "ext" / "captcha2" # 指向验证码扩展目录
self.chromium_data_dir.mkdir(parents=True, exist_ok=True) # 确保数据目录存在，必要时自动创建
self._validate_templates() # 检查各模板目录是否齐备
def _validate_templates(self) -> None: # 校验模板目录的私有方法
"""检测指纹浏览器扩展模板目录
参数:
无
返回:
None
""" # 文档注释，说明校验作用
if not self.default_template_dir.exists(): # 判断Default模板是否存在
raise FileNotFoundError(f"缺少扩展模板: {self.default_template_dir}") # 缺失时抛出错误
if not self.proxy_template_dir.exists(): # 判断proxy模板是否存在
raise FileNotFoundError(f"缺少代理扩展模板: {self.proxy_template_dir}") # 缺失时抛出错误
if not self.captcha_dir.exists(): # 判断captcha目录是否存在
raise FileNotFoundError(f"缺少captcha扩展目录: {self.captcha_dir}") # 缺失时抛出错误
def _copy_template(self, src: Path, dest: Path) -> None: # 复制模板目录的工具方法
"""复制扩展模板
参数:
src (Path): 模板源目录
dest (Path): 目标目录
返回:
None
""" # 文档注释，说明参数与返回
if dest.exists(): # 如果目标目录已存在
shutil.rmtree(dest) # 删除旧目录以避免残留数据
shutil.copytree(src, dest) # 复制模板目录到目标位置
def _prepare_profile(self, signup: SignupXModel) -> tuple[Path, Path]: # 构建用户数据目录与代理扩展
"""创建用户浏览器数据目录并生成代理配置
参数:
signup (SignupXModel): 待注册的用户信息
返回:
tuple[Path, Path]: 返回用户目录以及代理扩展目录
""" # 文档注释，说明生成内容
user_dir = self.chromium_data_dir / str(signup.id) # 根据用户ID确定专属目录
user_dir.mkdir(parents=True, exist_ok=True) # 确保用户目录存在
default_target = user_dir / "Default" # 构造Default扩展目标路径
proxy_target = user_dir / "proxy" # 构造proxy扩展目标路径
self._copy_template(self.default_template_dir, default_target) # 复制Default扩展模板
self._copy_template(self.proxy_template_dir, proxy_target) # 复制proxy扩展模板
self._write_proxy_env(signup, proxy_target) # 写入代理配置文件
return user_dir, proxy_target # 返回准备好的目录路径
# 因为我找到浏览器内核不支持代理，我使用自己写的扩展实现，每个代理插件要写入一份额外的代理配置文件
def _write_proxy_env(self, signup: SignupXModel, proxy_dir: Path) -> Path: # 写env.json的私有方法
"""生成代理扩展的env配置文件
参数:
signup (SignupXModel): 用户信息，用于生成账号
proxy_dir (Path): 用户专属proxy目录
返回:
Path: env.json文件路径
""" # 文档注释，描述写入内容
payload = { # 构造代理配置字典
"scheme": "http", # 指定代理协议
"host": "us.res.proxy-seller.com", # 代理服务器主机名
"port": 10020, # 代理端口号
"user": f"apif2bfce8188a35091_city_{signup.city}_s_testx{signup.id}", # 根据城市与ID定制用户名
"password": "RNW78Fm5", # 代理密码
}
env_path = proxy_dir / "env.json" # 计算env.json目标路径
env_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") # 写入配置文件
return env_path # 返回生成的文件路径
# 扫描可用端口的私有方法
def _find_available_port(self) -> int:
"""扫描可用远程调试端口
参数:
无
返回:
int: 可用端口号
""" # 文档注释，说明扫描行为
start, end = env.PORT_RANGE # 读取端口范围配置
for port in range(start, end + 1): # 遍历端口区间
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: # 创建TCP套接字检测占用情况
sock.settimeout(0.2) # 设置连接超时时间，避免长时间阻塞
if sock.connect_ex(("127.0.0.1", port)) != 0: # 非0表示端口未占用
return port # 返回可用端口
raise RuntimeError("没有找到可用端口") # 遍历完仍无可用端口则报错
# 等待端口开放的工具方法
async def _wait_for_debug_port(self, port: int, timeout: float = 20.0) -> None:
"""等待浏览器开放远程调试端口
参数:
port (int): 端口号
timeout (float): 超时时间，单位秒
返回:
None
""" # 文档注释，描述等待逻辑
deadline = time.monotonic() + timeout # 计算终止时间点
while time.monotonic() < deadline: # 在超时前循环检测
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: # 创建临时套接字检测端口
sock.settimeout(0.5) # 每次检测等待0.5秒
if sock.connect_ex(("127.0.0.1", port)) == 0: # 返回0表示端口已开放
return # 端口可用直接退出
await asyncio.sleep(0.5) # 未开放时短暂休眠后重试
raise TimeoutError(f"端口{port}在{timeout}秒内未就绪") # 超时仍未开放则抛出超时异常
# 构造启动命令
def _build_command(self, signup: SignupXModel, port: int, user_dir: Path, proxy_dir: Path) -> list[str]:
"""构建fingerprint-chromium启动命令
参数:
signup (SignupXModel): 用户数据
port (int): 调试端口
user_dir (Path): 用户数据目录
proxy_dir (Path): 用户代理扩展目录
返回:
list[str]: 启动命令参数列表
""" # 文档注释，说明命令构造
proxy_path = str(proxy_dir) # 将代理目录转成字符串供命令行使用
captcha_path = str(self.captcha_dir) # 获取captcha扩展路径字符串
user_dir_str = str(user_dir) # 用户目录字符串形式
args = [ # 汇总所有命令行参数
str(self.browser_path), # 浏览器可执行文件路径
f"--user-data-dir={user_dir_str}", # 指定用户数据目录
"--no-first-run", # 禁用首次运行提示
"--no-default-browser-check", # 禁用默认浏览器检查
"--disable-webrtc-multiple-routes", # 限制WebRTC路由
"--force-webrtc-ip-handling-policy=disable_non_proxied_udp", # 强制WebRTC走代理
f"--fingerprint={signup.id}", # 注入指纹参数
f"--lang={signup.lang}", # 设置界面语言
f"--accept-lang={signup.accept_lang}", # 设置HTTP语言
f"--timezone={signup.timezone}", # 设置时区参数
f"--disable-extensions-except={proxy_path},{captcha_path}", # 仅加载指定扩展
f"--load-extension={proxy_path},{captcha_path}", # 加载扩展目录
f"--remote-debugging-port={port}", # 指定远程调试端口
]
return args # 返回参数列表
# 启动单个浏览器并建立连接，这里是启动浏览器并返回连接对象
async def launch_session(self, signup: SignupXModel) -> BrowserSession:
"""为单个signup_x记录启动浏览器
参数:
signup (SignupXModel): 待注册用户信息
返回:
BrowserSession: 浏览器会话对象
""" # 文档注释，说明流程输出
user_dir, proxy_dir = await asyncio.to_thread(self._prepare_profile, signup) # 在线程池中准备目录，避免阻塞事件循环
port = self._find_available_port() # 获取一个空闲端口
command = self._build_command(signup, port, user_dir, proxy_dir) # 构造启动命令
# 浏览器在这里启动
process = await asyncio.create_subprocess_exec(*command) # 异步启动指纹浏览器进程
playwright: Playwright | None = None # 预声明Playwright变量，便于异常处理
browser: Browser | None = None # 预声明浏览器变量
try: # 尝试建立Playwright连接
await self._wait_for_debug_port(port) # 等待远程调试端口开放
playwright = await async_playwright().start() # 启动Playwright异步客户端
browser = await playwright.chromium.connect_over_cdp(f"http://127.0.0.1:{port}") # 通过CDP连接浏览器
except Exception: # 捕获连接过程中的所有异常
if process.returncode is None: # 若进程仍然存活
process.terminate() # 主动终止指纹浏览器
with suppress(ProcessLookupError): # 忽略进程已退出的异常
await process.wait() # 等待进程真正退出
if playwright is not None: # 如果Playwright实例已创建
await playwright.stop() # 停止Playwright，避免资源泄露
raise # 将异常继续抛出供上层处理
contexts = browser.contexts # 读取浏览器内的上下文列表
if not contexts: # 如果没有上下文说明启动异常
raise RuntimeError("未检测到可用的浏览器上下文") # 抛出运行时错误
context = contexts[0] # 默认取第一个上下文
if context.pages: # 检查是否存在默认页面
page = context.pages[0] # 使用首个已有页面
else: # 若不存在页面
page = await context.new_page() # 主动创建新页面以便后续使用
logger.info("启动浏览器成功，端口: %s，用户: %s,%s", port, signup.email, signup.id) # 输出成功日志
return BrowserSession( # 构建BrowserSession返回给调用方
signup=signup, # 绑定signup记录
port=port, # 记录使用的端口
process=process, # 记录浏览器进程
playwright=playwright, # 保存Playwright实例
browser=browser, # 保存浏览器对象
context=context, # 保存上下文对象
page=page, # 保存用于操作的页面
profile_dir=user_dir, # 保存用户数据目录
)
# 启动单个浏览器会话并返回句柄
async def run_signup(self, signup: SignupXModel | None = None) -> BrowserSession | None: # 启动单实例返回会话
"""拉取一条待注册记录并启动一个指纹浏览器实例
参数:
signup (SignupXModel | None): 预先指定的数据库记录，默认自动提取状态为1的数据
返回:
BrowserSession | None: 返回成功启动的会话实例，若无可用数据则返回None
""" # 文档注释，说明单实例启动过程
record = signup or fetch_next_signup_x() # 优先使用调用方指定的记录，否则自动获取
if not record: # 如果没有可用记录
logger.info("没有可用的signup_x数据，跳过启动") # 输出提示日志
return None # 返回空结果
update_signup_x_status(record.id, 2) # 将该记录标记为运行中，避免被重复调度
try: # 保护启动流程，出现异常时回滚状态
session = await self.launch_session(record) # 调用底层方法真正创建浏览器会话
except Exception: # 捕获启动异常
update_signup_x_status(record.id, 1) # 恢复状态为待注册，方便后续重试
raise # 继续抛出异常供上层处理
return session # 返回成功启动的会话对象
# 长循环调度任务，按需在运行时增减浏览器实例
async def schedule_signups( # noqa: C901
self,
initial_instances: int = 2,
max_concurrent: int = 3,
interval_range: tuple[int, int] = (120, 300),
handler: Callable[[BrowserSession], Awaitable[None]] | None = None,
stop_event: asyncio.Event | None = None,
) -> None: # 异步调度函数，管理整个运行期
"""调度指纹浏览器实例，可动态拉起与销毁
参数:
initial_instances (int): 启动时预置的实例数量
max_concurrent (int): 同时允许存在的最大实例数
interval_range (tuple[int, int]): 新实例拉起的随机间隔范围，单位秒
handler (Callable[[BrowserSession], Awaitable[None]] | None): 会话处理协程，签名需为`async def handler(session: BrowserSession) -> None`
stop_event (asyncio.Event | None): 外部停止信号，为空时按数据耗尽结束
返回:
None
""" # 文档注释，描述调度行为
async def _handle_session(session: BrowserSession) -> None: # 会话包装器，负责执行任务并清理
try: # 确保无论任务是否成功都能关闭会话
if handler: # 如果调用方提供处理逻辑
await handler(session) # 执行自定义任务
finally: # 任务完成或异常时进入收尾
await session.close() # 关闭浏览器与Playwright连接
update_signup_x_status(session.signup.id, 4) # 将记录状态标记为结束
active_tasks: set[asyncio.Task] = set() # 跟踪当前活动任务，便于动态管理
async def _launch_one() -> bool: # 启动单实例并注册到任务集合
session = await self.run_signup() # 调度一个浏览器实例
if not session: # 若无可用数据
return False # 通知外层没有成功启动
task = asyncio.create_task(_handle_session(session)) # 为会话启动独立协程
active_tasks.add(task) # 将任务加入活动集合
task.add_done_callback(lambda t: active_tasks.discard(t)) # 任务结束时自动移除引用
return True # 启动成功
for _ in range(initial_instances): # 先启动指定数量的实例
launched = await _launch_one() # 尝试启动
if not launched: # 如果没有更多数据
break # 结束初始化阶段
while True: # 调度主循环，直到外部条件满足退出
if stop_event and stop_event.is_set(): # 检查外部停止信号
break # 收到停止请求后跳出循环
active_tasks = {task for task in active_tasks if not task.done()} # 刷新任务集合，仅保留未完成任务
if not active_tasks: # 没有正在运行的任务
if not await _launch_one(): # 尝试继续拉起一个实例
break # 若拉起失败，说明无数据可用，结束调度
continue # 启动成功后继续下一轮
await asyncio.sleep(random.uniform(*interval_range)) # 等待随机时间控制拉起节奏，支持2-5分钟间隔
if len(active_tasks) >= max_concurrent: # 若已达到并发上限
continue # 跳过启动，等待下次循环
await _launch_one() # 尝试再启动一个实例，保持持续产出
if active_tasks: # 循环退出后，如果还有未完成任务
await asyncio.gather(*active_tasks, return_exceptions=True) # 等待所有任务收尾
# 测试调用
async def test( # noqa: C901
count: int = 1,
interval_seconds: int = 0,
handler: Callable[[BrowserSession], Awaitable[None]] | None = None,
) -> None: # 示例入口，可启动多个浏览器进行调试
"""示例：按需启动若干浏览器实例并执行指定任务
参数:
count (int): 需要启动的浏览器数量
interval_seconds (int): 每次启动之间的固定间隔秒数
handler (Callable[[BrowserSession], Awaitable[None]] | None): 任务处理协程
返回:
None
"""
from core.signup_x import run_signup_x_flow # 延迟导入，避免循环依赖
manager = FingerprintBrowserManager() # 创建浏览器管理器实例
sessions: List[BrowserSession] = [] # 用于保存已启动的会话，方便统一关闭
for index in range(count): # 按照请求的数量依次启动
session = await manager.run_signup() # 启动单个浏览器实例
if session: # 若成功启动
sessions.append(session) # 记录会话
if interval_seconds > 0 and index < count - 1: # 如果设定了间隔且不是最后一次
await asyncio.sleep(interval_seconds) # 等待指定时间再启动下一台
if not sessions: # 若没有任何会话启动成功
logger.warning("没有成功启动任何浏览器实例") # 输出警告日志
return # 直接返回
try: # 保证后续finally中能统一清理
for session in sessions: # 遍历所有会话
try: # 捕获任务执行异常
# 这里决定具体执行什么任务
# await session.page.goto("https://ipbrowser.net/", timeout=30000) # 使用已有标签访问检测站点
await run_signup_x_flow(session) # 执行指定任务流程
# 等待输入
await asyncio.to_thread(input, "按回车后开始关闭……")
except Exception as exc: # 如果任务执行失败
update_signup_x_status(session.signup.id, 3) # 恢复状态，便于后续重试
logger.exception("任务执行失败，端口 %s: %s", session.port, exc) # 记录异常详细信息
raise # 继续抛出异常，方便调试
await asyncio.to_thread(input, "任务已执行，按回车后开始关闭……") # 在后台线程等待用户按回车
finally: # 不论是否出错都执行清理逻辑
await asyncio.gather(*(session.close() for session in sessions), return_exceptions=True) # 并发关闭所有会话
for session in sessions: # 遍历所有会话
update_signup_x_status(session.signup.id, 4) # 将状态标记为结束
logger.info("已关闭全部浏览器并恢复状态") # 输出收尾日志
if __name__ == "__main__": # 当脚本直接运行时执行示例
asyncio.run(test()) # 启动事件循环并执行测试协程
具体爬虫任务，一种任务一个文件core\signup_x.py:
py
from __future__ import annotations # 未来注解特性，便于类型前向引用
import asyncio # 引入异步IO，配合to_thread执行阻塞调用
import logging # 日志记录，方便排查异常
from typing import TYPE_CHECKING # 条件导入辅助，避免循环依赖
from playwright.async_api import Locator, Page # Playwright的页面类型，便于类型提示
from core.human import human_type, human_click, human_delay # 引入人类操作函数
from core.ser import get_email_captcha # 引入验证码获取函数
if TYPE_CHECKING: # 仅在类型检查时导入，运行时避免循环依赖
from core.manager import BrowserSession # 引入BrowserSession类型提示
logger = logging.getLogger(__name__) # 创建模块级日志对象
async def _select_birthdate(page: Page, year: int, month: int, day: int) -> None:
"""在注册表单中选择出生年月日""" # 函数文档注释：说明作用
# 定位包含三个下拉框的容器
container = page.locator("div[data-viewportview='true']")
selects = container.locator("select") # 获取所有下拉框
if await selects.count() < 3: # 如果数量不足则抛出异常
raise RuntimeError("未找到完整的出生日期下拉框")
# 分别选择年、月、日
await selects.nth(2).select_option(str(year))
await human_delay()
await selects.nth(0).select_option(str(month))
await human_delay()
await selects.nth(1).select_option(str(day))
# 注册成功后随季节浏览操作
async def _random_browse(page: Page) -> None:
"""注册成功后随季节浏览操作""" # 函数文档注释：说明作用
pass
async def _is_fun_captcha_present(page: Page):
"""检测当前页面是否出现FunCaptcha挑战。
参数:
page (Page): Playwright页面对象，用于查找iframe及其内容。
返回:
bool: 若检测到包含 ``id="FunCaptcha-Token"`` 的输入框则返回True，否则返回False。
"""
iframe_locator = page.locator("#FunCaptcha > iframe") # 定位页面上所有iframe元素
iframe_count = await iframe_locator.count() # 计算iframe数量
if iframe_count == 0: # 若不存在iframe则直接返回
return False
for index in range(iframe_count): # 遍历所有iframe
iframe_handle = await iframe_locator.nth(index).element_handle() # 获取iframe句柄
if iframe_handle is None: # 如果当前iframe无法获取句柄，跳过
continue
frame = await iframe_handle.content_frame() # 获取iframe对应的frame对象
if frame is None: # 若frame尚未就绪，继续检查下一个
continue
try:
token_locator = frame.locator("input#FunCaptcha-Token") # 在frame内查找目标输入框
if await token_locator.count() > 0: # 找到目标输入框即认为进入FunCaptcha
return True
except Exception as exc: # 捕获可能的跨域或访问异常
logger.debug("检测FunCaptcha时发生异常: %s", exc) # 记录调试日志帮助定位问题
continue
return False # 遍历结束仍未找到，认为未进入FunCaptcha
# 供外部调用，执行X平台注册流程
async def run_signup_x_flow(session: "BrowserSession") -> None:
"""执行X平台注册流程""" # 函数文档注释：描述核心流程
page = session.page # 获取当前浏览器页面
signup = session.signup # 读取数据库中的注册信息
# 设置默认超时时间，避免网络波动导致过早报错
page.set_default_navigation_timeout(60_000)
page.set_default_timeout(60_000)
await asyncio.to_thread(input, "如果需要调试，请按F12键，然后回车键继续...")
# 打开目标站点
logger.info("打开目标站点")
await page.goto("https://x.com/", wait_until="domcontentloaded")
await human_delay(2, 5) # 页面加载后延迟一会儿
# 点击注册按钮
logger.info("点击注册按钮")
signup_btn = page.locator('a[href*="/signup"]')
await signup_btn.first.wait_for(state="visible")
# await signup_btn.first.click()
await human_click(page, signup_btn.first)
await human_delay()
# 输入昵称
logger.info("输入昵称")
name_input = page.locator("input[name='name'][autocomplete='name']")
await name_input.wait_for(state="visible")
# await name_input.fill(signup.name)
await human_type(name_input, signup.name)
await human_delay()
# 切换使用邮箱注册
logger.info("切换使用邮箱注册")
email_toggle = page.locator("div[data-viewportview='true'] button:has(span)")
if await email_toggle.count() > 0:
await email_toggle.first.wait_for(state="visible")
# await email_toggle.first.click()
await human_click(page, email_toggle.first)
await human_delay()
# 输入邮箱
logger.info("输入邮箱")
email_input = page.locator("input[name='email'][autocomplete='email']")
await email_input.wait_for(state="visible")
# await email_input.fill(signup.email)
await human_type(email_input, signup.email)
await human_delay()
# 选择出生年月日
logger.info("选择出生年月日")
await _select_birthdate(page, signup.year, signup.month, signup.day)
await human_delay()
# 点击下一步按钮
logger.info("点击下一步按钮")
next_btn = page.locator('div[data-viewportview="true"] + div button')
await next_btn.wait_for(state="visible")
# await next_btn.click()
await human_click(page, next_btn)
await human_delay(15, 20) # 这里等待的就因为要等验证码
# 检查是否在机器人挑战页：FunCaptcha包含特定的iframe与输入框
logger.info("检测是否出现FunCaptcha挑战页")
if await _is_fun_captcha_present(page):
logger.warning("检测到FunCaptcha挑战")
await asyncio.to_thread(input, "检测到FunCaptcha挑战，请手动完成后按回车继续...")
# 读取验证码，从远程服务器获取,如果返回空字符串，这里重试2次，如果接收到非空则读取验证码继续
logger.info("读取验证码，这里会重复三次请求服务器")
for _ in range(3):
code = await get_email_captcha(signup.email, "x")
if not code:
# 如果没有验证码，延迟10-15秒后重试
await human_delay(12, 15)
continue
break
# 如果依旧获取验证码失败，这里应该要点击重新获取验证码
if not code:
logger.error("用户 %s 验证码获取失败", signup.email)
# 使用to_thread在后台线程执行input，避免阻塞事件循环
await asyncio.to_thread(input, "一直没有获取到验证码，按回车键执行退出逻辑...")
# 假设这里是重新点击验证码的过程
return
# 输入验证码
logger.info("输入验证码")
code_inp = page.locator("input[name='verfication_code'][autocomplete='on']")
await code_inp.wait_for(state="visible")
await human_type(code_inp, code)
await human_delay()
# 点击确认验证码按钮
logger.info("点击确认验证码按钮")
next_btn = page.locator('div[data-viewportview="true"] + div button')
await next_btn.wait_for(state="visible")
await human_click(page, next_btn)
await human_delay(2, 5)
# 检查是否在机器人挑战页：FunCaptcha包含特定的iframe与输入框
logger.info("检测是否出现FunCaptcha挑战页")
if await _is_fun_captcha_present(page): # 若检测到FunCaptcha
logger.warning("检测到FunCaptcha挑战")
await asyncio.to_thread(input, "检测到FunCaptcha挑战，请手动完成后按回车继续...")
# 进入到输入密码的逻辑
logger.info("输入密码")
password_inp = page.locator("input[type='password'][name='password']")
await password_inp.wait_for(state="visible")
await human_type(password_inp, signup.password)
await human_delay()
# 输入密码后下一步
logger.info("输入密码后点击下一步")
password_btn = page.locator('div[data-testid="LoginForm_Footer_Container"] div:not(:has(a)) button')
password_btn.wait_for(state="visible")
await human_click(page, password_btn)
await human_delay(4, 8)
# 跳过设置头像步骤
logger.info("跳过设置头像步骤")
skip_btn = page.locator('button[data-testid$="SkipForNowButton"]')
skip_btn.wait_for(state="visible")
await human_click(page, skip_btn)
await human_delay(2, 5)
logger.info("用户 %s 已完成基础注册表单", signup.email)
# 再次等待用户输入，同样放入线程以确保提示输出
await asyncio.to_thread(input, "按回车键执行退出逻辑...")
## 这里再最好随机做一些浏览操作，停留一分钟，这样看起来不是注册完就走人了
目录结构：
scss
playwright_browser/
├─ core/
│ ├─ human.py 模拟人类键鼠操作
│ ├─ signup_x.py 注册X账户，具体的爬虫逻辑都使用单独的文件来写
│ └─ manager.py 浏览器调度启动核心
├─ database/ 数据库类
│ ├─ __init__.py
│ ├─ db.py 数据库操作
│ └─ model.py 数据表模型
├─ ext/ 浏览器扩展公共目录，用来拷贝给每个浏览器使用
│ ├─ Default/ 过验证码扩展的设置文件(每个浏览器要拷贝一份)
│ ├─ captcha/ 过验证码的扩展(公共使用)
│ └─ proxy/ 自己写的代理扩展(每个浏览器要拷贝一份，再额外由代码生成一份env.json配置)
├─ test/ 测试代码/参考代码
│ ├─ cdp.py
│ └─ playwright启动.py
├─ chromium_data/ 浏览器用户隔离
└─ env.py 统一从环境变量读取
自写浏览器代理扩展
因为chromium不支持带账号密码的代理，而代理池要通过这个来限制地区
ext\proxy\目录创建两个文件：
ext\proxy\manifest.json：
json
{
"name": "Global Speed",
"version": "1.0.0",
"manifest_version": 3,
"description": "Global Speed",
"permissions": [
"proxy",
"webRequest",
"webRequestAuthProvider",
"privacy",
"storage"
],
"host_permissions": ["<all_urls>"],
"background": {
"service_worker": "background.js",
"type": "module"
},
"web_accessible_resources": [
{
"resources": ["env.json"],
"matches": ["<all_urls>"]
}
],
"incognito": "spanning"
}
ext\proxy\background.js：
js
// ========== MV3 Service Worker 初始化 ==========
// 全局变量存储配置
let ENV = null;
/**
* 从扩展包内读取 env.json 配置
*/
async function loadConfig() {
try {
const url = chrome.runtime.getURL('env.json');
console.log('[ext_auth] 正在加载配置:', url);
const response = await fetch(url);
if (!response.ok) {
throw new Error(`HTTP ${response.status}: ${response.statusText}`);
}
const config = await response.json();
console.log('[ext_auth] 配置加载成功:', {
scheme: config.scheme,
host: config.host,
port: config.port,
hasUser: !!config.user,
hasPassword: !!config.password
});
return config;
} catch (error) {
console.error('[ext_auth] ❌ 加载 env.json 失败:', error);
return null;
}
}
/**
* 应用代理配置到指定作用域
*/
function applyProxyConfigTo(scope) {
if (!ENV) {
console.warn('[ext_auth] ENV 未初始化，跳过代理设置');
return;
}
const config = {
mode: "fixed_servers",
rules: {
singleProxy: {
scheme: ENV.scheme,
host: ENV.host,
port: parseInt(ENV.port)
},
bypassList: [] // 不留任何直连白名单
}
};
chrome.proxy.settings.set({ value: config, scope }, () => {
if (chrome.runtime.lastError) {
console.error(`[ext_auth] ❌ 设置代理失败 (${scope}):`, chrome.runtime.lastError);
} else {
console.log(`[ext_auth] ✓ 代理已设置 (${scope}):`, `${ENV.scheme}://${ENV.host}:${ENV.port}`);
}
});
}
/**
* 设置代理和隐私选项
*/
async function setProxy() {
if (!ENV) {
console.warn('[ext_auth] ENV 未初始化，跳过代理设置');
return;
}
// regular（普通窗口）
applyProxyConfigTo("regular");
// 尝试对隐身窗口生效：仅当用户已允许扩展在隐身模式运行
try {
const allowed = await new Promise(resolve =>
chrome.extension.isAllowedIncognitoAccess(resolve)
);
if (allowed) {
applyProxyConfigTo("incognito_persistent");
}
} catch (e) {
// 未被允许或环境不支持时忽略（避免报 Unchecked runtime.lastError）
console.log('[ext_auth] 隐身模式未启用或不支持');
}
// 关闭非代理 UDP 的 WebRTC，避免泄露
chrome.privacy?.network?.webRTCIPHandlingPolicy?.set(
{ value: "disable_non_proxied_udp" },
() => console.log('[ext_auth] WebRTC 已配置')
);
// 可选硬化
chrome.privacy?.network?.networkPredictionEnabled?.set({ value: false });
chrome.privacy?.network?.hyperlinkAuditingEnabled?.set({ value: false });
}
// ========== 初始化流程 ==========
(async () => {
// 1. 加载配置
ENV = await loadConfig();
if (!ENV) {
console.error('[ext_auth] ❌ 无法加载配置，扩展将不工作');
return;
}
// 2. 立即设置代理
await setProxy();
// 3. 注册事件监听器
chrome.runtime.onInstalled.addListener(() => {
console.log('[ext_auth] 扩展已安装/更新，重新设置代理');
setProxy();
});
chrome.runtime.onStartup.addListener(() => {
console.log('[ext_auth] 浏览器启动，重新设置代理');
setProxy();
});
chrome.proxy.settings.onChange.addListener((details) => {
console.log('[ext_auth] 代理设置被改动，重新应用');
setProxy();
});
// // 关闭非代理 UDP 的 WebRTC（已做）
// chrome.privacy?.network?.webRTCIPHandlingPolicy?.set({ value: "disable_non_proxied_udp" });
// // 关闭预连接/预测
// chrome.privacy?.network?.networkPredictionEnabled?.set({ value: false });
// // 关闭 DNS 预取（有的内核还保留这个开关）
// chrome.privacy?.network?.dnsPrefetchEnabled?.set?.({ value: false });
})();
// ========== 代理认证处理 ==========
chrome.webRequest.onAuthRequired.addListener(
(details, callback) => {
if (details.isProxy && ENV?.user && ENV?.password) {
console.log('[ext_auth] 提供代理认证凭据:', details.url);
callback({
authCredentials: {
username: ENV.user,
password: ENV.password
}
});
} else {
callback();
}
},
{ urls: ["<all_urls>"] },
["asyncBlocking"] // MV3 用 asyncBlocking；无需 webRequestBlocking 权限
);
一份env.json示范这个要脚本去生成，先创建这个浏览器的文件夹，然后将扩展复制到浏览器中，然后用脚本去生成这个文件，因为每个人的user都不一样，通过它来限定IP：
json
{
"scheme": "http",
"host": "us.res.proxy-seller.com",
"port": 10020,
"user": "<api>_city_Tokyo_s_0001",
"password": "<pass>"
}
说明
- 建议先执行企业策略减少开销和增加安全性
- 如果有过验证码要将验证码程序一起加载
- 此demo使用的是https://github.com/adryfish/fingerprint-chromium
- 浏览器备份了一份139版在阿里云盘
下面是一些在demo里多余的代码
生成注册信息，中日英，三种名字+英语邮箱等信息
py
from faker import Faker
from database.db import random_language, random_language_region, add_signup_x,check_signup_x_email
from database.model import SignupXModel
from pypinyin import lazy_pinyin
from pykakasi import kakasi
import random
import argparse
# 缓存日文转写器，避免重复初始化带来的性能开销
_kakasi = kakasi()
EMAIL_DOMAIN = ["fzmail.net","lvline.com"]
def _romanize_japanese(text: str) -> str:
"""将日文文本转换为赫本式罗马音。
:param text: 需要转换的日文文本。
:return: 转换后的罗马音文本（首字母大写）。
"""
# 使用新的 API 进行转换
result = _kakasi.convert(text)
# 提取赫本式罗马音并拼接
romaji_parts = []
for item in result:
# 优先使用赫本式罗马音，如果没有则尝试其他形式
hepburn = item.get("hepburn") or item.get("kunrei") or item.get("kana") or item.get("orig")
if hepburn:
romaji_parts.append(hepburn)
# 返回首字母大写的罗马音
return " ".join(romaji_parts).title()
# 根据传入的语言生成名字和英文名字，目前支持英语、日语、中文
def _generate_name(lang: str | None = None) -> tuple[str, str, str]:
"""生成指定语言的人名并返回对应的英文转写。
:param lang: 语言代码，例如 ``zh_CN``、``ja_JP``、``en_US``。
若为 ``None`` 则自动随机选择支持的语言。
:return: 包含原语言名字与英文名字的二元组 ``(origin_name, english_name)``。
"""
# 如果语言为空，则随机生成语言
if lang is None:
lang = random_language()
gender = random.choice(["boy", "girl"])
fake = Faker(lang)
if lang == "zh_CN":
# 常见中文男/女名字字符，可自行扩充
surname = fake.last_name()
male_chars = list("伟刚勇毅俊峰强军平保东文辉力明永健世广志义兴坤良海山轩仁波宁贵福生龙元全国胜学祥才发武新利飞彬富顺信子勤杰涛昌成康星光天达安岩中茂进林有坚和彪博诚先敬震振壮会思群邦承乐豪绍功松善厚庆磊民友裕河哲江超浩亮政谦亨奇固之轮翰朗伯宏言若鸣朋斌梁栋维启克伦翔旭鹏泽晨辰士建家致树炎德行好毅治易悠")
female_chars = list("秀娟英华慧巧美娜静淑惠珠翠雅芝玉萍红娥玲芬芳燕彩菊兰凤洁梅琳素云莲真淼环雪荣爱妹霞妙香月雨清莺媛蕊艳心瑞凡茗芯佳嘉琼珍贞莉桂娣叶璐娅琦晶妍茜秋珊莎锦黛青倩姣婉娴瑾颖露瑶怡婵雁蓓桦柳璇冬曼丹雅彬倩婷丽")
given_len = random.choice([1, 2]) # 随机生成1或2个字符
if gender == "boy": # 男
given = "".join(random.choices(male_chars, k=given_len))
else: # 女
given = "".join(random.choices(female_chars, k=given_len))
name = surname + given # 拼出完整中文姓名
# 使用拼音作为英文转写
english_name = " ".join(lazy_pinyin(name))
elif lang == "ja_JP":
surname = fake.last_name() # 日文姓氏
if gender == "boy": # 男
given = fake.first_name_male() # 日文男性名
else: # 女
given = fake.first_name_female() # 日文女性名
name = f"{surname}{given}" # 日文姓名常用空格分隔
# 使用 pykakasi 将日文转换成罗马音
english_name = _romanize_japanese(name)
else:
if gender == "boy": # 男
name = fake.name_male() # 英文男名
else: # 女
name = fake.name_female() # 英文女名
english_name = name # 英文名无需再转写
return name, english_name, gender
# 差异化邮箱
def _generate_email(name: str) -> str:
"""生成差异化邮箱地址。
根据传入的英文名生成邮箱，包含以下随机变换：
- 取消空格、或使用下划线替代空格
- 随机添加1-3位数字后缀
- 随机删除首尾1-2个字符
:param name: 英文名字。
:return: 生成并验证的唯一邮箱地址。如果邮箱已存在则重新生成。
"""
while True:
base_name = name
# 随机选择处理方式：取消空格或使用下划线
if random.choice([True, False]):
base_name = base_name.replace(" ", "") # 取消空格
else:
base_name = base_name.replace(" ", "_") # 使用下划线替代空格
# 随机删除首尾1-2个字符
if len(base_name) > 2:
# 随机选择是否删除首字符
if random.choice([True, False]) and len(base_name) > 1:
base_name = base_name[1:]
# 随机选择是否删除尾字符
if random.choice([True, False]) and len(base_name) > 1:
base_name = base_name[:-1]
# 随机添加1-3位数字后缀
if random.choice([True, False]):
suffix_length = random.choice([1, 2])
suffix = "".join([str(random.randint(0, 9)) for _ in range(suffix_length)])
# 随机决定是加前缀还是后缀
if random.choice([True, False]):
base_name = suffix + base_name
else:
base_name = base_name + suffix
# 生成完整邮箱地址
email = f"{base_name}@{random.choice(EMAIL_DOMAIN)}".lower()
# 检查邮箱是否已存在
if not check_signup_x_email(email):
return email
# 生成信息
def generate_info(lang: str | None = None):
"""生成单条用户信息"""
# 如果指定了语言，使用指定的语言；否则随机选择
if lang is not None:
# 找到对应的语言区域
from database.db import sy_db_session
from sqlalchemy import select
from database.model import LanguageRegionModel
with sy_db_session() as session:
result = session.execute(select(LanguageRegionModel).where(LanguageRegionModel.lang == lang))
language_region = result.scalars().first()
if language_region is None:
raise ValueError(f"不支持的语言代码: {lang}")
else:
# 随机取一个语言，城市，时区组合
language_region = random_language_region()
# 生成名字
name, english_name, gender = _generate_name(language_region.lang_code)
# 生成邮箱
email = _generate_email(english_name)
# 生成密码
password = Faker().password(length=random.randint(8, 16))
# 生成生日
birthday = Faker().date_of_birth(minimum_age=18, maximum_age=60)
year = birthday.year
month = birthday.month
day = birthday.day
# 生成性别
# 写入SignupXModel数据
signup_x_model = SignupXModel(
name=english_name,
email=email,
password=password,
year=year,
month=month,
day=day,
lang=language_region.lang,
accept_lang=language_region.accept_lang,
timezone=language_region.timezone,
city=language_region.city,
status=1,
real_name=name,
gender=gender,
)
# 写入数据库
add_signup_x(signup_x_model)
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="生成用户信息并写入数据库")
parser.add_argument(
"--lang",
type=str,
default=None,
help="指定语言代码 (例如: zh_CN, ja_JP, en_US)，不指定则随机选择"
)
parser.add_argument(
"--num",
type=int,
default=1,
help="生成的用户信息数量 (默认: 1)"
)
args = parser.parse_args()
# 根据指定数量生成用户信息
for i in range(args.num):
try:
generate_info(lang=args.lang)
print(f"成功生成第 {i+1}/{args.num} 条用户信息")
except Exception as e:
print(f"生成第 {i+1} 条用户信息时出错: {e}")
break
三个数据库表：
ini
# 语言地区映射表
class LanguageRegionModel(Base):
__tablename__ = "language_region"
id = Column(Integer, primary_key=True)
city = Column(String(255), nullable=False,comment="英文城市,首字母大写,空格为-")
language = Column(String(255), nullable=False,comment="语言，英文全称")
lang_code = Column(String(255), nullable=False,comment="语言代码i18n，zh_CN")
lang = Column(String(255), nullable=False,comment="语言代码，浏览器简写zh-CN")
accept_lang = Column(String(255), nullable=False,comment="Accept-Language，zh-CN,en-US")
timezone = Column(String(255), nullable=False,comment="时区")
__table_args__ = {"comment":"语言地区映射表"}
# 预注册x用户表
class SignupXModel(Base):
__tablename__ = "signup_x"
id = Column(Integer, primary_key=True)
name = Column(String(255), nullable=False,comment="昵称")
email = Column(String(255), nullable=False, index=True, unique=True,comment="邮箱")
password = Column(String(255), nullable=False,comment="密码")
# 生日
year = Column(Integer, nullable=False,comment="年")
month = Column(Integer, nullable=False,comment="月")
day = Column(Integer, nullable=False,comment="日")
# 语言
lang = Column(String(255), nullable=False,comment="语言,浏览器简写，zh-CN")
accept_lang = Column(String(255), nullable=False,comment="Accept-Language，zh-CN,en-US")
# 时区
timezone = Column(String(255), nullable=False,comment="时区")
# 城市
city = Column(String(255), nullable=False,comment="城市")
# status
status = Column(SmallInteger, nullable=False,comment="1:待注册 2运行中 3:已注册 4:已失败删除")
# 真名
real_name = Column(String(255), nullable=False,comment="真名")
# 性别
gender = Column(String(10), nullable=False,comment="男 女")
__table_args__ = {"comment":"预注册x用户表"}
# 已注册x用户表
class RegisteredXModel(Base):
__tablename__ = "registered_x"
id = Column(Integer, primary_key=True)
username = Column(String(255), nullable=False, index=True, unique=True,comment="用户名")
email = Column(String(255), nullable=False, index=True, unique=True,comment="邮箱")
password = Column(String(255), nullable=False,comment="密码")
# 生日
year = Column(Integer, nullable=False,comment="年")
month = Column(Integer, nullable=False,comment="月")
day = Column(Integer, nullable=False,comment="日")
# 语言
lang = Column(String(255), nullable=False,comment="语言,浏览器简写，zh-CN")
accept_lang = Column(String(255), nullable=False,comment="Accept-Language，zh-CN,en-US")
# 时区
timezone = Column(String(255), nullable=False,comment="时区")
# 城市
city = Column(String(255), nullable=False,comment="城市")
# 浏览器id
browser_id = Column(Integer, nullable=False,comment="浏览器id,代表用户目录和随机种子")
# 真名
real_name = Column(String(255), nullable=False,comment="真名")
# 性别
gender = Column(String(10), nullable=False,comment="男 女")
# 注册时间
created_at = Column(DateTime(timezone=True), default=datetime.now,comment="注册时间")
__table_args__ = {"comment":"已注册x用户表"}