Oopz 平台 Python SDK,提供 HTTP API、文件上传与 WebSocket 事件能力。
Project description
Oopz SDK
Oopz 平台 Python SDK,提供频道消息、私信、文件上传、平台查询与 WebSocket 实时事件能力。
v0.4 在 v0.3 的统一异常和结果模型基础上,继续补了三件事:
- 真实联调入口
smoke/smoke_test.py - 更统一的读接口重试与缓存回退语义
- 高频查询结果模型与更明确的 WebSocket 认证生命周期
安装
从 PyPI 安装
pip install oopz-sdk
从源码安装
pip install -e .
开发依赖:
pip install -e .[dev]
运行环境
- Python
3.10+ requestscryptographywebsocket-clientpillow
配置
创建 OopzConfig 时,以下字段为必填:
device_idperson_uidjwt_tokenprivate_key
如果你要依赖默认上下文发送频道消息,还需要配置:
default_areadefault_channel
当前稳定配置面还包括:
request_timeoutrate_limit_intervalarea_members_cache_ttlarea_members_stale_ttlquery_cache_ttlquery_cache_stale_ttl
from oopz import OopzConfig
config = OopzConfig(
device_id="你的设备ID",
person_uid="你的用户UID",
jwt_token="你的JWT Token",
private_key="-----BEGIN RSA PRIVATE KEY-----\n...\n-----END RSA PRIVATE KEY-----",
default_area="默认域ID",
default_channel="默认频道ID",
)
快速开始
发送消息
from oopz import MessageSendResult, OopzConfig, OopzSender
with OopzSender(config) as sender:
result: MessageSendResult = sender.send_message("Hello Oopz!")
print(result.message_id)
发送私信
from oopz import OopzSender
with OopzSender(config) as sender:
result = sender.send_private_message("目标UID", "你好")
print(result.channel, result.message_id)
上传并发图
from oopz import OopzSender, UploadResult
with OopzSender(config) as sender:
upload: UploadResult = sender.upload_file("demo.png", file_type="IMAGE", ext=".png")
print(upload.attachment.url)
sender.upload_and_send_image("demo.png", text="图片说明")
发布版最小示例
下面的示例只依赖公开导出接口,适合通过 pip install oopz-sdk 安装后的最小验证:
from oopz import OopzConfig, OopzSender
config = OopzConfig(
device_id="你的设备ID",
person_uid="你的用户UID",
jwt_token="你的JWT Token",
private_key="-----BEGIN RSA PRIVATE KEY-----\n...\n-----END RSA PRIVATE KEY-----",
default_area="默认域ID",
default_channel="默认频道ID",
)
with OopzSender(config) as sender:
result = sender.send_message("Hello from PyPI")
print(result.message_id)
WebSocket 实时事件
聊天消息回调接收 ChatMessageEvent,生命周期回调接收 LifecycleEvent。
v0.4 的生命周期状态包括:
connectingconnectedauth_sentauth_okauth_failedreconnectingclosederror
其中 auth_sent 表示已发起认证,auth_ok / auth_failed 才表示协议级认证结果。
from oopz import ChatMessageEvent, LifecycleEvent, OopzClient, OopzSender
sender = OopzSender(config)
def on_message(event: ChatMessageEvent) -> None:
if event.content.strip().lower() == "ping":
sender.send_message("pong", area=event.area, channel=event.channel)
def on_lifecycle(event: LifecycleEvent) -> None:
print(event.state, event.reason, event.error)
client = OopzClient(
config,
on_chat_message=on_message,
on_lifecycle_event=on_lifecycle,
)
client.start()
公开返回值
发送、上传和管理类接口返回:
MessageSendResultUploadResultUploadAttachmentPrivateSessionResultOperationResult
高频查询接口返回:
ChannelGroupsResultJoinedAreasResultSelfDetailPersonDetailChannelSettingVoiceChannelMembersResultDailySpeechResultAreaBlocksResultlist[ChannelMessage]
其中带缓存回退能力的读接口会通过 from_cache 明确标识缓存命中,而不是让调用方猜测:
get_area_membersget_area_channelsget_joined_areasget_self_detail
异常处理
异常层级:
OopzError
├─ OopzAuthError
├─ OopzConnectionError
└─ OopzApiError
└─ OopzRateLimitError
from oopz import OopzApiError, OopzConnectionError, OopzRateLimitError, OopzSender
try:
with OopzSender(config) as sender:
sender.send_message("test")
except OopzRateLimitError as exc:
print(f"被限流,建议 {exc.retry_after}s 后重试")
except OopzConnectionError as exc:
print(f"网络失败: {exc}")
except OopzApiError as exc:
print(f"请求失败: {exc} (HTTP {exc.status_code})")
示例
examples/ 目录包含:
send_message.py:最小发送消息reply_bot.py:接收消息并自动回复upload_private_image.py:上传图片并通过私信发送
真实联调
smoke/smoke_test.py 用于跑真实账号联调,不会进入默认 CI。
联调前准备
- 准备测试域和测试频道
- 准备可用的
device_id / person_uid / jwt_token / private_key - 如需私信联调,准备
OOPZ_TARGET_UID - 如需严格验证 WebSocket 收消息,准备第二个账号在等待窗口内发消息
环境变量
OOPZ_DEVICE_IDOOPZ_PERSON_UIDOOPZ_JWT_TOKENOOPZ_PRIVATE_KEY或OOPZ_PRIVATE_KEY_FILEOOPZ_AREA_IDOOPZ_CHANNEL_IDOOPZ_TARGET_UID可选OOPZ_SMOKE_IMAGE可选OOPZ_SMOKE_EXPECT_WS_MESSAGE=1可选OOPZ_SMOKE_WS_WAIT_SECONDS可选,默认20
执行方式
python smoke/smoke_test.py
通过标准
- 域列表、自身信息、频道列表、域成员查询成功
- 频道发消息和撤回成功
- 文件上传和发图成功
- WebSocket 至少完成
auth_ok - 提供
OOPZ_TARGET_UID时私信成功 - 设置
OOPZ_SMOKE_EXPECT_WS_MESSAGE=1时,等待窗口内收到真实消息事件
常见失败原因
jwt_token过期,导致 HTTP 或 WebSocket 认证失败private_key与账号不匹配,导致签名异常default_area/default_channel或环境变量填写错误- 给自己发私信,触发平台已知异常
- 仅用单账号联调时,WebSocket 默认无法验证“收到他人消息”这条链路
本地发布前检查
运行单元测试:
python -m pytest tests -q
构建源码包和 wheel:
python -m build
校验发布产物元数据和 README 渲染:
python -m twine check dist/*
可选:在干净虚拟环境验证 wheel 安装和导入:
pip install dist/oopz_sdk-0.4.3-py3-none-any.whl
python -c "import oopz; print(oopz.__version__)"
可选:先上传到 TestPyPI 做一次发布演练:
python -m twine upload --repository testpypi dist/*
版本变更
v0.3 -> v0.4变化见 CHANGELOG.md
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file oopz_sdk-0.4.3.tar.gz.
File metadata
- Download URL: oopz_sdk-0.4.3.tar.gz
- Upload date:
- Size: 37.3 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.2.0 CPython/3.12.3
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
c00ea5aaf2c5897e9ba19ef4e0561b5aaf38b49f4bdfe85a34bfd0814aeaedbc
|
|
| MD5 |
7a8d4b4c0e09fc7144bff4939516b5bc
|
|
| BLAKE2b-256 |
579cff7f550c7e61a1b0967d13606a475f71808930d75bd4c097dc98efb3dfb6
|
File details
Details for the file oopz_sdk-0.4.3-py3-none-any.whl.
File metadata
- Download URL: oopz_sdk-0.4.3-py3-none-any.whl
- Upload date:
- Size: 34.7 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.2.0 CPython/3.12.3
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
d5c5b63845c3ed400a71528d71d19f47d094682af869463688b5899c944eefbd
|
|
| MD5 |
81c3659d6e36dedf43e30ad82dbade18
|
|
| BLAKE2b-256 |
47f1c25744d9ef33b98c6458bde65603e68d4d756146135f8beca193d693c576
|