Spaces:
Runtime error
Runtime error
File size: 2,493 Bytes
9fd6e20 a60b3fc bb5dde5 005a292 9fd6e20 bb5dde5 9fd6e20 bb5dde5 9fd6e20 005a292 9fd6e20 005a292 9fd6e20 bb5dde5 9fd6e20 67436c8 9fd6e20 bb5dde5 67436c8 9fd6e20 a60b3fc 9fd6e20 a60b3fc 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 |
from datetime import datetime
from json import dumps
from pydantic import BaseModel, ConfigDict, PositiveInt, PrivateAttr
from types import MappingProxyType
from typing import Any, Literal, Mapping, Optional, Self
from ctp_slack_bot.models.base import Chunk, Content
class SlackEventPayload(BaseModel):
"""Represents a general event payload from Slack."""
type: str
event_ts: str
model_config = ConfigDict(extra='allow', frozen=True)
class SlackEvent(BaseModel):
"""Represents a general event from Slack."""
token: str
team_id: str
api_app_id: str
event: SlackEventPayload
type: str
event_id: str
event_time: int
authed_users: tuple[str, ...]
model_config = ConfigDict(frozen=True)
class SlackUserTimestampPair(BaseModel):
"""Represents a Slack user-timestamp pair."""
user: str
ts: str
model_config = ConfigDict(frozen=True)
class SlackReaction(BaseModel):
"""Represents a Slack reaction information."""
name: str
count: PositiveInt
users: tuple[str, ...]
model_config = ConfigDict(frozen=True)
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."""
text: str
channel: Optional[str]
thread_ts: Optional[str] = None
|