Merge pull request #4 from ValeriyMaslenikov/telethon-v1

Updated telethon version and the application was rewritten to support asyncio

.env.example

@@ -5,3 +5,14 @@ TELEGRAM_API_HASH = 0123456789abcdef0123456789abcdef
 
 # Logging level, available values: https://docs.python.org/3/library/logging.html#levels
 LOGGING_LEVEL = CRITICAL
+
+# Correct values:
+# '' – You will not be notified about deleted ongoing messages
+# '1' – You will be notified about deleted ongoing messages
+#
+# Enabled option is useful, when your companion deletes the bunch of his and your messages
+NOTIFY_ONGOING_MESSAGES='1'
+
+# How many days messages will be stored in the SQLite database
+# Warning: Database is not constrained by memory it'll occupy, you need to monitor your free disk space manually
+MESSAGES_TTL_DAYS = 14

requirements.txt

@@ -1,2 +1,3 @@
-Telethon==0.19.0.1
-python-dotenv==v0.8.2
+Telethon==1.14.0
+python-dotenv==v0.13.0
+pylint==2.5.3

src/helpers.py

@@ -0,0 +1,138 @@
+import logging
+import os
+import pickle
+import sqlite3
+import asyncio
+from datetime import datetime, timedelta
+from pathlib import Path
+from typing import List
+
+from dotenv import load_dotenv
+from telethon.events import NewMessage, MessageDeleted
+from telethon import TelegramClient
+from telethon.hints import Entity
+from telethon.tl.types import Message
+
+CLEAN_OLD_MESSAGES_EVERY_SECONDS = 60 # 1 minute
+
+
+def load_env(dot_env_folder):
+ env_path = Path(dot_env_folder) / ".env"
+
+ if os.path.isfile(env_path):
+ load_dotenv(dotenv_path=env_path)
+ logging.debug('`.env` file is loaded')
+ else:
+ logging.debug('`.env` file is absent, using system environment variables')
+
+
+def initialize_messages_db():
+ connection = sqlite3.connect("db/messages_v2.db")
+ cursor = connection.cursor()
+
+ cursor.execute("""CREATE TABLE IF NOT EXISTS messages
+ (message_id INTEGER PRIMARY KEY, message_from_id INTEGER, message TEXT, media BLOB, created DATETIME)""")
+
+ cursor.execute("CREATE INDEX IF NOT EXISTS messages_created_index ON messages (created DESC)")
+
+ connection.commit()
+
+ return cursor, connection
+
+
+sqlite_cursor, sqlite_connection = initialize_messages_db()
+
+
+def get_on_new_message(client: TelegramClient):
+ async def on_new_message(event: NewMessage.Event):
+ sqlite_cursor.execute(
+ "INSERT INTO messages (message_id, message_from_id, message, media, created) VALUES (?, ?, ?, ?, ?)",
+ (
+ event.message.id,
+ event.message.from_id,
+ event.message.message,
+ sqlite3.Binary(pickle.dumps(event.message.media)),
+ str(datetime.now())))
+ sqlite_connection.commit()
+
+ return on_new_message
+
+
+def load_messages_from_event(event: MessageDeleted.Event) -> List[Message]:
+ sql_message_ids = ",".join(str(deleted_id) for deleted_id in event.deleted_ids)
+
+ db_results = sqlite_cursor.execute(
+ f"SELECT message_id, message_from_id, message, media FROM messages WHERE message_id IN ({sql_message_ids})"
+ ).fetchall()
+
+ messages = []
+ for db_result in db_results:
+ messages.append({
+ "id": db_result[0],
+ "message_from_id": db_result[1],
+ "message": db_result[2],
+ "media": pickle.loads(db_result[3]),
+ })
+
+ return messages
+
+
+async def get_mention_username(user: Entity):
+ if user.first_name or user.last_name:
+ mention_username = \
+ (user.first_name + " " if user.first_name else "") + \
+ (user.last_name if user.last_name else "")
+ elif user.username:
+ mention_username = user.username
+ elif user.phone:
+ mention_username = user.phone
+ else:
+ mention_username = user.id
+
+ return mention_username
+
+
+def get_on_message_deleted(client: TelegramClient):
+ async def on_message_deleted(event: MessageDeleted.Event):
+ messages = load_messages_from_event(event)
+
+ log_deleted_usernames = []
+
+ for message in messages:
+ user = await client.get_entity(message['message_from_id'])
+ mention_username = await get_mention_username(user)
+
+ log_deleted_usernames.append(mention_username + " (" + str(user.id) + ")")
+ text = "🔥🔥🔥🤫🤐🤭🙊🔥🔥🔥\n**Deleted message from: **[{username}](tg://user?id={id})\n".format(
+ username=mention_username, id=user.id)
+
+ if message['message']:
+ text += "**Message:** " + message['message']
+
+ await client.send_message(
+ "me",
+ text,
+ file=message['media']
+ )
+
+ logging.info(
+ "Got {deleted_messages_count} deleted messages. Has in DB {db_messages_count}. Users: {users}".format(
+ deleted_messages_count=str(len(event.deleted_ids)),
+ db_messages_count=str(len(messages)),
+ users=", ".join(log_deleted_usernames))
+ )
+
+ return on_message_deleted
+
+
+async def cycled_clean_old_messages():
+ messages_ttl_days = int(os.getenv('MESSAGES_TTL_DAYS', 14))
+
+ while True:
+ delete_from_time = str(datetime.now() - timedelta(days=messages_ttl_days))
+ sqlite_cursor.execute("DELETE FROM messages WHERE created < ?", (delete_from_time,))
+ logging.info(
+ f"Deleted {sqlite_cursor.rowcount} messages older than {delete_from_time} from DB"
+ )
+
+ await asyncio.sleep(CLEAN_OLD_MESSAGES_EVERY_SECONDS)

src/monitor.py

@@ -0,0 +1,50 @@
+import logging
+import os
+import pathlib
+import sys
+
+from telethon import TelegramClient, events
+from helpers import load_env, get_on_new_message, get_on_message_deleted, cycled_clean_old_messages
+
+BASE_DIR = (pathlib.Path(__file__).parent / '..').absolute()
+
+# Configure logging level, based on the system environment variables
+logging.basicConfig(level=os.getenv("LOGGING_LEVEL", logging.INFO))
+
+# Loading environment variables
+load_env(BASE_DIR)
+
+# Configure logging level, based on the `.env` file and on the system environment variables
+logging.basicConfig(level=os.getenv("LOGGING_LEVEL", logging.INFO))
+
+if os.getenv("TELEGRAM_API_ID") is None or os.getenv("TELEGRAM_API_HASH") is None:
+ logging.critical('Please, read `README.md` and set-up environment variables (you can create a copy of '
+ '`.env.example` file with new name `.env` and fill correct values')
+ exit(1)
+
+
+async def main():
+ if len(sys.argv) > 1 and sys.argv[1] == 'auth':
+ # TODO: perform logout in the code, in case the user use `auth` argument
+ logging.critical('You successfully authorized, please, run the same command without `auth` argument to '
+ 'start monitoring your messages. If you want to log-out, remove the file `db/user.session`, '
+ 'to log-out and re-execute this command')
+ exit(0)
+
+ if not await client.is_user_authorized():
+ logger.critical('Please, execute `auth` command before starting the daemon (see `README.md` file)')
+ exit(1)
+
+ if bool(os.getenv('NOTIFY_ONGOING_MESSAGES', '1')):
+ new_message_event = events.NewMessage()
+ else:
+ new_message_event = events.NewMessage(incoming=True, outgoing=False)
+
+ client.add_event_handler(get_on_new_message(client), new_message_event)
+ client.add_event_handler(get_on_message_deleted(client), events.MessageDeleted())
+
+ await cycled_clean_old_messages()
+
+
+with TelegramClient('db/user', os.getenv("TELEGRAM_API_ID"), os.getenv("TELEGRAM_API_HASH")) as client:
+ client.loop.run_until_complete(main())