Metadata-Version: 2.4
Name: py-qunjielong-toolkit
Version: 1.0.0
Summary: 企业微信 SDK，提供企业微信 API 的 Python 封装，支持同步和异步调用。
Author-email: Guolei <174000902@qq.com>
Maintainer-email: Guolei <174000902@qq.com>
License: MIT License
        
        Copyright (c) 2026 郭磊
        
        Permission is hereby granted, free of charge, to any person obtaining a copy
        of this software and associated documentation files (the "Software"), to deal
        in the Software without restriction, including without limitation the rights
        to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
        copies of the Software, and to permit persons to whom the Software is
        furnished to do so, subject to the following conditions:
        
        The above copyright notice and this permission notice shall be included in all
        copies or substantial portions of the Software.
        
        THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
        IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
        FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
        AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
        LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
        OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
        SOFTWARE.
        
Project-URL: Homepage, https://gitee.com/guolei19850528/py_wecom_toolkit
Project-URL: Repository, https://gitee.com/guolei19850528/py_wecom_toolkit.git
Project-URL: Documentation, https://gitee.com/guolei19850528/py_wecom_toolkit
Keywords: wecom,python,client,api,企业微信,异步调用,server,消息推送,文件上传,webhook,机器人
Classifier: License :: OSI Approved :: MIT License
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
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: Operating System :: OS Independent
Classifier: Topic :: Communications
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: pydantic>=2.0
Requires-Dist: diskcache>=5.6.3
Requires-Dist: redis>=4.6.0
Requires-Dist: py-httpx-toolkit>=1.0.1
Provides-Extra: dev
Requires-Dist: pytest>=7.0; extra == "dev"
Requires-Dist: pytest-cov>=4.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.21.0; extra == "dev"
Requires-Dist: setuptools>=61.0; extra == "dev"
Requires-Dist: twine>=4.0; extra == "dev"
Requires-Dist: build>=1.0; extra == "dev"
Dynamic: license-file

# py-qunjielong-toolkit

群接龙开放平台 Python SDK，提供便捷的 API 调用方式，支持同步和异步模式，内置令牌缓存机制。

## 官方文档

群接龙开放平台 API 文档：[https://open-api-doc.qunjielong.com/8481217m0](https://open-api-doc.qunjielong.com/8481217m0)

## 安装

```bash
pip install py-qunjielong-toolkit
```

```bash
uv add py-qunjielong-toolkit
```

## 依赖包

| 依赖包 | 版本 | 说明 |
| :--- | :--- | :--- |
| httpx | ^0.27.0 | HTTP 客户端，支持同步和异步 |
| py-httpx-toolkit | ^1.0.0 | HTTP 工具包 |
| pydantic | ^2.0.0 | 数据模型验证 |
| jsonpath-ng | ^1.5.3 | JSONPath 表达式解析 |
| jsonschema | ^4.20.0 | JSON Schema 验证 |
| diskcache | ^5.6.0 | 本地磁盘缓存（可选） |
| redis | ^5.0.0 | Redis 缓存（可选） |

## 快速开始

### 基本使用

```python
from py_qunjielong_toolkit.open import Open

# 初始化客户端
client = Open(secret="your_secret_key")

# 刷新访问令牌
client.refresh_access_token()

# 调用 API
response = client.ghome_getGhomeInfo()
print(response.json())
```

### 异步使用

```python
import asyncio
from py_qunjielong_toolkit.open import Open

async def main():
    client = Open(secret="your_secret_key")
    await client.async_refresh_access_token()
    response = await client.async_ghome_getGhomeInfo()
    print(response.json())

asyncio.run(main())
```

### 使用缓存

```python
import diskcache
from py_qunjielong_toolkit.open import Open

# 使用 diskcache
cache = diskcache.Cache("./cache")
client = Open(
    secret="your_secret_key",
    cache_config={
        "instance": cache,
        "expire": 7100  # 缓存过期时间（秒）
    }
)

# 或使用 Redis
import redis
redis_client = redis.Redis(host="localhost", port=6379, db=0)
client = Open(
    secret="your_secret_key",
    cache_config={
        "instance": redis_client,
        "expire": 7100
    }
)
```

## API 说明

### Open 类

#### 初始化参数

| 参数 | 类型 | 默认值 | 说明 |
| :--- | :--- | :--- | :--- |
| base_url | Optional[str] | "https://openapi.qunjielong.com" | API 基础地址 |
| secret | Optional[str] | None | 企业密钥 |
| cache_config | Optional[dict] | None | 缓存配置 |
| client_kwargs | Optional[dict] | None | HTTP 客户端配置 |

#### 主要方法

| 方法 | 说明 |
| :--- | :--- |
| `auth_token()` | 获取访问令牌（同步） |
| `ghome_getGhomeInfo()` | 获取企业/组织信息（同步） |
| `refresh_access_token()` | 刷新访问令牌（同步） |
| `request_with_access_token()` | 带令牌的通用请求（同步） |
| `async_auth_token()` | 获取访问令牌（异步） |
| `async_ghome_getGhomeInfo()` | 获取企业/组织信息（异步） |
| `async_refresh_access_token()` | 刷新访问令牌（异步） |
| `async_request_with_access_token()` | 带令牌的通用请求（异步） |

### auth_token

获取访问令牌接口（同步），调用 `/open/auth/token` 接口获取 access_token。

```python
response = client.auth_token()
print(response.json())
```

### ghome_getGhomeInfo

获取企业/组织信息接口（同步），调用 `/open/api/ghome/getGhomeInfo` 接口。

```python
response = client.ghome_getGhomeInfo()
print(response.json())
```

### refresh_access_token

刷新访问令牌（同步），支持缓存机制。

```python
client.refresh_access_token()
```

### request_with_access_token

带令牌的通用请求方法（同步），自动将 access_token 添加到请求参数中。

```python
response = client.request_with_access_token(
    method="GET",
    url="/open/api/custom/endpoint",
    params={"param1": "value1"}
)
print(response.json())
```

### async_auth_token

获取访问令牌接口（异步版本）。

```python
response = await client.async_auth_token()
print(response.json())
```

### async_ghome_getGhomeInfo

获取企业/组织信息接口（异步版本）。

```python
response = await client.async_ghome_getGhomeInfo()
print(response.json())
```

### async_refresh_access_token

刷新访问令牌（异步版本）。

```python
await client.async_refresh_access_token()
```

### async_request_with_access_token

带令牌的通用请求方法（异步版本）。

```python
response = await client.async_request_with_access_token(
    method="POST",
    url="/open/api/custom/endpoint",
    json={"key": "value"}
)
print(response.json())
```

## Utils 工具函数

### json_find_first

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

```python
from py_qunjielong_toolkit.open.utils import json_find_first

data = {"result": {"items": [1, 2, 3]}}
value = json_find_first("$.result.items[0]", data)
print(value)  # 输出: 1
```

### json_is_valid

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

```python
from py_qunjielong_toolkit.open.utils import json_is_valid

schema = {"type": "object", "properties": {"name": {"type": "string"}}}
data = {"name": "test"}
print(json_is_valid(schema, data))  # 输出: True
```

### success_is_valid

校验 API 响应是否成功（code 是否为 200）。

```python
from py_qunjielong_toolkit.open.utils import success_is_valid

print(success_is_valid({"code": 200}))    # 输出: True
print(success_is_valid({"code": "200"}))  # 输出: True
print(success_is_valid({"code": "400"}))  # 输出: False
```

### build_success_instance

将 HTTP 响应或字典转换为 Success 响应模型。

```python
from py_qunjielong_toolkit.open.utils import build_success_instance

result = build_success_instance({"code": 200, "message": "ok", "data": "xxx"})
print(result.code)  # 输出: 200
```

## 响应模型

### Base

基础响应模型：
- `code`: 错误码（整数或字符串）
- `message`: 错误信息（可选）

### Success

成功响应模型：
- `code`: 固定为 200

## 缓存机制

支持两种缓存方式：

### diskcache（本地缓存）

```python
import diskcache

cache = diskcache.Cache("./cache")
client = Open(
    secret="your_secret_key",
    cache_config={
        "instance": cache,
        "key": "custom_cache_key",
        "expire": 7100
    }
)
```

### Redis（分布式缓存）

```python
import redis

redis_client = redis.Redis(host="localhost", port=6379, db=0)
client = Open(
    secret="your_secret_key",
    cache_config={
        "instance": redis_client,
        "key": "custom_cache_key",
        "expire": 7100
    }
)
```

## 示例代码

### 完整示例

```python
from py_qunjielong_toolkit.open import Open
from py_qunjielong_toolkit.open.utils import success_is_valid

# 初始化客户端
client = Open(
    secret="your_secret_key",
    client_kwargs={
        "timeout": 30,
        "verify": True
    }
)

try:
    # 刷新令牌
    client.refresh_access_token()
    
    # 调用接口
    response = client.ghome_getGhomeInfo()
    
    # 校验响应
    if success_is_valid(response):
        data = response.json()
        print("企业信息:", data)
    else:
        print("请求失败:", response.json())
        
except Exception as e:
    print(f"发生错误: {e}")
```

### 异步完整示例

```python
import asyncio
from py_qunjielong_toolkit.open import Open
from py_qunjielong_toolkit.open.utils import success_is_valid

async def main():
    client = Open(secret="your_secret_key")
    
    try:
        await client.async_refresh_access_token()
        response = await client.async_ghome_getGhomeInfo()
        
        if success_is_valid(response):
            data = response.json()
            print("企业信息:", data)
        else:
            print("请求失败:", response.json())
            
    except Exception as e:
        print(f"发生错误: {e}")

asyncio.run(main())
```

## 项目主页

[https://gitee.com/guolei19850528/py_qunjielong_toolkit](https://gitee.com/guolei19850528/py_qunjielong_toolkit)

## 作者

- **作者**: Guolei
- **邮箱**: 174000902@qq.com

## 许可证

MIT License
