notifications

feishu.integrations.notifications

WorkflowNotificationIntegration dataclass

Mount a generic service-authenticated notification route on a Feishu gateway.

源代码位于： feishu/integrations/notifications.py

@dataclass(frozen=True)
class WorkflowNotificationIntegration:
    r"""Mount a generic service-authenticated notification route on a Feishu gateway."""

    path: str = "/notifications"
    store: EventMessageStore | None = None
    text_summarizer: TextSummarizer | None = None
    card_builder: NotificationCardBuilder | None = None
    max_summary_chars: int = 1200

    def routes(self, context: GatewayContext) -> list[Route]:
        return [
            create_notification_route(
                context.config,
                context.client,
                path=self.path,
                store=self.store,
                text_summarizer=self.text_summarizer,
                card_builder=self.card_builder,
                max_summary_chars=self.max_summary_chars,
            )
        ]

create_notification_route

Python

create_notification_route(config: GatewayConfig, client: Any, *, path: str = '/notifications', store: EventMessageStore | None = None, text_summarizer: TextSummarizer | None = None, card_builder: NotificationCardBuilder | None = None, max_summary_chars: int = 1200) -> Route

Create a service-authenticated workflow notification route.

源代码位于： feishu/integrations/notifications.py

def create_notification_route(
    config: GatewayConfig,
    client: Any,
    *,
    path: str = "/notifications",
    store: EventMessageStore | None = None,
    text_summarizer: TextSummarizer | None = None,
    card_builder: NotificationCardBuilder | None = None,
    max_summary_chars: int = 1200,
) -> Route:
    r"""Create a service-authenticated workflow notification route."""
    return Route(
        path,
        _notification_endpoint(
            config,
            client,
            store=store,
            text_summarizer=text_summarizer,
            card_builder=card_builder,
            max_summary_chars=max_summary_chars,
        ),
        methods=["POST"],
    )

build_notification_card

Python

build_notification_card(payload: dict[str, Any], summary: str) -> dict[str, Any]

Render a normalized workflow/status notification as a Feishu card.

源代码位于： feishu/integrations/notifications.py

def build_notification_card(payload: dict[str, Any], summary: str) -> dict[str, Any]:
    r"""Render a normalized workflow/status notification as a Feishu card."""
    title = _text(payload.get("title") or payload.get("name") or "流程通知")
    status = _text(payload.get("status"))
    source = _text(payload.get("source") or payload.get("system"))
    template = _template(status)
    title_parts = [part for part in (status.upper() if status else "", title) if part]
    card = Card().header(" - ".join(title_parts), template=template).markdown(summary or title)

    metadata = []
    event_id = notification_event_id(payload)
    if event_id:
        metadata.append(f"**ID**: `{event_id}`")
    if source:
        metadata.append(f"**Source**: {source}")
    if status:
        metadata.append(f"**Status**: {status}")
    if metadata:
        card.divider().markdown("\n".join(metadata))

    links = _links(payload.get("links"))
    url = _text(payload.get("url") or payload.get("source_url"))
    if url:
        links = [*links, {"title": "Open source", "url": url}]
    if links:
        card.markdown("\n".join(f"[{item['title']}]({item['url']})" for item in links[:5]))
    return card.to_dict()

notification_event_id

Python

notification_event_id(payload: Mapping[str, Any]) -> str

Return the stable notification ID used for card upserts.

源代码位于： feishu/integrations/notifications.py

def notification_event_id(payload: Mapping[str, Any]) -> str:
    r"""Return the stable notification ID used for card upserts."""
    for key in ("event_id", "idempotency_key", "notification_id", "id"):
        value = _text(payload.get(key))
        if value:
            return value
    source = _text(payload.get("source") or payload.get("system"))
    title = _text(payload.get("title") or payload.get("name"))
    status = _text(payload.get("status"))
    return "/".join(part for part in (source, title, status) if part)

notification_summary async

Python

notification_summary(payload: Mapping[str, Any], *, text_summarizer: TextSummarizer | None, max_chars: int = 1200) -> str

Summarize a notification payload with a fast text summarizer when configured.

源代码位于： feishu/integrations/notifications.py

async def notification_summary(
    payload: Mapping[str, Any],
    *,
    text_summarizer: TextSummarizer | None,
    max_chars: int = 1200,
) -> str:
    r"""Summarize a notification payload with a fast text summarizer when configured."""
    explicit = _text(payload.get("summary") or payload.get("message"))
    text = _text(payload.get("summary_input") or payload.get("text") or payload.get("description"))
    if not text:
        text = _compact_json(payload.get("details") or payload)
    if text_summarizer is None:
        return _truncate(explicit or text, max_chars=max_chars)
    instruction = _text(payload.get("summary_instruction")) or (
        "Summarize this workflow/status event for the target recipient. "
        "Keep it concise, include the current state, important risks, and any next action."
    )
    request = TextSummaryRequest(
        kind=_text(payload.get("kind") or payload.get("type") or "workflow_notification"),
        instruction=instruction,
        text=f"{explicit}\n\n{text}".strip(),
        max_chars=max_chars,
        language=_text(payload.get("language")) or "zh-CN",
    )
    result = text_summarizer(request)
    if inspect.isawaitable(result):
        result = await result
    return _truncate(_text(result) or explicit or text, max_chars=max_chars)