initial commit

.gitignore

@@ -0,0 +1,2 @@
+.idea
+__pycache__

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2023 RuslanUC
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1 @@
+## Telegram export tool.

pyproject.toml

@@ -0,0 +1,17 @@
+[tool.poetry]
+name = "t-export"
+version = "0.1.0"
+description = ""
+authors = ["RuslanUC <dev_ruslan_uc@protonmail.com>"]
+readme = "README.md"
+
+[tool.poetry.dependencies]
+python = "^3.9"
+pyrogram = "^2.0.106"
+tgcrypto = "^1.2.5"
+click = "^8.1.7"
+
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

texport/__main__.py

@@ -0,0 +1,4 @@
+from .main import main
+
+if __name__ == "__main__":
+    main()

texport/export_config.py

@@ -0,0 +1,38 @@
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Union
+
+from pyrogram.enums import MessageMediaType
+
+EXPORT_MEDIA = {
+    "photos": MessageMediaType.PHOTO,
+    "videos": MessageMediaType.VIDEO,
+    "voice": MessageMediaType.VOICE,
+    "video_notes": MessageMediaType.VIDEO_NOTE,
+    "stickers": MessageMediaType.STICKER,
+    "gifs": MessageMediaType.ANIMATION,
+    "files": MessageMediaType.DOCUMENT,
+}
+
+
+@dataclass
+class ExportConfig:
+    chat_id: Union[str, int] = "me"
+    output_dir: str = "./telegram_export"
+    export_photos: bool = True
+    export_videos: bool = True
+    export_voice: bool = True
+    export_video_notes: bool = True
+    export_stickers: bool = True
+    export_gifs: bool = True
+    export_files: bool = True
+    size_limit: int = 32  # In megabytes
+    from_date: datetime = datetime(1970, 1, 1)
+    to_date: datetime = datetime.now()
+
+    def excluded_media(self) -> set[MessageMediaType]:
+        result = set()
+        for media_type in EXPORT_MEDIA:
+            if not getattr(self, f"export_{media_type}"):
+                result.add(EXPORT_MEDIA[media_type])
+        return result

texport/exporter.py

@@ -0,0 +1,82 @@
+import asyncio
+from datetime import date
+from os.path import relpath
+from typing import Union, Optional
+
+from pyrogram import Client
+from pyrogram.types import Message as PyroMessage
+
+from texport.export_config import ExportConfig
+from texport.media import MEDIA_TYPES
+from texport.messages_saver import MessagesSaver
+
+
+class ExportStatus:
+    def __init__(self):
+        self.biggest_message_id = None
+        self.last_message_id = None
+        self.last_date = None
+
+
+class Exporter:
+    def __init__(self, client: Client, export_config: ExportConfig=None):
+        self._config = export_config or ExportConfig()
+        self._client = client
+        self._task = None
+        self.status: Optional[ExportStatus] = None
+        self._messages: list[PyroMessage] = []
+        self._media: dict[Union[int, str], str] = {}
+        self._saver = MessagesSaver(self._messages, self._media, export_config)
+        self._excluded_media = self._config.excluded_media()
+
+    async def _export_media(self, message: PyroMessage) -> None:
+        if message.media not in MEDIA_TYPES or message.media in self._excluded_media:
+            return
+        m = MEDIA_TYPES[message.media]
+        media = m.get_media(message)
+        if media.file_size > self._config.size_limit * 1024 * 1024:
+            return
+
+        path = await message.download(file_name=f"{self._config.output_dir}/{m.dir_name}/")
+        path = relpath(path, self._config.output_dir)
+        self._media[message.id] = path
+
+        if hasattr(media, "thumbs") and media.thumbs:
+            path = await self._client.download_media(media.thumbs[0].file_id,
+                                                     file_name=f"{self._config.output_dir}/thumbs/")
+            path = relpath(path, self._config.output_dir)
+            self._media[f"{message.id}_thumb"] = path
+
+    async def _export(self, chat_id: Union[int, str]):
+        offset_date = None if self._config.to_date.date() >= date.today() else self._config.to_date
+        # TODO: fix offset_date
+        async for message in self._client.get_chat_history(chat_id):
+            if message.date < self._config.from_date:
+                break
+            if self.status.biggest_message_id is None:
+                self.status.biggest_message_id = message.id
+            self.status.last_message_id = message.id
+            self.status.last_date = message.date
+            if message.media:
+                await self._export_media(message)
+
+            if not message.text and not message.caption and message.id not in self._media:
+                continue
+
+            self._messages.append(message)
+            if len(self._messages) > 5000:
+                await self._saver.save()
+
+        if self._messages:
+            await self._saver.save()
+        self.status = self._task = None
+
+    async def export(self, block: bool=True) -> None:
+        if self._task is not None or self.status is not None:
+            return
+        self.status = ExportStatus()
+        coro = self._export(self._config.chat_id)
+        if block:
+            await coro
+        else:
+            asyncio.get_event_loop().create_task(coro)

texport/html/animation.py

@@ -0,0 +1,34 @@
+from typing import Optional
+from pyrogram.types import Message as PyroMessage
+
+from .base import HtmlMedia
+from .utils import file_size_str
+
+
+class Animation(HtmlMedia):
+    def __init__(self, media_path: str, media_thumb: Optional[str], message: PyroMessage):
+        self.path = media_path
+        self.thumb = media_thumb or media_path
+        self.message = message
+
+    def no_media(self) -> str:
+        return f"""
+        <div class="media clearfix pull_left media_video">
+            <div class="fill pull_left"></div>
+            <div class="body">
+                <div class="title bold">Animation</div>
+                <div class="description">Not included, change data exporting settings to download.</div>
+                <div class="status details">{file_size_str(self.message.animation.file_size)}</div>
+            </div>
+        </div>
+        """
+
+    def to_html(self) -> str:
+        return f"""
+        <a class="animated_wrap clearfix pull_left" href="{self.path}">
+            <div class="video_play_bg">
+                <div class="gif_play">GIF</div>
+            </div>
+            <img class="animated" src="{self.thumb}" style="width: 192px; height: auto">
+        </a>
+        """ if self.path else self.no_media()