Spaces:
Runtime error
Runtime error
File size: 2,511 Bytes
9fd6e20 bb7c9a3 a60b3fc bb5dde5 005a292 a1a6d79 9fd6e20 bb7c9a3 9fd6e20 bb7c9a3 9fd6e20 bb7c9a3 9fd6e20 bb5dde5 9fd6e20 bb7c9a3 9fd6e20 bb7c9a3 9fd6e20 bb5dde5 9fd6e20 005a292 9fd6e20 005a292 9fd6e20 bb5dde5 9fd6e20 67436c8 9fd6e20 bb5dde5 67436c8 9fd6e20 a60b3fc 9fd6e20 a60b3fc 9fd6e20 bb7c9a3 9fd6e20 bb7c9a3 9fd6e20 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 |
from datetime import datetime
from json import dumps
from pydantic import BaseModel, ConfigDict, PositiveInt
from types import MappingProxyType
from typing import Any, Literal, Mapping, Optional, Self
from .base import Chunk, Content
class SlackEventPayload(BaseModel):
"""Represents a general event payload from Slack."""
model_config = ConfigDict(extra='allow', frozen=True)
type: str
event_ts: str
class SlackEvent(BaseModel):
"""Represents a general event from Slack."""
model_config = ConfigDict(frozen=True)
token: str
team_id: str
api_app_id: str
event: SlackEventPayload
type: str
event_id: str
event_time: int
authed_users: tuple[str, ...]
class SlackUserTimestampPair(BaseModel):
"""Represents a Slack user-timestamp pair."""
model_config = ConfigDict(frozen=True)
user: str
ts: str
class SlackReaction(BaseModel):
"""Represents a Slack reaction information."""
model_config = ConfigDict(frozen=True)
name: str
count: PositiveInt
users: tuple[str, ...]
class SlackMessage(Content):
"""Represents a message from Slack after adaptation."""
type: Literal["app_mention", "message"]
subtype: Optional[str] = None
channel: str
channel_type: Optional[str] = None
user: Optional[str] = None
bot_id: Optional[str] = None
thread_ts: Optional[str] = None
text: str
ts: str
edited: Optional[SlackUserTimestampPair] = None
event_ts: str
deleted_ts: Optional[str] = None
hidden: bool = False
is_starred: Optional[bool] = None
pinned_to: Optional[tuple[str, ...]] = None
reactions: Optional[tuple[SlackReaction, ...]] = None
def get_id(self: Self) -> str:
"""Unique identifier for this message."""
return f"slack-message:{self.channel}:{self.ts}"
def get_chunks(self: Self) -> tuple[Chunk]:
return (Chunk(text=self.text, parent_id=self.get_id(), chunk_id="", metadata=self.get_metadata()), )
def get_metadata(self: Self) -> Mapping[str, Any]:
return MappingProxyType({
"modificationTime": datetime.fromtimestamp(float(self.ts))
})
class SlackResponse(BaseModel): # TODO: This should also be based on Content as it is a SlackMessage―just not one for which we know the identity yet.
"""Represents a response message to be sent to Slack."""
model_config = ConfigDict(frozen=True)
text: str
channel: Optional[str]
thread_ts: Optional[str] = None
|