-
Notifications
You must be signed in to change notification settings - Fork 29
/
Copy pathscrapper.py
107 lines (99 loc) · 4.17 KB
/
scrapper.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
# Copyright (C) @TheSmartBisnu
# Channel: https://t.me/itsSmartDev
import re
import os
import asyncio
from urllib.parse import urlparse
from pyrogram.enums import ParseMode
from pyrogram import Client, filters
from config import API_ID, API_HASH, SESSION_STRING, BOT_TOKEN, ADMIN_IDS, DEFAULT_LIMIT, ADMIN_LIMIT
# Initialize the bot and user clients
bot = Client(
"bot_session",
api_id=API_ID,
api_hash=API_HASH,
bot_token=BOT_TOKEN,
workers=1000,
parse_mode=ParseMode.HTML
)
user = Client(
"user_session",
session_string=SESSION_STRING,
workers=1000
)
scrape_queue = asyncio.Queue()
def remove_duplicates(messages):
unique_messages = list(set(messages))
duplicates_removed = len(messages) - len(unique_messages)
return unique_messages, duplicates_removed
async def scrape_messages(client, channel_username, limit, start_number=None):
messages = []
count = 0
pattern = r'\d{16}\D*\d{2}\D*\d{2,4}\D*\d{3,4}'
async for message in user.search_messages(channel_username):
if count >= limit:
break
text = message.text if message.text else message.caption
if text:
matched_messages = re.findall(pattern, text)
if matched_messages:
formatted_messages = []
for matched_message in matched_messages:
extracted_values = re.findall(r'\d+', matched_message)
if len(extracted_values) == 4:
card_number, mo, year, cvv = extracted_values
year = year[-2:]
formatted_messages.append(f"{card_number}|{mo}|{year}|{cvv}")
messages.extend(formatted_messages)
count += len(formatted_messages)
if start_number:
messages = [msg for msg in messages if msg.startswith(start_number)]
messages = messages[:limit]
return messages
@bot.on_message(filters.command(["scr"]))
async def scr_cmd(client, message):
args = message.text.split()[1:]
if len(args) < 2 or len(args) > 3:
await message.reply_text("<b>⚠️ Provide channel username and amount to scrape</b>")
return
channel_identifier = args[0]
limit = int(args[1])
max_lim = ADMIN_LIMIT if message.from_user.id in ADMIN_IDS else DEFAULT_LIMIT
if limit > max_lim:
await message.reply_text(f"<b>Sorry Bro! Amount over Max limit is {max_lim} ❌</b>")
return
start_number = args[2] if len(args) == 3 else None
parsed_url = urlparse(channel_identifier)
channel_username = parsed_url.path.lstrip('/') if not parsed_url.scheme else channel_identifier
try:
chat = await user.get_chat(channel_username)
channel_name = chat.title
except Exception:
await message.reply_text("<b>Hey Bro! 🥲 Incorrect username ❌</b>")
return
temporary_msg = await message.reply_text("<b>Scraping in progress wait.....</b>")
scrapped_results = await scrape_messages(user, chat.id, limit, start_number)
unique_messages, duplicates_removed = remove_duplicates(scrapped_results)
if unique_messages:
file_name = f"x{len(unique_messages)}_{channel_name.replace(' ', '_')}.txt"
with open(file_name, 'w') as f:
f.write("\n".join(unique_messages))
with open(file_name, 'rb') as f:
caption = (
f"<b>CC Scrapped Successful ✅</b>\n"
f"<b>━━━━━━━━━━━━━━━━</b>\n"
f"<b>Source:</b> <code>{channel_name}</code>\n"
f"<b>Amount:</b> <code>{len(unique_messages)}</code>\n"
f"<b>Duplicates Removed:</b> <code>{duplicates_removed}</code>\n"
f"<b>━━━━━━━━━━━━━━━━</b>\n"
f"<b>Card-Scrapper By: <a href='https://t.me/itsSmartDev'>Smart Dev</a></b>\n"
)
await temporary_msg.delete()
await client.send_document(message.chat.id, f, caption=caption)
os.remove(file_name)
else:
await temporary_msg.delete()
await client.send_message(message.chat.id, "<b>Sorry Bro ❌ No Credit Card Found</b>")
if __name__ == "__main__":
user.start()
bot.run()