Python SDK API 说明

Ti 云存 API 位于 tirtc.storage：

import tirtc
import tirtc.storage as storage

Ti 云存 Client、CloudStorage、Replay 和 Export 独立于 TiRTC 连接，同时复用根 tirtc package 的 ClientOptions、Frame、媒体枚举、临时文件和异常。

Client、CloudStorage 与鉴权

class Client:
    def __init__(
        self,
        options: tirtc.ClientOptions,
        *,
        access_key_id: str | None = None,
        access_key_secret: str | None = None,
    ) -> None: ...
    def open_with_token(self, token: str) -> CloudStorage: ...
    def open_device(self, device_id: str) -> CloudStorage: ...
    def upload_logs(self) -> str: ...
    def close(self) -> None: ...

class CloudStorage:
    def update_token(self, token: str) -> None: ...
    def close(self) -> None: ...

ClientOptions.cache_dir 必须是可写绝对路径。省略 AK/SK 时 Client 使用 External Token 模式，通过 open_with_token() 接收业务服务端签发并绑定设备的短期 APP Access Token。同时提供 access_key_id 和 access_key_secret 时使用 Access Key 模式，通过 open_device() 打开目标设备，由 Runtime 签发和刷新短期 Token。

只提供一项 Access Key、传入空字符串或 NUL 会在创建 Native 资源前抛出 ValueError。两个鉴权模式的反向 open 方法抛出 InUseError。External Token 过期时，当前操作抛出 TokenExpiredError；调用 update_token() 后需要显式重做失败操作。Access Key 模式调用 update_token() 抛出 InUseError。

Client 可以多实例。Access Key Client 可以使用不同 App ID、AK/SK 和 Endpoint；同时活跃的 External Token Client 必须使用相同 App ID 和 Endpoint。一个进程中的所有 TiRTC 与 Ti 云存 Client 仍须共享 cache 目录和 console log 设置。

Client.close() 会取消自己的在途查询、Replay、Recording 和 Export，Detach 关联 Output，关闭 CloudStorage 树，并等待已经接受的回调退出。它不关闭应用独立持有的 Output 或已经返回的媒体文件。

查询日期与录像范围

@dataclass(frozen=True, slots=True)
class RecordingDay:
    date: date
    has_recording: bool

@dataclass(frozen=True, slots=True)
class RecordingRange:
    start_time: datetime
    end_time: datetime

def list_recording_days(
    self,
    start_date: date,
    end_date: date,
    *,
    timezone: str | ZoneInfo = "Asia/Shanghai",
    timeout: float | None = None,
) -> list[RecordingDay]: ...

def list_recordings(
    self,
    start_time: datetime,
    end_time: datetime,
    *,
    timeout: float | None = None,
) -> list[RecordingRange]: ...

日期查询的起止日期都包含，使用显式 IANA 时区，默认 Asia/Shanghai。录像范围使用 [start_time, end_time)，按开始时间升序返回，裁剪到请求窗口，并合并相邻或重叠范围；没有录像时返回空 list。

录像时间必须是精确到整毫秒的 aware datetime。timeout 单位为秒，None 表示没有调用方 deadline；非正数、NaN 和 Infinity 抛出 ValueError。同一 CloudStorage 同时只允许一个 List 请求，第二个请求抛出 InUseError。超时会取消 Native 请求并等待清理完成，再抛出 OperationTimeoutError。

Replay

class CloudStorage:
    def create_replay(
        self,
        *,
        on_time_changed=None,
        on_completed=None,
        on_error=None,
        on_recording_gap=None,
    ) -> Replay: ...

class Replay:
    def play(
        self,
        start_time: datetime,
        end_time: datetime,
        *,
        initial_time: datetime | None = None,
    ) -> None: ...
    def pause(self) -> None: ...
    def resume(self) -> None: ...
    def seek(self, target: datetime) -> None: ...
    def set_speed(self, speed: ReplaySpeed) -> None: ...
    @property
    def speed(self) -> ReplaySpeed: ...
    @property
    def current_time(self) -> datetime | None: ...
    def stop(self) -> None: ...
    def close(self) -> None: ...

播放范围使用 [start_time, end_time)。initial_time 和 seek() 目标必须位于该范围。倍速包括 0.125x、0.25x、0.5x、1x、2x、4x 和 8x。

自然播放结束只调用一次 on_completed。主动停止、被新 Play 替换或来源失败不会调用 completed；来源失败通过 on_error 报告。on_recording_gap 交付已经确认的录像缺口，Replay 可以继续播放缺口后的媒体。在 Replay 或关联 Output 的回调中同步调用 play()、seek()、stop() 或 close() 会抛出 InUseError。

Output 与 Frame

storage.AudioOutput、VideoOutput、EncodedAudioOutput 和 EncodedVideoOutput 分别交付解码或编码媒体帧。四类 Output 都提供 attach(replay, channel_id)、detach()、只读 state 和幂等 close()；VideoOutput 还提供 take_snapshot()。

Channel ID 范围是 0..255，音视频可以使用相同数值。Attach 只建立绑定，不转移所有权。Replay、CloudStorage 或 Client 关闭时会 Detach 相关 Output 并等待已接受的回调退出，Output 本身仍由应用关闭。

媒体数据是只读 memoryview。Cloud Frame 的 source_time 是录像 UTC 时间。Output 背压丢帧后，下一帧的 discontinuity 为 True。

播放中保存、独立导出与截图

class Replay:
    def start_recording(
        self, *, video_channel_id: int, audio_channel_id: int | None = None
    ) -> RecordingTask: ...

class RecordingTask:
    def stop(self) -> tirtc.RecordingFile: ...

class CloudStorage:
    def export_recording(
        self,
        start_time: datetime,
        end_time: datetime,
        *,
        video_channel_id: int,
        audio_channel_id: int | None = None,
        on_progress: Callable[[ExportProgress], None] | None = None,
        on_recording_gap: Callable[[RecordingGap], None] | None = None,
    ) -> ExportTask: ...

class ExportTask:
    @property
    def progress(self) -> float: ...
    @property
    def progress_detail(self) -> ExportProgress: ...
    @property
    def report(self) -> ExportReport: ...
    def cancel(self) -> None: ...
    def wait(self, *, timeout: float | None = None) -> tirtc.RecordingFile: ...
    def stop(self) -> tirtc.RecordingFile: ...

RecordingTask 只保存创建后实际进入当前 Replay 的所选媒体。ExportTask 独立读取指定录像范围，不需要 Replay 或 Output。wait(timeout=...) 超时时只停止当前等待，Task 仍继续运行，可以再次等待；cancel() 发出取消请求后，仍须调用 wait() 收取终局和报告。

完整和部分可播放的文件都可以成功返回。使用 task.report.complete 判断是否完整覆盖；报告还提供 requested_range、covered_duration_ms、segments、gaps、unprocessed_ranges、termination 和 cause。即使终局抛出异常，仍可以读取已知报告。

Snapshot、Recording 和 Export 返回根 package 的 SnapshotFile 或 RecordingFile。文件位于 SDK cache。应在文件 context 内检查报告并复制到业务目录；离开 context 会删除临时文件。

原始音视频诊断采集

Replay 可以按 Channel ID 采集用于问题排查的原始音视频数据：

options = storage.RawDumpOptions(
    audio_channel_ids=(audio_channel_id,),
    video_channel_ids=(video_channel_id,),
)

with replay.start_raw_dump(options) as capture:
    # 复现问题后停止采集。
    archive = capture.stop()

两个 Channel ID 元组至少填写一个。停止后检查 RawDumpArchive 的 capture_complete、stop_reason、unsaved_packet_count、unsaved_byte_count 和 empty，再调用 Client.upload_logs() 将采集文件随下一次日志上传提交。完整流程、限制和隐私要求见接入客户端诊断能力。

释放、异常与进程

类型和值域错误分别使用 TypeError 和 ValueError。Native 错误与调用方 deadline 使用根 package 的 TiRTCError 体系。不要解析异常文本或 Runtime 日志来驱动业务逻辑。

正常释放顺序是完成 RecordingTask 和 ExportTask、Detach 并关闭 Output、停止并关闭 Replay、关闭 CloudStorage、关闭 Client。Client 也会级联收拢自己的资源树。在当前 Client 或 opaque Cloud task 回调中同步关闭 Client，会在 teardown 前抛出 InUseError。

import 不创建线程、目录、网络连接或 Runtime 状态。父进程只 import 后 fork，子进程可以创建自己的 Client；任一 TiRTC 或 Ti 云存 Client 激活后再 fork，子进程首次调用 SDK 会抛出 UnsupportedError(name="forked_process")。

覆盖查询、Replay、Output、截图、播放中保存和独立导出的可运行程序见 Python API 示例。