py-chajet-toolkit

畅捷通 T+ 系列产品的 Python 客户端库，提供与畅捷通 T+ 系统的 API 交互能力。

功能特性

• 提供畅捷通 T+ 智慧社区 WebService 接口的同步和异步调用
• 封装 SOAP 请求的构建和响应解析
• 提供数据校验、日期处理、SQL 生成等辅助工具函数
• 支持 JSONPath 查询和 JSON Schema 校验

安装方式

使用 pip 安装

pip install py-chajet-toolkit

使用 uv 安装

uv add py-chajet-toolkit

快速开始

基础使用示例

from py_chajet_toolkit.tplus.zkhb import ForcelandEstateService
from py_chajet_toolkit.tplus.zkhb.utils import (
    build_get_data_set_post_xml,
    build_get_data_set_results,
    build_actual_payment_items_query_sql
)

# 初始化服务客户端
service = ForcelandEstateService(base_url="http://your-server/estate")

# 构建 SOAP 请求体
xml_data = build_get_data_set_post_xml({"sql": "SELECT * FROM EstateDetail"})

# 同步调用接口
response = service.get_data_set(content=xml_data)

# 解析响应数据
results = build_get_data_set_results(response)
print(results)

异步调用示例

import asyncio
from py_chajet_toolkit.tplus.zkhb import ForcelandEstateService
from py_chajet_toolkit.tplus.zkhb.utils import build_get_data_set_post_xml, build_get_data_set_results

async def main():
    service = ForcelandEstateService(base_url="http://your-server/estate")
    xml_data = build_get_data_set_post_xml({"sql": "SELECT * FROM ChargeMasterList"})
    
    # 异步调用接口
    response = await service.async_get_data_set(content=xml_data)
    results = build_get_data_set_results(response)
    print(results)

asyncio.run(main())

API 说明

ForcelandEstateService

畅捷通T+智慧社区服务客户端类，封装了与畅捷通T+智慧社区系统的WebService接口交互。

初始化

service = ForcelandEstateService(
    base_url="http://your-server/estate",
    client_kwargs={"timeout": 120}
)

同步方法

• get_data_set(client=None, client_kwargs=None, **kwargs) - 调用 GetDataSet 接口

异步方法

• async_get_data_set(client=None, client_kwargs=None, **kwargs) - 异步调用 GetDataSet 接口

Utils 工具函数

json_find_first

使用 JSONPath 表达式从嵌套的 JSON 数据中查找第一个匹配项。

from py_chajet_toolkit.tplus.zkhb.utils import json_find_first

data = {"users": [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]}
result = json_find_first("$.users[0].name", data)
# 返回: 'Alice'

json_is_valid

校验 JSON 数据是否符合指定的 JSON Schema 规范。

from py_chajet_toolkit.tplus.zkhb.utils import json_is_valid

schema = {
    "type": "object",
    "properties": {"id": {"type": "string"}},
    "required": ["id"]
}
is_valid = json_is_valid(schema, {"id": "123"})
# 返回: True

build_get_data_set_post_xml

将请求数据转换为 SOAP GetDataSet 接口所需的 XML 格式。

from py_chajet_toolkit.tplus.zkhb.utils import build_get_data_set_post_xml

xml = build_get_data_set_post_xml({"sql": "SELECT * FROM EstateDetail"})

build_get_data_set_results

从 GetDataSet 接口响应中提取实际数据列表。

from py_chajet_toolkit.tplus.zkhb.utils import build_get_data_set_results

results = build_get_data_set_results(response)

build_actual_payment_items_query_sql

生成查询实际支付项目的 SQL 语句。

from py_chajet_toolkit.tplus.zkhb.utils import build_actual_payment_items_query_sql

sql = build_actual_payment_items_query_sql(
    conditions="and cml.EstateID='E001'",
    order_by="order by cml.CreateTime desc"
)

payment_date_range_is_valid

验证支付金额与日期范围是否匹配。

from py_chajet_toolkit.tplus.zkhb.utils import payment_date_range_is_valid

is_valid = payment_date_range_is_valid(
    daily_fee=10,
    total_amount=300,
    start="2024-01-01",
    end="2024-01-31"
)
# 返回: True

filter_actual_payment_items

过滤实际支付记录，移除配对的充值和退款记录。

from py_chajet_toolkit.tplus.zkhb.utils import filter_actual_payment_items

filtered_items = filter_actual_payment_items(actual_payment_items)

项目结构

py_chajet_toolkit/
├── tplus/
│   ├── zkhb/
│   │   ├── __init__.py    # 智慧社区服务客户端类
│   │   ├── utils.py       # 工具函数集合
│   │   └── responses.py   # 响应处理模块
│   └── __init__.py
└── __init__.py

依赖

• arrow - 日期时间处理
• httpx - HTTP 客户端
• jsonpath_ng - JSONPath 查询
• jsonschema - JSON Schema 校验
• xmltodict - XML 与字典转换
• py_httpx_toolkit - HTTP 工具封装

项目主页

https://gitee.com/guolei19850528/py_chanjet_toolkit

作者

Author: Lei Guo
Email: guolei@example.com

许可证

MIT License