Metadata-Version: 2.4
Name: xiaobo-task
Version: 1.0.5
Summary: 自用多线程任务模块
Author-email: xiaobo <tb791549729@gmail.com>
Requires-Python: >=3.10
Requires-Dist: curl-cffi>=0.13.0
Requires-Dist: loguru>=0.7.3
Requires-Dist: pydantic-settings>=2.12.0
Requires-Dist: python-dotenv>=1.2.1
Requires-Dist: tenacity>=9.1.2
Description-Content-Type: text/markdown

# Xiaobo Task [![X (formerly Twitter) Follow](https://img.shields.io/twitter/follow/0xiaobo888)](https://x.com/0xiaobo888)

![Python Version from PEP 621 TOML](https://img.shields.io/python/required-version-toml?logo=python&tomlFilePath=https%3A%2F%2Fraw.githubusercontent.com%2FXiaobooooo%2Fxiaobo-task%2Frefs%2Fheads%2Fmain%2Fpyproject.toml)
![Project Version from TOML](https://img.shields.io/badge/dynamic/toml?logo=semanticweb&color=orange&label=version&query=project.version&url=https%3A%2F%2Fraw.githubusercontent.com%2FXiaobooooo%2Fxiaobo-task%2Frefs%2Fheads%2Fmain%2Fpyproject.toml)

## 简介

`xiaobo-task` 是一个通用的多线程 / 异步任务管理模块，面向需要批量执行任务、重试控制、回调处理与统计的场景。支持同步与异步两套接口，提供
`retry`、`callback`、代理池、任务统计等能力。

## 特性

- 同步 `XiaoboTask` 与异步 `AsyncXiaoboTask` 双接口
- 支持批量提交、失败重试、回调处理（on_success/on_error/on_cancel）
- 内置代理池（支持 IPv4 / IPv6 / API 拉取）
- 任务执行统计与错误汇总
- 配置自动从 `.env` / 环境变量加载（可在构造时覆盖）

## 安装

```bash
pip install xiaobo-task
```

> Python >= 3.10

## 快速开始（同步）

```python
import random
import time
from typing import Any

from loguru import logger
from xiaobo_task import XiaoboTask, Target, TaskFailed

APPNAME = "XiaoboTaskExample"


def example_task_processor(target: Target):
    target.logger.info(f"开始处理任务，数据: {target.data}")

    sleep_time = random.uniform(1, 3)
    time.sleep(sleep_time)

    # 抛出 TaskFailed 将不再重试
    if target.data == "data-6":
        raise TaskFailed("任务失败，不进行重试")

    return f"{target.data} 处理完毕，耗时 {sleep_time:.2f} 秒"


def on_task_success(target: Target, result: Any):
    target.logger.success(f"成功回调 -> {result}")


def on_task_error(target: Target, error: Exception):
    target.logger.error(f"失败回调 -> {error}")


def main():
    task_data_list = [f"data-{i}" for i in range(10)]
    with XiaoboTask(APPNAME, shuffle=False) as task_manager:
        task_manager.submit_tasks(
            task_func=example_task_processor,
            source=task_data_list,
            on_success=on_task_success,
            on_error=on_task_error,
            retries=1,
        )
        task_manager.wait()
        task_manager.statistics()


if __name__ == "__main__":
    main()
```

## 异步示例

```python
import asyncio
import random
from typing import Any

from loguru import logger
from xiaobo_task import Target, AsyncXiaoboTask, TaskFailed

APPNAME = "XiaoboTaskAsyncExample"


async def example_async_task_processor(target: Target):
    target.logger.info(f"开始处理任务，数据: {target.data}")
    await asyncio.sleep(random.uniform(1, 3))

    if target.data == "data-6":
        raise TaskFailed("任务失败，不进行重试")

    return f"{target.data} 处理完毕"


def on_task_success(target: Target, result: Any):
    target.logger.info(f"成功回调 -> {result}")


async def main():
    async with AsyncXiaoboTask(APPNAME, shuffle=False, retries=1) as task_manager:
        task_manager.submit_tasks(
            task_func=example_async_task_processor,
            source=[f"data-{i}" for i in range(10)],
            on_success=on_task_success,
        )
        await task_manager.wait()
        await task_manager.statistics()


if __name__ == "__main__":
    asyncio.run(main())
```

## 运行示例

```bash
python examples/example.py
python examples/example_async.py
```

异步示例会读取 `examples/example.txt` 作为任务源。

## 从文件批量读取任务

支持通过 `submit_tasks_from_file` 直接读取文本文件批量提交：

```python
task_manager.submit_tasks_from_file(
    task_func=example_task_processor,
    filename="example",  # 自动补全 .txt
)
```

## `.env` 配置项

配置默认从 `.env` / 环境变量加载，大小写不敏感，空字符串会自动回退到默认值。

可用配置项如下：

| 配置项              | 默认值     | 说明                                                                |
|------------------|---------|-------------------------------------------------------------------|
| `MAX_WORKERS`    | `5`     | 最大线程数                                                             |
| `SHUFFLE`        | `false` | 是否打乱任务顺序                                                          |
| `RETRIES`        | `2`     | 重试次数（抛出 `TaskFailed` 不重试）                                         |
| `RETRY_DELAY`    | `0`     | 重试延迟（秒）                                                           |
| `PROXY`          | *(空)*   | IPv4 代理，支持 `host:port` / `user:pass@host:port` / `protocol://...` |
| `PROXY_IPV6`     | *(空)*   | IPv6 代理                                                           |
| `PROXY_API`      | *(空)*   | IPv4 代理 API 地址                                                    |
| `PROXY_IPV6_API` | *(空)*   | IPv6 代理 API 地址                                                    |
| `USE_PROXY_IPV6` | `false` | 是否优先使用 IPv6 代理                                                    |
| `DISABLE_PROXY`  | `false` | 是否禁用代理                                                            |

示例 `.env`：

```dotenv
MAX_WORKERS=5
SHUFFLE=false
RETRIES=2
RETRY_DELAY=0
PROXY=
PROXY_IPV6=
PROXY_API=
PROXY_IPV6_API=
USE_PROXY_IPV6=false
DISABLE_PROXY=false
```

> 也可以在初始化时通过关键字参数覆盖任意配置，例如 `XiaoboTask(max_workers=10, retries=3)`。

## data 目录说明

读取 `.txt` 文件时会自动补全后缀，并按以下顺序查找：

1) 脚本所在目录
    ```
    project_root/
      data/
        example.txt
    ```
2) 项目根目录同级的 `data/` 目录
    ```
    project_root/
      example.txt
      main.py
    ```

## 依赖

核心依赖包括：`curl-cffi`、`loguru`、`pydantic-settings`、`python-dotenv`、`tenacity`。

## 许可证

暂无开源协议（No License）。
