Metadata-Version: 2.4
Name: py_chanjet_toolkit
Version: 3.1.0
Summary: 畅捷通 T+ Python 扩展包，提供 T+ 云 API 调用封装
Author-email: guolei <174000902@qq.com>
License: MIT
Project-URL: Homepage, https://gitee.com/guolei19850528/py_chanjet_toolkit
Keywords: chanjet,tplus,chanjet_toolkit,畅捷通,tplus_toolkit
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Intended Audience :: Developers
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.8
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: httpx>=0.24.0
Requires-Dist: pydantic>=2.0
Provides-Extra: dev
Requires-Dist: pytest>=7.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.23.0; extra == "dev"
Requires-Dist: python-dotenv>=1.0.0; extra == "dev"
Dynamic: license-file

# py-chanjet-toolkit

畅捷通 T+ API Python 工具包

**作者**：郭磊  
**邮箱**：17490002@qq.com  
**项目主页**：https://pypi.org/project/py-chanjet-toolkit/

## 功能特性

- **T+ 模块** (`py_chanjet_toolkit.tplus`) - T+ 云 API 客户端（预留模块）
- **ZKHB 模块** (`py_chanjet_toolkit.tplus.zkhb`) - ZKHB SOAP Web Service 客户端
  - 基于 `{base_url}{endpoint_path}` (需自行配置)
  - 实现 `GetDataSet` 接口（执行 SQL 查询返回 DataSet）
  - 支持 SOAP 1.1 和 SOAP 1.2
  - 基于 httpx + pydantic 实现
  - 提供同步客户端 `ZKHBClient` 和异步客户端 `AsyncZKHBClient`
  - 提供 SQL 构建函数 `build_charge_fee_items_query_sql`
  - 提供数据过滤函数 `filter_charge_fee_items_data`

## 安装

推荐使用 [uv](https://github.com/astral-sh/uv) 快速安装：

```bash
uv add py_chanjet_toolkit
```

或从源码安装：

```bash
git clone https://gitee.com/guolei19850528/py_chanjet_toolkit.git
cd py_chanjet_toolkit
uv sync
```

或者使用 pip 安装：

```bash
pip install py_chanjet_toolkit
```

或从源码安装：

```bash
git clone https://gitee.com/guolei19850528/py_chanjet_toolkit.git
cd py_chanjet_toolkit
pip install -e .
```

开发模式安装（含测试依赖）：

```bash
# 使用 uv
uv sync --dev

# 或使用 pip
pip install -e ".[dev]"
```

## 快速开始

### ZKHB SOAP Web Service - GetDataSet (同步)

```python
from py_chanjet_toolkit.tplus.zkhb import ZKHBClient, ZKHBClientConfig

# 方式 1: 使用自定义配置
config = ZKHBClientConfig(
    base_url="http://your-server:port",
    endpoint_path="/estate/webService/ForcelandEstateService.asmx",
    timeout=30.0,
    soap_version="1.1",  # 或 "1.2"
)

with ZKHBClient(config=config) as client:
    result = client.get_dataset("SELECT * FROM ChargeFee")
    if result.ok and result.data:
        for row in result.data.to_dict_list():
            print(row)
```

### ZKHB SOAP Web Service - GetDataSet (异步)

```python
import asyncio
from py_chanjet_toolkit.tplus.zkhb import AsyncZKHBClient, ZKHBClientConfig

config = ZKHBClientConfig(
    base_url="http://your-server:port",
    endpoint_path="/estate/webService/ForcelandEstateService.asmx",
    timeout=30.0,
    soap_version="1.1",  # 或 "1.2"
)

async def async_example():
    async with AsyncZKHBClient(config=config) as client:
        result = await client.get_dataset("SELECT * FROM ChargeFee")
        if result.ok and result.data:
            for row in result.data.to_dict_list():
                print(row)

asyncio.run(async_example())
```

### 返回结果说明

`get_dataset()` 返回 `ZKHBDataSetResponse` 对象，包含：
- `ok`: bool - 是否成功
- `data`: ZKHBGetDataSetResult - 结果数据（查询失败时为 None）

`ZKHBGetDataSetResult` 提供：
- `to_dict_list()` - 将 DataSet XML 转换为字典列表
- `schema_xml`: str - 表结构定义 XML
- `data_xml`: str - 原始数据 XML

## ZKHB 模块 API

### ZKHBClientConfig
配置类，支持以下参数：
- `base_url`: SOAP 服务基础地址（如 `http://your-server:port`），无默认值，需配置
- `endpoint_path`: 服务端点路径（如 `/estate/webService/ForcelandEstateService.asmx`），无默认值，需配置
- `timeout`: 请求超时时间（秒），默认 30.0
- `verify_ssl`: 是否验证 SSL 证书，默认 False
- `soap_version`: SOAP 版本，"1.1" 或 "1.2"，默认 "1.1"
- `default_op`: 默认操作名，默认 "GetDataSet"

### ZKHBClient (同步客户端)
主要客户端类：
- `get_dataset(sql, url)` - 执行 SQL 查询获取 DataSet
  - `sql`: SQL 查询语句
  - `url`: 可选的 URL 参数，默认为空
- `close()` - 关闭底层 httpx 客户端
- 支持上下文管理器 (`with` 语句)

### AsyncZKHBClient (异步客户端)
异步客户端类（基于 `httpx.AsyncClient`）：
- `get_dataset(sql, url)` - 异步执行 SQL 查询获取 DataSet
  - `sql`: SQL 查询语句
  - `url`: 可选的 URL 参数，默认为空
- `close()` - 异步关闭底层 httpx 客户端
- 支持异步上下文管理器 (`async with` 语句)

### ZKHBGetDataSetResult
结果数据类：
- `to_dict_list()` - 转换为字典列表
- `schema_xml`: str - 表结构定义 XML
- `data_xml`: str - 原始数据 XML

### ZKHBDataSetResponse
完整响应类：
- `ok`: bool - 是否成功
- `data`: ZKHBGetDataSetResult - 结果数据（查询失败时为 None）

### SQL 构建工具函数

#### `build_charge_fee_items_query_sql(columns, conditions, order_by)`
生成查询实际支付项目的 SQL 语句，用于物业收费多表关联查询。

**参数：**
- `columns`: str - 额外查询列，添加到默认列前面（默认空）
- `conditions`: str - 查询条件，拼接到 WHERE 后（默认空）
- `order_by`: str - 排序方式（默认 `"order by cfi.ChargeFeeItemID"`）

**返回：**
- `str` - 完整 SQL 语句

**默认查询表关联：**
- `chargeMasterList` (cml) - 收费主列表（主表）
- `EstateDetail` (ed) - 小区详情
- `ChargeFeeItem` (cfi) - 收费项目
- `RoomDetail` (rd) - 房间详情
- `ChargeBillItem` (cbi) - 收费账单项目

**使用示例：**
```python
from py_chanjet_toolkit.tplus.zkhb import build_charge_fee_items_query_sql, ZKHBClient

# 生成查询某小区收费记录的 SQL
sql = build_charge_fee_items_query_sql(
    conditions="and cml.EstateID='E001' and cml.ChargeTime >= '2024-01-01'"
)

with ZKHBClient("http://your-server:port", "/estate/webService/ForcelandEstateService.asmx") as client:
    result = client.get_dataset(sql)
    for row in result.data.to_dict_list():
        print(row)
```

### 数据过滤工具函数

#### `filter_charge_fee_items_data(records)`
过滤收费数据，排除相同缴费周期且缴费和退费金额相等的数据。

**参数：**
- `records`: List[Dict] - 包含 ActualAmount、SDate、EDate 字段的记录列表

**返回：**
- `List[Dict]` - 过滤后的记录列表

**逻辑说明：**
- 按 (SDate, EDate) 缴费周期分组
- 在每个周期内，查找正数（缴费）和负数（退费）且金额绝对值相等的配对
- 将配对的数据全部排除
- 未配对的数据保留

**使用示例：**
```python
from py_chanjet_toolkit.tplus.zkhb import filter_charge_fee_items_data

raw_data = [
    {'ChargeMListID': '1', 'SDate': '2026-03-15', 'EDate': '2026-09-15', 'ActualAmount': '600.00'},
    {'ChargeMListID': '2', 'SDate': '2026-03-15', 'EDate': '2026-09-15', 'ActualAmount': '-600.00'},
    {'ChargeMListID': '3', 'SDate': '2026-03-15', 'EDate': '2026-09-15', 'ActualAmount': '600.00'},
]

filtered = filter_charge_fee_items_data(raw_data)
# 第1条和第2条被排除（同周期正负相等），第3条保留
print(len(filtered))  # 输出 2
```

## 示例代码

### 同步客户端示例 (`examples/tplus/zkhb/example_client_usage.py`)

演示从 `.env` 加载配置、创建客户端、使用上下文管理器等用法。

### 异步客户端示例 (`examples/tplus/zkhb/example_async_client_usage.py`)

演示异步客户端的创建、配置、上下文管理器等用法。

### SQL 构建示例 (`examples/tplus/zkhb/example_sql_build.py`)

演示如何使用 `build_charge_fee_items_query_sql` 构建各种查询 SQL。

### 完整工作流示例

- `examples/tplus/zkhb/example_full_workflow.py` - 同步完整工作流（配置加载 + SQL 构建 + 数据过滤 + 调用服务）
- `examples/tplus/zkhb/example_async_full_workflow.py` - 异步完整工作流（配置加载 + SQL 构建 + 数据过滤 + 异步调用服务）

所有示例均支持从当前目录的 `.env` 文件加载配置。

## 依赖

- Python >= 3.8
- httpx >= 0.24.0
- pydantic >= 2.0.0

## 开发

```bash
# 使用 uv
uv sync --dev
uv run pytest tests/tplus/zkhb/ -v

# 或使用 pip
pip install -e ".[dev]"
pytest tests/tplus/zkhb/ -v

# 代码格式化
ruff check --fix .

# 类型检查
mypy src/
```

## 许可证

MIT License
