Гайд Создаем многопоточный чекер для «Fragment»

[wlux.net]

Суть​

Для начала устанавливаем библиотеки для парсинга и самого чека (также beauty улучшение):

pip install bs4 requests tqdm

Отлично, теперь нам надо узнать, как чекать. Я сделал так. Открыл Fragment.



Ввел свой юзернейм и меня перебросило на

Пожалуйста, войдите или зерегистрируйтесь, чтобы увидеть скрытый текст.

.



а если перейти по кнопке юзернейма, то откроется

Пожалуйста, войдите или зерегистрируйтесь, чтобы увидеть скрытый текст.

.



Я подумал, что если поставить желаемый юзер в

Пожалуйста, войдите или зерегистрируйтесь, чтобы увидеть скрытый текст.

, то нас перенесет на страницу покупки.

Возьмем юзернейм, который никто бы не выбрал. Например @csdjkhhbchsbdjch. Таким образом, наша ссылка будет выглядеть так:

Пожалуйста, войдите или зерегистрируйтесь, чтобы увидеть скрытый текст.

. Она перенесет нас на

Пожалуйста, войдите или зерегистрируйтесь, чтобы увидеть скрытый текст.

.

О чем это говорит? Только что, мы с вами поняли, что если юзернейм свободный, то нас переносит на

Пожалуйста, войдите или зерегистрируйтесь, чтобы увидеть скрытый текст.

.

Приступаем к написанию кода.

Импортируем нужные библиотеки:

pip install bs4 requests tqdm
from bs4 import BeautifulSoup # сам парс
import requests # запросы
from tqdm import tqdm # для бара
from concurrent.futures import ThreadPoolExecutor, Executor # многопоточность
import time # для времени

Теперь напишем сам класс чекера:

class UserName():



  def __init__(self, lines: list, threads: int) -> None:



    self.lines = lines # строчки



    self.threads = threads # колво потоков



    self.result = [] # результаты







  def run(self): # запуск



    total_lines = len(self.lines) # колво строк



    with ThreadPoolExecutor(max_workers=self.threads) as executor: # модуль для потоков



      list(tqdm(executor.map(self.parse, self.lines), total=total_lines, ascii=True, desc=" Check ")) # сам чек с визуализацией







  def parse(self, name): # сам парсинг



    name = name.strip() # удаляем лишние пробелы



    self.r = requests.get(f"https://fragment.com/username/{name}") # кидаем запрос на url



    if self.r.url == f"https://fragment.com/?query={name}": # про то что я вам говорил



      self.result.append(f" Username @{name} | Free\n")



      return # просто пустой возврат чтобы код не продолжал работу



    else: # если у нас не идет переброс на другой юрл



      try:



        bs = BeautifulSoup(self.r.text, "lxml") # парсим сайт



        res = bs.find("span", "tm-section-header-status").text # ищем текст внутри маленького окошка



        if res.lower() in ["sold", "available", "taken"]: # проверяем текст



          self.result.append(f" Username @{name} | {res}\n") # добавляем результат



          return



      except:



        self.result.append(f" Username @{name} | Unknown error\n") # просто обработка ошибок

Все расписано в комментах к коду

А теперь переходим к самому интерфейсу и запуску:

def main():



  start_time = time.time() # для подсчета времени



  _lines = input(" Загрузи файл со строками: ")



  _lines2 = input(" Загрузи файл со строками для результата: ")



  threads = input(" Введи кол-во потоков: ")



  lines = open(_lines, "r").readlines()



  checker = UserName(lines, int(threads))



  checker.run() # запуск чекера



  open(_lines2, "w").writelines(checker.result) # запись результата в файл



  print(f" Lead time: {time.time() - start_time}")







main()