py-qunjielong-toolkit

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

官方文档

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

安装

pip install py-qunjielong-toolkit

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 缓存（可选）

快速开始

基本使用

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

异步使用

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

使用缓存

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。

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

ghome_getGhomeInfo

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

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

refresh_access_token

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

client.refresh_access_token()

request_with_access_token

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

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

async_auth_token

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

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

async_ghome_getGhomeInfo

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

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

async_refresh_access_token

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

await client.async_refresh_access_token()

async_request_with_access_token

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

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 表达式从数据中查找第一个匹配项。

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。

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）。

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 响应模型。

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（本地缓存）

import diskcache

cache = diskcache.Cache("./cache")
client = Open(
    secret="your_secret_key",
    cache_config={
        "instance": cache,
        "key": "custom_cache_key",
        "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,
        "key": "custom_cache_key",
        "expire": 7100
    }
)

示例代码

完整示例

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}")

异步完整示例

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

作者

• 作者: Guolei
• 邮箱: 174000902@qq.com

许可证

MIT License