Код:
Сливаю сюда Fix версию 2025 а не то что сливал на соседнем бордеПредистория:
Понадобилось как то спарсить юзеров в чат, а тг тут залипуху новую придумала скрывать юзеров, присел покубатурил и написал небольшой скриптик
который
позволит нам обойти это.
Как? Да просто! Мы будем парсить историю чата потом вытягивать оттуда юзеров и записывать в файл
2 варианта скрипта:
1 вариант парсим username
- первый вариант доработал там ты сам выбираешь чат откуда парсить (по желанию сделаете под 2 вариант)
2 вариант парсим id
Какой юзать решать вам.
ЯП Python
Библиотека: Telethon (P.S. Скоро перепишу на Pyrogram)
Вставляем данные с my.telegram.org в строки 9-10 в 11 номер аккаунта парсера (важно должен быть в чате откуда парсим)
Строка 108 limit чем выше лимит тем больше истории чата будет спаршено (влияет на время выполнения скрипта)
У вас должно быть более 30 реакций для просмотра скрытого контента.
Python:
import logging
import asyncio
from telethon import TelegramClient
from telethon.tl.types import PeerUser, PeerChat, PeerChannel, InputPeerSelf
from telethon.tl.functions.messages import GetDialogsRequest, GetHistoryRequest
from datetime import datetime, timedelta
from colorama import init, Fore, Back, Style
import time
# Initialize colorama
init(autoreset=True)
# Define the necessary variables: API ID, API Hash, phone number, and session name
api_id = 12345678
api_hash = '123123123123123'
phone_number = '888888888888888'
session_name = 'Fury2025'
# Configure the logging with colors
logging.basicConfig(
level=logging.INFO,
format=f'{Fore.YELLOW}%(asctime)s{Style.RESET_ALL} - {Fore.BLUE}%(levelname)s{Style.RESET_ALL} - %(message)s'
)
# Create the client
client = TelegramClient(session_name, api_id, api_hash)
async def print_banner():
banner = f"""
{Fore.RED}╔══════════════════════════════════════════════════╗
{Fore.RED}║{Fore.YELLOW} TELEGRAM USER SCRAPER WITH @doker420 {Fore.RED}║
{Fore.RED}║{Fore.CYAN} Safe and Rate-Limited Scraping {Fore.RED}║
{Fore.RED}╚══════════════════════════════════════════════════╝
{Style.RESET_ALL}"""
print(banner)
async def main():
await print_banner()
# Connect to Telegram with colored output
print(f"{Fore.GREEN}Connecting to Telegram...{Style.RESET_ALL}")
await client.start()
if not await client.is_user_authorized():
print(f"{Fore.YELLOW}Authorization required...{Style.RESET_ALL}")
await client.send_code_request(phone_number)
code = input(f"{Fore.CYAN}Enter the code received: {Style.RESET_ALL}")
await client.sign_in(phone_number, code)
print(f"{Fore.GREEN}Successfully authorized!{Style.RESET_ALL}")
# Get information about all chats
print(f"{Fore.MAGENTA}Fetching your chats...{Style.RESET_ALL}")
dialogs = await client(GetDialogsRequest(
offset_date=None,
offset_id=0,
offset_peer=InputPeerSelf(),
limit=100,
hash=0
))
# Display the list of chats with colors
print(f"\n{Fore.CYAN}Available Chats:{Style.RESET_ALL}")
for i, dialog in enumerate(dialogs.chats, start=1):
print(f"{Fore.YELLOW}{i}. {Fore.WHITE}{dialog.title}{Style.RESET_ALL}")
try:
chat_index = int(input(f"\n{Fore.CYAN}Enter the number of the chat to parse: {Style.RESET_ALL}")) - 1
selected_chat = dialogs.chats[chat_index]
print(f"{Fore.GREEN}Selected chat: {Fore.YELLOW}{selected_chat.title}{Style.RESET_ALL}")
except (IndexError, ValueError):
print(f"{Fore.RED}Invalid selection!{Style.RESET_ALL}")
return
# Calculate the date for scraping (90 days back)
one_day_ago = int((datetime.now() - timedelta(days=90)).timestamp())
usernames = []
offset_id = 0
limit = 100 # Smaller batches to stay under rate limits
total_users = 0
request_count = 0
start_time = time.time()
# Open the file for writing
with open('usernames.txt', 'w', encoding='utf-8') as f:
try:
print(f"\n{Fore.MAGENTA}Starting to scrape users...{Style.RESET_ALL}")
print(f"{Fore.CYAN}This may take some time. Please wait...{Style.RESET_ALL}")
while True:
# Rate limiting: 1 request per second
if request_count > 0:
await asyncio.sleep(1) # Respect Telegram's rate limits
messages = await client(GetHistoryRequest(
peer=selected_chat,
offset_id=offset_id,
offset_date=one_day_ago,
add_offset=0,
limit=limit,
max_id=0,
min_id=0,
hash=0
))
request_count += 1
if not messages.messages:
break
for message in messages.messages:
if message.from_id is not None:
try:
user = await client.get_entity(message.from_id)
if hasattr(user, 'username') and user.username:
username = user.username
if username not in usernames:
usernames.append(username)
total_users += 1
# Color-coded logging
logging.info(f"{Fore.GREEN}New user found: {Fore.CYAN}@{username}{Style.RESET_ALL} (Total: {Fore.YELLOW}{total_users}{Style.RESET_ALL})")
f.write(f'@{username}\n')
f.flush() # Ensure immediate write to file
except Exception as e:
logging.error(f"{Fore.RED}Error processing user: {str(e)}{Style.RESET_ALL}")
await asyncio.sleep(2) # Additional delay on error
offset_id = messages.messages[-1].id
# Progress update every 10 requests
if request_count % 10 == 0:
elapsed = time.time() - start_time
print(f"{Fore.BLUE}Progress: {Fore.YELLOW}{total_users}{Fore.BLUE} users found | {Fore.YELLOW}{elapsed:.1f}{Fore.BLUE} seconds elapsed{Style.RESET_ALL}")
# Final summary
elapsed = time.time() - start_time
print(f"\n{Fore.GREEN}╔══════════════════════════════════════════════════╗")
print(f"{Fore.GREEN}║{Fore.YELLOW} SCRAPING COMPLETED SUCCESSFULLY! {Fore.GREEN}║")
print(f"{Fore.GREEN}║{Fore.CYAN} Total users collected: {Fore.YELLOW}{total_users}{Fore.CYAN} {Fore.GREEN}║")
print(f"{Fore.GREEN}║{Fore.CYAN} Time elapsed: {Fore.YELLOW}{elapsed:.1f}{Fore.CYAN} seconds {Fore.GREEN}║")
print(f"{Fore.GREEN}║{Fore.CYAN} Usernames saved to {Fore.YELLOW}usernames.txt{Fore.CYAN} {Fore.GREEN}║")
print(f"{Fore.GREEN}╚══════════════════════════════════════════════════╝{Style.RESET_ALL}")
except Exception as e:
print(f"{Fore.RED}An error occurred: {str(e)}{Style.RESET_ALL}")
logging.error(f"Error: {str(e)}")
# Disconnect the client
await client.disconnect()
print(f"{Fore.GREEN}Disconnected from Telegram.{Style.RESET_ALL}")
if __name__ == '__main__':
try:
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
except KeyboardInterrupt:
print(f"\n{Fore.RED}Script interrupted by user.{Style.RESET_ALL}")
except Exception as e:
print(f"{Fore.RED}Fatal error: {str(e)}{Style.RESET_ALL}")У вас должно быть более 30 реакций для просмотра скрытого контента.
