diff --git a/Desktop/Telegram_Bot/.gitignore b/Desktop/Telegram_Bot/.gitignore index b138efca41b1b7de034909d6261e027c267a173d..4e40d92ad7171640b3c874b3fea1ce60587eee21 100644 --- a/Desktop/Telegram_Bot/.gitignore +++ b/Desktop/Telegram_Bot/.gitignore @@ -1,3 +1,4 @@ .env .venv session_name.session +/output \ No newline at end of file diff --git a/Desktop/Telegram_Bot/requirements.txt b/Desktop/Telegram_Bot/requirements.txt index 49d2a986e0c15ed8989f285e4c696f4399bfb10b..62102d5b4039e3d1a394d14c03a1a5f3e94094b7 100644 --- a/Desktop/Telegram_Bot/requirements.txt +++ b/Desktop/Telegram_Bot/requirements.txt @@ -1,8 +1,9 @@ --extra-index-url https://download.pytorch.org/whl/cpu telethon python-dotenv -transformers==4.44.2 -torch==2.5.1+cpu +transformers>=4.50.0,<5.0 +torch==2.9.1+cpu accelerate==0.33.0 +langdetect diff --git a/Desktop/Telegram_Bot/services/telegram_listener.py b/Desktop/Telegram_Bot/services/telegram_listener.py index e16e21a0783b9e0a9894d2fd818dbde9373a303e..520322abbdc01bb2e398a347578a9a204bb91009 100644 --- a/Desktop/Telegram_Bot/services/telegram_listener.py +++ b/Desktop/Telegram_Bot/services/telegram_listener.py @@ -1,12 +1,13 @@ # python import asyncio import csv -import os import json +import os from pathlib import Path -from dotenv import load_dotenv,find_dotenv +from dotenv import load_dotenv, find_dotenv from datetime import datetime +from langdetect import detect from telethon import TelegramClient, events from telethon.errors import UserAlreadyParticipantError @@ -17,6 +18,7 @@ from telethon.tl.types import InputChannel, Channel, ReactionEmoji, ReactionCust from transformers import pipeline load_dotenv(find_dotenv()) +# load_dotenv(dotenv_path=r"{Path for the env file}") # Please create a .env file. Add the environment variables over there. And never the env file on github or gitlab. API_ID = int(os.getenv("API_ID")) @@ -24,19 +26,18 @@ API_HASH = os.getenv("API_HASH") SESSION = os.getenv("SESSION") INVITE_LINK = os.getenv("INVITE_LINK") +sentiment_analyzer = pipeline("sentiment-analysis") -DATA_DIR = Path(os.getenv("DATA_DIR", "/app/data")) -DATA_DIR.mkdir(parents=True, exist_ok=True) +# These are the headers for the csv file. +HEADER = ["chat_id", "message_id", "sender_id", "timestamp", "text", + "has_media", "media_type", "file_name", "file_size_kb", + "views", "reactions", "language", "sentiment"] timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") -JSON_PATH = DATA_DIR / f"messages_{timestamp}.json" -sentiment_analyzer = pipeline("sentiment-analysis") +JSON_PATH = Path(f"output/messages_{timestamp}.json") -# These are the headers for the Json file. -HEADER = ["chat_id", "message_id", "sender_id", "timestamp", "text", - "has_media", "media_type", "file_name", "file_size_kb", - "views", "reactions", "sentiment"] +# JSON_PATH = Path(r"messages_.json") # e.g., "output/messages.json" async def list_joined_channels(client): channels = [] @@ -47,7 +48,7 @@ async def list_joined_channels(client): return channels -async def process_messages_for_all_channels(client, process_message, limit): +async def process_last_messages_for_all(client, process_message, limit): channels = await list_joined_channels(client) print(f"No INVITE_URL provided. Listening to all chats. Found {len(channels)} channels.") @@ -63,19 +64,14 @@ async def process_messages_for_all_channels(client, process_message, limit): # Register one global handler for new messages (all chats) @client.on(events.NewMessage()) async def handle_all(event): - await process_message(event.message) - - @client.on(events.MessageEdited()) - async def handle_edit(event): - message = event.message - await process_message(message) + await process_message(event.message) - return channels # optional, if you need the list + return channels -async def process_messages_for_one_channel(client, process_message, target, limit): +async def process_last_message_for_one_channel(client, process_message, target, limit): - msgs = [msg async for msg in client.iter_messages(target, limit=limit)] + msgs = [msg async for msg in client.iter_messages(target, limit=10)] for m in reversed(msgs): await process_message(m) @@ -100,61 +96,84 @@ def write_csv_row(row_dict): writer.writeheader() writer.writerow(row_dict) - -def safe_serialize_reactions(message): - r = getattr(message, "reactions", None) - if not r or not getattr(r, "results", None): - return "" - parts = [] - for rc in r.results: - label = render_reaction_label(rc.reaction) - parts.append(f"{label}:{rc.count}") - return ";".join(parts) +# def write_json_entry(entry): +# """Append a single JSON entry to a file.""" +# try: +# existing_data = [] +# if os.path.exists(JSON_PATH): +# with open(JSON_PATH, "r", encoding="utf-8-sig") as f: +# try: +# existing_data = json.load(f) +# except json.JSONDecodeError: +# existing_data = [] + +# existing_data.append(entry) +# with open(JSON_PATH, "w", encoding="utf-8") as f: +# json.dump(existing_data, f, ensure_ascii=False, indent=4) +# except Exception as e: +# print(f"JSON write error: {e}") + +def detect_language(text): + """Detect language and map it to short format.""" + if not text or not text.strip(): + return "Unknown" + try: + lang_code = detect(text) + lang_map = { + 'en': 'En', 'de': 'De', 'fr': 'Fr', 'es': 'Es', 'it': 'It', + 'pt': 'Pt', 'nl': 'Nl', 'ru': 'Ru', 'zh-cn': 'Zh', 'ja': 'Ja', + 'ar': 'Ar', 'hi': 'Hi' + } + return lang_map.get(lang_code, lang_code.upper()) + except Exception: + return "Unknown" def write_json_entry_per_post(entry, message_id, sender_id): """Create a separate JSON file for each post.""" try: - JSON_PATH = DATA_DIR / f"output/post_{message_id}_{sender_id}.json" - JSON_PATH.parent.mkdir(parents=True, exist_ok=True) # <-- ADD THIS LINE ✅ - existing_data = [] - if JSON_PATH.exists(): - with open(JSON_PATH, "r", encoding="utf-8-sig") as f: - try: - existing_data = json.load(f) - except json.JSONDecodeError: - existing_data = [] - - existing_data.append(entry) - with open(JSON_PATH, "w", encoding="utf-8") as f: - json.dump(existing_data, f, ensure_ascii=False, indent=4) - print(f"JSON file saved: {JSON_PATH}") + os.makedirs("output", exist_ok=True) + json_path = Path(f"output/post_{message_id}_{sender_id}.json") + with open(json_path, "w", encoding="utf-8-sig") as f: + json.dump(entry, f, ensure_ascii=False, indent=4) + print(f"JSON file saved: {json_path}") except Exception as e: - print(f"JSON write error: {e}") print(f"Error writing JSON for post {message_id}: {e}") def write_text_file_per_post(message,sender_id): """Save only the text of a message to a separate .txt file.""" - try: - FILE_PATH = DATA_DIR / f"output/post_{message.id}_{sender_id}.txt" - FILE_PATH.parent.mkdir(parents=True, exist_ok=True) + os.makedirs("output", exist_ok=True) text_content = message.text or "" if not text_content.strip(): return # skip empty or non-text messages - with open(FILE_PATH, "w", encoding="utf-8-sig") as f: + + file_path = Path(f"output/post_{message.id}_{sender_id}.txt") + with open(file_path, "w", encoding="utf-8-sig") as f: f.write(text_content.strip()) - print(f"Text file saved: {FILE_PATH}") + print(f"Text file saved: {file_path}") except Exception as e: print(f"Error writing text file for post {message.id}: {e}") -async def process_message_for_json(message): +def safe_serialize_reactions(message): + r = getattr(message, "reactions", None) + if not r or not getattr(r, "results", None): + return "" + parts = [] + for rc in r.results: + label = render_reaction_label(rc.reaction) + parts.append(f"{label}:{rc.count}") + return ";".join(parts) + + +async def process_message_for_csv(message): sentiment = await asyncio.to_thread(sentiment_analyzer, message.text or "") sender_id = getattr(message, "sender_id", None) + language = detect_language(message.text or "") row = { "chat_id": message.chat_id, "message_id": message.id, - "sender_id": sender_id, + "sender_id": getattr(message, "sender_id", None), "timestamp": message.date.isoformat() if getattr(message, "date", None) else None, "text": (message.text[:3000] if message.text else ""), "has_media": bool(message.media), @@ -163,14 +182,15 @@ async def process_message_for_json(message): "file_size_kb": round(getattr(message.file, "size", 0) / 1024, 2) if getattr(message, "file", None) else "", "views": getattr(message, "views", None), "reactions": safe_serialize_reactions(message), + "language" : language, "sentiment": sentiment[0]["label"] if sentiment else "" } try: - #write_csv_row(row) + # write_csv_row(row) write_json_entry_per_post(row, message.id, sender_id) write_text_file_per_post(message,sender_id) except Exception as e: - print(f"CSV write error: {e}") + print(f"JSON write error: {e}") def render_reaction_label(r): if isinstance(r, ReactionEmoji): @@ -210,8 +230,15 @@ async def ensure_joined(client, invite_url: str): except Exception: pass return entity + +async def main(): + client = TelegramClient(SESSION, API_ID, API_HASH) + await client.start() + + # Resolve and ensure we are re in the channel/group + target = await ensure_joined(client, INVITE_LINK) -async def process_message(message): + async def process_message(message): # 1. Basic Info print("--- New Message ---") @@ -240,22 +267,16 @@ async def process_message(message): print(f" - {label}: {reaction_count.count}") print("------\n") - await process_message_for_json(message) - -async def main(): - client = TelegramClient(SESSION, API_ID, API_HASH) - await client.start() - - # Resolve and ensure we are re in the channel/group - target = await ensure_joined(client, INVITE_LINK) + await process_message_for_csv(message) if target: print("Listening to single channel ") - await process_messages_for_one_channel(client, process_message, target, limit= 5) + await process_last_message_for_one_channel(client, process_message, target, limit= 5) + # After collecting else: print("No INVITE_URL provided. Listening to all chats.") - await process_messages_for_all_channels(client, process_message, limit=5) + await process_last_messages_for_all(client, process_message, limit=5) print("Listening for messages...") await client.run_until_disconnected()