Metadata-Version: 2.4
Name: multi-proto-agent
Version: 0.1.5
Summary: 用统一的 “Player” 抽象管理 TCP / WebSocket（WS/WSS）连接、请求发送、响应接收与断言流程。
Author: 石峰
Author-email: 330550742@qq.com
Requires-Python: >=3.9
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: beautifulsoup4>=4.12.3
Requires-Dist: gevent>=25.9.1
Requires-Dist: jsonpath-ng>=1.7.0
Requires-Dist: paramiko>=4.0.0
Requires-Dist: protobuf>=5.28.3
Requires-Dist: pytest>=8.3.3
Requires-Dist: PyYAML>=6.0.2
Requires-Dist: requests>=2.32.3
Requires-Dist: websockets>=12.0
Requires-Dist: xmindparser>=1.0.9
Dynamic: author
Dynamic: author-email
Dynamic: description
Dynamic: description-content-type
Dynamic: license-file
Dynamic: requires-dist
Dynamic: requires-python
Dynamic: summary

# Multi-Proto-Agent

多协议通信工具库，面向自动化测试与高并发连接场景。  
用统一的 `Player` 抽象管理 TCP / WebSocket（WS/WSS）连接、请求发送、响应接收与断言流程。

---

## 1. 适合谁用

- 需要在 Python 中统一管理 TCP/WS 长连接
- 需要可插拔的 protobuf 报文打包/解包策略
- 需要在测试框架里做响应断言与参数提取

---

## 2. 安装

```bash
pip install multi-proto-agent
```

或源码安装：

```bash
git clone https://gitee.com/flinttina/multi-proto-agent.git
cd multi-proto-agent
pip install -e .
```

---

## 3. 快速开始

```python
from multi_proto_agent.utils.player import Player
from multi_proto_agent.utils.protocol_type import ProtocolType

player = Player(
    account_id="u001",
    protocol_type=ProtocolType.WS,  # 也支持 "ws"
)

player.set_ap_address("ws://127.0.0.1:20000")
player.set_secret_key("your_secret")
player.connect()

# req_obj 为 protobuf 消息对象
# ok = player.send_request("Demo.ReqPing", req_obj)

rsp_data = player.receive_response()  # 默认返回 bytes
print(rsp_data)

player.tear_down()
```

---

## 4. 核心概念

### 4.1 协议类型

```python
from multi_proto_agent.utils.protocol_type import ProtocolType

ProtocolType.TCP
ProtocolType.WS
ProtocolType.WSS
```

`Player` 支持传字符串或枚举：

- `Player(protocol_type="ws")`
- `Player(protocol_type=ProtocolType.WS)`

### 4.2 编解码策略（关键）

框架默认协议中立，不绑定业务包头：

- 未传 `msg_packer`：发送裸 protobuf bytes
- 未传 `msg_unpacker`：接收原始 bytes

有业务外层协议时（如项目自定义 BaseMsg），在初始化注入：

```python
player = Player(
    account_id="u001",
    protocol_type=ProtocolType.WS,
    msg_packer=my_packer,
    msg_unpacker=my_unpacker,
)
```

策略对象约定：

- `pack(proto_msg_name, obj_to_send, trace_id="") -> bytes`
- `unpack(rsp_data: bytes) -> Any`

---

## 5. 对外 API（稳定优先）

### 5.1 `player.py`

- `Player(account_id, role_name=None, role_uid=None, protocol_type='ws', heartbeat_interval=30, heartbeat_class_name="", heartbeat_obj=None, msg_packer=None, msg_unpacker=None)`
  - 作用：创建连接与收发控制对象，统一管理 TCP/WS/WSS 通信、心跳和编解码策略。
  - 入参：
    - `account_id`：账号标识（必填）。
    - `role_name` / `role_uid`：角色信息（可选）。
    - `protocol_type`：协议类型，支持 `tcp` / `ws` / `wss` 或 `ProtocolType` 枚举。
    - `heartbeat_interval`：心跳发送间隔（秒）。
    - `heartbeat_class_name` / `heartbeat_obj`：心跳消息名与对象（配置后自动心跳）。
    - `msg_packer` / `msg_unpacker`：自定义打包/解包策略（可选）。
  - 示例：
    ```python
    from multi_proto_agent.utils.player import Player
    from multi_proto_agent.utils.protocol_type import ProtocolType

    player = Player(account_id="u001", protocol_type=ProtocolType.WS)
    ```

- `connect(max_retry_times=3)`
  - 作用：根据 `protocol_type` 建立连接，并初始化响应队列；连接成功返回 `True`。
  - 入参：
    - `max_retry_times`：连接失败后的最大重试次数，默认 `3`。
  - 示例：
    ```python
    player.set_ap_address("ws://127.0.0.1:20000")
    player.set_secret_key("your_secret")
    player.connect(max_retry_times=5)
    ```

- `send_request(req_msg_name, req_obj, is_heartbeat=False)`
  - 作用：发送请求消息；非心跳请求会自动生成并记录 `trace_id`。
  - 入参：
    - `req_msg_name`：protobuf 消息名（如 `Demo.ReqPing`）。
    - `req_obj`：protobuf 请求对象实例。
    - `is_heartbeat`：是否为心跳消息，默认 `False`。
  - 示例：
    ```python
    ok = player.send_request("Demo.ReqPing", req_obj)
    ```

- `receive_response()`
  - 作用：从内部响应队列非阻塞读取一条消息；队列为空返回 `None`。
  - 入参：无。
  - 示例：
    ```python
    rsp_data = player.receive_response()
    if rsp_data is not None:
        print("got response")
    ```

- `tear_down()`
  - 作用：关闭连接并清理资源（含心跳任务）。
  - 入参：无。
  - 示例：
    ```python
    player.tear_down()
    ```

- `set_ap_address(ap_address)`
  - 作用：设置服务端地址（连接前必设）。
  - 入参：
    - `ap_address`：地址字符串；TCP 格式通常为 `ip:port`，WS/WSS 为 URL。
  - 示例：
    ```python
    player.set_ap_address("127.0.0.1:9000")  # tcp
    ```

- `set_secret_key(secret_key)`
  - 作用：设置连接鉴权所需密钥（连接前必设）。
  - 入参：
    - `secret_key`：密钥字符串。
  - 示例：
    ```python
    player.set_secret_key("my_secret_key")
    ```

### 5.2 `translator.py`

- `handle_send_data(proto_msg_name, obj_to_send, trace_id="", packer=None)`
  - 作用：将请求对象转为可发送的 bytes；若传入 `packer` 则走自定义打包。
  - 入参：
    - `proto_msg_name`：消息名（如 `Demo.ReqPing`）。
    - `obj_to_send`：protobuf 对象。
    - `trace_id`：链路标识（可选）。
    - `packer`：实现 `pack(...)` 的策略对象（可选）。
  - 示例：
    ```python
    req_bytes = handle_send_data("Demo.ReqPing", req_obj)
    ```

- `handle_rsp_data(rsp_data, unpacker=None)`
  - 作用：处理响应数据；未传 `unpacker` 时原样返回 bytes。
  - 入参：
    - `rsp_data`：原始响应 bytes。
    - `unpacker`：实现 `unpack(...)` 的策略对象（可选）。
  - 示例：
    ```python
    rsp_obj = handle_rsp_data(rsp_bytes, unpacker=my_unpacker)
    ```

- `deserialize_proto_object(proto_msg_name, bytes_data)`
  - 作用：按消息名动态反序列化 bytes 为 protobuf 对象。
  - 入参：
    - `proto_msg_name`：消息名（模块名需在环境变量中可映射到 protobuf 包）。
    - `bytes_data`：待反序列化的二进制数据。
  - 示例：
    ```python
    proto_obj = deserialize_proto_object("Demo.RspPing", rsp_bytes)
    ```

- `get_string_of_req_obj(req_obj)`
  - 作用：将 protobuf 对象转为 JSON 字符串，便于日志打印。
  - 入参：
    - `req_obj`：protobuf 对象。
  - 示例：
    ```python
    print(get_string_of_req_obj(req_obj))
    ```

### 5.3 `assert_util.py`

- `is_player_passed(player, assertion_dict, trace_id="", trace_desc="", max_timeout=5, is_isNotGot_assertion=False)`
  - 作用：消费 `player` 的响应队列并执行断言规则，返回 `AssertResult`（是否通过、提取值、断言信息）。
  - 入参：
    - `player`：`Player` 实例。
    - `assertion_dict`：断言配置（按消息名组织 `assertion_rule_list` / `value_to_get_list`）。
    - `trace_id` / `trace_desc`：断言失败时上报追踪信息（可选）。
    - `max_timeout`：最大等待时长（秒）。
    - `is_isNotGot_assertion`：是否启用 `isNotGot` 语义控制（通常默认即可）。
  - 示例：
    ```python
    result = is_player_passed(player, assertion_dict, max_timeout=10)
    print(result.get_is_player_pass(), result.get_value_dict())
    ```

- `get_supported_assertions()`
  - 作用：返回当前支持的断言类型列表（如 `==`、`in`、`isNotNone`、`isGot` 等）。
  - 入参：无。
  - 示例：
    ```python
    print(get_supported_assertions())
    ```

### 5.4 `config_util.py`

- `set_config(env_type, config_path=None)`
  - 作用：从环境配置文件读取指定环境并写入环境变量。
  - 入参：
    - `env_type`：环境标识（如 `test`、`prod`）。
    - `config_path`：环境配置 YAML 路径。
  - 示例：
    ```python
    set_config(env_type="test", config_path="./config/env_config.yaml")
    ```

- `set_protos_config(config_path=None)`
  - 作用：读取 protobuf 相关配置并写入环境变量（用于动态反序列化映射）。
  - 入参：
    - `config_path`：`protos_config.yaml` 路径。
  - 示例：
    ```python
    set_protos_config("./config/protos_config.yaml")
    ```

- `add_python_protos_to_path(python_protos_path=None)`
  - 作用：将 protobuf 生成目录及其子目录加入 `sys.path`，便于动态导入。
  - 入参：
    - `python_protos_path`：protobuf Python 文件根目录。
  - 示例：
    ```python
    add_python_protos_to_path("./python_protos")
    ```

- `find_root_directory(start_file=None, root_marker_file="i_am_the_root.md")`
  - 作用：向上查找项目根目录（通过根标记文件定位）。
  - 入参：
    - `start_file`：起始路径或文件（默认当前工作目录）。
    - `root_marker_file`：根标记文件名，默认 `i_am_the_root.md`。
  - 示例：
    ```python
    root_dir = find_root_directory(__file__)
    ```

### 5.5 其他常用工具

- `trace_id_util.py`
  - `generate_trace_id()`
    - 作用：生成追踪 ID（十六进制字符串）。
    - 入参：无。
    - 示例：`trace_id = generate_trace_id()`
  - `send_trace_id(url, trace_id, trace_desc)`
    - 作用：将 trace 信息上报到指定 URL（GET 请求）。
    - 入参：`url` 上报地址，`trace_id` 追踪 ID，`trace_desc` 描述信息。
    - 示例：`resp = send_trace_id(report_url, trace_id, "断言失败")`

- `random_util.py`
  - 作用：提供随机文本/随机数值生成能力（用于构造测试数据）。
  - 示例：
    ```python
    # 具体函数名以 random_util.py 中导出为准
    from multi_proto_agent.utils import random_util
    ```

- `logger_config.py`
  - `setup_logging(level=None, log_dir="./logs")`
    - 作用：初始化日志系统（控制台 + 文件）。
    - 入参：`level` 日志级别，`log_dir` 日志目录。
    - 示例：`setup_logging(level="INFO", log_dir="./logs")`
  - `get_logger(name=None)`
    - 作用：获取 logger 实例；未配置时会自动初始化。
    - 入参：`name` logger 名称（可选）。
    - 示例：`logger = get_logger(__name__)`

- `protocol_type.py`
  - `ProtocolType`
    - 作用：协议类型枚举，包含 `TCP` / `WS` / `WSS`。
    - 示例：`ProtocolType.WS`
  - `ProtocolType.from_value(value)`
    - 作用：将字符串或枚举值规范化为 `ProtocolType`。
    - 入参：`value`，如 `"ws"` 或 `ProtocolType.WS`。
    - 示例：`pt = ProtocolType.from_value("wss")`

---

## 6. 常见使用模式

### 6.1 只做通信（最简）

- 使用 `Player + ProtocolType`
- 不注入 `msg_packer/msg_unpacker`
- 自己处理收到的裸 bytes

### 6.2 业务协议封装（推荐）

- 项目侧实现 `packer/unpacker`
- 在 `Player` 初始化注入
- 断言和日志统一处理解包后的消息

### 6.3 自动化断言

- 用 `is_player_passed(...)` 消费 `player.rsp_queue`
- 通过 `assertion_dict` 声明断言规则与提取字段

---

## 7. FAQ

### Q1: 为什么断言拿不到字段？

通常因为未注入 `msg_unpacker`，响应仍是裸 bytes。  
请先在 `Player(...)` 注入项目侧解包策略。

### Q2: 协议类型大小写敏感吗？

不敏感。`ProtocolType.from_value()` 会统一归一化并校验。

### Q3: 何时直接用 `TcpClient/WebSocketClient`？

当你需要绕过 `Player`，做更底层的连接控制时。常规场景优先用 `Player`。

---

## 8. 文档索引

- 使用者文档：当前文件 `README.md`
- 开发者文档：`README.dev.md`（内部结构、分层、关键实现）

---

## 9. 许可证

MIT License

