cogvideo 业务集成 (#44)

Author: glide-the

tests/integration_tests/test_videos.py

@@ -0,0 +1,43 @@
+from zhipuai import ZhipuAI
+import zhipuai
+
+import logging
+import logging.config
+
+
+def test_videos(logging_conf):
+    logging.config.dictConfig(logging_conf)  # type: ignore
+    client = ZhipuAI()  # 填写您自己的APIKey
+    try:
+        response = client.videos.generations(
+            model="cogvideo",
+            prompt="一些相信光的人，举着奥特曼"
+        )
+        print(response)
+
+
+
+    except zhipuai.core._errors.APIRequestFailedError as err:
+        print(err)
+    except zhipuai.core._errors.APIInternalError as err:
+        print(err)
+    except zhipuai.core._errors.APIStatusError as err:
+        print(err)
+
+
+def test_retrieve_videos_result(logging_conf):
+    logging.config.dictConfig(logging_conf)  # type: ignore
+    client = ZhipuAI()  # 填写您自己的APIKey
+    try:
+        response = client.videos.retrieve_videos_result(
+            id="1014908828144164571910908"
+        )
+        print(response)
+
+
+    except zhipuai.core._errors.APIRequestFailedError as err:
+        print(err)
+    except zhipuai.core._errors.APIInternalError as err:
+        print(err)
+    except zhipuai.core._errors.APIStatusError as err:
+        print(err)

zhipuai/_client.py

@@ -60,6 +60,7 @@ def __init__(
         self.batches = api_resource.Batches(self)
         self.knowledge = api_resource.Knowledge(self)
         self.tools = api_resource.Tools(self)
+        self.videos = api_resource.Videos(self)
 
     @property
     @override

zhipuai/api_resource/__init__.py

@@ -27,7 +27,11 @@
 from .tools import (
     Tools
 )
+from .videos import (
+    Videos,
+)
 __all__ = [
+    'Videos',
     'AsyncCompletions',
     'Chat',
     'Completions',

zhipuai/api_resource/videos/__init__.py

@@ -0,0 +1,7 @@
+from .videos import (
+    Videos,
+)
+__all__ = [
+    'Videos',
+
+]

zhipuai/api_resource/videos/videos.py

@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+
+from typing import TYPE_CHECKING, List, Mapping, cast, Optional, Dict
+from typing_extensions import Literal
+
+from ...types.video import video_create_params
+from ...types.video import VideoObject
+from ...core import BaseAPI, maybe_transform
+from ...core import NOT_GIVEN, Body, Headers, NotGiven
+
+import httpx
+
+from ...core import (
+    make_request_options,
+)
+from ...core import deepcopy_minimal, extract_files
+
+if TYPE_CHECKING:
+    from ..._client import ZhipuAI
+
+__all__ = ["Videos"]
+
+
+class Videos(BaseAPI):
+
+    def __init__(self, client: "ZhipuAI") -> None:
+        super().__init__(client)
+
+    def generations(
+            self,
+            model: str,
+            prompt: str,
+            *,
+            request_id: str = None,
+            extra_headers: Headers | None = None,
+            extra_body: Body | None = None,
+            timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
+    ) -> VideoObject:
+
+        if not model and not model:
+            raise ValueError("At least one of `model` and `prompt` must be provided.")
+        body = deepcopy_minimal(
+            {
+                "model": model,
+                "prompt": prompt,
+                "request_id": request_id,
+            }
+        )
+        return self._post(
+            "/videos/generations",
+            body=maybe_transform(body, video_create_params.VideoCreateParams),
+            options=make_request_options(
+                extra_headers=extra_headers, extra_body=extra_body, timeout=timeout
+            ),
+            cast_type=VideoObject,
+        )
+
+    def retrieve_videos_result(
+            self,
+            id: str,
+            *,
+            extra_headers: Headers | None = None,
+            extra_body: Body | None = None,
+            timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
+    ) -> VideoObject:
+
+        if not id:
+            raise ValueError("At least one of `id` must be provided.")
+
+        return self._get(
+            f"/async-result/{id}",
+            options=make_request_options(
+                extra_headers=extra_headers, extra_body=extra_body, timeout=timeout
+            ),
+            cast_type=VideoObject,
+        )

zhipuai/types/video/__init__.py

@@ -0,0 +1,7 @@
+
+from .video_object import (
+    VideoObject,
+    VideoResult
+)
+
+__all__ = ["VideoObject", "VideoResult"]

zhipuai/types/video/video_create_params.py

@@ -0,0 +1,16 @@
+from __future__ import annotations
+
+from typing import List, Optional
+
+from typing_extensions import Literal, Required, TypedDict
+
+__all__ = ["VideoCreateParams"]
+
+
+class VideoCreateParams(TypedDict, total=False):
+    model: str
+    """模型编码"""
+    prompt: str
+    """所需视频的文本描述"""
+    request_id: str
+    """由用户端传参，需保证唯一性；用于区分每次请求的唯一标识，用户端不传时平台会默认生成。"""

zhipuai/types/video/video_object.py

@@ -0,0 +1,28 @@
+from typing import Optional, List
+
+from ...core import BaseModel
+
+__all__ = ["VideoObject", "VideoResult"]
+
+
+class VideoResult(BaseModel):
+    url: str
+    """视频url"""
+
+
+class VideoObject(BaseModel):
+    id: Optional[str] = None
+    """智谱 AI 开放平台生成的任务订单号，调用请求结果接口时请使用此订单号"""
+
+    model: str
+    """模型名称"""
+
+    video_result: List[VideoResult]
+    """视频生成结果"""
+
+    task_status: str
+    """处理状态，PROCESSING（处理中），SUCCESS（成功），FAIL（失败）
+    注：处理中状态需通过查询获取结果"""
+
+    request_id: str
+    """用户在客户端请求时提交的任务编号或者平台生成的任务编号"""