Optional periodic import of history while listening for messages

Author: Programie

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## [3.2.0] - pending
+
+* Optional periodic import of history while listening for messages
+
 ## [3.1.2] - 2023-11-08
 
 * Fixed crash if username, first_name or last_name of sender are not defined (thanks to @cololi)
@@ -45,4 +49,4 @@ Chat Import: Fixed not correctly resolving entities if ID is given, prefix names
 
 ## [1.0] - 2020-10-22
 
-Initial release
+Initial release

README.md

@@ -82,6 +82,34 @@ output_map:
   "message.translated": "translated_text"
 ```
 
+## Periodically import history
+
+In some cases you might want to periodically import the history while listening for messages.
+
+This feature is disabled by default but can be enabled by adding the following configuration to your `config.yml`:
+
+```yaml
+periodic_import:
+  interval: 1d
+  range: 7d
+```
+
+At least an interval is required. If no range is given, everything is imported.
+
+You might also specify multiple time units like "1d12h" which means "1 days and 12 hours".
+
+The following units are supported:
+
+| Unit | Meaning | Example |
+|------|---------|---------|
+| y    | Years   | 1y      |
+| mo   | Months  | 12mo    |
+| w    | Weeks   | 30w     |
+| d    | Days    | 7d      |
+| h    | Hours   | 24h     |
+| m    | Minutes | 60m     |
+| s    | Seconds | 60s     |
+
 ## Initial setup
 
 When started for the first time, the application will ask you to connect with your Telegram account.
@@ -90,4 +118,4 @@ When started for the first time, the application will ask you to connect with yo
 
 * `listen` - Listen for chat messages and write them to the configured outputs
 * `import-history` - Import the chat history (the complete history or only for specific chats or a specific time range)
-* `list-chats` - List available chats
+* `list-chats` - List available chats

config.sample.yml

@@ -18,6 +18,17 @@ telegram:
     - <id 2>
     - ...
 
+# Periodically import history
+# The time properties support multiple units: y (years), mo (months), w (weeks), d (days), h (hours), m (minutes), s (seconds)
+# Those units can be combined, i.e. "1d12h" means "1 days and 12 hours".
+periodic_import:
+  # Interval between periodic imports
+  interval: 1d
+
+  # Range to import, e.g. 7d to import the last 7 days
+  # Omit to always import everything
+  range: 7d
+
 # Configure whether messages should be translated into the specified language
 # Use the two-letter ISO 639-1 language code (examples: "de", "en", "es", "it")
 # Omit or keep empty to disable translations
@@ -154,4 +165,4 @@ outputs:
       chat: "get_display_name(await message.get_chat())"
       message: "message.text"
       translated_message: "translated_text"
-      media: "media.filename if media else None"
+      media: "media.filename if media else None"

telegram2elastic.py

@@ -14,7 +14,7 @@
 
 import yaml
 
-from datetime import datetime
+from datetime import datetime, timedelta
 from telethon import TelegramClient, events
 from telethon.tl import types
 from telethon.tl.functions.messages import TranslateTextRequest
@@ -84,6 +84,66 @@ def set(self, path, value):
         node[path_parts[-1]] = value
 
 
+class TimeInterval:
+    seconds: int
+
+    def __init__(self, seconds: int):
+        self.seconds = seconds
+
+    @staticmethod
+    def parse(string: str):
+        matches = re.findall(r"(\d+)(y|mo|w|d|h|m|s)", re.sub(r"[\s,_-]+", "", string.lower()))
+        if not matches:
+            return None
+
+        total_seconds = 0
+
+        for value, unit in matches:
+            value = int(value)
+
+            if unit == "y":
+                total_seconds += value * 365 * 24 * 3600
+            elif unit == "mo":
+                total_seconds += value * 30 * 24 * 3600
+            elif unit == "w":
+                total_seconds += value * 7 * 24 * 3600
+            elif unit == "d":
+                total_seconds += value * 24 * 3600
+            elif unit == "h":
+                total_seconds += value * 3600
+            elif unit == "m":
+                total_seconds += value * 60
+            elif unit == "s":
+                total_seconds += value
+
+        return TimeInterval(total_seconds)
+
+    def format_human_readable(self):
+        total_seconds = self.seconds
+        parts = []
+
+        units = [
+            ("year", 365 * 24 * 3600),
+            ("month", 30 * 24 * 3600),
+            ("week", 7 * 24 * 3600),
+            ("day", 24 * 3600),
+            ("hour", 3600),
+            ("minute", 60),
+            ("second", 1),
+        ]
+
+        for name, unit_seconds in units:
+            value, total_seconds = divmod(total_seconds, unit_seconds)
+            if value:
+                label = name if value == 1 else name + "s"
+                parts.append(f"{value} {label}")
+
+        return ", ".join(parts) if parts else "0 seconds"
+
+    def timedelta(self):
+        return timedelta(seconds=self.seconds)
+
+
 def json_default(value):
     if isinstance(value, bytes):
         return base64.b64encode(value).decode("ascii")
@@ -439,12 +499,7 @@ def __init__(self, config: dict, output_handler: OutputHandler):
         self.additional_chats = config.get("additional_chats", [])
         self.chat_types = config.get("chat_types", [])
 
-    async def import_history(self, start_date, chats):
-        if start_date:
-            offset_date = datetime.strptime(start_date, "%Y-%m-%d")
-        else:
-            offset_date = None
-
+    async def import_history(self, start_date: datetime = None, chats=None):
         if chats:
             chats = await self.client.get_entity(TelegramReader.prepare_chats(chats))
         else:
@@ -453,12 +508,12 @@ async def import_history(self, start_date, chats):
         for chat in chats:
             display_name = get_display_name(chat)
 
-            if offset_date:
-                logging.log(LOG_LEVEL_INFO, "Importing history for chat '{}' starting at {}".format(display_name, offset_date.strftime("%c")))
+            if start_date:
+                logging.log(LOG_LEVEL_INFO, "Importing history for chat '{}' starting at {}".format(display_name, start_date.strftime("%c")))
             else:
                 logging.log(LOG_LEVEL_INFO, "Importing full history for chat '{}'".format(display_name))
 
-            async for message in self.client.iter_messages(chat, offset_date=offset_date, reverse=True):
+            async for message in self.client.iter_messages(chat, offset_date=start_date, reverse=True):
                 await self.output_handler.write_message(message, self.is_chat_enabled)
 
         logging.log(LOG_LEVEL_INFO, "Import finished")
@@ -479,6 +534,31 @@ async def handler(event):
 
         await self.client.catch_up()
 
+    async def periodic_import(self, config: dict):
+        interval = TimeInterval.parse(config.get("interval", "1d"))
+
+        time_range = config.get("range")
+        if time_range:
+            time_range = TimeInterval.parse(time_range)
+            time_range_string = time_range.format_human_readable()
+        else:
+            time_range = None
+            time_range_string = "all"
+
+        logging.log(LOG_LEVEL_INFO, f"Scheduling periodic import (interval: {interval.format_human_readable()}, range: {time_range_string})")
+
+        while True:
+            await asyncio.sleep(interval.seconds)
+
+            logging.log(LOG_LEVEL_INFO, "Starting periodic import")
+
+            start_date = None
+            if time_range:
+                start_date = datetime.now() - time_range.timedelta()
+            await self.import_history(start_date)
+
+            logging.log(LOG_LEVEL_INFO, f"Periodic import completed, next periodic import at {datetime.now() + interval.timedelta()}")
+
     def is_chat_enabled(self, chat, chat_types=None):
         if chat.id in self.additional_chats:
             return True
@@ -559,12 +639,20 @@ def main():
         loop = telegram_reader.client.loop
 
         if arguments.command == "import-history":
-            loop.run_until_complete(telegram_reader.import_history(arguments.start_date, arguments.chats))
+            start_date = arguments.start_date
+            if start_date:
+                start_date = datetime.strptime(start_date, "%Y-%m-%d")
+
+            loop.run_until_complete(telegram_reader.import_history(start_date, arguments.chats))
         elif arguments.command == "list-chats":
             loop.run_until_complete(telegram_reader.list_chats(arguments.types))
         elif arguments.command == "listen":
             loop.create_task(telegram_reader.listen())
 
+            periodic_import_config = config.get("periodic_import", {})
+            if periodic_import_config.get("interval") is not None:
+                loop.create_task(telegram_reader.periodic_import(periodic_import_config))
+
             try:
                 loop.run_forever()
             except KeyboardInterrupt:

test_telegram2elastic.py

@@ -1,6 +1,6 @@
 import pytest
 
-from telegram2elastic import FileSize, DottedPathDict
+from telegram2elastic import FileSize, DottedPathDict, TimeInterval
 
 
 class TestFileSize:
@@ -29,3 +29,25 @@ def test(self):
 
         with pytest.raises(TypeError):
             dotted_path_dict.set("foo.bar.baz", "other value")
+
+
+class TestTimeInterval:
+    def test_parse(self):
+        assert TimeInterval.parse("1h1m").seconds == 60*60 + 60
+        assert TimeInterval.parse("1h5m").seconds == 60*60 + 60*5
+        assert TimeInterval.parse("1d").seconds == 60*60*24
+        assert TimeInterval.parse("1d3h10m").seconds == 60*60*24 + 60*60*3 + 60*10
+        assert TimeInterval.parse("1y2mo").seconds == 60*60*24*365 + 60*60*24*60
+        assert TimeInterval.parse("1mo2m").seconds == 60*60*24*30 + 60*2
+        assert TimeInterval.parse("2m1mo").seconds == 60*60*24*30 + 60*2
+
+    def test_format(self):
+        assert TimeInterval(1).format_human_readable() == "1 second"
+        assert TimeInterval(25).format_human_readable() == "25 seconds"
+        assert TimeInterval(60).format_human_readable() == "1 minute"
+        assert TimeInterval(90).format_human_readable() == "1 minute, 30 seconds"
+        assert TimeInterval(60*2).format_human_readable() == "2 minutes"
+        assert TimeInterval(60*60).format_human_readable() == "1 hour"
+        assert TimeInterval(60*60*24).format_human_readable() == "1 day"
+        assert TimeInterval(60*60*24*2).format_human_readable() == "2 days"
+        assert TimeInterval(60*60*24 + 60*60*12 + 60 + 35).format_human_readable() == "1 day, 12 hours, 1 minute, 35 seconds"