import asyncio import os from pathlib import Path import requests from telethon import TelegramClient, events from telethon.sessions import StringSession API_ID = int(os.getenv("TG_API_ID", "0")) API_HASH = os.getenv("TG_API_HASH", "") GROUP_ID = int(os.getenv("TG_GROUP_ID", "0")) N8N_WEBHOOK_URL = os.getenv("N8N_WEBHOOK_URL", "http://n8n:5678/webhook/tg-inbox") HISTORY_CHECK_FILE = os.getenv("HISTORY_CHECK_FILE", "/app/session/history_done.flag") SESSION_FILE_PATH = os.getenv("SESSION_FILE_PATH", "/app/session/session.session") SESSION_STRING = os.getenv("TG_SESSION_STRING", "") TG_USE_IPV6 = os.getenv("TG_USE_IPV6", "false").lower() in {"1", "true", "yes"} TG_CONNECT_TIMEOUT = int(os.getenv("TG_CONNECT_TIMEOUT", "20")) TG_REQUEST_RETRIES = int(os.getenv("TG_REQUEST_RETRIES", "10")) TG_CONNECTION_RETRIES = int(os.getenv("TG_CONNECTION_RETRIES", "20")) TG_RETRY_DELAY = int(os.getenv("TG_RETRY_DELAY", "5")) TG_RECONNECT_WAIT = int(os.getenv("TG_RECONNECT_WAIT", "15")) def load_session_string() -> str: if SESSION_STRING: return SESSION_STRING.strip() session_path = Path(SESSION_FILE_PATH) if not session_path.exists(): raise FileNotFoundError( f"Session file not found: {SESSION_FILE_PATH}. " "Provide TG_SESSION_STRING or mount session file." ) return session_path.read_text(encoding="utf-8").strip() def validate_config() -> None: if API_ID <= 0: raise ValueError("TG_API_ID is required and must be > 0") if not API_HASH: raise ValueError("TG_API_HASH is required") if GROUP_ID == 0: raise ValueError("TG_GROUP_ID is required") if not N8N_WEBHOOK_URL: raise ValueError("N8N_WEBHOOK_URL is required") validate_config() client = TelegramClient( StringSession(load_session_string()), API_ID, API_HASH, use_ipv6=TG_USE_IPV6, timeout=TG_CONNECT_TIMEOUT, request_retries=TG_REQUEST_RETRIES, connection_retries=TG_CONNECTION_RETRIES, retry_delay=TG_RETRY_DELAY, auto_reconnect=True, ) def build_author(sender) -> dict: if not sender: return { "id": "0", "username": "", "display_name": "Unknown", "label": "Unknown", } username = getattr(sender, "username", "") or "" display_name = ( f"{getattr(sender, 'first_name', '')} {getattr(sender, 'last_name', '')}".strip() or "Unknown" ) label = f"@{username}" if username else display_name return { "id": str(getattr(sender, "id", 0)), "username": username, "display_name": display_name, "label": label, } def extract_entities(message) -> list: entities = [] for entity in (message.entities or []): entities.append( { "type": entity.__class__.__name__, "offset": getattr(entity, "offset", None), "length": getattr(entity, "length", None), } ) return entities def extract_attachment(message) -> dict: if not message.media: return {"has_media": False} return { "has_media": True, "media_type": message.media.__class__.__name__, "file_name": getattr(message.file, "name", None) if message.file else None, "mime_type": getattr(message.file, "mime_type", None) if message.file else None, } async def send_to_n8n(message): sender = message.sender or await message.get_sender() author = build_author(sender) reply_message = await message.get_reply_message() if message.reply_to_msg_id else None reply_sender = await reply_message.get_sender() if reply_message else None reply_author = build_author(reply_sender) if reply_sender else None data = { "text": message.text, "metadata": { "date": str(message.date), "message_id": message.id, "chat_id": str(message.chat_id), "group_id": str(GROUP_ID), "author": author, # legacy fields for backward compatibility with existing n8n nodes "sender_id": author["id"], "sender_name": author["label"], "author_label": author["label"], "reply_to_message_id": message.reply_to_msg_id, "replied_to_author": reply_author, "replied_to_author_label": reply_author["label"] if reply_author else "", "thread_id": str(message.reply_to.reply_to_top_id) if getattr(message, "reply_to", None) and getattr(message.reply_to, "reply_to_top_id", None) else "", "edit_date": str(message.edit_date) if message.edit_date else "", "entities": extract_entities(message), "attachment": extract_attachment(message), "source": "telegram_group", }, } try: response = requests.post(N8N_WEBHOOK_URL, json=data, timeout=10) return response.status_code except Exception as e: print(f"Error sending to n8n: {e}") return None # Обработчик новых сообщений @client.on(events.NewMessage(chats=GROUP_ID)) async def handler(event): if event.message.text: await send_to_n8n(event.message) async def main(): print( "Connecting to Telegram with settings:", { "use_ipv6": TG_USE_IPV6, "connect_timeout": TG_CONNECT_TIMEOUT, "request_retries": TG_REQUEST_RETRIES, "connection_retries": TG_CONNECTION_RETRIES, "retry_delay": TG_RETRY_DELAY, }, ) while True: try: await client.start() break except Exception as e: print(f"Telegram connect failed: {e}. Retrying in {TG_RECONNECT_WAIT}s...") await asyncio.sleep(TG_RECONNECT_WAIT) print("UserBot started...") if not os.path.exists(HISTORY_CHECK_FILE): print("Starting massive history sync...") count = 0 async for message in client.iter_messages(GROUP_ID, reverse=True): if message.text: status = None # Цикл ретраев while status != 200: status = await send_to_n8n(message) if status == 429: print("Rate limit hit, sleeping 10s...") await asyncio.sleep(10) elif status != 200: print(f"Error {status}, retrying in 2s...") await asyncio.sleep(2) count += 1 if count % 20 == 0: print(f"Processed {count} messages...") await asyncio.sleep(0.3) Path(HISTORY_CHECK_FILE).write_text("done", encoding="utf-8") print("History sync complete.") await client.run_until_disconnected() if __name__ == '__main__': asyncio.run(main())