Python SDK 接入
Python SDK 可以查询录像日期和时间段、回放多路录像、取得解码或编码媒体帧、截图、保存播放片段,并把指定范围独立导出为 MP4。它支持业务服务签发的 External Token,也支持受控环境中的 Access Key 托管鉴权。
确认运行环境
- CPython 3.11 或更高的常规 GIL 版本;
- Apple Silicon,macOS 11.5 或更高版本;
- Linux x86_64,glibc 2.35 或更高版本,不支持 Alpine/musl;
- 不支持 Windows、macOS x86_64、Linux arm64、free-threaded Python 或 subinterpreter。
安装 SDK
在虚拟环境中安装已发布的二进制 wheel:
python -m pip install --only-binary=:all: tirtc==VERSION将 VERSION 替换为更新记录中需要接入的版本。
如果 pip 报告找不到匹配版本,先用 python -VV 和 python -c "import platform; print(platform.machine())" 检查 Python 实现、版本和 CPU 架构。
选择鉴权方式
External Token 适合由业务服务端为目标设备签发短期 APP Access Token 的应用。SDK 不持有应用长期密钥:
from pathlib import Path
import tirtc
import tirtc.storage as storage
options = tirtc.ClientOptions(
app_id=app_id,
cache_dir=Path("/absolute/path/to/tirtc-cache"),
)
with storage.Client(options) as client:
with client.open_with_token(app_access_token) as cloud:
# 在这里查询、回放或导出录像。
...Access Key 只用于能够妥善保管应用密钥的可信进程。Runtime 会为指定设备签发和刷新短期 Token:
with storage.Client(
options,
access_key_id=access_key_id,
access_key_secret=access_key_secret,
) as client:
with client.open_device(device_id) as cloud:
...不要把 Access Key 写进源码、命令行、日志、浏览器、移动 App 或交付给终端用户的程序。External Token 过期后,从业务服务取得同一设备的新 Token,调用 cloud.update_token(new_token),再重新发起失败的操作。Access Key 模式由 Runtime 管理 Token,不调用 update_token()。
app_id 和可写的绝对 cache_dir 必填。普通接入省略 endpoint,使用 SDK 内置的正式服务地址;只有开通时收到其他地址,才把完整地址传给 ClientOptions(endpoint=...)。同一进程中的所有 TiRTC 与 Ti 云存 Client 必须使用相同的 cache 目录和 console log 设置。
查询并导出一段录像
时间参数必须是带时区的 datetime,录像查询和导出使用左闭右开区间 [start_time, end_time):
from datetime import datetime, timedelta, timezone
from pathlib import Path
import shutil
end = datetime.now(timezone.utc)
start = end - timedelta(hours=1)
ranges = cloud.list_recordings(start, end, timeout=30)
if not ranges:
raise RuntimeError("查询时间内没有录像")
selected = ranges[-1]
task = cloud.export_recording(
selected.start_time,
selected.end_time,
video_channel_id=11,
)
with task.wait(timeout=900) as recording:
if not task.report.complete:
raise RuntimeError(f"录像导出不完整:{task.report}")
shutil.copyfile(recording.path, Path("recording.mp4"))video_channel_id 必须与设备上传录像时使用的 Channel 一致,范围是 0..255。需要同时导出声音时再传 audio_channel_id;同一个数值可以同时承载一路视频和一路音频。录像查询不会返回 Channel 映射,应从设备上传配置或业务服务取得。
快速验证今天的录像导出
下面的 Smoke 使用 Access Key 模式查询 Asia/Shanghai 的当天录像,并尝试把最后一个可用时间段中的 30 秒视频导出为 MP4。示例使用视频 Channel 11;如果设备配置不同,请替换 VIDEO_CHANNEL_ID。
mkdir tirtc-python-storage-smoke
cd tirtc-python-storage-smoke
python -m venv .venv
source .venv/bin/activate
python -m pip install --only-binary=:all: tirtc==VERSION通过环境变量传入凭据,值不会出现在进程命令行中:
export TI_CLOUD_STORAGE_APP_ID='<app-id>'
export TI_CLOUD_STORAGE_ACCESS_KEY_ID='<access-key-id>'
export TI_CLOUD_STORAGE_ACCESS_KEY_SECRET='<access-key-secret>'
export TI_CLOUD_STORAGE_DEVICE_ID='<device-id>'创建 export_today.py:
from datetime import datetime, timedelta
import os
from pathlib import Path
import shutil
from zoneinfo import ZoneInfo
import tirtc
import tirtc.storage as storage
VIDEO_CHANNEL_ID = 11
root = Path.cwd()
cache_dir = (root / ".cache").resolve()
output_dir = (root / "output").resolve()
cache_dir.mkdir(parents=True, exist_ok=True)
output_dir.mkdir(parents=True, exist_ok=True)
shanghai = ZoneInfo("Asia/Shanghai")
current = datetime.now(shanghai)
now = current.replace(microsecond=(current.microsecond // 1000) * 1000)
today_start = now.replace(hour=0, minute=0, second=0, microsecond=0)
options = tirtc.ClientOptions(
app_id=os.environ["TI_CLOUD_STORAGE_APP_ID"],
cache_dir=cache_dir,
endpoint=os.environ.get("TI_CLOUD_STORAGE_ENDPOINT") or None,
)
last_error = None
with storage.Client(
options,
access_key_id=os.environ["TI_CLOUD_STORAGE_ACCESS_KEY_ID"],
access_key_secret=os.environ["TI_CLOUD_STORAGE_ACCESS_KEY_SECRET"],
) as client:
with client.open_device(os.environ["TI_CLOUD_STORAGE_DEVICE_ID"]) as cloud:
ranges = cloud.list_recordings(today_start, now, timeout=30)
if not ranges:
raise RuntimeError("今天没有可导出的录像")
for selected in reversed(ranges):
export_start = selected.start_time
export_end = min(selected.end_time, export_start + timedelta(seconds=30))
task = cloud.export_recording(
export_start,
export_end,
video_channel_id=VIDEO_CHANNEL_ID,
)
try:
recording = task.wait(timeout=900)
except tirtc.OperationTimeoutError:
recording = task.stop()
except tirtc.TiRTCError as error:
last_error = error
continue
destination = output_dir / f"today-channel-{VIDEO_CHANNEL_ID}.mp4"
with recording:
shutil.copyfile(recording.path, destination)
report = task.report
print(
f"saved={destination} duration={recording.duration} "
f"complete={report.complete} gaps={len(report.gaps)} "
f"unprocessed={len(report.unprocessed_ranges)}"
)
break
else:
raise RuntimeError(
f"今天的录像范围没有 Channel {VIDEO_CHANNEL_ID} 的可播放视频"
) from last_error运行并检查 MP4 是否含有可解析、正时长的视频流:
python export_today.py
ffprobe -v error \
-show_entries format=duration:stream=codec_type,codec_name \
-of json output/today-channel-11.mp4report.complete 为 False 表示这段 MP4 存在已经确认的来源缺口或未处理范围。Smoke 可以用 ffprobe 验证实际得到的可播放片段;产品流程应继续读取 gaps 和 unprocessed_ranges,按业务要求决定是否接收部分录像。
若提示今天没有录像,先确认设备当天已上传录像、系统时间正确,并检查 App ID、设备归属和查询时区。若录像范围存在但没有 MP4,检查实际视频 Channel。凭据或设备归属错误时不要持续重试。
需要同时验证日期查询、Replay、四类 Output、截图、播放中保存和完整范围导出时,运行公共 Python API 示例。该示例把 report.complete == False 视为门禁失败,适合验证已知完整的测试录像。
接入客户端诊断能力(推荐)
创建 Client 后,建议在开发阶段为应用预留日志上传入口。这样在需要平台协助诊断时,可以直接上传同一次复现的日志;入口设计、上传结果处理和可选的原始音视频诊断采集见接入客户端诊断能力。
选择后续能力
- 查询、回放、Output、截图和播放中保存:看查询与播放录像;
- 独立导出和完整性报告:看下载录像;
- 对象、回调、临时文件和异常语义:看 Python SDK API 说明。
优先使用 with 关闭 Client、CloudStorage 和临时文件。Client 关闭时会收拢自己的查询、Replay、Recording 和 Export;已经返回给应用的临时媒体文件仍由应用复制到业务目录或删除。