卡片模块

卡片模块提供了钉钉卡片消息的功能，包括AI卡片和私有数据构建器。

基础类

BaseCard

卡片基类。

class BaseCard:
    data: Optional[dict[str, str]] = None
    
    def __init__(self):
        pass

MarkDown

Markdown字符串类。

class MarkDown(str):
    pass

私有数据构建器

PrivateDataBuilder

私有数据构建器，用于构建卡片的私有数据。

class PrivateDataBuilder:
    def __init__(self, data: dict[str, dict]):
        """
        初始化私有数据构建器
        
        Args:
            data: 私有数据字典
        """

方法

to_json(): 转换为JSON格式
solve(): 解析数据

使用示例

from dingraia.card import PrivateDataBuilder

# 创建私有数据
private_data = PrivateDataBuilder({
    "user1": {"name": "张三", "age": 25},
    "user2": {"name": "李四", "age": 30}
})

# 转换为JSON格式
json_data = private_data.to_json()
# 结果:
# {
#     "user1": {"cardParamMap": {"name": "张三", "age": 25}},
#     "user2": {"cardParamMap": {"name": "李四", "age": 30}}
# }

AI卡片

AICard

AI卡片类，用于发送AI流式卡片消息。

class AICard(BaseCard):
    response: Optional[Union[SizedIterable[str], list]] = None
    """流式回复内容，必须是可以被迭代的"""
    
    content_type: Literal["auto", "full", "stream"] = None
    """流式回复内容的类型，默认为auto自动判断"""
    
    _texts: List[str] = []
    """流式回复内容的列表"""
    
    text: str = ""
    """最新的完整文本"""
    
    def __init__(self):
        super().__init__()
        self.content_type: str = "auto"

方法

set_response

设置流式回复内容。

def set_response(
    self,
    response: Union[SizedIterable[Union[str, MarkDown]], Generator, AsyncGenerator],
    content_type: Literal["auto", "full", "stream"] = "auto"
):
    """
    设置流式回复内容
    
    Args:
        response: 回复内容，可以是字符串、Generator、AsyncGenerator
        content_type: response输出内容的方式，默认为auto自动判断
            - auto: 自动判断
            - stream: 每次追加
            - full: 每次全量
    """

streaming_string

流式输出字符串。

async def streaming_string(self, length_limit: int = 0, full_content: bool = True):
    """
    流式输出
    
    Args:
        length_limit: 最小输出长度，默认为0，即不限制
        full_content: 是否输出完整文本，默认为True
        
    Yields:
        str: 流式输出的字符串
    """

completed_string

获取完整的字符串。

async def completed_string(self) -> str:
    """
    获取完整的字符串
    
    Returns:
        str: 完整的字符串
    """

withPostUrl

从POST请求获取流式回复。

def withPostUrl(
    self, 
    post_url: str, 
    json: dict, 
    data_handler: Callable[[dict], Optional[str]], 
    headers: dict = None,
    timeout: Optional[float] = None
):
    """
    从POST请求获取流式回复
    
    Args:
        post_url: POST请求URL
        json: 请求体
        data_handler: 数据处理函数
        headers: 请求头
        timeout: 超时时间
    """

属性

data: 卡片数据，返回 {"content": self.text}

使用示例

基本使用

from dingraia.card import AICard, MarkDown

# 创建AI卡片
card = AICard()

# 设置流式回复内容
async def generate_response():
    yield "你好"
    yield "，我是"
    yield "AI助手"

card.set_response(generate_response())

# 流式输出
async for text in card.streaming_string():
    print(text)

使用字符串列表

from dingraia.card import AICard

card = AICard()

# 设置字符串列表
card.set_response(["你好", "，我是", "AI助手"], content_type="stream")

# 获取完整字符串
full_text = await card.completed_string()
print(full_text)  # 输出: 你好，我是AI助手

从POST请求获取回复

from dingraia.card import AICard

card = AICard()

# 定义数据处理函数
def data_handler(data: dict) -> str:
    return data.get("content", "")

# 从POST请求获取流式回复
card.withPostUrl(
    post_url="https://api.example.com/chat",
    json={"message": "你好"},
    data_handler=data_handler,
    headers={"Authorization": "Bearer token"},
    timeout=30.0
)

发送AI卡片消息

from dingraia.DingTalk import Dingtalk
from dingraia.element import OpenConversationId
from dingraia.card import AICard

app = Dingtalk()
app.prepare()

# 创建AI卡片
card = AICard()

# 设置流式回复
async def generate_response():
    for i in range(10):
        yield f"消息 {i}\n"

card.set_response(generate_response())

# 发送AI卡片
target = OpenConversationId("cid123456")
template_id = "your_template_id"

await app.send_ai_card(
    target=target,
    cardTemplateId=template_id,
    card=card,
    update_limit=0.5,  # 更新频率限制（秒）
    key="content"  # 卡片数据在模板中的键名
)

内容类型

auto

自动判断内容类型：

如果内容是累积的（每次包含之前的内容），则使用 full 模式
如果内容是增量的（每次只包含新增内容），则使用 stream 模式

full

全量模式：每次更新都发送完整的内容。

适用于：

AI模型返回完整响应
内容更新频率较低

stream

流式模式：每次更新只发送新增的内容。

适用于：

AI模型流式返回响应
打字机效果
实时更新

最佳实践

选择合适的content_type
- 如果不确定，使用 auto
- 如果AI模型返回完整响应，使用 full
- 如果AI模型流式返回响应，使用 stream
设置合理的更新频率
- 使用 update_limit 参数控制更新频率
- 避免过于频繁的更新导致性能问题
处理异常情况
- 捕获流式处理中的异常
- 设置合理的超时时间
- 提供降级方案
优化用户体验
- 使用 streaming_string 的 length_limit 参数
- 避免过于频繁的更新
- 提供加载状态提示