Source code for langchain.chat_loaders.slack
import json
import logging
import re
import zipfile
from pathlib import Path
from typing import Dict, Iterator, List, Union
from langchain.chat_loaders.base import BaseChatLoader
from langchain.schema import AIMessage, HumanMessage
from langchain.schema.chat import ChatSession
logger = logging.getLogger(__name__)
[docs]class SlackChatLoader(BaseChatLoader):
    """Load `Slack` conversations from a dump zip file."""
[docs]    def __init__(
        self,
        path: Union[str, Path],
    ):
        """
        Initialize the chat loader with the path to the exported Slack dump zip file.
        :param path: Path to the exported Slack dump zip file.
        """
        self.zip_path = path if isinstance(path, Path) else Path(path)
        if not self.zip_path.exists():
            raise FileNotFoundError(f"File {self.zip_path} not found")
    def _load_single_chat_session(self, messages: List[Dict]) -> ChatSession:
        results: List[Union[AIMessage, HumanMessage]] = []
        previous_sender = None
        for message in messages:
            if not isinstance(message, dict):
                continue
            text = message.get("text", "")
            timestamp = message.get("ts", "")
            sender = message.get("user", "")
            if not sender:
                continue
            skip_pattern = re.compile(
                r"<@U\d+> has joined the channel", flags=re.IGNORECASE
            )
            if skip_pattern.match(text):
                continue
            if sender == previous_sender:
                results[-1].content += "\n\n" + text
                results[-1].additional_kwargs["events"].append(
                    {"message_time": timestamp}
                )
            else: