Metadata-Version: 2.5
Name: hagworm
Version: 5.11.4
Summary: Network Development Suite
Author-email: "Shaobo.Wang" <wsb310@gmail.com>
Requires-Python: >=3.11
Description-Content-Type: text/markdown
License-Expression: Apache-2.0
Classifier: Programming Language :: Python :: 3.11
Classifier: Operating System :: POSIX :: Linux
License-File: LICENSE
Requires-Dist: APScheduler==3.11.3
Requires-Dist: Cython==3.2.8
Requires-Dist: Pillow==12.3.0
Requires-Dist: PyJWT==2.13.0
Requires-Dist: PyYAML==6.0.3
Requires-Dist: SQLAlchemy==2.0.51
Requires-Dist: SnakeMD==2.4.1
Requires-Dist: ahocorasick-rs==1.0.3
Requires-Dist: aio-pika==10.0.1
Requires-Dist: aiosmtplib==5.1.2
Requires-Dist: asyncmy==0.2.11
Requires-Dist: asyncstdlib==3.14.0
Requires-Dist: cachetools==7.1.4
Requires-Dist: confluent-kafka==2.15.0
Requires-Dist: coredis==5.7.0
Requires-Dist: cryptography==49.0.0
Requires-Dist: elasticsearch==8.12.1
Requires-Dist: fastapi==0.139.2
Requires-Dist: filelock==3.31.2
Requires-Dist: grpcio==1.82.1
Requires-Dist: gunicorn==26.0.0
Requires-Dist: httpx==0.28.1
Requires-Dist: igraph==1.0.0
Requires-Dist: json5==0.15.0
Requires-Dist: loguru==0.7.3
Requires-Dist: motor==3.7.1
Requires-Dist: msgpack==1.2.1
Requires-Dist: ntplib==0.4.0
Requires-Dist: psutil==7.2.2
Requires-Dist: pymongo==4.17.0
Requires-Dist: pytest-asyncio==1.4.0
Requires-Dist: python-dateutil==2.9.0
Requires-Dist: python-multipart==0.0.32
Requires-Dist: python-stdnum==2.2
Requires-Dist: qrcode==8.2
Requires-Dist: tzdata==2026.3
Requires-Dist: ujson==5.13.0
Requires-Dist: uvicorn[standard]==0.51.0
Requires-Dist: uvicorn-worker==0.4.0
Requires-Dist: uvloop==0.22.1;sys_platform!='win32'
Requires-Dist: wrapt==2.2.2
Requires-Dist: xmltodict==1.0.4
Import-Name: hagworm

# Hagworm

> Network Development Suite - 异步网络开发套件

## 项目理念

Hagworm 是一个面向异步编程的 Python 工具库，旨在提供高效、可靠的网络开发基础设施。项目以 `asyncio` 为核心，封装了从基础工具函数到分布式组件的全栈开发能力，帮助开发者快速构建高性能的异步应用。

核心理念：

- **异步优先**：所有 I/O 操作均基于 `asyncio`，支持 `uvloop` 加速
- **开箱即用**：提供 FastAPI/gRPC/RabbitMQ/Nacos 等框架的集成方案
- **生产就绪**：包含日志、配置、事务、缓存、限流等生产级组件
- **优雅抽象**：通过上下文管理器、装饰器等模式简化资源管理

## 安装

```bash
pip install hagworm
```

要求 Python >= 3.11。

---

## 工程目录结构

```
hagworm/
├── hagworm/                          # 核心代码
│   ├── __init__.py                   # 版本信息与品牌标识
│   ├── extend/                       # 基础扩展模块
│   │   ├── asyncio/                  # 异步扩展模块
│   │   │   ├── base.py               # 异步基础工具类
│   │   │   ├── buffer.py             # 队列缓冲与文件缓冲
│   │   │   ├── command.py            # 多进程命令框架
│   │   │   ├── file.py               # 带缓存的文件加载器
│   │   │   ├── future.py             # Future/线程/子进程工具
│   │   │   ├── mail.py               # SMTP 邮件客户端
│   │   │   ├── mongo.py              # MongoDB 连接管理
│   │   │   ├── mysql.py              # MySQL 连接管理
│   │   │   ├── net.py                # HTTP 客户端
│   │   │   ├── pool.py               # 通用对象池
│   │   │   ├── process.py            # 子进程托管
│   │   │   ├── redis.py              # Redis 连接管理
│   │   │   ├── socket.py             # TCP/Unix Socket 服务端与客户端
│   │   │   ├── task.py               # 定时任务/并发任务/限流器
│   │   │   └── transaction.py        # 异步事务
│   │   ├── base.py                   # 同步基础工具类
│   │   ├── cache.py                  # 堆栈缓存
│   │   ├── compile.py                # PYC 编译工具
│   │   ├── config.py                 # 配置管理
│   │   ├── crypto.py                 # RSA 签名/验签
│   │   ├── error.py                  # 异常处理
│   │   ├── event.py                  # 同步事件总线
│   │   ├── igraph.py                 # 图数据结构
│   │   ├── interface.py              # 接口定义
│   │   ├── logging.py                # 日志系统
│   │   ├── metaclass.py              # 单例元类
│   │   ├── ntp.py                    # NTP 时间同步
│   │   ├── process.py                # 多进程管理
│   │   ├── qrcode.py                 # 二维码生成
│   │   ├── struct.py                 # 数据结构
│   │   ├── text.py                   # 文本处理
│   │   ├── trace.py                  # 链路追踪
│   │   ├── transaction.py            # 同步事务
│   │   └── validator.py              # 验证器
│   ├── frame/                        # 应用框架
│   │   ├── fastapi/                  # FastAPI 集成
│   │   │   ├── base.py               # FastAPI 应用创建与中间件
│   │   │   ├── field.py              # 自定义字段类型
│   │   │   ├── model.py              # 数据模型
│   │   │   └── response.py           # 响应封装
│   │   ├── gunicorn.py               # Gunicorn Worker
│   │   └── stress_tests.py           # 压力测试框架
│   ├── static/                       # 静态资源（Swagger UI）
│   └── third/                        # 第三方集成
│       ├── grpc/                     # gRPC 框架
│       │   ├── bridge.py             # gRPC 网桥
│       │   ├── client.py             # gRPC 客户端
│       │   └── server.py             # gRPC 服务端
│       ├── nacos/                    # Nacos 服务发现
│       │   ├── client.py             # Nacos 客户端
│       │   └── config.py             # Nacos 配置
│       └── rabbitmq/                 # RabbitMQ 消息队列
│           ├── consume.py            # 消费者
│           ├── publish.py            # 生产者
│           └── rpc.py                # RPC 实现
├── example/                          # 使用示例
│   ├── fastapi_demo/                 # FastAPI 示例项目（main.py / setting.py / controller / service / model / config / build.sh / main.sh / gunicorn.config.py）
│   ├── grpc_demo/                    # gRPC 示例（server.py / client.py）
│   ├── main_rabbitmq.py              # RabbitMQ 示例
│   ├── main_tcp.py                   # TCP 服务示例
│   └── main_test.py                  # 压力测试示例
├── testing/                          # 单元测试（含 main.py 入口与 setup.cfg pytest 配置）
├── .gitignore                        # Git 忽略规则
├── LICENSE                           # 许可证
├── README.md                         # 项目说明
├── install.sh                        # 安装脚本
├── upload.sh                         # 发布脚本
└── pyproject.toml                    # 项目配置
```

---

## 模块详解

---

### hagworm.extend.base — 同步基础工具类

#### `Utils`

聚合工具类，继承自 `UtilsSystem`、`UtilsCodec`、`UtilsTime`、`UtilsConvert`、`UtilsCrypto`、`UtilsMisc`，可直接通过 `Utils.xxx` 访问全部方法。此外提供以下常用类属性与静态方法：

| 名称 | 类型 | 说明 |
|------|------|------|
| `log` | `loguru.Logger` | 全局日志对象 |
| `path` | 模块 | `os.path` 路径工具 |
| `math` / `random` / `textwrap` / `itertools` | 模块 | 常用标准库模块引用 |
| `func_wraps(func)` | 静态方法 | 等价于 `functools.wraps` |
| `func_partial(func, *args, **kwargs)` | 静态方法 | 等价于 `functools.partial` |
| `randint(a, b)` | 静态方法 | 随机整数 |
| `randstr(population, k)` | 静态方法 | 随机采样（`random.sample`） |
| `getenv(key)` / `getpid()` / `getppid()` | 静态方法 | 环境变量 / 进程 ID |
| `date_parse(str)` | 静态方法 | 解析日期字符串 |
| `urlparse(url)` / `deepcopy(obj)` | 静态方法 | URL 解析 / 深拷贝 |
| `suppress(*exceptions)` | 静态方法 | 上下文管理器，抑制指定异常 |

#### `UtilsSystem`

系统与环境相关工具。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `get_class_name(val)` | `val: Type` | `str` | 获取类的全限定名 |
| `environment()` | 无 | `Dict` | 获取 Python 和系统环境信息 |
| `print_slogan()` | 无 | `None` | 打印 Hagworm 品牌 Logo |
| `deprecation_warning(val)` | `val: str` | `None` | 发出弃用警告 |
| `get_node_id()` | 无 | `str` | 获取基于 PID 和 MAC 的节点 ID（MD5） |
| `get_cpu_percent()` | 无 | `float` | 获取 CPU 使用率（1秒缓存） |
| `kill_process(*names)` | `*names: str` | `None` | 按名称杀死进程 |

#### `UtilsCodec`

编码与序列化相关工具。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `utf8(val)` | `val: AnyStr` | `bytes` | 将字符串或 bytes 转为 UTF-8 bytes |
| `basestring(val)` | `val: AnyStr` | `str` | 将 bytes 或 str 转为 str |
| `vague_string(val, vague_str, vague_len, min_vague, max_vague)` | `val: str` 等 | `str` | 字符串脱敏，如 `"13812345678"` → `"138******78"` |
| `json_encode(val, **kwargs)` | `val: Any` | `str` | JSON 编码（使用 ujson） |
| `json_decode(val, **kwargs)` | `val: AnyStr` | `Any` | JSON 解码（ujson 优先，回退 json5） |
| `msgpack_encode(val, **kwargs)` | `val: Any` | `bytes` | MessagePack 编码 |
| `msgpack_decode(val, **kwargs)` | `val: AnyStr` | `Any` | MessagePack 解码 |
| `xml_encode(dict_val, root_tag)` | `dict_val: Dict` 等 | `xml.dom.minidom.Document` | 字典转 XML |
| `xml_decode(val)` | `val: str` | `Dict` | XML 转字典 |
| `b32_encode(val, standard)` | `val: AnyStr, standard: bool=False` | `str` | Base32 编码 |
| `b32_decode(val, standard, for_bytes)` | `val: AnyStr` 等 | `AnyStr` | Base32 解码 |
| `b64_encode(val, standard, for_bytes)` | `val: AnyStr` 等 | `AnyStr` | Base64 编码（默认 urlsafe） |
| `b64_decode(val, standard, for_bytes)` | `val: AnyStr` 等 | `AnyStr` | Base64 解码（默认 urlsafe） |
| `yaml_encode(data, stream)` | `data: Any` 等 | `str` | YAML 编码 |
| `yaml_decode(stream)` | `stream: Any` | `Any` | YAML 解码 |
| `pickle_dumps(val)` | `val: Any` | `bytes` | Pickle 序列化（带 zlib 压缩） |
| `pickle_loads(val)` | `val: bytes` | `Any` | Pickle 反序列化（带 zlib 解压） |

> 注：`utf8` / `basestring` 对 `None` 入参原样返回 `None`（不会转为 `bytes` / `str`）。

#### `UtilsTime`

时间与日期相关工具。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `today()` | 无 | `datetime` | 获取今天零点时间 |
| `yesterday()` | 无 | `datetime` | 获取昨天零点时间 |
| `localnow()` | 无 | `datetime` | 获取当前本地时间（分钟级截断，秒与微秒清零） |
| `localnow_on_hour()` | 无 | `datetime` | 获取当前本地时间（小时级截断） |
| `utcnow()` | 无 | `datetime` | 获取当前 UTC 时间（分钟级截断，秒与微秒清零） |
| `utcnow_on_hour()` | 无 | `datetime` | 获取当前 UTC 时间（小时级截断） |
| `timestamp()` | 无 | `int` | 获取当前秒级时间戳 |
| `timestamp_ms()` | 无 | `int` | 获取当前毫秒级时间戳 |
| `time2stamp(time_str, format_type, timezone_)` | `time_str: str, format_type: str='%Y-%m-%d %H:%M:%S', timezone_: Optional[str]=None` | `int` | 时间字符串转时间戳 |
| `stamp2time(time_int, format_type, timezone_)` | `time_int: Optional[int]=None, format_type: str='%Y-%m-%d %H:%M:%S', timezone_: Optional[str]=None` | `str` | 时间戳转时间字符串 |
| `get_today_region(today)` | `today: datetime=None` | `Tuple[int, int]` | 获取当天的时间戳范围 |
| `get_month_region(today)` | `today: datetime=None` | `Tuple[int, int]` | 获取当月的时间戳范围 |
| `get_week_region(today)` | `today: datetime=None` | `Tuple[int, int]` | 获取本周的时间戳范围 |

#### `UtilsConvert`

类型转换与字符串相关工具。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `convert_bool(val)` | `val: Any` | `bool` | 将值转为布尔值，`null/none/nil/false/'0'/''/False/0` 为 False |
| `convert_int(val, default)` | `val: Any, default: int=0` | `int` | 将 int/float/str 转为 int |
| `convert_float(val, default)` | `val: Any, default: float=0` | `float` | 将值转为 float |
| `interval_limit(val, min_val, max_val)` | `val, min_val, max_val` | `int/float` | 将值限制在区间内 |
| `split_float(val, sep, min_split, max_split)` | `val: str` 等 | `List[float]` | 按分隔符拆分字符串为浮点数列表 |
| `join_float(val, sep)` | `val: Iterable, sep: str` | `str` | 将浮点数列表合并为字符串 |
| `split_int(val, sep, min_split, max_split)` | `val: str` 等 | `List[int]` | 按分隔符拆分字符串为整数列表 |
| `join_int(val, sep)` | `val: Iterable, sep: str` | `str` | 将整数列表合并为字符串 |
| `split_str(val, sep, min_split, max_split)` | `val: str` 等 | `List[str]` | 按分隔符拆分字符串 |
| `join_str(val, sep)` | `val, sep` | `str` | 将值列表合并为字符串 |
| `list_extend(iterable, val)` | `iterable: List, val: Any` | `None` | 智能扩展列表，可迭代则 extend，否则 append |
| `str_len(str_val)` | `str_val: str` | `int` | 计算字符串显示宽度（中文占2） |
| `sub_str(val, length, suffix)` | `val: str, length: int` 等 | `str` | 按显示宽度截断字符串 |
| `re_match(pattern, value)` | `pattern: str, value: str` | `bool` | 正则匹配 |
| `ordered_dict(val)` | `val: Dict=None` | `OrderedDict` | 创建按键排序的有序字典 |
| `is_iterable(obj, standard)` | `obj: Any, standard: bool=False` | `bool` | 判断是否可迭代 |
| `params_join(params, excludes)` | `params: Dict, excludes: List` | `str` | 参数排序拼接 |
| `params_sign(*args, **kwargs)` | `*args, **kwargs` | `str` | 参数签名（MD5） |

#### `UtilsCrypto`

进制、哈希与加密相关工具。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `radix24(val, align)` | `val: int, align: int=0` | `str` | 10 进制转 24 进制 |
| `radix24_to_10(val)` | `val: str` | `int` | 24 进制转 10 进制 |
| `radix36(val, align)` | `val: int, align: int=0` | `str` | 10 进制转 36 进制 |
| `radix36_to_10(val)` | `val: str` | `int` | 36 进制转 10 进制 |
| `radix62(val, align)` | `val: int, align: int=0` | `str` | 10 进制转 62 进制 |
| `radix62_to_10(val)` | `val: str` | `int` | 62 进制转 10 进制 |
| `radix_n(val, base, radix, align)` | `val: int` 等 | `str` | 通用进制转换 |
| `radix_n_to_10(val, base, radix)` | `val: str` 等 | `int` | 通用进制转 10 进制 |
| `jwt_encode(val, key, algorithms)` | `val: dict, key: str, algorithms: str='HS256'` | `str` | JWT 编码 |
| `jwt_decode(val, key, algorithms)` | `val: str, key: str, algorithms: str='HS256'` | `Dict` | JWT 解码 |
| `uuid1(node, clock_seq)` | `node, clock_seq` | `str` | 生成 UUID1（hex 格式） |
| `uuid1_urn(node, clock_seq)` | `node, clock_seq` | `str` | 生成 UUID1（URN 格式） |
| `crc32(val)` | `val: AnyStr` | `int` | CRC32 校验 |
| `md5(val)` | `val: AnyStr` | `str` | MD5 哈希 |
| `md5_u32(val)` | `val: AnyStr` | `int` | MD5 取前 32 位整数 |
| `md5_u64(val)` | `val: AnyStr` | `int` | MD5 取前 64 位整数 |
| `sha1(val)` | `val: AnyStr` | `str` | SHA1 哈希 |
| `sha256(val)` | `val: AnyStr` | `str` | SHA256 哈希 |
| `sha512(val)` | `val: AnyStr` | `str` | SHA512 哈希 |
| `hmac_md5(key, data, b64_std)` | `key, data: AnyStr` 等 | `str` | HMAC-MD5 签名 |
| `hmac_sha1(key, data, b64_std)` | `key, data: AnyStr` 等 | `str` | HMAC-SHA1 签名 |

#### `UtilsMisc`

网络、随机、校验与文件相关工具。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `rand_hit(val, prob)` | `val: List, prob: List/Callable` | `Any` | 按概率随机命中 |
| `urandom_seed()` | 无 | `int` | 使用 os.urandom 生成随机种子并播种 random |
| `read_urandom(size)` | `size: int` | `bytes` | 使用 os.urandom 读取随机字节 |
| `get_host_ip()` | 无 | `str` | 获取本机默认路由的出口 IP（本地接口地址，通常为内网地址） |
| `ip2int(val)` | `val: str` | `int` | IP 地址转整数 |
| `int2ip(val)` | `val: int` | `str` | 整数转 IP 地址 |
| `luhn_valid(val)` | `val: str` | `bool` | Luhn 算法校验（银行卡） |
| `luhn_sign(val)` | `val: str` | `str` | Luhn 算法签名 |
| `identity_card(val)` | `val: str` | `bool` | 中国身份证号校验 |
| `zip_file(zip_file, *file_paths)` | `zip_file: str, *file_paths: str` | `None` | 压缩文件 |
| `unzip_file(zip_file, file_paths)` | `zip_file: str, file_paths: str` | `None` | 解压文件 |

> ⚠️ **易混淆点**：`b64_encode`/`b64_decode` 默认使用 urlsafe 模式；编码时会去除 `=` 填充，解码时会自动补全 `=` 填充。如需标准 Base64 请设置 `standard=True`。

```python
from hagworm.extend.base import Utils

# 字符串脱敏
Utils.vague_string("13812345678")  # "138******78"

# 进制转换
Utils.radix62(12345)  # "3d7"
Utils.radix62_to_10("3d7")  # 12345

# 时间范围
start, end = Utils.get_today_region()
start, end = Utils.get_month_region()

# Base64 编码（注意 urlsafe 与 standard 的区别）
Utils.b64_encode("hello")  # urlsafe, 无填充
Utils.b64_encode("hello", standard=True)  # 标准 Base64

# JSON 编解码
Utils.json_encode({"key": "value"})
Utils.json_decode('{"key": "value"}')

# 身份证校验
Utils.identity_card("110101199001011234")  # True/False

# 参数签名
Utils.params_sign("arg1", "arg2", key1="val1", key2="val2")
```

#### `FuncWrapper`

将多个函数包装成一个可调用对象，调用时依次执行所有注册的函数。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `__call__(*args, **kwargs)` | 任意参数 | `None` | 依次调用所有注册函数 |
| `is_valid` (属性) | 无 | `bool` | 是否有注册函数 |
| `add(func)` | `func: Callable` | `bool` | 添加函数，已存在返回 False |
| `remove(func)` | `func: Callable` | `bool` | 移除函数，不存在返回 False |

```python
from hagworm.extend.base import FuncWrapper

wrapper = FuncWrapper()

def handler1(data):
    print(f"handler1: {data}")

def handler2(data):
    print(f"handler2: {data}")

wrapper.add(handler1)
wrapper.add(handler2)

wrapper("test")  # 依次调用 handler1 和 handler2
```

#### `TimerMS`

代码块耗时统计（毫秒）。作为上下文管理器使用，退出 `with` 时统计耗时；传入限时 `limit`（毫秒）后，耗时超过限时时通过 `Utils.log` 以警告级别打印，默认 `0` 或不超限时不打印。构造时传入标签用于日志区分。同时可当数值使用（读取、比较、格式化）。

| 成员 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `__init__(label, limit)` | `label: str, limit: float=0` | `TimerMS` | 创建计时器，`label` 用于日志区分；`limit` 为限时(毫秒)，`0` 表示不处理 |
| `__enter__()` | 无 | `TimerMS` | 进入 `with` 块并重置计时起点（实例可复用） |
| `__exit__(...)` | 异常信息 | `None` | 退出 `with` 块统计耗时；超过 `limit` 时以警告级别打印 `[label] cost xxx.xxxms exceed limit Xms` |
| `cost` (属性) | 无 | `float` | 只读耗时(毫秒)；块内为 `0.0`，退出后有效 |
| `int(timer)` / `float(timer)` | 无 | `int` / `float` | 取耗时（`int` 为四舍五入） |
| `str(timer)` / `format(timer, spec)` | `spec: str` | `str` | 按耗时格式化（`str` 固定三位小数，如 `12.345`） |
| 比较运算 | `int`/`float`/`TimerMS` | `bool` | 如 `timer > 200`、`timer == other_timer` |

```python
from hagworm.extend.base import TimerMS

with TimerMS("load config") as timer:
    load_config()
# 默认 limit=0：不打印日志

# 传入限时(毫秒)：超过则以警告级别输出
with TimerMS("load config", limit=100) as timer:
    load_config()
# 超时输出: [load config] cost 152.001ms exceed limit 100ms

print(timer.cost)    # 12.345
print(int(timer))    # 12
assert timer > 10
```

> ⚠️ **注意**：耗时在退出 `with` 后才确定，块内 `cost` 恒为 `0.0`；因此对象**不可哈希**（不能作为 `set`/`dict` 的键）。

---

### hagworm.extend.struct — 数据结构

#### `Enum` / `IntEnum`

增强版的枚举类，提供字典转换和键值查询功能。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `items()` | 无 | `Iterable` | 返回 `(name, value)` 对 |
| `keys()` | 无 | `Iterable` | 返回所有 name |
| `values()` | 无 | `Iterable` | 返回所有 value |
| `to_dict()` | 无 | `Dict` | 返回 `{name: value}` 字典 |
| `to_keys_dict()` | 无 | `Dict` | 返回 `{name: enum_item}` 字典 |
| `to_values_dict()` | 无 | `Dict` | 返回 `{value: enum_item}` 字典 |
| `has_key(key)` | `key` | `bool` | 是否包含指定 name |
| `has_value(value)` | `value` | `bool` | 是否包含指定 value |

```python
from hagworm.extend.struct import Enum, IntEnum

class Color(Enum):
    RED = "red"
    GREEN = "green"
    BLUE = "blue"

Color.to_dict()       # {"RED": "red", "GREEN": "green", "BLUE": "blue"}
Color.has_key("RED")  # True
Color.has_value("red")  # True

class Status(IntEnum):
    ACTIVE = 1
    INACTIVE = 0

Status.items()  # dict_items([("ACTIVE", 1), ("INACTIVE", 0)])
```

#### `RoundRobin` / `RoundRobinSimple`

轮询调度器。`RoundRobin` 为命名轮询（成员以 `name` 关联，`get()` 返回 `(name, member)` 元组）；`RoundRobinSimple` 为无命名轮询，API 不同。

##### `RoundRobin`

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `RoundRobin(members)` | `members: Dict[str, T]=None` | `RoundRobin` | 创建命名轮询器 |
| `append(name, member)` | `name: str, member: T` | `Optional[T]` | 添加成员，返回被替换的旧成员 |
| `clear(names)` | `names: List[str]=None` | `Dict` | 清除成员，返回被清除的成员 |
| `reset(members)` | `members: Dict=None` | `Dict` | 重置成员列表，返回被替换的旧成员 |
| `get()` | 无 | `Tuple[str, T]` | 获取下一个 `(name, member)` |
| `__len__()` | 无 | `int` | 成员数量，支持 `len(rr)` |
| `__contains__(name)` | `name: str` | `bool` | 是否包含指定名称成员，支持 `name in rr` |

##### `RoundRobinSimple`

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `RoundRobinSimple(members)` | `members: List[T]=None` | `RoundRobinSimple` | 创建简单轮询器 |
| `append(member)` | `member: T` | `None` | 添加成员 |
| `clear()` | 无 | `List` | 清空成员，返回被清空的成员列表 |
| `get()` | 无 | `T` | 获取下一个成员 |
| `__len__()` | 无 | `int` | 成员数量，支持 `len(rrs)` |

```python
from hagworm.extend.struct import RoundRobin, RoundRobinSimple

# 命名轮询
rr = RoundRobin({"server1": "10.0.0.1", "server2": "10.0.0.2"})
name, member = rr.get()  # 轮询获取 (name, member)

# 简单轮询
rrs = RoundRobinSimple([1, 2, 3])
val = rrs.get()  # 轮询获取单个成员
```

#### `ThreadList` / `ThreadDict`

多线程安全的列表和字典，基于 `threading.local` 实现。

```python
from hagworm.extend.struct import ThreadList, ThreadDict

tls_list = ThreadList()
tls_list.data.append("value")

tls_dict = ThreadDict()
tls_dict.data["key"] = "value"
```

#### `ByteArrayAbstract` / `ByteArray`

`ByteArrayAbstract` 是结构化二进制读写的抽象基类，下列 `read_*` / `write_*` 方法均定义于其中（`SharedByteArray` 等也基于它）；`ByteArray` 继承 `BytesIO` 与 `ByteArrayAbstract`，是可实例化的内存缓冲实现，支持结构化二进制数据的读写。

支持的字段读写方法：`read_char`/`write_char`、`read_signed_char`/`write_signed_char`、`read_unsigned_char`/`write_unsigned_char`、`read_short`/`write_short`、`read_unsigned_short`/`write_unsigned_short`、`read_int`/`write_int`、`read_unsigned_int`/`write_unsigned_int`、`read_long`/`write_long`、`read_unsigned_long`/`write_unsigned_long`、`read_long_long`/`write_long_long`、`read_unsigned_long_long`/`write_unsigned_long_long`、`read_float`/`write_float`、`read_double`/`write_double`、`read_bool`/`write_bool`、`read_bytes`/`write_bytes`、`read_string`/`write_string`、`read_pascal_bytes`/`write_pascal_bytes`、`read_pascal_string`/`write_pascal_string`、`read_pad_byte`/`write_pad_byte`（补位）。

字节序通过 `get_endian()` / `set_endian(value)` 读取与设置，支持字节序：`NETWORK`(!)、`NATIVE`(=)、`NATIVE_ALIGNMENT`(@)、`LITTLE_ENDIAN`(<)、`BIG_ENDIAN`(>)。

```python
from hagworm.extend.struct import ByteArray

buf = ByteArray()
buf.set_endian(ByteArray.NETWORK)
buf.write_int(42)
buf.write_string("hello")
buf.write_float(3.14)

# 读取需要重新定位
buf.seek(0)
val_int = buf.read_int()    # 42
val_str = buf.read_string(5)  # "hello"
val_float = buf.read_float()  # 3.14
```

#### `KeyLowerDict`

将字典的 CamelCase 键自动转为 snake_case 小写键。

```python
from hagworm.extend.struct import KeyLowerDict

data = KeyLowerDict({"MySQLHost": "localhost", "RedisPort": 6379})
# {"my_sqlhost": "localhost", "redis_port": 6379}
```

> ⚠️ **注意**：连续大写字母之间不会插入下划线，例如 `MySQLHost` 会得到 `my_sqlhost`（而非 `my_sql_host`）。

> ⚠️ **注意**：`KeyLowerDict` 会递归处理嵌套字典，所有层级的键都会被转换。

---

### hagworm.extend.cache — 堆栈缓存

#### `StackCache`

基于 `TTLCache` 的内存高速缓存。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `__init__(maxsize, ttl)` | `maxsize: float=0xff, ttl: float=60` | `StackCache` | 初始化缓存 |
| `has(key)` | `key: str` | `bool` | 是否存在缓存 |
| `get(key, default)` | `key: str, default: Any=None` | `Any` | 获取缓存值 |
| `set(key, val)` | `key: str, val: Any` | `None` | 设置缓存 |
| `incr(key, val)` | `key: str, val: float=1` | `float` | 自增 |
| `decr(key, val)` | `key: str, val: float=1` | `float` | 自减 |
| `delete(key)` | `key: str` | `None` | 删除缓存（key 不存在时抛出 `KeyError`） |
| `size()` | 无 | `int` | 缓存大小 |
| `clear()` | 无 | `None` | 清空缓存 |

```python
from hagworm.extend.cache import StackCache

cache = StackCache(maxsize=100, ttl=60)

cache.set("user:1", {"name": "Alice"})
cache.get("user:1")  # {"name": "Alice"}
cache.has("user:1")  # True
cache.incr("counter")  # 1
cache.incr("counter")  # 2
cache.decr("counter")  # 1
```

---

### hagworm.extend.config — 配置管理

#### `Field`

配置字段定义，用于声明配置项所属 section 和默认值。

```python
from hagworm.extend.config import Field, HostType, StrListType

Port: int = Field("Base", default=8080)
Debug: bool = Field("Base", default=False)
Server: HostType = Field("Server")  # 自动解析 "host:port" 为 (host, port)
Tags: StrListType = Field("App")    # 自动按 "|" 分割为字符串列表
```

内置类型解码器（通过 `decode` 类方法将配置字符串解析为目标类型）：

| 类型 | 说明 |
|------|------|
| `HostType` | 解析 `"host:port"` 为 `(host, port)` |
| `JsonType` | 解析 JSON 字符串 |
| `StrListType` | 按 `\|` 解析为字符串列表 |
| `IntListType` | 按 `,` 解析为整数列表 |
| `FloatListType` | 按 `,` 解析为浮点数列表 |

配置基类 `ConfigureBase`：`Field` 字段通过元类 `ConfigureMetaclass`（内部以 `dataclasses.dataclass(init=False)` 包装）自动转为 dataclass 字段；`KeySection` 为描述单个配置键的 `NamedTuple`（含 `key` / `type` / `section` / `default` 四个字段）。

#### `Configure`

配置管理类，支持从 INI、JSON、YAML、环境变量等多种来源加载配置。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `read(path, encoding)` | `path: str, encoding: str='utf-8'` | `None` | 从 INI 文件读取 |
| `read_env()` | 无 | `None` | 从环境变量读取 |
| `read_str(val)` | `val: str` | `None` | 从字符串读取 |
| `read_dict(val)` | `val: Dict` | `None` | 从字典读取 |
| `read_json(path, encoding)` | `path: str, encoding: str='utf-8'` | `None` | 从 JSON 文件读取 |
| `read_yaml(path, encoding)` | `path: str, encoding: str='utf-8'` | `None` | 从 YAML 文件读取 |
| `get_option(section, option)` | `section: str, option: str` | `Any` | 获取指定配置项 |
| `get_options(section)` | `section: str` | `Dict` | 获取指定 section 的全部配置 |
| `set_options(section, **options)` | `section: str, **options: str` | `None` | 动态设置配置 |
| `to_dict(section)` | `section=None` | `Dict` | 导出为字典 |
| `to_yaml(section)` | `section=None` | `str` | 导出为 YAML |

> ⚠️ **注意**：环境变量优先级高于配置文件。环境变量格式为 `{SECTION}_{KEY}` 全大写。

```python
from hagworm.extend.config import Configure, Field, HostType, StrListType

class AppConfig(Configure):
    Port: int = Field("Base", default=8080)
    Debug: bool = Field("Base", default=False)
    Server: HostType = Field("Server")
    Tags: StrListType = Field("App")

config = AppConfig()
config.read_yaml("./config.yaml")
# config.Port, config.Debug, config.Server 等可直接访问
```

---

### hagworm.extend.error — 异常处理

#### `Ignore`

可忽略的异常，用于 `with` 语句块跳出或多层逻辑跳出。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `__init__(*args, layers)` | `layers: int=1` | `Ignore` | `layers` 控制跳出几层 with |
| `log()` | 无 | `None` | 打印 warning 日志 |
| `throw()` | 无 | `bool` | 返回是否继续抛出 |

> ⚠️ **易混淆点**：`layers=1`（默认）表示跳出当前 with 块后不再继续抛出；`layers=2` 表示跳出两层 with 块。

#### `catch_warning()` / `catch_error()`

上下文管理器，捕获异常并打印日志。

```python
from hagworm.extend.error import Ignore, catch_error, catch_warning

# 基本用法
with catch_error():
    result = 1 / 0  # 异常被捕获并打印 error 日志

# 使用 Ignore 跳出 with 块
with catch_error():
    if should_skip:
        raise Ignore("skip this block")  # 安全跳出，打印 warning

# 多层跳出
with catch_error():
    with catch_error():
        raise Ignore(layers=2)  # 跳出两层 with 块
```

---

### hagworm.extend.event — 同步事件总线

#### `EventDispatcher`

同步事件总线，基于发布-订阅模式。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `dispatch(_type, *args, **kwargs)` | `_type: str` 等 | `None` | 触发事件 |
| `add_listener(_type, _callable)` | `_type: str, _callable: Callable` | `bool` | 添加事件监听器 |
| `remove_listener(_type, _callable)` | `_type: str, _callable: Callable` | `bool` | 移除事件监听器 |

```python
from hagworm.extend.event import EventDispatcher

dispatcher = EventDispatcher()

def on_login(user_id):
    print(f"User {user_id} logged in")

dispatcher.add_listener("login", on_login)
dispatcher.dispatch("login", "user_123")
dispatcher.remove_listener("login", on_login)
```

---

### hagworm.extend.interface — 接口定义

#### `RunnableInterface`

可运行接口，需实现 `run()` 方法。

#### `TaskInterface`

任务接口，需实现 `start()`、`stop()`、`is_running()` 方法。

#### `ContextManager` / `AsyncContextManager`

上下文资源管理器基类，子类实现 `_context_release()` 即可使用 `with`/`async with` 语句管理资源；进入上下文时会调用可覆写的 `_context_initialize()`（默认空实现），退出时调用 `_context_release()`。支持 `Ignore` 异常的安全跳出。

```python
from hagworm.extend.interface import ContextManager

class MyResource(ContextManager):
    def _context_release(self):
        print("资源释放")

with MyResource() as res:
    print("使用资源")
# 自动调用 _context_release
```

---

### hagworm.extend.metaclass — 单例元类

#### `Singleton` / `SafeSingleton`

单例基类。`Singleton` 非线程安全，`SafeSingleton` 线程安全。

> ⚠️ **注意**：在多进程 fork 场景下，单例可能会因内存复制导致资源冲突，会发出警告。

`Singleton` / `SafeSingleton` 分别由元类 `SingletonMetaclass` / `SafeSingletonMetaclass` 实现；`SafeSingletonMetaclass` 通过加锁保证线程安全，并在跨进程序列化时通过 `check_process_id` 发出告警。

```python
from hagworm.extend.metaclass import Singleton, SafeSingleton

class Database(Singleton):
    def __init__(self):
        self.connection = "connected"

db1 = Database()
db2 = Database()
assert db1 is db2  # True，同一个实例
```

---

### hagworm.extend.text — 文本处理

#### `StrUtils`

字符串工具类。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `to_half_width(value)` | `value: str` | `str` | 全角转半角 |
| `to_full_width(value)` | `value: str` | `str` | 半角转全角 |
| `to_camel(val)` | `val: str` | `str` | snake_case → camelCase |
| `to_snake(val)` | `val: str` | `str` | CamelCase → snake_case |
| `to_upper_camel(val)` | `val: str` | `str` | snake_case → UpperCamelCase |
| `to_camel_dict(val)` | `val: Any` | `Any` | 递归转换字典键为 camelCase |
| `to_snake_dict(val)` | `val: Any` | `Any` | 递归转换字典键为 snake_case |
| `to_upper_camel_dict(val)` | `val: Any` | `Any` | 递归转换字典键为 UpperCamelCase |

```python
from hagworm.extend.text import StrUtils

StrUtils.to_camel("hello_world")       # "helloWorld"
StrUtils.to_snake("helloWorld")        # "hello_world"
StrUtils.to_upper_camel("hello_world") # "HelloWorld"
StrUtils.to_half_width("Ｈｅｌｌｏ")    # "Hello"

StrUtils.to_snake_dict({"firstName": "Alice", "lastName": "Bob"})
# {"first_name": "Alice", "last_name": "Bob"}
```

#### `TextFinder`

基于 Aho-Corasick 算法的关键词查找与替换。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `__init__(words)` | `words: List[str]` | `TextFinder` | 初始化关键词列表 |
| `find_all(content)` | `content: str` | `List[str]` | 查找所有匹配关键词 |
| `replace_all(content, _chars)` | `content: str, _chars: str='*'` | `str` | 替换所有匹配关键词 |

```python
from hagworm.extend.text import TextFinder

finder = TextFinder(["敏感词1", "敏感词2", "测试"])
finder.find_all("这是一个测试文本")       # ["测试"]
finder.replace_all("这是一个测试文本")     # "这是一个**文本"
```

---

### hagworm.extend.validator — 验证器

提供常用格式校验函数（IPv4/IPv6/CIDR 基于标准库 `ipaddress`，UUID 基于标准库 `uuid`，其余为正则实现）。

| 函数 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `asc_visible(val)` | `val: str` | `bool` | ASC 可见字符验证 |
| `uuid(val)` | `val: str` | `bool` | UUID 格式验证 |
| `email(val)` | `val: str` | `bool` | 邮箱格式验证 |
| `domain(val)` | `val: str` | `bool` | 域名格式验证 |
| `url(val)` | `val: str` | `bool` | URL 格式验证 |
| `mac_addr(val)` | `val: str` | `bool` | MAC 地址验证 |
| `ipv4(val)` | `val: str` | `bool` | IPv4 地址验证 |
| `ipv4_cidr(val)` | `val: str` | `bool` | IPv4 CIDR 验证 |
| `ipv6(val)` | `val: str` | `bool` | IPv6 地址验证 |
| `ipv6_cidr(val)` | `val: str` | `bool` | IPv6 CIDR 验证 |

```python
from hagworm.extend.validator import email, url, ipv4

email("test@example.com")  # True
url("https://example.com")  # True
ipv4("192.168.1.1")        # True
ipv4_cidr("192.168.1.0/24")  # True
```

---

### hagworm.extend.trace — 链路追踪

| 函数/装饰器 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `get_trace_id()` | 无 | `Optional[str]` | 获取当前 trace_id（未初始化时为 None） |
| `refresh_trace_id(trace_id)` | `trace_id: str=None` | `str` | 刷新 trace_id |
| `@trace_wrapper` | 函数 | 函数 | 为函数自动生成 trace_id |
| `@tracing` | 函数 | 函数 | 同步函数耗时追踪日志 |
| `@async_tracing` | 函数 | 函数 | 异步函数耗时追踪日志 |

`TRACE_ID_CONTEXT` 为保存当前 trace_id 的 `ContextVar`（默认 `None`）。

```python
from hagworm.extend.trace import trace_wrapper, async_tracing, refresh_trace_id

@trace_wrapper
async def handle_request():
    # 自动生成新的 trace_id
    pass

@async_tracing
async def slow_operation():
    # 自动记录执行耗时
    pass

refresh_trace_id("custom-trace-id-123")
```

---

### hagworm.extend.logging — 日志系统

#### `init_logger`

初始化日志系统。

`init_logger(level, *, handler=None, file_path=None, file_name=DEFAULT_LOG_FILE_NAME, file_rotation=DEFAULT_LOG_FILE_ROTATOR, file_retention=0xff, extra=None, enqueue=False, debug=False)`，除 `level` 外全部为关键字参数。

| 入参 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `level` | `str` | 必填 | 日志级别 |
| `handler` | `logging.Handler` | `None` | 自定义日志处理器 |
| `file_path` | `str` | `None` | 日志文件路径 |
| `file_name` | `str` | `runtime_{time}.log` | 日志文件名 |
| `file_rotation` | `Callable` | 默认轮转器 | 文件轮转策略 |
| `file_retention` | `int` | `0xff` | 保留文件数 |
| `extra` | `Dict` | `None` | 额外上下文 |
| `enqueue` | `bool` | `False` | 是否队列化 |
| `debug` | `bool` | `False` | 调试模式 |

行为说明：

- 写入文件时，文件名会被改写为 `{name}.pid-{pid}{ext}`（如 `runtime_2026-01-01_00-00-00.pid-1234.log`），便于多进程区分。
- 未传 `handler` 且未传 `file_path` 时，日志回退输出到 `sys.stdout`。

模块级常量：

| 名称 | 默认值 | 说明 |
|------|--------|------|
| `DEFAULT_LOG_FILE_NAME` | `runtime_{time}.log` | 默认日志文件名 |
| `DEFAULT_LOG_FILE_ROTATOR` | `LogFileRotator.make()` | 默认文件轮转策略 |
| `DEFAULT_LOG_MAXSIZE` | `0xaff` | 单条日志最大长度（超出截断） |
| `DEFAULT_BUFFER_MAXSIZE` | `0xfff` | 缓冲上限 |
| `DEFAULT_BULK_MAXSIZE` | `0x800` | 批量写入条数上限 |
| `DEFAULT_WRITE_MAX_DELAY` | `5` | 批量写入最大延迟（秒） |

#### `LogFileRotator`

日志文件轮转策略，支持按大小和时间轮转。

```python
from hagworm.extend.logging import init_logger, LogFileRotator

init_logger(
    "INFO",
    file_path="/var/log/app",
    file_rotation=LogFileRotator.make(500, "00:00"),  # 500MB 或每天0点轮转
    file_retention=30,
)
```

#### `KafkaSink` / `ElasticsearchSink`

日志投递到 Kafka 或 Elasticsearch。

- `KafkaSink(servers, topic, *, log_maxsize=DEFAULT_LOG_MAXSIZE, buffer_maxsize=DEFAULT_BUFFER_MAXSIZE, bulk_maxsize=DEFAULT_BULK_MAXSIZE, write_max_delay=DEFAULT_WRITE_MAX_DELAY, **kwargs)`：`servers` 为 Kafka 地址，`topic` 为投递主题。
- `ElasticsearchSink(hosts, stream_name, *, log_maxsize=..., buffer_maxsize=..., bulk_maxsize=..., write_max_delay=..., rollover_max_age='7d', rollover_max_primary_shard_size='50gb', delete_min_age='30d', refresh_interval='5s', number_of_replicas=0, **kwargs)`：`hosts` 为 ES 地址，`stream_name` 为 Data Stream 名称，其余为 ILM/索引模板参数。

```python
from hagworm.extend.logging import KafkaSink, ElasticsearchSink

# Kafka 日志投递
kafka_sink = KafkaSink("localhost:9092", "app-logs")

# Elasticsearch 日志投递
es_sink = ElasticsearchSink("http://localhost:9200", "app-logs")
```

#### `InterceptHandler`

将标准库 `logging` 的日志转发到 `loguru`。

#### `QueuedFileSink`

队列化的文件 sink，由后台线程按 `bulk_maxsize` / `write_max_delay` 批量写入日志文件（与 asyncio 无关）。

`QueuedFileSink(path, *, log_maxsize=DEFAULT_LOG_MAXSIZE, buffer_maxsize=DEFAULT_BUFFER_MAXSIZE, bulk_maxsize=DEFAULT_BULK_MAXSIZE, write_max_delay=DEFAULT_WRITE_MAX_DELAY, **kwargs)`

#### `ElasticsearchDataStreamUtil` / `AsyncElasticsearchDataStreamUtil`

Elasticsearch Data Stream 索引模板与 ILM 策略初始化工具。

---

### hagworm.extend.crypto — RSA 签名/验签

#### `RsaUtil`

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `gen_rsa_key(rsa_key, private)` | `rsa_key: str, private: bool=False` | `str` | 生成 PEM 格式密钥 |
| `rsa_sign(rsa_key, sign_data)` | `rsa_key: str, sign_data: str` | `str` | RSA 签名 |
| `rsa_verify(public_key, verify_data, verify_sign)` | 三个 `str` | `bool` | RSA 验签 |

> 说明：`rsa_sign` / `rsa_verify` 还支持可选参数 `hash_alg`（默认 `SHA256`），用于指定签名哈希算法。

```python
from hagworm.extend.crypto import RsaUtil

private_key = "MIIEvQ..."  # 裸密钥字符串
public_key = "MIIBIj..."

pem_private = RsaUtil.gen_rsa_key(private_key, private=True)
pem_public = RsaUtil.gen_rsa_key(public_key)

signature = RsaUtil.rsa_sign(private_key, "data to sign")
is_valid = RsaUtil.rsa_verify(public_key, "data to sign", signature)
```

---

### hagworm.extend.transaction — 同步事务

#### `TransactionAbstract`

事务抽象基类，提供 `add_commit_callback` / `add_rollback_callback` / `commit` / `rollback` / `bind` 方法。

#### `Transaction`

基于上下文管理器的事务对象，未显式 commit 则自动 rollback。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `add_commit_callback(_callable, *args, **kwargs)` | `_callable: Callable` 等 | `None` | 添加提交回调 |
| `add_rollback_callback(_callable, *args, **kwargs)` | `_callable: Callable` 等 | `None` | 添加回滚回调 |
| `commit()` | 无 | `None` | 提交事务，执行所有提交回调 |
| `rollback()` | 无 | `None` | 回滚事务，执行所有回滚回调 |
| `bind(trx)` | `trx: TransactionAbstract` | `None` | 绑定另一个事务（级联提交/回滚） |

`TransactionAbstract(*, commit_callback: Optional[Callable] = None, rollback_callback: Optional[Callable] = None)` 可在构造时注入初始提交/回滚回调。

```python
from hagworm.extend.transaction import Transaction

with Transaction() as trx:
    trx.add_commit_callback(lambda: print("committed"))
    trx.add_rollback_callback(lambda: print("rolled back"))
    # 如果不调用 trx.commit()，退出 with 时自动 rollback
    trx.commit()
```

---

### hagworm.extend.compile — PYC 编译

#### `compile_pyc`

将 Python 源码编译为 PYC 字节码。

| 入参 | 类型 | 说明 |
|------|------|------|
| `file_path` | `str` | 源码目录 |
| `pyc_file_path` | `str` | 输出目录（默认同源码目录） |
| `exclude` | `str` | 排除目录（用 `\|` 分隔） |

```bash
python -m hagworm.extend.compile -i ./src -o ./dist -e "test|tmp"
```

模块同时暴露 `main()` CLI 入口，可通过 `python -m hagworm.extend.compile ...` 直接调用。

---

### hagworm.extend.qrcode — 二维码

#### `QRCode`

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `make(data, **kwargs)` | `data: Any` | `bytes` | 生成 PNG 二维码 |
| `make_svg(data, **kwargs)` | `data: Any` | `str` | 生成 SVG 二维码 |

```python
from hagworm.extend.qrcode import QRCode

png_data = QRCode.make("https://example.com")
svg_str = QRCode.make_svg("https://example.com")
```

---

### hagworm.extend.igraph — 图数据结构

#### `Graph` / `DiGraph`

基于 `igraph`（包名 igraph）的图数据结构封装。`Graph` 构造为 `Graph(directed, name=None)`；`DiGraph(name=None)` 为有向图快捷类（内部 `directed=True`）。顶点标识类型别名为 `VERTEX_PARAMETER_TYPE = Union[Vertex, str, int]`；`len(graph)` 返回已登记顶点数，`repr(graph)` 形如 `<Graph|DiGraph: name, vertices: n, edges: m>`。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `Graph(directed, name)` | `directed: bool, name: str` | `Graph` | 创建图 |
| `add_vertex(name, **kwargs)` | `name: str=None` | `Vertex` | 添加顶点 |
| `find_vertex(name)` | `name: str` | `Vertex` | 查找顶点 |
| `find_vertices(names)` | `names: List[str]` | `List[Vertex]` | 批量查找顶点 |
| `add_edge(source, target)` | 顶点标识 | `None` | 添加边 |
| `init_edges(edges)` | `edges: List[Tuple]` | `None` | 批量初始化边 |
| `get_edge(source, target)` | 顶点标识 | `Edge` | 获取边 |
| `del_edge(source, target)` | 顶点标识 | `None` | 删除边 |
| `del_vertex_edge(name, mode)` | `name: str, mode: str='all'` | `None` | 删除顶点的关联边 |
| `clear()` | 无 | `None` | 清空图 |
| `deep_search(name, mode)` | `name: str, mode: str='all'` | `List[str]` | 深度搜索 |
| `vertices` (属性) | 无 | `Dict[str, Vertex]` | 顶点字典 |
| `name` (属性) | 无 | `str` | 图名称 |
| `data` (属性) | 无 | `igraph.Graph` | 底层 igraph 对象 |
| `vertex_seq` (属性) | 无 | `VertexSeq` | 顶点序列 |
| `edge_seq` (属性) | 无 | `EdgeSeq` | 边序列 |

`DiGraph` 额外方法：

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `find_in(name)` | `name: str` | `List[Vertex]` | 查找入边顶点 |
| `find_out(name)` | `name: str` | `List[Vertex]` | 查找出边顶点 |
| `find_tree(name, depth, mode)` | `name: str, depth: int=0xff, mode: str='out'` | `Dict` | 查找依赖树 |

```python
from hagworm.extend.igraph import DiGraph

graph = DiGraph("service_deps")
graph.add_vertex("service_a")
graph.add_vertex("service_b")
graph.add_vertex("service_c")
graph.add_edge("service_a", "service_b")
graph.add_edge("service_b", "service_c")

graph.find_out("service_a")  # [service_b]
graph.deep_search("service_a", mode="out")  # ["service_b", "service_c"]
```

---

### hagworm.extend.ntp — NTP 时间同步

#### `NTPClient`

NTP 客户端（基于后台线程调度器，非 asyncio 异步），定期同步时间偏移。

`NTPClient(host, *, version=4, port='ntp', timeout=5, interval=3600, sampling=5)`，`host` 之后均为关键字参数。

| 入参 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `host` | `str` | `ntp.aliyun.com` | NTP 服务器 |
| `version` | `int` | `4` | NTP 协议版本 |
| `port` | `str` | `ntp` | NTP 端口 |
| `timeout` | `int` | `5` | 请求超时（秒） |
| `interval` | `int` | `3600` | 同步间隔（秒） |
| `sampling` | `int` | `5` | 每次采样次数（取中位数作为偏移） |

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `start()` | 无 | `None` | 启动调度器 |
| `stop()` | 无 | `None` | 停止调度器 |

| 属性 | 类型 | 说明 |
|------|------|------|
| `offset` | `float` | 时间偏移量（多次采样的中位数，未成功同步时为 `0`） |
| `timestamp` | `float` | 校准后的秒级时间戳 |
| `timestamp_ms` | `int` | 校准后的毫秒级时间戳 |

```python
from hagworm.extend.ntp import NTPClient

ntp = NTPClient("ntp.aliyun.com", interval=1800, sampling=5)
ntp.start()

calibrated_time = ntp.timestamp
calibrated_time_ms = ntp.timestamp_ms

ntp.stop()
```

---

### hagworm.extend.process — 多进程管理

> ⚠️ **注意**：导入 `hagworm.extend.process` 会执行 `multiprocessing.set_start_method('spawn', force=True)`，强制改变整个进程的多进程启动方式。此外 `fork_processes()` 依赖 `os.fork`，仅 POSIX 可用。

#### `fork_processes()`

封装 `os.fork()`，在子进程中重新播种随机数。返回值与 `os.fork()` 一致：父进程返回子进程 pid，子进程返回 0。

#### `Daemon`

守护进程管理器，管理多个子进程。

`Daemon(target: Callable, sub_process_num: int, *, cpu_affinity: bool = False, join_timeout: int = 10, keep_alive: bool = False, **kwargs)`

| 入参 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `target` | `Callable` | 必填 | 子进程入口函数，实际以 `target(idx, **kwargs)` 调用，`idx` 为槽位序号（`0 ~ sub_process_num-1`） |
| `sub_process_num` | `int` | 必填 | 子进程数量 |
| `cpu_affinity` | `bool` | `False` | 是否绑定 CPU 亲和性（按槽位绑定到同序号核心，数量超过核心数时构造即报错） |
| `join_timeout` | `int` | `10` | 等待子进程退出的超时（秒） |
| `keep_alive` | `bool` | `False` | 是否在子进程退出后自动补齐 |
| `**kwargs` | `Any` | 无 | 仅原样透传给 `target`，不参与主进程逻辑 |

> `cpu_affinity` / `join_timeout` / `keep_alive` 均为**关键字参数**（`sub_process_num` 之后的位置参数非法）。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `run()` | 无 | `None` | 启动子进程；`keep_alive=True` 时子进程退出后自动补齐并持续运行，否则等所有子进程退出后返回 |

#### `SharedByteArray`

基于共享内存的字节数组，支持跨进程读写。

`SharedByteArray(name=None, create=False, size=0)`，实现 `ByteArrayAbstract` 的 `read(size) -> memoryview` 与 `write(buffer)`，其余 `read_*` / `write_*` 方法均继承自该基类。

提供 `release()` 方法：关闭共享内存句柄，创建模式下同时 `unlink`。

> ⚠️ **注意**：创建方（`create=True`）退出 `with` 块时会关闭并 `unlink` 共享内存段，因此读取方必须在创建方进程存活期间、于**另一个进程**中并发打开同一 `name`。

```python
from hagworm.extend.process import SharedByteArray

# 创建模式（进程 A）：必须在 with 块内完成写入并保持进程存活，
# 读取方（进程 B）需在此期间并发打开同名共享内存
with SharedByteArray(name="shared_buf", create=True, size=1024) as buf:
    buf.write_int(42)
    # ... 保持进程运行并协调进程 B 读取 ...
    # 退出 with 后会 unlink，进程 B 将无法再打开

# 读取模式（进程 B，需在进程 A 仍处于上述 with 块期间并发执行）
with SharedByteArray(name="shared_buf", create=False) as buf:
    val = buf.read_int()  # 42
```

---

### hagworm.extend.asyncio.base — 异步基础工具类

#### `Utils`（异步版）

继承同步 `Utils`，增加异步工具方法。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `sleep(delay, result=None)` | `delay: float` 等 | `None` | 异步等待（等价于 `asyncio.sleep`） |
| `is_coroutine_function(func)` | `func: Callable` | `bool` | 判断是否协程函数（支持 partial） |
| `awaitable_wrapper(obj)` | `obj: Any` | `Any` | 自适应 awaitable 对象 |
| `wait_frame(count)` | `count: int=10` | `Generator` | 暂停指定帧数 |
| `get_event_loop()` | 无 | `AbstractEventLoop` | 获取事件循环 |
| `get_current_loop()` | 无 | `AbstractEventLoop` | 获取当前任务的事件循环 |
| `loop_time()` | 无 | `float` | 获取事件循环时间 |
| `wait_any_completed(*args)` | 协程/任务 | `None` | 等待任一完成，取消其余 |
| `call_soon(callback, *args, loop, context)` | `callback: Callable` 等 | `Handle` | 尽快调度（隔离上下文） |
| `call_later(delay, callback, *args, loop, context)` | `delay: float` 等 | `Handle` | 延时指定秒数调用 |
| `call_at(when, callback, *args, loop, context)` | `when: float` 等 | `Handle` | 指定时间调用 |
| `create_task(coro)` | `coro: Coroutine` | `Task` | 创建任务 |
| `async_timeout(timeout)` | `timeout: float` | 异步上下文管理器 | 异步超时等待 |

#### `install_uvloop()`

尝试安装 uvloop 加速事件循环。

#### `async_adapter(func)`

异步函数适配装饰器，使异步函数可以在同步代码中非阻塞调用。

#### `WeakContextVar`

弱引用版的 `ContextVar`，避免上下文变量阻止对象回收。

`WeakContextVar(name: str)`；`get() -> Any`（引用已被回收时返回 `None`）；`set(value) -> Token`（不可弱引用的对象退化为强引用存储）。

#### `robust_iterator`

异步循环器，用于需要重复执行的场景。

| 入参 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `timeout` | `int/float` | `0` | 超时时间（0=不限） |
| `interval` | `int/float` | `1` | 间隔秒数 |
| `max_times` | `int/None` | `None` | 最大执行次数（None=不限） |

产出 `index` 从 `1` 开始计数；`timeout` 为整段循环的总时限（`> 0` 才生效）；每轮 `yield` 之后会 `await Utils.sleep(interval)`。

```python
from hagworm.extend.asyncio.base import Utils, robust_iterator, ShareFuture, FuncCache

# 异步循环
async for index in robust_iterator(timeout=10, interval=1):
    # 每秒执行一次，最多10秒
    await check_status()

# 共享 Future（并发调用共享计算结果）
@ShareFuture()
async def fetch_data(key):
    # 同一时刻相同参数的调用共享结果
    return await expensive_operation(key)

# 函数缓存
@FuncCache(maxsize=100, ttl=10)
async def get_config(key):
    return await load_config(key)
```

#### `ShareFuture`

共享 Future 装饰器：同一时刻以相同函数签名并发调用时，只执行一次并共享结果（内部用 `Utils.params_sign` 生成签名，复用 `asyncio.Task`，并以 `asyncio.shield` 避免调用方取消影响其他等待者）。

`ShareFuture()` 无构造参数，用法为 `@ShareFuture()`。

#### `FuncCache`

继承 `ShareFuture` 的函数缓存，基于 `StackCache`，在有效期内函数签名一致即命中缓存。

`FuncCache(maxsize: int = 0xffff, ttl: float = 10)`

> 注意：函数返回 `None` 时不会写入缓存，因此每次都需重新计算。

#### `NonBlockingFuncWrapper`

非阻塞异步函数包装器，继承自 `base.FuncWrapper`，将多个同步或异步函数包装成一个可调用对象。调用时通过 `call_soon` 非阻塞式执行所有注册的函数。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `__call__(*args, **kwargs)` | 任意参数 | `None` | 非阻塞调用所有注册函数 |
| `is_valid` (属性) | 无 | `bool` | 是否有注册函数 |
| `add(func)` | `func: Callable` | `bool` | 添加函数，已存在返回 False |
| `remove(func)` | `func: Callable` | `bool` | 移除函数，不存在返回 False |

```python
from hagworm.extend.asyncio.base import NonBlockingFuncWrapper

wrapper = NonBlockingFuncWrapper()
wrapper.add(sync_handler)
wrapper.add(async_handler)

wrapper("event_data")  # 非阻塞，通过 call_soon 调度
```

#### `AsyncConstructor`

支持异步构造函数的基类。

```python
from hagworm.extend.asyncio.base import AsyncConstructor

class AsyncClient(AsyncConstructor):
    async def __async_init__(self):
        self.connection = await create_connection()

client = await AsyncClient()  # 异步构造
```

#### `AsyncFuncWrapper`

阻塞式异步函数包装器，将多个同步/异步函数包装为一个可 await 调用对象。

```python
from hagworm.extend.asyncio.base import AsyncFuncWrapper

wrapper = AsyncFuncWrapper()
wrapper.add(async_handler1)
wrapper.add(sync_handler2)
await wrapper("event_data")  # 依次 await 所有处理器
```

---

### hagworm.extend.asyncio.future — Future 与线程工具

#### `SignalManager`

信号管理器（静态类）。asyncio 的 `loop.add_signal_handler` 对同一信号是**覆盖式注册**，多个组件各自注册会互相覆盖（只有最后注册者生效）。`SignalManager` 在进程内只安装一个信号处理器，所有回调集中登记，信号到达时广播给全部回调，因此同进程内的多个组件（`WaitForever`、`MainProcess`/`SubProcess`、`AsyncTcpServer`、`Worker` 等）可以各自注册而不互相覆盖。

处理器直接挂在系统层（`signal.signal`，要求主线程、且需在运行中的事件循环内登记），再通过 `loop.call_soon_threadsafe` 唤醒事件循环广播。这样即使事件循环被同步代码阻塞（长计算、阻塞 IO、`time.sleep`），信号依然会被处理——这是不用 `loop.add_signal_handler` 的原因（它会用 noop 处理器接管信号，只在循环迭代时才响应）。

重复信号语义：**同一信号第二次到达**说明上一次广播后进程仍未退出（回调可能只关闭了资源、没有结束事件循环），此时强制退出：事件循环仍在运行时，`SIGINT` 抛 `KeyboardInterrupt`，其余信号抛 `SystemExit(128 + 信号编号)`（如 SIGTERM 退出码 143），两者都会执行 `finally` 与 `atexit`，因此子进程兜底清理不会丢；若此时事件循环已停止或关闭（进程已进入收尾阶段，可能正在执行 GC/弱引用回调），则改用 `os._exit(128 + 信号编号)` 直接强制退出，避免抛异常被解释器当作 `Exception ignored` 吞掉并打印堆栈。由于处理器在系统层，事件循环被阻塞时重复信号同样生效。

退出事件语义：首次登记时 `SignalManager` 会自行 `atexit.register` 一个钩子；解释器退出时把**退出事件**走同一份登记集合广播一次，让仍登记着的组件有机会收尾，随后 `set_wakeup_fd(-1)` 并恢复 SIGINT/SIGTERM 的默认处理、清空登记。注意退出阶段没有运行中的事件循环，回调需自行保证在无事件循环环境下安全（框架内回调已处理：`Worker` 会跳过、`MainProcess`/`SubProcess` 仅在循环运行时才 stop）。

信号广播与退出广播都对每个回调做**异常隔离**（`except Exception` + warning），单个回调失败不会阻断其余回调与进程退出。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `register(func, *, loop=None)` | `func: Callable` | `None` | 登记 SIGINT/SIGTERM 回调；`loop` 缺省取当前事件循环（信号广播所在循环，切换时重置重复信号计数）；安装系统层处理器；同一回调只登记一次；首次登记时挂上退出钩子 |
| `unregister(func)` | `func: Callable` | `None` | 注销回调，按等值比较（兼容绑定方法） |
| `dispatch()` | 无 | `None` | 在事件循环内广播给所有已登记回调，快照遍历（允许回调中注销自己） |

注册与注销示例（框架内各组件即按此方式使用）：

```python
from hagworm.extend.asyncio.future import SignalManager

SignalManager.register(handler, loop=loop)   # 登记回调，首次调用时安装分发器
SignalManager.unregister(handler)            # 组件结束时注销，避免残留回调
```

#### `WaitForever`

永远等待的 Future，通过 SIGINT/SIGTERM 信号退出。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `__init__(loop)` | `loop: AbstractEventLoop=None` | `WaitForever` | 初始化并通过 `SignalManager` 注册信号处理 |
| `__await__()` | 无 | 无 | `await WaitForever()` 阻塞等待直到 `exit()` 被触发 |
| `done()` | 无 | `bool` | 是否已退出 |
| `exit()` | 无 | `None` | 触发退出并注销信号处理 |

#### `DelayTask`

延迟启动的协程任务，在调用 `start()` 之前不会执行，完成后可通过 `result()` 和 `exception()` 获取结果。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `__init__(coroutine, **kwargs)` | `coroutine: Coroutine` | `DelayTask` | 初始化，传入协程对象 |
| `start()` | 无 | `None` | 启动任务执行 |
| `cancel()` | 无 | `bool` | 取消任务 |
| `running()` | 无 | `bool` | 是否尚未启动 |
| `done()` | 无 | `bool` | 任务是否已完成 |
| `result()` | 无 | `Any` | 获取任务结果（完成前调用会异常） |
| `exception()` | 无 | `Optional[BaseException]` | 获取任务异常 |
| `__await__()` | 无 | `Any` | `await task` 等待任务结果 |

```python
from hagworm.extend.asyncio.future import DelayTask

task = DelayTask(some_coroutine())
# ... 暂未执行
task.start()         # 启动
result = await task  # 等待结果
```

#### `Thread`

继承 `threading.Thread` 的增强版线程（固定 `daemon=True`，不可关闭），支持优雅退出。构造签名为 `Thread(group=None, target=None, name=None, args=(), kwargs=None)`。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `stop(timeout)` | `timeout: Optional[int]=None` | `None` | 设置退出标记；`timeout` 非 `None` 时额外 `join(timeout)` 等待线程退出 |
| `is_stopped()` | 无 | `bool` | 是否已标记退出 |

#### `ThreadPool`

线程池，桥接线程与协程。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `__init__(max_workers)` | `max_workers: int=None` | `ThreadPool` | 初始化 |
| `run(_callable, *args, **kwargs)` | `_callable: Callable` | `Any` | 在线程中执行函数并 await 结果（不支持协程函数） |

#### `ThreadWorker`

装饰器，将同步函数转为非阻塞异步函数。

```python
from hagworm.extend.asyncio.future import ThreadWorker, ThreadPool

@ThreadWorker(max_workers=4)
def blocking_io(path):
    with open(path) as f:
        return f.read()

# 在异步代码中调用
result = await blocking_io("/path/to/file")

# 也可以直接使用 ThreadPool
pool = ThreadPool(max_workers=8)
result = await pool.run(os.path.exists, "/path/to/file")
```

#### `SubProcess`

异步子进程管理，通过命令方式启动子进程。

`SubProcess(program: str, *args, stdin=asyncio.subprocess.DEVNULL, stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL, **kwargs)`

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `create(program, *args, stdin=None, stdout=None, stderr=None, **kwargs)` | `program: str` 等 | `SubProcess` | 创建并启动子进程（三个流参数默认值为 `None`，与构造函数不同） |
| `start()` | 无 | `bool` | 启动子进程（已运行返回 False） |
| `stop()` | 无 | `None` | 结束子进程 |
| `kill()` | 无 | `None` | 强制结束子进程 |
| `wait(timeout)` | `timeout: float=None` | `None` | 等待子进程结束 |
| `is_running()` | 无 | `bool` | 是否运行中 |
| `create_stdout_task()` | 无 | `Task` | 创建 stdout 日志任务 |
| `create_stderr_task()` | 无 | `Task` | 创建 stderr 日志任务 |

| 属性 | 类型 | 说明 |
|------|------|------|
| `pid` | `int` | 进程 ID |
| `process` | `asyncio.subprocess.Process` | 底层进程对象 |
| `stdin` | `StreamWriter` | 标准输入流 |
| `stdout` | `StreamReader` | 标准输出流 |
| `stderr` | `StreamReader` | 标准错误流 |

生命周期：`start()` 会注册 `atexit` 兜底钩子，`stop()` / `kill()` 自动注销；解释器退出时若外部命令仍存活则强制 `kill`，避免残留孤儿进程。

```python
from hagworm.extend.asyncio.future import SubProcess

proc = await SubProcess.create("python", "script.py", stdout=asyncio.subprocess.PIPE)
await proc.wait(timeout=30)
```

---

### hagworm.extend.asyncio.task — 定时任务与并发控制

#### `TaskAbstract`

任务基类，实现 `TaskInterface`，`IntervalTask` / `CronTask` 均由它派生（`DCSCronTask` 直接实现 `TaskInterface`，与本类无继承关系）。

`TaskAbstract(scheduler: Optional[AsyncIOScheduler] = None)`：不传 `scheduler` 时按 `DEFAULT_SCHEDULER_CONFIG` 创建 `AsyncIOScheduler`（时区 `DEFAULT_SCHEDULER_TIMEZONE`，即 `Asia/Shanghai`，`job_defaults = {coalesce: False, max_instances: 1, misfire_grace_time: 10}`）。

| 成员 | 说明 |
|------|------|
| `scheduler` | 属性，底层 `AsyncIOScheduler` |
| `start()` / `stop()` | 启动 / 关闭调度器（重复调用为空操作） |
| `is_running()` | 调度器是否运行中 |
| `add_job(...)` | 抽象方法，由子类实现 |
| `remove_job(job_id)` | 移除指定任务 |
| `remove_all_jobs()` | 移除全部任务 |

#### `IntervalTask`

间隔任务，按固定秒数重复执行。

`IntervalTask.add_job(interval, func, *args, **kwargs)`；由于内部设置 `next_run_time=now`，**启动后会立即执行一次**，之后每隔 `interval` 秒重复。

```python
from hagworm.extend.asyncio.task import IntervalTask

task = IntervalTask.create(60, my_func, arg1, arg2)
task.start()
```

#### `CronTask`

Cron 定时任务。

`CronTask.add_job(crontab, func, *args, **kwargs)`，`crontab` 为标准 5 段表达式，时区 `Asia/Shanghai`。

```python
from hagworm.extend.asyncio.task import CronTask

task = CronTask.create("0 8 * * *", morning_job)  # 每天8点
task.start()
```

#### `DCSCronTask`

分布式 Cron 任务，基于 Redis 实现分布式锁，确保多实例环境下任务只执行一次。锁通过 `SET key name NX EX 3600` 获取，任务结束后 `DELETE`；未获得锁时打 debug 日志 `dcs cron task idle`。

`DCSCronTask(redis_client, name, crontab, func, *args, **kwargs)`：直接实现 `TaskInterface`，仅提供 `start()` / `stop()` / `is_running()`，**不具备** `TaskAbstract` 的 `scheduler` / `add_job` / `remove_job` 等成员。

```python
from hagworm.extend.asyncio.task import DCSCronTask
from hagworm.extend.asyncio.redis import RedisPool

dcs_task = DCSCronTask(redis_pool, "daily_report", "0 9 * * *", generate_report)
dcs_task.start()
```

#### `MultiTasks`

多任务并发管理器，通过信号量控制并发数。

提供 `append(coroutine)` 与 `extend(coroutines)` 方法，分别用于追加单个协程与批量追加协程。

```python
from hagworm.extend.asyncio.task import MultiTasks

tasks = MultiTasks(tasks_num=10)
tasks.append(fetch_url("http://example.com/1"))
tasks.append(fetch_url("http://example.com/2"))
tasks.append(fetch_url("http://example.com/3"))
results = await tasks  # 最多 10 个并发
```

#### `RateLimiter`

流量控制器，保护计算资源。

进入队列的任务，排队或执行超时会通过 Future 引发 `TimeoutError`；仅当任务被显式 `cancel` 时才会引发 `CancelledError`。

`RateLimiter(task_limit: int, wait_limit: int = 256, timeout: int = 60)`

| 入参 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `task_limit` | `int` | 必填 | 最大并发任务数 |
| `wait_limit` | `int` | `256` | 等待队列长度 |
| `timeout` | `int` | `60` | 等待超时（秒） |

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `append(func, *args, **kwargs)` | `func: Callable` | `bool` | 提交任务（不等待结果） |
| `call(func, *args, **kwargs)` | `func: Callable` | `Any` | 提交任务并等待结果 |
| `start()` | 无 | `None` | 启动消费协程 |
| `size()` | 无 | `int` | 当前队列长度 |
| `join()` | 无 | `None` | 等待所有任务完成 |
| `close()` | 无 | `None` | 关闭限流器 |

```python
from hagworm.extend.asyncio.task import RateLimiter

limiter = RateLimiter(task_limit=5, wait_limit=10, timeout=60)

# 不等待结果
await limiter.append(process_data, data1)
await limiter.append(process_data, data2)

# 等待结果
result = await limiter.call(process_data, data3)

await limiter.join()
limiter.close()
```

---

### hagworm.extend.asyncio.buffer — 缓冲区

#### `QueueBuffer`

异步队列缓冲区，批量消费数据。

提供 `append(data, timeout=None)`（`await` 入队，达到 `slice_size` 时唤醒消费协程；`timeout` 为入队等待超时）、`safe_append(data)`（线程安全的入队）、`size()`（当前队列长度）与 `close()`（异步，排空队列后关闭消费协程与内部 `RateLimiter`）方法。

| 入参 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `handler` | `Callable` | 必填 | 批量处理函数 |
| `slice_size` | `int` | 必填 | 每批大小 |
| `slice_time` | `float` | 必填 | 最大等待时间 |
| `task_limit` | `int` | 必填 | 并发处理数 |
| `wait_limit` | `int` | `0` | 等待队列长度（0=不限） |
| `timeout` | `int` | `60` | 超时 |

```python
from hagworm.extend.asyncio.buffer import QueueBuffer

async def batch_insert(records):
    await db.insert_many(records)

buffer = QueueBuffer(batch_insert, slice_size=100, slice_time=5, task_limit=4)
await buffer.append(record1)
await buffer.append(record2)
# 满100条或5秒后自动批量处理
await buffer.close()
```

#### `FileBuffer`

文件缓冲区，支持大文件分片读写。

提供 `close()` 方法：关闭并释放所有分片临时文件。

```python
from hagworm.extend.asyncio.buffer import FileBuffer

with FileBuffer(slice_size=0x1000000) as fb:
    fb.write(b"large data...")
    data = fb.read(1024)
```

---

### hagworm.extend.asyncio.net — HTTP 客户端

#### `HTTPClientPool`

基于 `httpx` 的异步 HTTP 客户端，内置重试机制。

`HTTPClientPool(retries: int = 5, verify: ssl.SSLContext | str | bool = False, limits: httpx.Limits = DEFAULT_LIMITS, timeout: TimeoutTypes = DEFAULT_TIMEOUT, **kwargs)`

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `get(url, **kwargs)` | `url: str` 等 | `bytes` | GET 请求 |
| `post(url, **kwargs)` | `url: str` 等 | `bytes` | POST 请求 |
| `put(url, **kwargs)` | `url: str` 等 | `bytes` | PUT 请求 |
| `patch(url, **kwargs)` | `url: str` 等 | `bytes` | PATCH 请求 |
| `delete(url, **kwargs)` | `url: str` 等 | `bytes` | DELETE 请求 |
| `options(url, **kwargs)` | `url: str` 等 | `Dict` | OPTIONS 请求，返回响应头 |
| `head(url, **kwargs)` | `url: str` 等 | `Dict` | HEAD 请求，返回响应头 |
| `send_request(method, url, **kwargs)` | `method: str` 等 | `Result` | 通用请求（`content`/`data`/`files`/`json`/`params`/`headers`/`cookies` 均为关键字参数） |
| `close()` | 无 | `None` | 异步关闭底层 httpx 客户端 |

`send_request` 返回 `Result` 命名元组，字段为 `status` / `headers` / `cookies` / `body`。

模块级常量：

| 名称 | 说明 |
|------|------|
| `DEFAULT_RETRIES` | 默认最大尝试次数 `5`（首次 + 最多 4 次重试） |
| `DEFAULT_TIMEOUT` | 默认请求超时 `60s` |
| `DEFAULT_LIMITS` | 默认连接池上限（`max_connections=200` / `max_keepalive_connections=50`） |
| `DOWNLOAD_TIMEOUT` | `Downloader` 的默认超时 `600s` |

#### `HTTPTextClientPool`

返回文本响应的 HTTP 客户端。

#### `HTTPJsonClientPool`

返回 JSON 响应的 HTTP 客户端。

#### `Downloader`

流式文件下载器，继承 `_ClientBase`（超时默认 `DOWNLOAD_TIMEOUT`，即 `600s`）。

`fetch(file: BinaryIO, url: URL | str, *, chunk_size: int = 65536, params=None, headers=None, cookies=None) -> None`，以流式方式将响应体按 `chunk_size` 写入 `file`（文件句柄由调用方自行打开与关闭）。

```python
from hagworm.extend.asyncio.net import HTTPClientPool, HTTPJsonClientPool, Downloader

# 基本用法
client = HTTPClientPool(retries=3)
body = await client.get("https://api.example.com/data")

# JSON 客户端
json_client = HTTPJsonClientPool()
data = await json_client.get("https://api.example.com/json")  # 返回 dict/list

# 文件下载
downloader = Downloader()
with open("output.zip", "wb") as f:
    await downloader.fetch(f, "https://example.com/file.zip")

await client.close()
```

---

### hagworm.extend.asyncio.redis — Redis 连接管理

#### `RedisPool` / `RedisClusterPool`

Redis 单机/集群连接池管理。

| 入参 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `host` | `str` | 必填 | 主机 |
| `port` | `int` | 必填 | 端口 |
| `db` | `int` | `0` | 数据库序号（仅 RedisPool） |
| `startup_nodes` | `Iterable[Node]` | `None` | 集群起始节点（仅 RedisClusterPool） |
| `username` | `str` | `None` | 用户名 |
| `password` | `str` | `None` | 密码 |
| `max_connections` | `int` | `32` | 最大连接数 |
| `max_idle_time` | `float` | `43200` | 最大空闲时间（秒） |
| `idle_check_interval` | `float` | `1` | 空闲检查间隔（秒） |
| `protocol_version` | `Literal[2, 3]` | `2` | RESP 协议版本 |

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `open()` | 无 | `self` | 初始化连接池 |
| `close()` | 无 | `None` | 关闭连接池 |
| `set_key_prefix(value)` | `value: str` | `None` | 设置键前缀 |
| `get_safe_key(key, *args, **kwargs)` | `key: str` 等 | `str` | 生成安全键名 |
| `allocate_lock(name, **kwargs)` | `name: str` 等 | `LuaLock` | 分配分布式锁 |
| `get_obj(name)` | `name: str` | `Any` | 获取 Pickle 对象 |
| `getset_obj(name, value)` | `name: str, value: Any` | `Any` | 设置并返回旧的 Pickle 对象 |
| `set_obj(name, value, seconds)` | `name: str, value: Any, seconds: Optional[float]=None` | `Any` | 存储 Pickle 对象 |
| `decode(value)` | `value: bytes / tuple / list / dict` | `Any` | 静态方法，对顶层及一层容器元素做 utf-8 解码（非递归） |

属性 `max_connections` 为连接池上限。

#### `LuaLock`

Redis 分布式锁（`coredis` 的 `LuaLock` 封装），实现 `AsyncContextManager`，可 `async with` 使用。

`LuaLock(client, name, timeout: Optional[float] = None, sleep: float = 0.1, blocking: bool = True, blocking_timeout: Optional[float] = None)`

- `acquire()`：本任务已持锁时改为 `extend(timeout)` 续期，避免重复加锁。
- `release()`：仅在本任务持锁时释放。
- `async with` 退出时自动 `release()`。

#### `RedisDelegate`

Redis 功能组件，简化初始化。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `init_redis_single(...)` | 同 `RedisPool` 构造参数 | `RedisPool` | 初始化单机连接池 |
| `init_redis_cluster(...)` | 同 `RedisClusterPool` 构造参数 | `RedisClusterPool` | 初始化集群连接池 |
| `set_redis_key_prefix(value)` | `value: str` | `None` | 设置键前缀 |
| `close_redis_pool()` | 无 | `None` | 关闭连接池 |
| `get_redis_client()` | 无 | `RedisPool` / `RedisClusterPool` | 获取连接池 |

属性 `redis_pool` 为当前连接池。注意 `init_redis_single` / `init_redis_cluster` 在 `username, password` 之后的关键字参数（`max_connections` / `max_idle_time` / `idle_check_interval` / `protocol_version`）均为关键字限定，与 `RedisPool` 构造函数（无 `*`）并不完全一致。

#### `ShareCache`

基于分布式锁的共享缓存，用于避免并发重复计算。注意：互斥并非强制，仅当调用方严格按 `get → 计算 → set → release` 流程使用时，才能保证同一时刻业务逻辑只执行一次。

`ShareCache(redis_client, share_key, lock_timeout=60, lock_blocking_timeout=60)`，实现 `AsyncContextManager`；属性 `result` 保存最近一次读到的值，提供 `get()`、`set(value, expire=None)` 与异步 `release()`。

#### `PeriodCounter`

周期计数器，基于 Redis 的 incrby/decrby。

`PeriodCounter(redis_client, key_prefix, time_slice)`，`time_slice` 为周期秒数（必填）。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `get()` | 无 | `Any` | 获取当前周期计数值 |
| `incr(val)` | `val: int=1` | `int` | 增加计数 |
| `decr(val)` | `val: int=1` | `int` | 减少计数 |

```python
from hagworm.extend.asyncio.redis import RedisPool, RedisDelegate, ShareCache

# 直接使用
redis = await RedisPool("localhost", 6379, max_connections=32).open()
await redis.set("key", "value")
val = await redis.get("key")

# 使用 Delegate
class MyService(RedisDelegate):
    async def init(self):
        await self.init_redis_single("localhost", 6379)

    async def get_data(self, key):
        return await self.redis_pool.get(key)

# 共享缓存
async with ShareCache(redis, "expensive_result", lock_timeout=60) as cache:
    result = await cache.get()
    if result is None:
        result = await expensive_computation()
        await cache.set(result, expire=300)
```

---

### hagworm.extend.asyncio.mysql — MySQL 连接管理

#### `MySQLPool`

MySQL 连接池管理。

| 入参 | 类型 | 说明 |
|------|------|------|
| `host` | `str` | 主机 |
| `port` | `int` | 端口 |
| `db` | `str` | 数据库名 |
| `user` | `str` | 用户名 |
| `password` | `str` | 密码 |
| `name` | `str` | 连接池名称（默认自动生成；`name` 之后参数均为关键字限定） |
| `minsize` | `int` | 最小连接数（默认 `8`） |
| `maxsize` | `int` | 最大连接数（默认 `32`） |
| `echo` | `bool` | 是否回显 SQL（默认 `False`） |
| `pool_recycle` | `int` | 连接回收时间（秒，默认 `21600`） |
| `charset` | `str` | 字符集（默认 `utf8`） |
| `autocommit` | `bool` | 是否自动提交（默认 `True`） |
| `cursor_class` | `cursors.Cursor` | 游标类型（默认 DictCursor） |
| `readonly` | `bool` | 是否只读（默认 `False`） |

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `open()` | 无 | `None` | 初始化连接池 |
| `close()` | 无 | `None` | 关闭连接池 |
| `acquire()` | 无 | `Connection` | 获取连接 |
| `release(connection)` | `connection: Connection` | `None` | 归还连接 |
| `get_client()` | 无 | `MySQLClient` | 获取客户端 |
| `get_transaction()` | 无 | `MySQLTransaction` | 获取事务对象 |

#### `MySQLClient`

MySQL 客户端，支持 SQLAlchemy 查询构建器。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `execute(query, args)` | `query: str, args=None` | `Cursor` | 执行 SQL |
| `select(query)` | `query: Select` | `List` | 查询列表 |
| `find(query)` | `query: Select` | `Any` | 查询单条 |
| `count(column, where_clause)` | `column, where_clause=None` | `int` | 计数 |
| `insert(query)` | `query: Insert` | `int` | 插入，返回 insert_id |
| `update(query)` | `query: Update` | `int` | 更新，返回影响行数 |
| `delete(query)` | `query: Delete` | `int` | 删除，返回影响行数 |
| `release()` | 无 | `None` | 释放连接 |
| `readonly` (属性) | 无 | `bool` | 是否只读 |
| `insert_id` (属性) | 无 | `int` | 最近一次插入的自增 ID |

#### `MySQLTransaction`

MySQL 事务对象，未显式 commit 则自动 rollback。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `commit()` | 无 | `None` | 提交事务 |
| `rollback()` | 无 | `None` | 回滚事务 |

#### `MySQLDelegate`

MySQL 功能组件。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `init_mysql_rw(...)` | 同 `MySQLPool` 构造参数 | `None` | 初始化读写连接池 |
| `init_mysql_ro(...)` | 同 `MySQLPool` 构造参数 | `None` | 初始化只读连接池 |
| `close_mysql()` | 无 | `None` | 关闭连接池 |
| `get_mysql_client(readonly)` | `readonly: bool=False` | `MySQLClient` | 获取客户端 |
| `get_mysql_transaction()` | 无 | `MySQLTransaction` | 获取事务对象 |

属性 `mysql_rw_pool` / `mysql_ro_pool` 分别为读写与只读连接池。

```python
from hagworm.extend.asyncio.mysql import MySQLPool, MySQLClient, MySQLDelegate
import sqlalchemy as sa

# 直接使用
pool = MySQLPool("localhost", 3306, "mydb", "root", "password", minsize=8, maxsize=32)
await pool.open()

async with pool.get_client() as client:
    users = await client.select(sa.select(sa.table("users")))
    user = await client.find(sa.select(sa.table("users")).where(sa.column("id") == 1))

# 事务
async with pool.get_transaction() as trx:
    await trx.insert(sa.insert(sa.table("users")).values(name="Alice"))
    await trx.insert(sa.insert(sa.table("logs")).values(action="create"))
    await trx.commit()  # 不调用则自动 rollback

# 使用 Delegate
class MyService(MySQLDelegate):
    async def init(self):
        await self.init_mysql_rw("localhost", 3306, "mydb", "root", "password")
        await self.init_mysql_ro("localhost", 3307, "mydb", "reader", "password")
```

> ⚠️ **易混淆点**：`readonly=True` 的连接池调用 `insert`/`update`/`delete` 或 `get_transaction()` 都会抛出 `MySQLReadOnlyError`。

属性：`name`（连接池名称）、`readonly`、`cursor_class`。

模块级常量与异常：

| 名称 | 说明 |
|------|------|
| `MYSQL_POOL_WATER_LEVEL_WARNING_LINE` | 连接池水位告警线（`0x10`） |
| `MYSQL_EXECUTE_ARGUMENTS_TYPE` | SQL 执行参数类型别名 `Union[None, Tuple, List, Dict]` |
| `MySQLReadOnlyError` | 只读连接池执行写操作时抛出，继承 `errors.MySQLError` |

---

### hagworm.extend.asyncio.mongo — MongoDB 连接管理

#### `MongoPool`

MongoDB 连接池管理。

`MongoPool(host: List[str], username: str, password: str, *, name=None, auth_source='admin', min_pool_size=8, max_pool_size=32, max_idle_time=3600, wait_queue_timeout=10, compressors='zlib', zlib_compression_level=6, **settings)`

下列方法均为**同步**方法（无需 `await`）：

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `close()`（同步） | 无 | `None` | 关闭连接池 |
| `get_database(db_name)`（同步） | `db_name: str` | `AsyncIOMotorDatabase` | 获取数据库 |

```python
from hagworm.extend.asyncio.mongo import MongoPool, MongoDelegate

pool = MongoPool(
    host=["mongo1:27017", "mongo2:27017"],
    username="admin", password="secret",
    min_pool_size=8, max_pool_size=32
)

db = pool.get_database("mydb")
collection = db["users"]
await collection.insert_one({"name": "Alice"})
```

#### `MongoDelegate`

Mongo 功能组件。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `init_mongo(...)`（同步） | 同 `MongoPool` 构造参数 | `None` | 初始化连接池 |
| `close_mongo_pool()`（同步） | 无 | `None` | 关闭连接池 |
| `get_mongo_database(db_name)`（同步） | `db_name: str` | `AsyncIOMotorDatabase` | 获取数据库 |
| `get_mongo_collection(db_name, collection)`（同步） | `db_name: str, collection: str` | `AsyncIOMotorCollection` | 获取集合 |

属性 `mongo_pool` 为当前连接池。模块级常量 `MONGO_POOL_WATER_LEVEL_WARNING_LINE`（`0x10`）当前仅定义、未被使用。

---

### hagworm.extend.asyncio.mail — SMTP 邮件

#### `SMTPClient`

异步 SMTP 邮件客户端。

`SMTPClient(username: str, password: str, hostname: str, port: int, retry_count: int = 5, **kwargs)`

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `send_messages(email_body_list)` | `email_body_list: List[EmailBody]` | `Optional[Dict]` | 批量发送邮件 |
| `send_message(sender, recipients, message)` | `sender: str, recipients: Sequence[str], message: EmailMessage` | `Optional[Dict]` | 发送单封邮件 |
| `send(sender, recipients, subject, content)` | 均为 `str` 等 | `Optional[Dict]` | 简单发送 |
| `format_address(nickname, mailbox)` | `nickname: str, mailbox: str` | `str` | 静态方法，格式化为 `nickname<mailbox>` |
| `format_addresses(addresses)` | `addresses: List[Tuple[str, str]]` | `str` | 格式化多个地址 |

#### `EmailBody`

`EmailBody(sender: str, recipients: Sequence[str], message: EmailMessage)`，提供 `sender` / `recipients` / `message` 属性，静态方法 `create_message(sender, recipients, subject, content, content_type='text/html')` 用于构建消息。

```python
from hagworm.extend.asyncio.mail import SMTPClient, EmailBody

client = SMTPClient("user@gmail.com", "password", "smtp.gmail.com", 587)

# 简单发送
await client.send(
    sender="sender@example.com",
    recipients=["receiver@example.com"],
    subject="Hello",
    content="<h1>Hi there!</h1>"
)

# 带附件或自定义消息
message = EmailBody.create_message(
    sender="sender@example.com",
    recipients=["receiver@example.com"],
    subject="Report",
    content="<p>Please find the report attached.</p>"
)
await client.send_message("sender@example.com", ["receiver@example.com"], message)
```

---

### hagworm.extend.asyncio.socket — TCP/Unix Socket

#### `AsyncTcpServer`

异步 TCP 服务器，实现 `RunnableInterface` 与 `async with` 协议。关闭兜底统一走 `SignalManager`（服务运行期间登记 `_exit`，收到信号或解释器退出时都会关闭全部 server），不再单独注册 `atexit` 钩子。

`AsyncTcpServer(client_connected_cb: Callable, address: Any, *, family: int = socket.AF_INET, backlog: Optional[int] = None, reuse_port: bool = True, buffer_limit: int = BUFFER_LIMIT, on_startup: Callable = None, on_shutdown: Callable = None)`

```python
from hagworm.extend.asyncio.socket import AsyncTcpServer

async def handle_connection(reader, writer):
    while True:
        data = await reader.readline()
        if not data:
            break
        writer.write(data)
        await writer.drain()
    writer.close()
    await writer.wait_closed()

server = AsyncTcpServer(handle_connection, ("0.0.0.0", 8080))
server.run()
```

`AsyncTcpServer.add_listener(client_connected_cb, address, *, family, backlog, reuse_port, buffer_limit)` 可追加监听地址（支持 `socket.AF_UNIX`）。

`on_startup` / `on_shutdown` 为无参协程，分别在建立监听之前、退出时执行；`run(*, debug=None)` 会通过 `SignalManager` 监听 SIGINT/SIGTERM，收到信号后关闭全部 server。内部 `_run()` 用 `try/finally` 收尾，因此正常关闭、信号中断、任务取消等任何路径下都会注销该回调并执行 `on_shutdown`。

#### `RobustConnection`

自动重连的 TCP 客户端连接。

`RobustConnection(host: Optional[str] = None, port: Optional[str] = None, *, limit: int = BUFFER_LIMIT, recycle: int = 86400, **kwargs)`，属性 `reader` / `writer` 为底层 `StreamReader` / `StreamWriter`。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `open()` | 无 | `None` | 建立连接 |
| `close()` | 无 | `None` | 关闭连接 |
| `reset()` | 无 | `None` | 关闭后重连 |
| `readline()` | 无 | `bytes` | 读取一行 |
| `readuntil(separator)` | `separator: bytes=b'\n'` | `bytes` | 读取到分隔符 |
| `read(n)` | `n: int=-1` | `bytes` | 读取 n 字节 |
| `write(data)` | `data: bytes` | `None` | 写入数据 |
| `writelines(data)` | `data: Iterable[bytes]` | `None` | 写入多行数据 |

> 说明：`open()` 失败只记日志、不抛异常，需以 `writer is None` 判断是否连接成功；每次读写前会 `_check_recycle()`，未连接或超过 `recycle` 周期（默认 86400 秒）先 `reset()` 重连，重连后仍无 `writer` 则抛 `ConnectionError`；读写失败时会先 `reset()` 再重新抛出原异常。

#### `UnixSocketServer` / `UnixSocketClient`

Unix Domain Socket 服务端与客户端。默认使用 msgpack 编解码（可在 StreamFramer 实例化时注入自定义编/解码器），线协议为 `MAGIC(8) + length(4) + body`。

| 类 | 构造签名 | 说明 |
|----|----------|------|
| `UnixSocketServer` | `(client_connected_cb: Callable, path: str, limit: int = BUFFER_LIMIT)` | `open()` 会先删除已存在的 `path`，再 `start_unix_server` 并 `start_serving()` |
| `UnixSocketClient` | `(path: str, limit: int = BUFFER_LIMIT)` | 属性 `reader` / `writer`；内部固定使用默认 `StreamFramer()`，如需自定义编解码器请直接使用 `StreamFramer` |

> `limit` 是 asyncio stream 缓冲区上限，并非消息大小上限。

`UnixSocketClient` 方法：

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `open()` | 无 | `None` | 建立连接 |
| `close()` | 无 | `None` | 关闭连接 |
| `recv_msg(timeout)` | `timeout: float=None` | `Any` | 接收一条消息 |
| `send_msg(data)` | `data: Any` | `None` | 发送一条消息 |

#### `StreamFramer`

实例类，封装 `recv_msg` / `send_msg`，线协议为 `MAGIC(8) + length(4, 大端) + body`。

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `StreamFramer(encoder, decoder)` | `encoder: Callable=msgpack.dumps, decoder: Callable=msgpack.loads` | `StreamFramer` | 编/解码器均为单参数可调用对象，可注入自定义实现 |
| `recv_msg(reader, timeout)` | `reader: StreamReader, timeout: float=None` | `Any` | 接收一条消息 |
| `send_msg(writer, data)` | `writer: StreamWriter, data: Any` | `None` | 发送一条消息 |

模块级常量与类型：

| 名称 | 说明 |
|------|------|
| `MESSAGE_LIMIT` | 单条消息 body 最大字节数（1 MiB） |
| `BUFFER_LIMIT` | asyncio stream 缓冲区上限（`2 * MESSAGE_LIMIT`） |
| `MAGIC` | 帧头魔数（8 字节） |
| `DEFAULT_UNIX_SOCKET_ENDPOINT` | 默认 Unix Domain Socket 路径 |
| `SocketConfig` | `NamedTuple`，描述单个监听配置（client_connected_cb/address/family/backlog/reuse_port/buffer_limit） |

---

### hagworm.extend.asyncio.command — 多进程命令框架

#### `MainProcess` / `SubProcess`

异步多进程框架，主进程管理子进程的生命周期。

`MainProcess(target: Callable, sub_process_num: int, *, cpu_affinity: bool = False, join_timeout: int = 10, keep_alive: bool = False, **kwargs)`，继承 `hagworm.extend.process.Daemon`，沿用同一套 `target(idx, **kwargs)` 调用约定与 `keep_alive` 语义。`run()` 内部安装 uvloop、通过 `SignalManager` 注册信号，并依次执行 `initialize()` → `_execute()` → `release()`，结束时注销该信号回调。

`SubProcess(process_num: int, *args, **kwargs)` 实现 `RunnableInterface`，`process_num` 即主进程分配的槽位序号（`Daemon` 以 `args=(idx,)` 传入）：

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `create(*args, **kwargs)` | 同构造函数 | `None` | 类方法，等价于 `cls(*args, **kwargs).run()`，常直接作为主进程的 `target` |
| `close(*_)` | 无 | `None` | 唤醒阻塞协程并关闭事件循环 |
| `initialize()` | 无 | `None` | 子进程启动钩子（可覆盖） |
| `release()` | 无 | `None` | 子进程退出钩子（可覆盖） |
| `_execute()` | 无 | `None` | 子进程主体逻辑，默认 `await WaitForever()` 常驻（可覆盖） |
| `run()` | 无 | `None` | 安装 uvloop、通过 `SignalManager` 注册信号并运行上述流程，结束时注销 |

```python
from hagworm.extend.asyncio.command import MainProcess, SubProcess

class MyWorker(SubProcess):
    async def initialize(self):
        print(f"Worker {self._process_id} initializing...")

    async def _execute(self):
        # 业务逻辑
        pass

    async def release(self):
        print(f"Worker {self._process_id} releasing...")

class MyMain(MainProcess):
    async def initialize(self):
        print("Main process initializing...")

    async def release(self):
        print("Main process releasing...")

if __name__ == "__main__":
    MyMain(MyWorker.create, sub_process_num=4).run()
```

#### `MainProcessWithIPC` / `SubProcessWithIPC`

基于 Unix Domain Socket 的进程间通信版主/子进程，主进程通过 `_client_connected_cb` 处理子进程连接。

| 类 | 构造签名 | 说明 |
|----|----------|------|
| `MainProcessWithIPC` | `(target, sub_process_num, *, cpu_affinity=False, join_timeout=10, keep_alive=False, unix_socket_path=DEFAULT_UNIX_SOCKET_ENDPOINT, unix_socket_limit=BUFFER_LIMIT, **kwargs)` | 额外维护 `UnixSocketServer`；子类需实现抽象协程 `_client_connected_cb(reader, writer)` |
| `SubProcessWithIPC` | `(process_num, unix_socket_path, unix_socket_limit=BUFFER_LIMIT)` | `unix_socket_path` 为必填位置参数（`process_num` 由主进程以 `args=(idx,)` 传入）；缓冲区上限须与主进程侧一致，否则读包会死锁 |

> `unix_socket_path` / `unix_socket_limit` 会经 `**kwargs` 自动透传给子进程入口，因此子类一般无需再手工传递。

---

### hagworm.extend.asyncio.process — 子进程托管

#### `Worker`

基于 `multiprocessing` 的异步子进程托管器，实现 `TaskInterface`。固定使用 `multiprocessing.get_context('spawn')` 启动（`PROCESS_CONTEXT`），不依赖 `hagworm.extend.process` 的 `set_start_method` 副作用，也避免 `fork` 复制父进程的事件循环与连接状态；因此 `target` 及其关键字参数必须可 pickle（模块级函数、类方法等），lambda 或局部函数会启动失败，入口脚本需保留 `if __name__ == "__main__"` 保护。

`Worker` 与子进程一一绑定：一个实例只启动一个子进程，启动后不可复用，子进程退出即终止。

多个实例各自注册信号回调，框架内部由 `SignalManager` 统一多播，因此主进程收到信号时多类子进程都会被强杀。

| 入参 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `name` | `str` | 必填 | 子进程名称，用于日志区分不同类型的子进程 |
| `target` | `Callable` | 必填 | 子进程入口函数，需可 pickle |
| `**kwargs` | `Any` | 无 | 透传给 `target` 的关键字参数（以 `Process(target=target, kwargs=kwargs)` 启动，不含位置参数） |

| 方法 | 入参 | 返回值 | 说明 |
|------|------|--------|------|
| `is_running()` | 无 | `bool` | 子进程是否存活（同步方法） |
| `start()` | 无 | `bool` | 启动子进程；每个实例只能启动一次，已启动过返回 `False`（协程） |
| `stop()` | 无 | `None` | 向子进程发送 `SIGINT` 并按 `CHECK_INTERVAL` 轮询等待其退出，超过 `STOP_TIMEOUT` 未退出则 `kill()`，最后 `join()`（协程） |

模块级常量：

| 名称 | 说明 |
|------|------|
| `CHECK_INTERVAL` | `stop()` 等待子进程退出的轮询间隔（秒），默认 `0.05` |
| `STOP_TIMEOUT` | `stop()` 等待子进程优雅退出的超时时间（秒），默认 `10`，超时强杀 |
| `PROCESS_CONTEXT` | 子进程启动上下文，固定为 `multiprocessing.get_context('spawn')` |

行为说明：

- `name` 会随启动、停止、信号日志一并打印，用于区分不同类型的子进程。
- `start()` 在子进程启动成功后通过 `SignalManager.register` 注册 SIGINT/SIGTERM 回调，收到信号后直接强杀子进程；框架内由 `SignalManager` 多播，多实例不会互相覆盖。
- `stop()` 先 `os.kill(pid, signal.SIGINT)` 通知子进程退出（在子进程中表现为 `KeyboardInterrupt`），再轮询至退出；若超过 `STOP_TIMEOUT` 仍未退出则 `process.kill()` 强杀，避免子进程忽略信号时无限等待；未启动或已退出时直接返回。
- 兜底与释放：构造时 `atexit.register(self._kill)`，解释器退出时若子进程仍存活则强制 `kill()` 并 `join()`；`stop()` 完成后会通过 `SignalManager.unregister` 注销信号回调并 `atexit.unregister(self._kill)`，避免长期持有实例引用。
- 启动失败（如 `target` 不可 pickle）时不会记录进程对象，`start()` 可再次调用重试。

```python
import asyncio

from hagworm.extend.asyncio.process import Worker

# 子进程入口必须是模块级函数（可 pickle）
def serve(port: int):
    # 子进程内的业务逻辑，通常阻塞运行
    ...

async def main():
    worker = Worker("http-server", serve, port=8080)

    if await worker.start():
        print("worker started")

    while worker.is_running():
        await asyncio.sleep(1)

    await worker.stop()

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

---

### hagworm.extend.asyncio.transaction — 异步事务

#### `Transaction`（异步版）

异步事务对象，支持同步和异步回调。

构造签名 `Transaction(*, commit_callback=None, rollback_callback=None)`；提供 `add_commit_callback(_callable, *args, **kwargs)`、`add_rollback_callback(_callable, *args, **kwargs)`、`commit()`、`rollback()` 与 `bind(trx)` 方法，语义与同步事务一致（异步回调会被 `await`）。

```python
from hagworm.extend.asyncio.transaction import Transaction

async with Transaction() as trx:
    trx.add_commit_callback(lambda: print("committed"))
    trx.add_commit_callback(async_cleanup)
    await trx.commit()
```

---

### hagworm.extend.asyncio.file — 文件加载器

#### `FileLoader`

带缓存的网络/本地文件加载器。

`FileLoader(maxsize=0xff, ttl=3600, thread=32)`；`read(file)` 读取本地文件，`fetch(url, params=None, *, cookies=None, headers=None)` 拉取网络文件。

```python
from hagworm.extend.asyncio.file import FileLoader

loader = FileLoader(maxsize=100, ttl=3600, thread=32)
content = await loader.read("/path/to/file")
html = await loader.fetch("https://example.com/page.html")
```

---

### hagworm.extend.asyncio.pool — 通用对象池

#### `ObjectPool`

通用异步对象池，需子类实现 `_create()` 方法；`ObjectInterface` 为其管理的对象基类，需实现抽象方法 `open(*args, **kwargs)` 与 `close(*args, **kwargs)`。

提供 `size` 属性（池容量）及 `open()` / `close()` / `get()` 方法（均可透传 `*args` / `**kwargs`）。

```python
from hagworm.extend.asyncio.pool import ObjectPool, ObjectInterface

class MyConnection(ObjectInterface):
    async def open(self):
        self.conn = await create_connection()

    async def close(self):
        await self.conn.close()

class MyConnectionPool(ObjectPool):
    def _create(self):
        return MyConnection()

pool = MyConnectionPool(maxsize=10)
await pool.open()

async with pool.get() as conn:
    # 使用连接
    pass

await pool.close()
```

---

### hagworm.frame.fastapi — FastAPI 集成

#### `create_fastapi`

创建 FastAPI 应用，自动配置日志、uvloop、中间件和异常处理。

`create_fastapi(log_level='info', log_handler=None, log_file_path=None, log_file_name=DEFAULT_LOG_FILE_NAME, log_file_rotation=DEFAULT_LOG_FILE_ROTATOR, log_file_retention=0xff, log_extra=None, log_enqueue=False, debug=False, routes=None, **setting) -> fastapi.FastAPI`

内部依次执行 `init_logger` → `Utils.print_slogan()` → `install_uvloop`，然后构造 `fastapi.FastAPI` 并注册 http 中间件 `http_default_middleware`，再注册 `BusinessException` / `HTTPException` / `RequestValidationError` 三个异常处理器（`business_exception_handler` / `http_exception_handler` / `request_validation_exception_handler`），最后挂载 `/static` 静态目录；`**setting` 原样透传给 `fastapi.FastAPI`。

```python
from hagworm.frame.fastapi.base import create_fastapi, uvicorn_run

app = create_fastapi(
    log_level="INFO",
    log_file_path="/var/log/app",
    debug=False,
)

if __name__ == "__main__":
    uvicorn_run(lambda: app, port=8080)
```

#### `uvicorn_run`

`uvicorn_run(app, host='0.0.0.0', port=8080)`：开发环境启动入口（生产环境建议使用 gunicorn）。

#### `BusinessException`

`BusinessException(code, content=None)`：业务异常，会被异常处理器转为 `ErrorResponse`（`code` 会透传为响应中的 `code`）。

#### `http_default_middleware`

默认 HTTP 中间件，注入 trace_id 并写入 `server` / `x-trace-id` / `x-request-ttl-ms` / `x-request-payload` 响应头。

#### `APIRouter`

支持末尾斜杠兼容的路由器。

#### `Request`

增强的请求对象，提供 `client_ip`、`referer` 等便捷属性。

#### `Response` / `ErrorResponse`

统一响应格式：`Response` 为 `{"code": 0, "data": ..., "trace_id": ...}`；`ErrorResponse` 为 `{"code": <构造传入的 code>, "error": ..., "trace_id": ...}`（内置 HTTP/校验异常处理器传 `-1`，业务异常处理器透传 `BusinessException.code`）。

#### `BaseModel` / `Depends`

增强的 Pydantic 模型（元类 `ModelMetaclass`），支持 `params` 方法生成 FastAPI 依赖签名。

`Depends(dependency=None, *, use_cache=True)`：`dependency` 为 `BaseModel` 子类时会被转换为该模型的 `params` 依赖。

#### `field` — 自定义字段类型

提供 `IDCardType`、`BankCardType`、`UUIDType`、`DateType`、`JsonType`、`Base64Type`、`StrListType`、`IntListType`、`FloatListType`、`ASCVisibleType`、`EmailType`、`DomainType`、`URLType`、`MacAddrType`、`IPvAnyType`、`IPv4Type`、`IPv4CidrType`、`IPv6Type`、`IPv6CidrType` 等验证字段。

```python
from hagworm.frame.fastapi.base import APIRouter, Request, BusinessException
from hagworm.frame.fastapi.field import EmailType, IntListType
from hagworm.frame.fastapi.model import BaseModel, Depends

router = APIRouter()

class QueryModel(BaseModel):
    email: EmailType
    ids: IntListType

@router.get("/users")
async def get_users(request: Request):
    return request.client_ip

@router.get("/error")
async def error(request: Request):
    raise BusinessException(-1, "something went wrong")  # 由异常处理器转为 ErrorResponse

@router.get("/search")
async def search(params: QueryModel = Depends(QueryModel)):
    # params.email 已验证为合法邮箱
    # params.ids 已解析为整数列表
    return {"email": params.email, "ids": params.ids}
```

---

### hagworm.frame.gunicorn — Gunicorn Worker

#### `Worker`

基于 `uvicorn-worker` 的 Gunicorn Worker 类（继承 `uvicorn_worker.UvicornWorker`，构造时强制 `factory=True`、`server_header=False`）。

因 `factory=True`，gunicorn 的 app 目标必须是**工厂函数**，例如：

```bash
gunicorn main:create_app -c gunicorn.config.py
```

```python
# gunicorn.conf.py
from hagworm.frame.gunicorn import DEFAULT_WORKER_STR, SIMPLE_LOG_CONFIG

worker_class = DEFAULT_WORKER_STR  # 等价于 "hagworm.frame.gunicorn.Worker"
logconfig_dict = SIMPLE_LOG_CONFIG
workers = 4
bind = "0.0.0.0:8080"
```

模块级常量：`DEFAULT_WORKER_STR`（`'hagworm.frame.gunicorn.Worker'`）、`SIMPLE_LOG_CONFIG`（仅含 `gunicorn.error`）与 `DEFAULT_LOG_CONFIG`（含 `gunicorn.error` / `gunicorn.access`），均将标准库日志经 `InterceptHandler` 转发到 loguru。

---

### hagworm.frame.stress_tests — 压力测试框架

#### `Launcher` / `RunnerAbstract`

多进程压力测试框架，自动收集样本并生成 Markdown 报表（含统计表格与 Mermaid 图表）。

报表包含：概览（同时存在成功与失败样本时才含请求结果占比饼图）、SLA 达标判定、吞吐/延迟统计表、时间趋势（总吞吐与成功吞吐 goodput、失败率、P50/P95/P99 耗时）、成功样本延迟分布、失败原因分布（全局与按事件拆分，含失败原因占比饼图）、失败样本延迟分布（`show_failure_dist=True` 时输出）。

样本在收集时即做流式聚合（定长对数直方图 + 计数器），**不保留原始样本，内存占用与样本量无关**；分位为 0.1% 相对精度估算，均值/标准差采用在线 Welford 算法。

统计口径：Qps 按各事件自身的「首末样本时间差」计算（`ALL` 行为全局窗口）；时间轴为主进程收包时刻（monotonic），反映样本到达节奏；耗时分位/分布基于 0.1% 相对精度直方图估算，用户自定义分桶为精确计数。

模块同时暴露 `LogHistogram`（定长对数直方图，`epsilon=HISTOGRAM_EPSILON`、`base=HISTOGRAM_BASE`，含 `count`/`minimum`/`maximum`/`mean`/`std` 属性与 `record`/`merge`/`quantile`/`bucket_counts` 方法）、`EventAggregator(keep_failure_latency=False)`、`TrendSeries(max_buckets=TREND_MAX_BUCKETS, histogram_epsilon=TREND_HISTOGRAM_EPSILON)` 与 `TrendSeriesResult`。

- `Launcher(target, sub_process_num, *, ..., **kwargs)`：`target` 为子进程执行体（通常传 `RunnerAbstract.create`）、`sub_process_num` 为子进程数，两者为必填位置参数；`**kwargs` 透传给 `MainProcessWithIPC`（如 `cpu_affinity` / `join_timeout` / `keep_alive`）。
- `RunnerAbstract.success(name, resp_time)` / `RunnerAbstract.failure(name, resp_time, error=None)`：在子进程内上报一次成功/失败请求。`resp_time` 为请求耗时（单位见 `time_unit`），`error` 为失败原因（可选，空值归入 `unknown`）。
- `Launcher` 构造参数：
  - `report_dir`：报表输出目录，默认 `./`，报表文件名为 `stress_report_年月日_时分秒.md`。
  - `time_unit`：延迟指标单位（仅用于展示，默认 `ms`）。
  - `shard_table_bins`：自定义延迟分桶边界（须严格递增），不传则按样本分位数自适应生成。
  - `percentiles`：统计分位数，默认 `(0.5, 0.9, 0.95, 0.99, 0.999)`。
  - `show_std`：是否展示标准差列（默认关闭，长尾延迟下分位数更直观）。
  - `show_failure_dist`：是否额外输出失败样本延迟分布表（默认关闭）。
  - `sla`：达标判定配置，如 `{'p95': 200, 'p99': 500, 'failure': 0.01}`，`pxx` 阈值单位与 `time_unit` 一致，`failure` 为失败率阈值(0~1)。
  - `sla_scope`：SLA 判定范围，`'global'`（仅全局）/ `'event'`（仅按事件）/ `'both'`（默认，全局表 + 按事件矩阵，按未达标数降序并附柱状图）。
  - `unstable_threshold`：每秒吞吐波动系数(标准差/均值)超过该值判定为不稳定（默认 `0.2`）。
  - `unix_socket_path`：主/子进程 IPC 的 Unix Domain Socket 路径，默认 `/tmp/unix_socket_endpoint`，须与子进程侧一致。
  - `unix_socket_limit`：IPC 流缓冲区上限（字节），默认 `BUFFER_LIMIT`（2 MiB），须与子进程侧一致。

```python
import time

from hagworm.extend.asyncio.base import Utils
from hagworm.frame.stress_tests import Launcher, RunnerAbstract

class MyRunner(RunnerAbstract):
    async def _execute(self):
        for _ in range(100):
            _begin = time.monotonic()
            await Utils.sleep(Utils.randint(10, 99) / 1000)
            _cost = (time.monotonic() - _begin) * 1000

            if Utils.rand_hit([True, False], [80, 20]):
                await self.success("API", _cost)
            else:
                await self.failure("API", _cost, error="timeout")

if __name__ == "__main__":
    Launcher(MyRunner.create, sub_process_num=4).run()
```

---

### hagworm.third.grpc — gRPC 框架

#### `GRPCServer`

`grpc.aio.server` 的服务端封装。属性 `server`；方法 `bind_router(router)`、`start(address, *, family=socket.AF_INET, server_credentials=None, enable_ping=False)`（支持 `AF_INET` / `AF_INET6` / `AF_UNIX`，默认开启 `SO_REUSEPORT`）、`stop(grace=None)`、`wait(timeout=None)`。

默认 msgpack 序列化由 `Router` 的 `request_deserializer=msgpack.loads` / `response_serializer=msgpack.dumps`（以及 `GRPCClient` 的对称默认值）提供；网桥转发时为原始字节透传。

#### `Router`

gRPC 路由器，支持装饰器注册方法：`unary_unary` / `unary_stream` / `stream_unary` / `stream_stream`，另提供 `register` / `include` / `get_generic_handlers`。

#### `StreamHandler`

`stream_stream` 场景的服务端处理基类：`StreamHandler(request, context, *, metadata_model=None)`，子类实现 `on_message(data)` / `on_connect()` / `on_close(context)`；提供 `join(*, timeout=0)`、`read()`、`write(message)`、`cancel()`、`close()`，以及方法 `metadata()` / `metadata_model()` 与属性 `peer`。

模块级常量：`UNARY_UNARY_MODE` / `UNARY_STREAM_MODE` / `STREAM_UNARY_MODE` / `STREAM_STREAM_MODE` 与 `GRPC_MODE_TYPE`；`RouteFunction` 为描述单个路由的 `NamedTuple`。

#### `GRPCClient`

gRPC 客户端，支持 RoundRobin 负载均衡。

#### `StreamClient` / `RobustStreamClient`

基于 stream_stream 的客户端封装；`RobustStreamClient` 额外支持消息回调（`on_connect` / `on_message`）。

#### `GRPCEntry`

worker 侧入口，将 router 注册到 bridge。

#### `GRPCBridge`

gRPC 网桥，实现请求转发（转发时 `request_serializer` / `response_deserializer` 均为 `None`，即原始字节透传）。

配套：`copy_metadata(remote_context)` 复制对端调用元数据并补充 `remote-peer`；`BridgeUnaryStream` / `BridgeStreamUnary` / `BridgeStreamStream` 为三种流转发器，均提供 `join()`（等待流转结束）。

#### `GRPCMainProcess` / `GRPCWorker`

gRPC 多进程框架。

```python
from hagworm.third.grpc.server import Router, StreamHandler
from hagworm.third.grpc.bridge import GRPCMainProcess, GRPCWorker
from hagworm.third.grpc.client import GRPCClient

# 服务端
router = Router("demo")

@router.unary_unary
async def echo(request, context):
    return {"echo": request}


class EchoHandler(StreamHandler):

    async def on_message(self, data):
        await self.write(data)

    async def on_connect(self):
        pass

    def on_close(self, context):
        pass


@router.stream_stream
async def stream_echo(request, context):
    await EchoHandler(request, context).join()

if __name__ == "__main__":
    GRPCMainProcess(("0.0.0.0", 8080), [router], GRPCWorker, 2).run()

# 客户端
client = GRPCClient()
await client.open(["127.0.0.1:8080"])
result = await client.unary_unary("/demo/echo", {"msg": "hello"})
```

---

### hagworm.third.rabbitmq — RabbitMQ 消息队列

#### `RabbitMQConsumer` / `RabbitMQConsumerForExchange`

RabbitMQ 消费者。`RabbitMQConsumer(connection, queue_name)`；`RabbitMQConsumerForExchange(connection, queue_name, exchange_name)`。

| 方法 | 说明 |
|------|------|
| `open(*, consume_func=None, consume_no_ack=False, channel_qos_config=None, queue_config=None[, routing_key=None])` | 打开消费者（`ForExchange` 版额外支持 `routing_key`） |
| `block_pull(consume_func, consume_no_ack=False, wait_time=1)` | 阻塞式拉取消费 |
| `close(with_connection=False)` | 关闭消费者（`with_connection=True` 时同时关闭连接） |

#### `RabbitMQProducer` / `RabbitMQProducerForExchange`

RabbitMQ 生产者。`RabbitMQProducer(connection)`；`RabbitMQProducerForExchange(connection, exchange_name)`。

| 方法 | 说明 |
|------|------|
| `open(*, channel_number=None, publisher_confirms=True, on_return_raises=False)` | 打开生产者 |
| `open(*, exchange_type=aio_pika.ExchangeType.FANOUT, exchange_config=None, channel_number=None, publisher_confirms=True, on_return_raises=False)` | 打开交换机生产者（`ForExchange` 版） |
| `publish(message, routing_key='', **kwargs)` | 发布消息 |
| `close(with_connection=False)` | 关闭生产者 |

#### `RabbitMQProducerPool` / `RabbitMQProducerForExchangePool`

RabbitMQ 生产者连接池。`RabbitMQProducerPool(pool_size, connection)`；`RabbitMQProducerForExchangePool(pool_size, connection, exchange_name)`；提供 `open()` / `publish(message, routing_key='', **kwargs)` / `close()`。

#### `RpcServer` / `RpcClient`

基于 RabbitMQ 的 RPC 实现，支持 Unary 和 Stream 两种模式。

`RpcServer(pool_size, connection, exchange_name, *, server_name=None, message_ttl=60, max_priority=10)`：`open()` / `close()` 管理生命周期（内置 `ping`），`register(name, func)` 注册被调用函数。

`RpcClient(pool_size, connection, exchange_name, message_ttl=60)`：`open()` / `close()`；`ping(priority=0, server_name=None)`；`call(func_name, *, func_args=None, func_kwargs=None, priority=0, server_name=None, request_timeout=60)`；`call_stream(..., request_timeout=3600)`。

配套：`RpcServerError`（服务端业务异常）、`FunctionType`（`Unary` / `Stream`，标识函数模式）、`get_routing_key(server_name)`（生成路由键），模块常量 `STREAM_EOF` / `ERROR_HEAD`（流结束与错误帧标记）。

#### `create_connection`

`create_connection(url, timeout=None, **kwargs) -> aio_pika.RobustConnection`：创建 `RobustConnection` 并内部 `asyncio.create_task(connection.connect(timeout))` 自动发起连接；如需等待连接完成，仍可再 `await connection.connect()`。

```python
from hagworm.third.rabbitmq import create_connection
from hagworm.third.rabbitmq.publish import RabbitMQProducerForExchange
from hagworm.third.rabbitmq.consume import RabbitMQConsumerForExchange
import aio_pika

# 创建连接
connection = create_connection("amqp://guest:guest@localhost/")  # 内部已自动发起连接

# 生产者
producer = RabbitMQProducerForExchange(connection, "my_exchange")
await producer.open(exchange_type=aio_pika.ExchangeType.TOPIC)
await producer.publish(b"hello", "routing_key")

# 消费者
async def handle_message(message: aio_pika.IncomingMessage):
    print(message.body)
    await message.ack()

consumer = RabbitMQConsumerForExchange(connection, "my_queue", "my_exchange")
await consumer.open(consume_func=handle_message, routing_key="routing_key")
```

---

### hagworm.third.nacos — Nacos 服务发现

#### `NacosConfig`

Nacos 配置中心客户端，支持长轮询监听配置变更。

`NacosConfig(listener, servers, data_id, *, username='', password='', protocol='http', endpoint='', group='DEFAULT_GROUP', namespace='', request_timeout=..., pulling_timeout=...)`

| 方法 | 说明 |
|------|------|
| `start()` / `stop()` / `is_running()` | 启动 / 停止 / 运行状态 |
| `get_config()` | 获取当前配置（`(content, content_hash)`） |

配置变更回调在构造时以 `listener` 传入（获取到配置后回调），不提供 `add_listener` / `clear_listeners` 方法。

#### `NacosInstanceRegister`

Nacos 服务实例注册，自动发送心跳。

`NacosInstanceRegister(servers, service_name, service_ip, service_port, heartbeat_interval=10, *, username='', password='', protocol='http', endpoint='', group='DEFAULT_GROUP', namespace='', cluster=None, weight=None, request_timeout=...)`

#### `NacosInstanceQuery`

Nacos 服务实例查询，监听实例变更。

`NacosInstanceQuery(servers, service_name, listener_interval=10, *, username='', password='', protocol='http', endpoint='', group='DEFAULT_GROUP', namespace='', cluster=None, request_timeout=...)`

| 方法 | 说明 |
|------|------|
| `start()` / `stop()` / `is_running()` | 启动 / 停止 / 运行状态 |
| `get_host()` | 按 `weight` 加权随机返回单个实例 `{'ip', 'port', 'weight', 'metadata'}` |
| `get_hosts()` | 返回全部实例列表 |
| `add_listener(callback)` / `clear_listeners()` | 追加 / 清空实例变更回调 |

#### `Configure`（Nacos 版）

从 Nacos 配置中心加载配置：`Configure(servers, data_id, *, username='', password='', protocol='http', endpoint='', group='DEFAULT_GROUP', namespace='', data_format='yaml')`，提供 `open()` / `close()`。

`data_format` 支持 `text` / `json` / `yaml`（默认 `yaml`）；当 `servers` 为空且 `data_id` 为本地文件路径时，直接从该文件加载。

```python
from hagworm.third.nacos.client import NacosConfig, NacosInstanceRegister, NacosInstanceQuery
from hagworm.third.nacos.config import Configure

# 配置中心
def on_config_change(content):
    print(f"Config updated: {content}")

nacos_config = NacosConfig(
    on_config_change, "localhost:8848", "my-app.yaml",
    group="DEFAULT_GROUP", namespace="dev"
)
nacos_config.start()

# 服务注册
register = NacosInstanceRegister(
    "localhost:8848", "my-service", "10.0.0.1", 8080,
    namespace="dev"
)
register.start()

# 服务发现
query = NacosInstanceQuery("localhost:8848", "my-service", namespace="dev")
instance = query.get_host()  # 按 weight 加权随机返回 {'ip','port','weight','metadata'}
hosts = query.get_hosts()    # 全部实例
```

---

## 完整示例

### FastAPI 项目示例

参见 `example/fastapi_demo/` 目录，包含完整的 FastAPI 项目结构：

- [main.py](example/fastapi_demo/main.py) — 应用入口
- [setting.py](example/fastapi_demo/setting.py) — 配置定义
- [controller/home.py](example/fastapi_demo/controller/home.py) — 路由控制器
- [gunicorn.config.py](example/fastapi_demo/gunicorn.config.py) — gunicorn 部署配置
- [main.sh](example/fastapi_demo/main.sh) — 启动脚本（`gunicorn main:create_app -c gunicorn.config.py`）
- [service/__init__.py](example/fastapi_demo/service/__init__.py) — 数据源与业务基类
- [model/__init__.py](example/fastapi_demo/model/__init__.py) — 数据模型
- [build.sh](example/fastapi_demo/build.sh) — PYC 编译打包脚本
- [config/config.yaml](example/fastapi_demo/config/config.yaml) — 配置文件

### gRPC 示例

参见 `example/grpc_demo/` 目录：

- [server.py](example/grpc_demo/server.py) — gRPC 服务端
- [client.py](example/grpc_demo/client.py) — gRPC 客户端

### TCP 服务示例

参见 [example/main_tcp.py](example/main_tcp.py)：

```python
import asyncio

from hagworm.extend.asyncio.base import Utils
from hagworm.extend.asyncio.socket import AsyncTcpServer

async def connection_handle(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:

    client_info = writer.transport.get_extra_info(r'socket').getpeername()
    Utils.log.info(f'create connection: {client_info}')

    try:
        while True:
            request = await reader.readline()
            if not request:
                break
            writer.write(request)
            await writer.drain()
    except Exception as err:
        Utils.log.error(err)
    finally:
        if not writer.is_closing():
            writer.close()
            await writer.wait_closed()

    Utils.log.info(f'lost connection: {client_info}')

def main() -> None:
    server = AsyncTcpServer(connection_handle, (r'0.0.0.0', 8080))
    server.run()

if __name__ == r'__main__':
    main()
```

### RabbitMQ 示例

参见 [example/main_rabbitmq.py](example/main_rabbitmq.py)：

```python
import asyncio

import aio_pika

from hagworm.extend.asyncio.base import Utils, install_uvloop
from hagworm.third.rabbitmq import create_connection
from hagworm.third.rabbitmq.publish import RabbitMQProducerForExchange
from hagworm.third.rabbitmq.consume import RabbitMQConsumerForExchange


async def main() -> None:

    connection = create_connection(r'')
    await connection.connect()  # create_connection 内部已自动发起连接，此处可再显式等待

    await asyncio.gather(publish_run(connection), consume_run_0(connection), consume_run_1(connection))


async def publish_run(connection: aio_pika.RobustConnection) -> None:

    producer = RabbitMQProducerForExchange(connection, r'test1983')
    await producer.open(exchange_type=aio_pika.ExchangeType.TOPIC)

    await asyncio.sleep(5)

    for _idx in range(100):
        await producer.publish('content_{}'.format(_idx).encode(), 'test_{}'.format(_idx % 3))

    await asyncio.sleep(5)

    await connection.close()


async def consume_run_0(connection: aio_pika.RobustConnection) -> None:

    async def _consume_func(message: aio_pika.abc.AbstractIncomingMessage) -> None:
        print(r'test_0', message.body)
        await message.ack()

    consumer = RabbitMQConsumerForExchange(connection, r'test0', exchange_name='test1983')
    await consumer.open(routing_key=r'test_0')
    await consumer.block_pull(_consume_func)


async def consume_run_1(connection: aio_pika.RobustConnection) -> None:

    async def _consume_func(message: aio_pika.abc.AbstractIncomingMessage) -> None:
        print(r'test_1', message.body)
        await message.ack()

    consumer = RabbitMQConsumerForExchange(connection, r'test1', exchange_name='test1983')
    await consumer.open(routing_key=[r'test_1', r'test_2'])
    await consumer.block_pull(_consume_func)


if __name__ == r'__main__':

    Utils.print_slogan()

    install_uvloop()

    asyncio.run(main())
```

### 压力测试示例

参见 [example/main_test.py](example/main_test.py)：

```python
from hagworm.extend.base import TimerMS
from hagworm.extend.asyncio.base import Utils
from hagworm.frame.stress_tests import Launcher, RunnerAbstract

class Runner(RunnerAbstract):
    async def _execute(self) -> None:
        for index in range(5):

            for _ in range(5):

                with TimerMS(f"Test{index}") as timer:
                    await Utils.sleep(Utils.randint(10, 99) / 1000)

                if Utils.rand_hit([True, False], [50, 50]):
                    await self.success(f"Test{index}", timer.cost)
                else:
                    await self.failure(f"Test{index}", timer.cost)

if __name__ == "__main__":
    Launcher(Runner.create, 5).run()
```

---

## 单元测试

运行测试：

```bash
cd testing
pytest
```

`testing/setup.cfg` 提供 `asyncio_mode=auto` 的 pytest 配置（异步用例依赖该配置）；也可直接运行 `testing/main.py` 启动 `pytest.main()`。

测试覆盖模块：

- `test_extend_base.py` — 基础工具、异常处理、上下文管理器、FuncWrapper
- `test_extend_cache.py` — StackCache 缓存
- `test_extend_struct.py` — KeyLowerDict 数据结构
- `test_extend_asyncio_base.py` — 异步工具、ShareFuture、FuncCache、robust_iterator、事务
- `test_extend_asyncio_future.py` — ThreadWorker 线程转协程
- `test_extend_asyncio_task.py` — MultiTasks 并发、RateLimiter 限流
- `test_extend_asyncio_net.py` — HTTP 客户端

---

## 许可证

Apache-2.0

