• XSS.stack #1 – первый литературный журнал от юзеров форума

Telegram parse

Fonada

CD-диск
Пользователь
Регистрация
03.04.2021
Сообщения
10
Реакции
4
Народ нужна помощь, имеется файл со ссылками типа t.me/+абонентский номер, кто может посоветовать чем можно спарсить или как извлечь данные аккаунтов телеги, из такого рода ссылок, selenium не видит строк с id аккаунта, и выдает ошибку при парсинге ссылки, остальными способами тоже не вариант, бот в телеге работает только если абонентский номер есть в контактах, но так как там ограничения самой телеги на 15 контактов в сутки вообще не катит такой вариант, открытие ссылок под учеткой телеги в декстоп приложении выдают инфу только userbot, но при наведении на ссылку с абон номером имеется аккаунт, нужно спарсить таким образом около 150 миллионов ссылок, у кого есть какие идеи на пробу напишите, буду благодарен
 
но так как там ограничения самой телеги на 15 контактов
Не совсем понятно что за ограничение?

кто может посоветовать чем можно спарсить или как извлечь данные аккаунтов телеги, из такого рода ссылок
Можно попробовать через Telethon (метод ImportContactsRequest, вот код реализации метода из доки телетона):
Python:
from telethon.sync import TelegramClient
from telethon import functions, types

with TelegramClient(name, api_id, api_hash) as client:
    result = client(functions.contacts.ImportContactsRequest(
        contacts=[types.InputPhoneContact(
            client_id=random.randrange(-2**63, 2**63),
            phone='some string here',
            first_name='some string here',
            last_name='some string here'
        )]
    ))
    print(result.stringify())

P.S. Если пользователь ограничил возможность найти его по номеру телефона в ТГ, метод не отработает.
 
Последнее редактирование:
Не совсем понятно что за ограничение?


Можно попробовать через Telethon (метод ImportContactsRequest, вот код реализации метода из доки телетона):
Python:
from telethon.sync import TelegramClient
from telethon import functions, types

with TelegramClient(name, api_id, api_hash) as client:
    result = client(functions.contacts.ImportContactsRequest(
        contacts=[types.InputPhoneContact(
            client_id=random.randrange(-2**63, 2**63),
            phone='some string here',
            first_name='some string here',
            last_name='some string here'
        )]
    ))
    print(result.stringify())

P.S. Если пользователь ограничил возможность найти его по номеру телефона в ТГ, метод не отработает.
нет суть ограничения в том, что в сутки можно добавить только 15 контактов, в последующем период ожидания в 5 минут, и в последующем добавляет только по 1 контакту, и опять ожидание 5 минут, но это не решит вопроса, с учетом количества обрабатываемых строк, парсинг будет длится несколько лет, а тут время выполнения скрипта играет роль, пробовал несколько сессий для добавления контактов, но тоже не особо будет работать нормально из-за ограничений самой Телеги, также пробовал уже и с временным добавлением контактов, тоже самое ограничение на 15 контактов, через гугл синхронизацию тоже не особо тянет, в сети бытует мнение что через CRM можно но это хер пойми как реализовать нормально никогда не работал с CRM, спасибо
 
нет суть ограничения в том, что в сутки можно добавить только 15 контактов, в последующем период ожидания в 5 минут, и в последующем добавляет только по 1 контакту, и опять ожидание 5 минут, но это не решит вопроса, с учетом количества обрабатываемых строк, парсинг будет длится несколько лет, а тут время выполнения скрипта играет роль, пробовал несколько сессий для добавления контактов, но тоже не особо будет работать нормально из-за ограничений самой Телеги, спасибо

А как именно пробовали? Что не тянет?
ограничение по количеству импортируемых контактов (Google) 500 штук, и проблема устройств, так как для обработки и парсинга всех ссылок и абон номеров, нужно в районе 300 000 аккаунтов гугл, даже с учетом импортирования в телегу максимум что можно сделать это охуеть от реализации данного процесса, поэтому и интересуюсь есть ли более экономичные способы отработать абон номера, на наличие тг аккаунтов
 
ограничение по количеству импортируемых контактов (Google) 500 штук, и проблема устройств, так как для обработки и парсинга всех ссылок и абон номеров, нужно в районе 300 000 аккаунтов гугл, даже с учетом импортирования в телегу максимум что можно сделать это охуеть от реализации данного процесса, поэтому и интересуюсь есть ли более экономичные способы отработать абон номера, на наличие тг аккаунтов
А не пробовали генерировать VCF (Vcard) с последующим импортом контактов в память телефона и уже после этого синхронизировать контакты с ТГ и чекать их? Как это будет выглядеть при большом массиве не представляю, но допустим блоками по 30к контактов на 1 акк телеги.
 
Последнее редактирование:
Нашел менее затратный способ чека номеров в телеге, ща пишу софт. Если все получится опубликую в паблике на форуме.
как опубликуешь, напиши в пм пожалуйста а так в целом было бы отлично если все получится
 
как опубликуешь, напиши в пм пожалуйста а так в целом было бы отлично если все получится
Да, пришлось перекурить доку по MTProto 100500 раз. Но таким способом в теории можно прочекать и 10000+ номеров с одного акка. Сразу скажу, не через добавление в контакты. =) Всё намного проще оказалось.

1729141684874.jpeg


А это беспонтовая версия на пирограме чекера через добавление в контакты (которая не заслуживает отдельного топика), на неё я забиваю хрен. Пошел писать другую затею на TDLib. Оставляю на память предкам:
Python:
from pyrogram import Client
from pyrogram.errors import PhoneNumberInvalid, FloodWait
from pyrogram.raw import functions, types
import random as rand
import asyncio
import os
import requests
from requests.exceptions import ProxyError, ConnectTimeout, ConnectionError, ReadTimeout
import socket


# Читаем список номеров телефона
with open('tel_base.txt', 'r') as f:
    PHONE_NUMBER = f.read().split("\n")

# Читаем список прокси из файла и преобразуем его в список словарей
def load_proxies(filename):
    proxies = []
    with open(filename, 'r') as file:
        for line in file:
            if line.strip():  # Пропускаем пустые строки
                ip, port = line.strip().split(':')
                proxies.append({"hostname": ip, "port": int(port)})
    return proxies

# Функция для проверки работоспособности прокси
def is_proxy_working(proxy):
    test_url = "http://ifconfig.me/ip"  # URL для проверки
    proxies = {
        "http": f"socks5://{proxy['hostname']}:{proxy['port']}",
        "https": f"socks5://{proxy['hostname']}:{proxy['port']}"
    }
    try:
        # Отправляем тестовый запрос через прокси
        response = requests.get(test_url, proxies=proxies, timeout=1)
        if response.status_code == 200:
            print(f"Прокси {proxy['hostname']}:{proxy['port']} работает.")
            return True
        else:
            print(f"Прокси {proxy['hostname']}:{proxy['port']} не отвечает.")
            return False
    except (ProxyError, ConnectTimeout, ConnectionError, ReadTimeout, socket.timeout):
        print(f"Прокси {proxy['hostname']}:{proxy['port']} недоступен.")
        return False

# Фильтруем рабочие прокси
def get_working_proxies(proxies):
    working_proxies = [proxy for proxy in proxies if is_proxy_working(proxy)]
    return working_proxies

# Функция проверки номеров телефона
async def check_phone_numbers(phone_batch, session, proxy):
    async with Client(
        session,
        app_version="5.6",
        device_model="PC",
        system_version="Linux Ubuntu",
        proxy=proxy  # Передаем параметры прокси
    ) as app:
        for tel in phone_batch:
            await asyncio.sleep(rand.randint(24, 68))  # Случайная задержка, для того чтобы ограничить количество запросов к API по интервалу времени
            try:
                # Добавляем контакт
                result = await app.invoke(
                    functions.contacts.ImportContacts(
                        contacts=[types.InputPhoneContact(
                            client_id=0,  # Уникальный ID для клиента, можно оставить 0
                            phone=tel,
                            first_name="Check",
                            last_name="User"
                        )]
                    )
                )

                # Проверка контакта
                contact_info = result.users
                if contact_info:
                    print(f"Номер {tel} активен в Telegram.")
                    # Если контакт валидный, записываем его в good_tel.txt
                    with open("good_tel.txt", "a") as file:
                        file.write(f"{tel}\n")

                    # Удаляем контакт
                    await app.invoke(
                        functions.contacts.DeleteContacts(
                            id=[types.InputUser(user_id=user.id, access_hash=user.access_hash) for user in contact_info]
                        )
                    )
                    print(f"Контакт с номером {tel} был удален.")
                else:
                    print(f"Номер {tel} не найден в Telegram.")
                    with open("bad_tel.txt", "a") as file:
                        file.write(f"{tel}\n")

            except PhoneNumberInvalid:
                print(f"Неверный формат телефонного номера: {tel}.")
            except FloodWait as e:
                print(f"Превышен лимит запросов. Ожидайте {e.value} секунд.")

# Основная функция для обработки номеров партиями по 12 номеров
async def process_numbers_in_batches(working_proxies):
    # Получаем список всех .session файлов из папки "sessions"
    session_folder = os.path.join(os.getcwd(), "sessions")
    session_files = [f for f in os.listdir(session_folder) if f.endswith(".session")]

    if not session_files:
        print("Не найдено ни одного файла сессии.")
        return


    batch_size = 12
    session_count = len(session_files)  # Количество доступных сессий
    required_sessions = (len(PHONE_NUMBER) + batch_size - 1) // batch_size  # Количество необходимых сессий

    if session_count < required_sessions:
        print(f"Недостаточное количество сессий для обработки всех номеров! Нужно: {required_sessions}, но доступно только: {session_count}.")
        return

    session_index = 0  # Индекс сессии

    for i in range(0, len(PHONE_NUMBER), batch_size):
        phone_batch = PHONE_NUMBER[i:i + batch_size]

        # Выбираем сессию для текущей партии номеров
        session = os.path.join(session_folder, session_files[session_index])

        # Выбираем случайный рабочий прокси
        proxy = rand.choice(working_proxies)

        # Подготавливаем параметры прокси для pyrogram
        proxy_params = {
            "scheme": "socks5",  # Указываем протокол SOCKS5
            "hostname": proxy["hostname"],
            "port": proxy["port"]
        }

        # Вызываем функцию для проверки номера с текущей сессией и прокси
        #await check_phone_numbers(phone_batch, session, proxy_params)

        # Обновляем индекс сессии
        session_index += 1

# Загружаем список прокси из файла и запускаем обработку номеров
proxies = load_proxies('proxy_list.txt')
# Проверяем рабочие прокси перед началом работы
print("Проверяем рабочие прокси перед началом работы")
working_proxies = get_working_proxies(proxies) # Получаем список валидных прокси из словарей
asyncio.run(process_numbers_in_batches(working_proxies)) # Запускаем функцию чека номеров по 12 штук.
 
Последнее редактирование:
Да, пришлось перекурить доку по MTProto 100500 раз. Но таким способом в теории можно прочекать и 10000+ номеров с одного акка. Сразу скажу, не через добавление в контакты. =) Всё намного проще оказалось.
Напишу тебе в личку позже обсудим, я вроде как написал тоже, обменяемся мнениями
 


Напишите ответ...
  • Вставить:
Прикрепить файлы
Верх