join.py

import csv
import asyncio
import time
from telethon import TelegramClient
from tqdm import tqdm  # Import tqdm for progress bar
from telethon.tl.functions.channels import JoinChannelRequest
from telethon.tl.functions.messages import ImportChatInviteRequest
from telethon.errors.rpcerrorlist import InviteHashExpiredError
from flask import Flask, jsonify, send_from_directory
# Directory for storing files
from flask import Flask, render_template, send_from_directory
from telethon.tl.functions.channels import GetParticipantsRequest
from telethon.tl.types import ChannelParticipantsSearch
from telethon.errors import FloodWaitError, UserAdminInvalidError
import json
import asyncio
import nest_asyncio
import logging
from telethon import TelegramClient, events
from supabase import create_client, Client
from flask import Flask, jsonify
from threading import Thread
from multiprocessing import Process, Queue
import unicodedata
from telegram.helpers import escape_markdown
import re
import os

from telethon.tl.functions.channels import JoinChannelRequest, InviteToChannelRequest
from telethon.tl.functions.channels import EditBannedRequest
from telethon.tl.types import ChatBannedRights
from telethon.errors.rpcerrorlist import UserAdminInvalidError, UserNotParticipantError
from telethon.errors.rpcerrorlist import InviteHashExpiredError, UserAlreadyParticipantError
from telethon.tl.types import Channel, Chat

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[
        logging.FileHandler("join_groups.log"),  # Log to a file
        logging.StreamHandler()  # Log to console
    ]
)
# Replace with your API credentials (from https://my.telegram.org/apps)
API_ID = 25216912  # Your API ID
API_HASH = "f65f6050fe9b342a4996c59e4283ab5e"
PHONE_NUMBER = "+967735201519"  # Your phone number with country code
OUTPUT_CSV = "groups_with_status.csv" 
# Path to your CSV file
CSV_FILENAME = "8.csv"
session_dir =  "mfoud73" 
FILE_DIRECTORY = os.getcwd()  # Current working directory

SLEEP_TIME = 280

# Flask App
app = Flask(__name__)

# 🔹 Flask API Endpoints
@app.route('/')
def index():
    """Show available files for download as an HTML page."""
    files = os.listdir(FILE_DIRECTORY)
    return render_template("index.html", files=files)


@app.route('/download/<filename>')
def download_file(filename):
    """Allow downloading any file from the directory."""
    return send_from_directory(FILE_DIRECTORY, filename, as_attachment=True)


def run_flask():
    app.run(host='0.0.0.0', port=7860)
    
    
USER_CSV = "user_list.csv"
# SLEEP_TIME = 280  # Delay between adding users

# Logging setup
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")

BATCH_SIZE = 30  # Fetch 200 users at a time (Telegram's limit)
MAX_USERS = 200  # Set your desired limit here



async def fetch_users_from_source_group(source_group, max_users=MAX_USERS):
    """
    Fetch up to `max_users` from a Telegram group using pagination and save them to a CSV file.
    
    :param source_group: The source group username or ID.
    :param max_users: The maximum number of users to fetch. Use `None` to fetch all users.
    """
    async with TelegramClient(session_dir, API_ID, API_HASH) as client:
        await client.start(PHONE_NUMBER)

        try:
            entity = await client.get_entity(source_group)
            offset = 0  # Pagination start
            total_users = 0
            csv_filename = f"{source_group}.csv"  # Save with group name

            with open(csv_filename, mode="a", newline="", encoding="utf-8") as file:
                writer = csv.writer(file)
                writer.writerow(["user_id", "username", "first_name", "last_name"])  # CSV headers

                with tqdm(desc=f"Fetching users from {source_group}", unit="user", ncols=100) as pbar:
                    while max_users is None or total_users < max_users:
                        remaining_users = max_users - total_users if max_users else BATCH_SIZE
                        batch_size = min(BATCH_SIZE, remaining_users)  # Adjust batch size if close to max

                        try:
                            participants = await client(GetParticipantsRequest(
                                entity, ChannelParticipantsSearch(''), offset, batch_size, hash=0
                            ))

                            if not participants.users:
                                break  # Stop when no more users are found

                            for user in participants.users:
                                writer.writerow([user.id, user.username or "N/A", user.first_name or "N/A", user.last_name or "N/A"])
                                pbar.update(1)  # Update progress bar
                                pbar.set_postfix(user=user.username or "N/A")  # Show last processed user

                            total_users += len(participants.users)
                            offset += len(participants.users)  # Move offset forward

                        except FloodWaitError as e:
                            pbar.set_postfix(waiting=f"{e.seconds}s")  # Show cooldown time
                            await asyncio.sleep(e.seconds)  # Wait for Telegram cooldown

                pbar.set_description(f"✅ Done! Fetched {total_users} users")

        except Exception as e:
            print(f"❌ Failed to fetch users from {source_group}: {e}")


async def fetch_users_from_source_group1(source_group, max_users=MAX_USERS):
    """
    Fetch up to `max_users` from a Telegram group using pagination and save them to a CSV file.
    
    :param source_group: The source group username or ID.
    :param max_users: The maximum number of users to fetch. Use `None` to fetch all users.
    """
    logging.info(f"Fetching users from {source_group} (Limit: {max_users if max_users else 'All'})...")

    async with TelegramClient(session_dir, API_ID, API_HASH) as client:
        await client.start(PHONE_NUMBER)

        try:
            entity = await client.get_entity(source_group)
            offset = 0  # Pagination start
            total_users = 0
            csv_filename = f"{source_group}.csv"  # Save with group name

            with open(csv_filename, mode="w", newline="", encoding="utf-8") as file:
                writer = csv.writer(file)
                writer.writerow(["user_id", "username", "first_name", "last_name"])  # CSV headers

                while max_users is None or total_users < max_users:
                    remaining_users = max_users - total_users if max_users else BATCH_SIZE
                    batch_size = min(BATCH_SIZE, remaining_users)  # Adjust batch size if close to max

                    try:
                        participants = await client(GetParticipantsRequest(
                            entity, ChannelParticipantsSearch(''), offset, batch_size, hash=0
                        ))

                        if not participants.users:
                            break  # Stop when no more users are found

                        for user in participants.users:
                            writer.writerow([user.id, user.username or "N/A", user.first_name or "N/A", user.last_name or "N/A"])
                            logging.info(f"✔ User saved: {user.id} | {user.username}")

                        total_users += len(participants.users)
                        offset += len(participants.users)  # Move offset forward

                    except FloodWaitError as e:
                        logging.warning(f"⚠️ Telegram rate limit hit! Waiting {e.seconds} seconds...")
                        await asyncio.sleep(e.seconds)  # Wait for Telegram cooldown

                    await asyncio.sleep(SLEEP_TIME)  # Avoid hitting limits

            logging.info(f"✅ Fetched {total_users} users from {source_group}. Saved to {csv_filename}.")

        except Exception as e:
            logging.error(f"❌ Failed to fetch users from {source_group}: {e}")

async def fetch_users_from_source_group1(source_group, max_users=MAX_USERS):
    """
    Fetch up to `max_users` from a Telegram group using client.iter_participants() and save them to a CSV file.
    
    :param source_group: The source group username or ID.
    :param max_users: The maximum number of users to fetch.
    """
    logging.info(f"Fetching up to {max_users} users from {source_group}...")

    async with TelegramClient(session_dir, API_ID, API_HASH) as client:
        await client.start(PHONE_NUMBER)

        try:
            entity = await client.get_entity(source_group)
            count = 0  # Counter for fetched users

            with open(USER_CSV, mode="w", newline="", encoding="utf-8") as file:
                writer = csv.writer(file)
                writer.writerow(["user_id", "username", "first_name", "last_name"])  # CSV headers

                async for user in client.iter_participants(entity, limit=None, aggressive=True):
                    if count >= max_users:
                        break  # Stop when max_users limit is reached
                    
                    writer.writerow([user.id, user.username or "N/A", user.first_name or "N/A", user.last_name or "N/A"])
                    logging.info(f"✔ User saved: {user.id} | {user.username}")

                    count += 1  # Increase user count
                    await asyncio.sleep(1)  # Delay to prevent rate limits

            logging.info(f"✅ Fetched {count}/{max_users} users from {source_group}. Saved to {USER_CSV}.")

        except Exception as e:
            logging.error(f"❌ Failed to fetch users from {source_group}: {e}")
async def add_users_to_destination_group(destination_group, csvfile):
    """
    Reads users from the CSV file and adds them to the destination group while handling rate limits.
    Before adding, it fetches current members of the destination group and filters out any users already present.
    
    :param destination_group: The destination group username or ID.
    :param csvfile: The CSV file containing the list of user IDs.
    """
    logging.info(f"Adding users to {destination_group} from {csvfile}...")

    async with TelegramClient(session_dir, API_ID, API_HASH) as client:
        await client.start(PHONE_NUMBER)

        try:
            # Get the destination group entity
            dest_entity = await client.get_entity(destination_group)

            # Fetch existing members in the destination group
            existing_user_ids = set()
            async for user in client.iter_participants(dest_entity, limit=None):
                existing_user_ids.add(user.id)
            logging.info(f"Fetched {len(existing_user_ids)} existing users from {destination_group}.")

            # Read users from CSV file and filter out those already in the destination group
            users = []
            with open(csvfile, mode="r", encoding="utf-8") as file:
                reader = csv.reader(file)
                # Skip header row
                header = next(reader, None)
                for row in reader:
                    # Check if the first cell is a valid integer (skip row if not)
                    try:
                        user_id = int(row[0].strip())
                    except ValueError:
                        logging.debug(f"Skipping row with non-numeric user_id: {row}")
                        continue
                    if user_id not in existing_user_ids:
                        users.append(user_id)

            logging.info(f"Filtered CSV: {len(users)} users to add after removing existing members.")

            count = 0
            for index, user_id in enumerate(users, start=1):
                try:
                    logging.info(f"[{index}/{len(users)}] Adding user {user_id} to {destination_group}...")
                    await client(InviteToChannelRequest(dest_entity, [user_id]))
                    logging.info(f"✅ Successfully added user {user_id}.")
                    
                    count += 1
                    if count % BATCH_SIZE == 0:  # Pause after each batch to avoid rate limits
                        logging.info(f"⏳ Waiting {SLEEP_TIME} seconds to avoid rate limits...")
                        await asyncio.sleep(SLEEP_TIME)

                except FloodWaitError as e:
                    logging.warning(f"⚠️ FloodWait: Waiting {e.seconds} seconds...")
                    await asyncio.sleep(e.seconds)

                except UserAdminInvalidError:
                    logging.error(f"❌ Cannot add {user_id}: Bot lacks admin rights.")

                except Exception as e:
                    logging.error(f"❌ Failed to add {user_id}: {e}")

            logging.info(f"✅ Process completed: Added {count} new users to {destination_group}.")

        except Exception as e:
            logging.error(f"❌ Failed to add users to {destination_group}: {e}")

async def get_user_groups(client):
    """Fetch all groups/channels the user is already a member of using get_dialogs."""
    joined_groups = set()
    dialogs = await client.get_dialogs()

    # Filter only groups and channels
    groups = [d for d in dialogs if d.is_group or d.is_channel]
    
    for group in groups:
        username = f"https://t.me/{group.entity.username}" if hasattr(group.entity, "username") and group.entity.username else "private_group"  # Get the group/channel ID
        joined_groups.add(username)
        logging.info(f"Joined group/channel: {group.entity.title} (ID: {username})")

    return joined_groups


async def join_groups():
    async with TelegramClient(session_dir, API_ID, API_HASH) as client:
        await client.start(PHONE_NUMBER)
        me = await client.get_me()
        logging.info(f"Logged in as {me.first_name} (ID: {me.id})")
      

        # Fetch all groups/channels the user is already a member of
        user_groups = await get_user_groups(client)
        logging.info(f"✅ Retrieved {len(user_groups)} joined groups/channels.")
        logging.info(f"✅ Retrieved {user_groups} joined groups/channels.")
        # Read the CSV file containing group information
        with open(CSV_FILENAME, mode="r", newline="", encoding="utf-8") as file:
            reader = csv.reader(file)
            header = next(reader)  # Skip header row
            groups = [row for row in reader]

        # Filter out groups the user is already a member of 1183631472
        filtered_groups = []
        for row in groups:
            phone_number, group_name, username, group_id = row
            
            if username and username in user_groups:
                logging.info(f"⚡ Already a member: {group_name} ({username}) - Skipping")
            else:
                filtered_groups.append(row)

        # Prepare output CSV file
        with open(OUTPUT_CSV, mode="a", newline="", encoding="utf-8") as output_file:
            writer = csv.writer(output_file)
            writer.writerow(header + ["status"])  # Add "status" column

            for index, row in enumerate(filtered_groups, start=2):
                phone_number, group_name, username, group_id = row
                status = ""

                try:
                    if username != "private_group":
                        # Join a public group/channel
                        await client(JoinChannelRequest(username))
                        status = "Joined (public)"
                        logging.info(f"[{index}/{len(filtered_groups)}] ✅ Joined public group: {group_name} ({username})")
                        
                        # Sleep only after a successful join
                        # time.sleep(SLEEP_TIME)
                    else:
                        # Join a private group using its invite hash (group_id)
                        await client(ImportChatInviteRequest(group_id))
                        status = "Joined (private)"
                        logging.info(f"[{index}/{len(filtered_groups)}] ✅ Joined private group: {group_name}")

                        # Sleep only after a successful join
                        # time.sleep(SLEEP_TIME)
                except FloodWaitError as e:
                    logging.warning(f"⚠️ FloodWait: Waiting {e.seconds} seconds...")
                    await asyncio.sleep(e.seconds)
                except UserAlreadyParticipantError:
                    status = "Already a member"
                    logging.info(f"[{index}/{len(filtered_groups)}] ⚡ Already a member: {group_name} ({username})")
                except InviteHashExpiredError:
                    status = "Failed (private) - Invite link expired"
                    logging.error(f"[{index}/{len(filtered_groups)}] ❌ Failed to join private group: {group_name} - Invite link expired")
                except Exception as e:
                    status = f"Failed - {e}"
                    logging.error(f"[{index}/{len(filtered_groups)}] ❌ Failed to join {group_name}: {e}")

                writer.writerow(row + [status])
                time.sleep(SLEEP_TIME)

        logging.info(f"✅ Process completed. Results saved to {OUTPUT_CSV}")

# async def join_groups():
#     async with TelegramClient(session_dir, API_ID, API_HASH) as client:
#         await client.start(PHONE_NUMBER)
#         me = await client.get_me()
#         user_id = me.id
#         logging.info(f"Logged in as {me.first_name} (ID: {user_id})")

#         # Read the CSV file containing group information
#         with open(CSV_FILENAME, mode="r", newline="", encoding="utf-8") as file:
#             reader = csv.reader(file)
#             header = next(reader)  # Skip header row
#             groups = [row for row in reader]

#         # Prepare output CSV file
#         with open(OUTPUT_CSV, mode="a", newline="", encoding="utf-8") as output_file:
#             writer = csv.writer(output_file)
#             writer.writerow(header + ["status"])  # Add "status" column

#             for index, row in enumerate(groups, start=1):
#                 phone_number, group_name, username, group_id = row
#                 status = ""

#                 try:
#                     if username != "private_group":
#                         # Join a public group/channel
#                         entity = await client.get_entity(username)
#                         participants = await client.get_participants(entity)
#                         if not any(p.id == user_id for p in participants):
#                             await client(JoinChannelRequest(username))
#                             status = "Joined (public)"
#                             logging.info(f"[{index}/{len(groups)}] ✅ Joined public group: {group_name} ({username})")
                            
#                             # Sleep only after a successful join
#                             time.sleep(SLEEP_TIME)
#                         else:
#                             status = "Already a member"
#                             logging.info(f"[{index}/{len(groups)}] ⚡ Already a member: {group_name} ({username})")
#                     else:
#                         # Join a private group using its invite hash (group_id)
#                         await client(ImportChatInviteRequest(group_id))
#                         status = "Joined (private)"
#                         logging.info(f"[{index}/{len(groups)}] ✅ Joined private group: {group_name}")
                        
#                         # Sleep only after a successful join
#                         time.sleep(SLEEP_TIME)

#                 except InviteHashExpiredError:
#                     status = "Failed (private) - Invite link expired"
#                     logging.error(f"[{index}/{len(groups)}] ❌ Failed to join private group: {group_name} - Invite link expired")
#                 except Exception as e:
#                     status = f"Failed - {e}"
#                     logging.error(f"[{index}/{len(groups)}] ❌ Failed to join {group_name}: {e}")

#                 writer.writerow(row + [status])

#         logging.info(f"✅ Process completed. Results saved to {OUTPUT_CSV}")


# Async function to join groups
# async def join_groups_tqdm():
#     async with TelegramClient(session_dir, API_ID, API_HASH) as client:
#         await client.start(PHONE_NUMBER)
#         me = await client.get_me()
#         user_id = me.id

#         # Read the CSV file containing group information
#         with open(CSV_FILENAME, mode="r", newline="", encoding="utf-8") as file:
#             reader = csv.reader(file)
#             header = next(reader)  # Skip header row
#             groups = [row for row in reader]

#         # Prepare output CSV file
#         with open(OUTPUT_CSV, mode="w", newline="", encoding="utf-8") as output_file:
#             writer = csv.writer(output_file)
#             writer.writerow(header + ["status"])  # Add "status" column

#             for row in tqdm(groups, desc="Joining groups", unit="group"):
#                 phone_number, group_name, username, group_id = row
#                 status = ""

#                 try:
#                     if username != "private_group":
#                         # Join a public group/channel
#                         entity = await client.get_entity(username)
#                         participants = await client.get_participants(entity)
#                         if not any(p.id == user_id for p in participants):
#                             await client(JoinChannelRequest(username))
#                             status = "Joined (public)"
#                             print(f"✅ Joined public group: {group_name} ({username})")
#                             time.sleep(SLEEP_TIME)
                            
#                         else:
#                             status = "Already a member"
#                             print(f"⚡ Already a member: {group_name} ({username})")
#                     else:
#                         # Join a private group using its invite hash (group_id)
#                         await client(ImportChatInviteRequest(group_id))
#                         status = "Joined (private)"
#                         print(f"✅ Joined private group: {group_name}")
#                         time.sleep(SLEEP_TIME)

#                 except InviteHashExpiredError:
#                     status = "Failed (private) - Invite link expired"
#                     print(f"❌ Failed to join private group: {group_name} - Invite link expired")
#                 except Exception as e:
#                     status = f"Failed - {e}"
#                     print(f"❌ Failed to join {group_name}: {e}")

#                 writer.writerow(row + [status])
              
                

#         print(f"\n✅ Process completed. Results saved to {OUTPUT_CSV}")

def run_telegram():
    asyncio.run(join_groups())

def run_telegram_mov():
    asyncio.run(fetch_users_from_source_group("UT_CHEM", None))
def run_telegram_mov2():
    asyncio.run(add_users_to_destination_group("@searchai090", 'UT_CHEM.csv'))  
    
if __name__ == "__main__":
    p1 = Process(target=run_flask)
    p2 = Process(target=run_telegram)
    # p2 = Process(target=run_telegram_mov2)
    p1.start()
    p2.start()

    p1.join()
    p2.join()