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 快速安装：

uv add py_chanjet_toolkit

或从源码安装：

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

或者使用 pip 安装：

pip install py_chanjet_toolkit

或从源码安装：

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

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

# 使用 uv
uv sync --dev

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

快速开始

ZKHB SOAP Web Service - GetDataSet (同步)

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 (异步)

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) - 收费账单项目

使用示例：

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) 缴费周期分组
• 在每个周期内，查找正数（缴费）和负数（退费）且金额绝对值相等的配对
• 将配对的数据全部排除
• 未配对的数据保留

使用示例：

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

开发

# 使用 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