Source code for langchain.document_loaders.telegram
from __future__ import annotations
import asyncio
import json
from pathlib import Path
from typing import TYPE_CHECKING, Dict, List, Optional, Union
from langchain.docstore.document import Document
from langchain.document_loaders.base import BaseLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
if TYPE_CHECKING:
    import pandas as pd
    from telethon.hints import EntityLike
[docs]def concatenate_rows(row: dict) -> str:
    """Combine message information in a readable format ready to be used."""
    date = row["date"]
    sender = row["from"]
    text = row["text"]
    return f"{sender} on {date}: {text}\n\n"
[docs]class TelegramChatFileLoader(BaseLoader):
    """Load from `Telegram chat` dump."""
[docs]    def __init__(self, path: str):
        """Initialize with a path."""
        self.file_path = path
[docs]    def load(self) -> List[Document]:
        """Load documents."""
        p = Path(self.file_path)
        with open(p, encoding="utf8") as f:
            d = json.load(f)
        text = "".join(
            concatenate_rows(message)
            for message in d["messages"]
            if message["type"] == "message" and isinstance(message["text"], str)
        )
        metadata = {"source": str(p)}
        return [Document(page_content=text, metadata=metadata)]
[docs]def text_to_docs(text: Union[str, List[str]]) -> List[Document]:
    """Convert a string or list of strings to a list of Documents with metadata."""
    if isinstance(text, str):
        # Take a single string as one page