xiaobo-tool 1.0.6

python version of xiaobo tool module

Xiaobo Tool

Python 通用工具库，提供任务执行器、代理池、X(Twitter) 客户端、临时邮箱等模块。

安装

uv add xiaobo-tool

模块概览

task_executor - 任务执行器

支持同步（线程池）和异步（asyncio）两种模式，内置重试、代理轮换、回调通知和统计。

from xiaobo_tool.task_executor import TaskExecutor, Target


def my_task(target: Target):
    print(f"执行任务: {target.data}, 代理: {target.proxy}")


with TaskExecutor(name="Demo", max_workers=3, retries=2) as executor:
    executor.submit_tasks(
        task_func=my_task,
        source=["item1", "item2", "item3"],
        on_success=lambda t, r: print(f"{t.data} 成功"),
        on_error=lambda t, e: print(f"{t.data} 失败: {e}"),
    )
    executor.wait()
    executor.statistics()

异步模式：

import asyncio
from xiaobo_tool.task_executor import AsyncTaskExecutor, Target


async def my_async_task(target: Target):
    print(f"执行任务: {target.data}")


async def main():
    async with AsyncTaskExecutor(name="AsyncDemo", max_workers=5) as executor:
        executor.submit_tasks(task_func=my_async_task, source=10)
        await executor.wait()
        await executor.statistics()


asyncio.run(main())

从文件批量提交任务：

# 读取 accounts.txt，每行按 "----" 分割
executor.submit_tasks_from_file(task_func=my_task, filename="accounts", separator="----")

配置通过 Settings 管理，支持构造参数、环境变量、.env 文件三种方式：

配置项	默认值	说明
MAX_WORKERS	5	最大线程数
PROXY	(空)	代理，支持 host:port / user:pass@host:port，占位符 ***** 自动替换为 index 或第一位数据
PROXY_IPV6	(空)	IPv6 代理，格式同 PROXY
PROXY_API	(空)	代理提取 API 地址（一行一个），格式同 PROXY
PROXY_IPV6_API	(空)	IPv6 代理提取 API 地址
RETRIES	2	重试次数（抛出 TaskFailed 不重试）
RETRY_DELAY	0	重试延迟（秒）
SHUFFLE	false	是否打乱任务顺序，按照数量运行的任务，支持布尔值或任务名称，多个任务用&拼接，如： task1&task2
USE_PROXY_IPV6	false	是否优先使用 IPv6 代理，支持布尔值或任务名称，多个任务用&拼接，如： task1&task2
DISABLE_PROXY	false	是否禁用代理，支持布尔值或任务名称，多个任务用&拼接，如：task1&task2

抛出 TaskFailed 异常可跳过重试，直接标记任务失败：

from xiaobo_tool.task_executor import TaskFailed


def my_task(target: Target):
    if invalid(target.data):
        raise TaskFailed("数据无效，无需重试")

proxy_pool - 代理池

管理代理获取与轮换，支持直连代理和 API 代理（带 3 分钟缓存）。

from xiaobo_tool.proxy_pool import ProxyPool

pool = ProxyPool(proxy_api="http://your-proxy-api.com/get")
proxy = pool.get_proxy()

x - X(Twitter) 客户端

基于 auth_token 操作 X API，支持发推和 OAuth2 授权。

from xiaobo_tool.x import XClient

client = XClient(auth_token="your_auth_token", proxy="http://127.0.0.1:7890")
tweet_url = client.send_tweet("Hello World!")

OAuth2 授权：

redirect_uri = client.authorize_oauth2("https://x.com/i/oauth2/authorize?client_id=xxx&...")

temp_email - 临时邮箱

创建临时邮箱并接收邮件，支持同步和异步两种模式。

from xiaobo_tool.temp_email import TempEmail

with TempEmail() as mail:
    domains = mail.query_domains()
    mailbox = mail.create_mailbox()
    result = mail.get_new_mail(mailbox)

异步模式：

from xiaobo_tool.temp_email import AsyncTempEmail

async with AsyncTempEmail() as mail:
    mailbox = await mail.create_mailbox(domain="example.com", mail_type=1)
    result = await mail.get_new_mail(mailbox, title="验证码")

utils - 工具函数

• read_txt_file_lines(filename) - 按行读取 txt 文件
• write_txt_file(filename, data) - 写入/追加 txt 文件
• get_session(proxy) / get_async_session(proxy) - 创建 HTTP 会话（curl_cffi）
• json_get(data, path, default) - 通过路径访问嵌套 JSON 数据
• raise_response_error(name, response) - 解析 HTTP 错误响应并抛出异常

开发

# 安装开发依赖
uv sync --dev

# 运行测试
uv run pytest

依赖

• curl-cffi - HTTP 客户端（浏览器指纹模拟）
• loguru - 日志
• pydantic-settings - 配置管理
• tenacity - 重试机制