[DEV-Abinash] Integrated Language Detection

Desktop/Telegram_Bot/requirements.txt

 --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
 
 
+

Desktop/Telegram_Bot/services/telegram_listener.py

 # 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.")
 
@@ -65,17 +66,12 @@ async def process_messages_for_all_channels(client, process_message, limit):
     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)  
-
-    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):
@@ -211,7 +231,14 @@ async def ensure_joined(client, invite_url: str):
             pass
     return entity
         
-async def process_message(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)
+
+    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()