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 штук.
