Spaces:
Runtime error
Runtime error
File size: 2,504 Bytes
9fd6e20 a60b3fc 005a292 9fd6e20 005a292 9fd6e20 005a292 9fd6e20 67436c8 9fd6e20 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, Dict, Literal, Mapping, Optional, Self, Sequence
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: Sequence[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: Sequence[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[Sequence[str]] = None
reactions: Optional[Sequence[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) -> Sequence[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
|