Разбор парсера Solana, EVM щитков
Вот и разбор нашего парсера щитков! Когда мы впервые рассказали о нем, мы и представить не могли, что он вызовет такой сильный отклик. Спасибо всем, кто проявил интерес к нашей работе ❤️
Этот пост будет насыщен полезной информацией, так что приготовьтесь впитывать. В статье мы будем разбирать довольно сложные конструкции, которые мы постарались максимально просто и понятно объяснить. Однако, если какие-то моменты покажутся сложными, не переживай — ты всегда можешь обратиться к нам в комментариях или спросить ChatGPT за дополнительными объяснениями.
Что делает наш парсер?
Наш скрипт обрабатывает список никнеймов аккаунтов X (ex-Twitter) из файла usernames.txt. Он последовательно проходит по каждому аккаунту, анализируя последние пять постов на наличие Solana или EVM контракта — проверяя как текст, так и изображения.
После проверки поста скрипт сохраняет его последний ID, чтобы при следующем запуске продолжить с того места, где остановился, тем самым анализируя только новые твиты. Если в посте найден контракт, пользователь получает уведомление в Telegram через бота.
Когда все аккаунты обработаны, скрипт переходит в режим ожидания до следующего парсинга.
Что ты должен уметь?
Как и говорили, мы написали статью, ориентированную на новичков. Все, что тебе нужно — базовые знания синтаксиса Python. А если ты хотя бы раз сталкивался с классами или слышал про ООП, то разбираться в материале будет еще проще, но это не обязательно.
Также ознакомься с нашей первой статьей, в которой мы подробно объясняем, как установить Python, настроить VS Code и написать свой первый скрипт. В другой статье рассказывается, как создать скрипт для распределения баланса с OKX на кошельки, включая работу с API, HTTP, HMAC и другое.
Где посмотреть код скрипта?
Весь исходный код парсера доступен на GitHub
Оглавление
2. Смотрим на структуру скрипта
Что такое классы
Перед тем как начать разбор, мы должны разобраться с тем, что такое классы.
Классы — это чертежи (шаблоны) для создания объектов. Они позволяют объединять данные (переменные) и функции (методы), работающие с этими данными, в одно целое.
Представь, что ты занимаешься разработкой игры, и тебе нужно моделировать персонажа. Персонаж может иметь различные характеристики, например, здоровье, силу и уровень. Он также может выполнять действия, например, атаковать врага или лечиться.
Ты можешь создать класс для описания такого персонажа. Например:
class Character:
def __init__(self, name, health, strength):
self.name = name
self.health = health
self.strength = strength
def attack(self, enemy):
print(f"{self.name} атакует {enemy.name} с силой {self.strength}!")
enemy.health -= self.strength
def heal(self):
print(f"{self.name} лечит себя!")
self.health += 10Здесь класс Character описывает персонажа с его характеристиками и действиями.
Теперь погнали создадим объект этого персонажа и повзаимодействуем с ним:
hero = Character("Герой", 100, 15)
villain = Character("Злодей", 80, 10)
hero.attack(villain)
villain.heal()Мы создаем переменные hero и villain, каждая из которых является экземпляром класса Character. При создании объекта передаем имя персонажа, его здоровье и силу. При вызове класса срабатывает его конструктор, который создает объект и возвращает его. Полученные объекты мы сохраняем в переменные и затем используем их для взаимодействия.
Пока можешь не вдаваться в подробности кода, так как мы еще разберем это все при дальнейшем разборе парсера.
Функции НЕ хуже классов
Теперь давай разберемся, почему функции не хуже классов — они просто применяются в разных ситуациях.
Если бы тебе нужно было просто посчитать урон, который наносит персонаж врагу, то проще было бы использовать функцию, например:
def calculate_damage(attack, defense):
return attack - defenseЗачем создавать сложный класс, если у тебя есть простая задача, которая не требует хранения состояния объектов и взаимодействия между ними?
Смотрим на структуру скрипта
Начнем наш разбор с структуры скрипта. Это важно, так как нужно обеспечить не только функциональность, но и грамотно организовать структуру проекта. Это помогает поддерживать код и упрощает его расширение.
Поэтому далее мы объясним, за что отвечает каждый файл и почему проект организован именно таким образом.
main.py: Главный исполняемый файл проекта. Именно с него начинается выполнение софта. В этом файле происходит инициализация необходимых компонентов и запуск основных процессов парсинга.
settings.py: Файл конфигурации, содержащий настройки проекта. Здесь определены параметры, которые можно легко изменить без необходимости редактирования кода.
utils.py: Файл с вспомогательными функциями и утилитами, которые используются в разных частях проекта.
requirements.txt: Список зависимостей проекта. В этом файле перечислены все внешние библиотеки и модули, которые необходимы для корректной работы скрипта.
x_usernames.txt: Текстовый файл, содержащий список имен пользователей X.
modules: Папка, предназначенная для хранения дополнительных модулей. Разделение кода на модули способствует его повторному использованию и облегчает поддержку.
modules/parser.py: Обрабатывает твиты полученные из модуля x.py, извлекает из них контрактные адреса (из текста или изображений), используя utils.py и отправляет уведомления пользователю через telegram.py
modules/telegram.py: Отправляет сообщения в Telegram-чат, используя API Telegram.
modules/tweet_id_storage.py: Отвечает за сохранение и извлечение идентификаторов последнего обработанного твита для каждого пользователя в JSON формате.
modules/x.py: Модуль для взаимодействия с API X, обрабатывая запросы для получения данных о пользователях и их твитах.
Подобная структура обеспечивает модульность, удобство настройки и поддержку кода. Логика разделена на запуск, конфигурацию, парсинг, взаимодействие с API, хранение данных и отправку уведомлений, что упрощает доработку и масштабирование.
Если бы все было в одном файле, то ты бы запарился его поддерживать и масштабировать. Когда скрипт не большой, то это допустимо, но в нашем же случае — нет.
Приступаем к разбору кода
Ну что, теперь наберись терпения, впереди тебя ждет разбор каждого файла и каждой строчки кода, так что ты точно во всем разберешься.
settings.py
Начнем с чего-то простого, например, с конфигурационного файла.
Этот файл содержит настройки для работы парсера. В нем определены параметры прокси (PROXY), поведение парсинга (MAX_RETRIES, CHECK_RETWEETS, PARSING_INTERVAL_MINUTES), а также доступ к API X (Twitter) через BEARER_TOKEN. Кроме того, файл включает настройки для отправки уведомлений в Telegram (TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID).
# Proxy
PROXY = (
"scheme://username:password@host:port" # оставь как есть, если прокси не нужен
)
# Parse settings
MAX_RETRIES = 3 # сколько раз повторять парсинг при ошибках
CHECK_RETWEETS = True # проверять ли ретвиты на наличие контрактов (увеличивает количество запросов)
PARSING_INTERVAL_MINUTES = 60 # сколько минут ждать после прохода всех аккаунтов перед повторным парсингом
# X settings
BEARER_TOKEN = "your_bearer_token"
# Telegram settings
TELEGRAM_BOT_TOKEN = "your_bot_token"
TELEGRAM_CHAT_ID = "your_chat_id"Думаю, пояснения к коду здесь не понадобятся, поскольку он достаточно понятен даже для новичка. Единственное, стоит отметить, что настройки мы записали в верхнем регистре. В Python нет строгих констант, но существует соглашение — писать неизменяемые значения в верхнем регистре. Поскольку эти настройки не будут изменяться в ходе выполнения программы, они считаются константами, поэтому мы записываем их в верхнем регистре.
main.py
Этот файл содержит основной код для запуска парсинга. Он загружает список имен пользователей из файла x_usernames.txt и периодически выполняет их обработку с помощью парсера (класс XParser). После завершения парсинга всех аккаунтов скрипт делает паузу на заданное время, а затем снова выполняет парсинг аккаунтов.
from time import sleep
from modules.parser import XParser
from settings import PARSING_INTERVAL_MINUTES
def main() -> None:
parser = XParser()
with open("x_usernames.txt", "r") as file:
usernames = file.readlines()
while True:
for username in usernames:
print(f"Начат парсинг аккаунта {username.strip()}")
parser.parse_account(username.strip())
print(f"Завершен парсинг аккаунта {username.strip()}")
print(
f"Парсинг завершен. Ждем {PARSING_INTERVAL_MINUTES} минут до следующего парсинга..."
)
sleep(PARSING_INTERVAL_MINUTES * 60)
if __name__ == "__main__":
main()from time import sleep from modules.parser import XParser from settings import PARSING_INTERVAL_MINUTES parser = XParser()
sleep— встроенная функция, приостанавливающая выполнение программы (используется для задержки между циклами).XParser— класс для парсинга аккаунтов (код находится вmodules/parser.py).PARSING_INTERVAL_MINUTES— настройка, определяющая интервал ожидания перед следующим запуском (берётся изsettings.py)
def main() -> None:
Создаем функцию main(), которая будет описывать основную логику работы программы.
Знак -> и любой тип данных за ним (как в нашем примере -> None) означает, что функция возвращает результат с типом None.
parser = XParser()
Создаем объект parser, используя класс XParser, тот самый шаблон, по которому строится объект. Когда мы его вызываем, то у нас создается объект XParser и записывается в переменную parser.
Ты еще увидишь класс XParser далее, когда будем разбирать модуль parser.py
with open("x_usernames.txt", "r") as file:
usernames = file.readlines()Файл x_usernames.txt открывается в режиме чтения ("r"), а все строки загружаются в список usernames. Использование with open(...) автоматически закрывает файл после завершения работы с ним.
while True:
Запускаем бесконечный цикл, который работает, пока скрипт не остановят вручную (Ctrl + C) или не произойдет какое нибудь событие, которые заставит скрипт выйти из цикла. В нашем же случае, мы будем ждать пока пользователь сам не остановит скрипт.
Обработка каждого пользователя
for username in usernames:
Проходим по списку пользователей, считанных из файла.
print(f"Начат парсинг аккаунта {username.strip()}")
parser.parse_account(username.strip())
print(f"Завершен парсинг аккаунта {username.strip()}")Метод username.strip() удаляет лишние пробелы и символы \n. Затем выводится сообщение о начале парсинга, после чего запускается метод parse_account() для обработки аккаунта. По завершении парсинга отображается сообщение об окончании процесса.
Ожидание перед следующим циклом
print(f"Парсинг завершен. Ждем {PARSING_INTERVAL_MINUTES} минут до следующего парсинга...")
sleep(PARSING_INTERVAL_MINUTES * 60)Выводится сообщение о завершении итерации, затем программа делает паузу (sleep) на указанное количество минут. Мы умножаем PARSING_INTERVAL_MINUTES на 60, так как sleep() ожидает время в секундах
if __name__ == "__main__":
main()Этот блок гарантирует, что main() запустится только если скрипт выполняется напрямую (python main.py), а не импортируется как модуль (import main).
utils.py
Здесь, как мы упомянали ранее, хранятся вспомогательные функции. Мы их будем использовать в модуле parser.py
В частности, здесь лежат функции по поиску адресов в тексте и картинке:
Поиск адреса контракта:
Функция анализирует текст, чтобы найти адреса контрактов Ethereum или Solana. Если такой адрес находится, он возвращается, иначе функция сообщает, что адресов нет.
Извлечение текста из изображения:
Функция загружает изображение по указанной ссылке, распознает текст на этом изображении с помощью технологии OCR и возвращает его в виде строки.
import re
from io import BytesIO
import pytesseract
import requests
from PIL import Image
def get_contract_address(text: str = "") -> str | None:
evm_pattern = r"0x[a-fA-F0-9]{40}"
solana_pattern = r"[1-9A-HJ-NP-Za-km-z]{32,44}"
evm_match = re.search(evm_pattern, text, re.MULTILINE | re.IGNORECASE | re.DOTALL)
solana_match = re.search(
solana_pattern, text, re.MULTILINE | re.IGNORECASE | re.DOTALL
)
if evm_match:
return evm_match.group(0)
elif solana_match:
return solana_match.group(0)
else:
return None
def extract_text_from_image(image_url: str) -> str:
with requests.request("GET", image_url) as response:
img = Image.open(BytesIO(response.content))
text = pytesseract.image_to_string(img)
return textimport re from io import BytesIO import pytesseract import requests from PIL import Image
re— модуль для работы с регулярными выражениями.BytesIO— позволяет работать с бинарными данными в оперативной памяти, как с файловым объектом.pytesseract— библиотека для распознавания текста на изображениях с помощью Tesseract OCR.requests— библиотека для выполнения HTTP-запросов.PIL.Image— класс из библиотеки Pillow, используется для работы с изображениями.
2. Функция get_contract_address()
def get_contract_address(text: str = "") -> str | None:
Эта функция ищет и извлекает из переданного текста адреса контрактов EVM и Solana.
Регулярные выражения для поиска
Регулярное выражение — это специальный шаблон для поиска текста в строках. Это своего рода "поиск по маске", который может быть очень гибким. Например, регулярка может искать все числа в тексте или проверять, соответствует ли строка конкретному формату (например, email-адресу).
Если ты никогда не сталкивался с регулярными выражениями, то, во-первых, стоит немного изучить эту тему. Во-вторых, ты можешь воспользоваться ChatGPT, который поможет тебе составить нужный шаблон.
evm_pattern = r"0x[a-fA-F0-9]{40}"
solana_pattern = r"[1-9A-HJ-NP-Za-km-z]{32,44}"evm_pattern— шаблон для EVM-адресов (начинается с0xи содержит 40 шестнадцатеричных символов).solana_pattern— шаблон для Solana-адресов (от 32 до 44 символов, исключая некоторые буквы, чтобы избежать ошибок в идентификации).
evm_match = re.search(evm_pattern, text, re.MULTILINE | re.IGNORECASE | re.DOTALL)
solana_match = re.search(
solana_pattern, text, re.MULTILINE | re.IGNORECASE | re.DOTALL
)re.search() ищет первое совпадение с регулярным выражением в строке.
Флаги MULTILINE, IGNORECASE, DOTALL обеспечивают поиск по всему тексту без учёта регистра. Они объединяются с помощью |, что позволяет применять сразу несколько флагов в одном вызове функции.
if evm_match:
return evm_match.group(0)
elif solana_match:
return solana_match.group(0)
else:
return NoneМы проверяем, есть ли результат для каждого поиска по регулярному выражению. Если evm_match содержит результат, то возвращаем первый найденный адрес. Если же нет, проверяем solana_match. В случае, когда ни одно из поисков не дало результата, возвращаем None.
3. Функция extract_text_from_image()
def extract_text_from_image(image_url: str) -> str:
Эта функция получает текст с изображения, загруженного по указанному URL. Этот URL мы получаем из твита при его обработке и принимаем в этой функции как аргумент.
with requests.request("GET", image_url) as response:
img = Image.open(BytesIO(response.content))Метод requests.request("GET", image_url) выполняет HTTP-запрос для загрузки изображения. Затем BytesIO(response.content) преобразует полученные бинарные данные в файловый объект. После этого Image.open(...) открывает изображение с помощью библиотеки Pillow.
text = pytesseract.image_to_string(img)
Чтобы извлечь текст с изображения, нужно воспользоваться сторонней библиотекой, такой как pytesseract.
Функция pytesseract.image_to_string(img) использует Tesseract OCR для распознавания текста.
Tesseract OCR — это инструмент, который читает текст с изображений и преобразует его в текстовый формат.
return text
Функция возвращает распознанный текст в виде строки.
modules/parser.py
Модуль parser.py использует другой модуль, x.py, для взаимодействия с API платформы X (Twitter). Он использует методы этого модуля для получения информации о пользователе, его твитах и медиафайлах, а затем анализирует твиты на наличие контрактов в тексте или изображениях.
from typing import List
from modules.telegram import send_message_to_user
from modules.tweet_id_storage import TweetIdStorage
from modules.x import XAPIHandler
from settings import CHECK_RETWEETS
from utils import extract_text_from_image, get_contract_address
class XParser:
def __init__(self):
self.x_api_handler = XAPIHandler()
self.tweet_id_storage = TweetIdStorage()
def parse_account(self, username: str) -> None:
user_id = self.x_api_handler.fetch_user_id_by_username(username)
fetched_tweets = self.x_api_handler.fetch_user_tweets(username, user_id)
if not fetched_tweets.get("data"):
print(f"Нет новых твитов от {username}")
return
for tweet in reversed(fetched_tweets["data"]):
if CHECK_RETWEETS and tweet.get("referenced_tweets"):
for ref_tweet in tweet["referenced_tweets"]:
ref_tweet = self.x_api_handler.get_tweet_by_id(ref_tweet["id"])
ref_tweet_data = ref_tweet.get("data", {})
media = ref_tweet.get("includes", {}).get("media", [])
image_urls = []
for media_item in media:
if media_item.get("url"):
image_urls.append(media_item["url"])
self._process_tweet(
username,
ref_tweet_data["id"],
ref_tweet_data.get("note_tweet", {}).get("text")
or ref_tweet_data.get("text", ""),
image_urls,
)
else:
self.tweet_id_storage.save_last_tweet_id(username, tweet["id"])
media = fetched_tweets.get("includes", {}).get("media", [])
media_keys = tweet.get("attachments", {}).get("media_keys", [])
image_urls = []
for media_item in media:
if media_item["media_key"] in media_keys and media_item.get("url"):
image_urls.append(media_item["url"])
self._process_tweet(
username,
tweet["id"],
tweet.get("note_tweet", {}).get("text") or tweet.get("text", ""),
image_urls,
)
def _process_tweet(
self,
username: str,
tweet_id: str = "",
tweet_text: str = "",
image_urls: List[str] = [],
) -> None:
if contract := get_contract_address(tweet_text):
self._notify_contract_found(contract, username, tweet_id, "text")
return
for image_url in image_urls:
image_text = extract_text_from_image(image_url)
if contract := get_contract_address(image_text):
self._notify_contract_found(contract, username, tweet_id, "image")
def _notify_contract_found(
self, contract: str, username: str, tweet_id: str, source: str
) -> None:
tweet_url = f"https://x.com/{username}/status/{tweet_id}"
if source == "text":
message = f"Контракт найден в тексте твита: `{contract}`\n\n[Ссылка на твит]({tweet_url})"
else:
message = f"Контракт найден в изображении твита и может быть неправильным: `{contract}`\n\n[Ссылка на твит]({tweet_url})"
send_message_to_user(message)from typing import List from modules.telegram import send_message_to_user from modules.tweet_id_storage import TweetIdStorage from modules.x import XAPIHandler from settings import CHECK_RETWEETS from utils import extract_text_from_image, get_contract_address
Listизtyping— для аннотации типов данных.send_message_to_user— функция для отправки сообщений пользователям.TweetIdStorage— класс для хранения ID обработанных твитов.XAPIHandler— класс для работы с API X.CHECK_RETWEETS— настройка изsettings, указывающая, нужно ли проверять ретвиты.extract_text_from_image— функция для извлечения текста из изображений.get_contract_address— функция для поиска контрактных адресов в тексте.
class XParser:
Этот класс является шаблоном для создания объектов, отвечающих за парсинг твитов.
def __init__(self):
self.x_api_handler = XAPIHandler()
self.tweet_id_storage = TweetIdStorage()При создании объекта XParser автоматически вызывается его конструктор __init__(), который инициализирует необходимые компоненты:
self.x_api_handler = XAPIHandler()— создаётся объект для работы с API X. КлассXAPIHandlerявляется основным для взаимодействия с этим API.self.tweet_id_storage = TweetIdStorage()— создаётся объект для работы с "базой данных" (обычный JSON файл), в котором хранятся последние ID твитов для каждого аккаунта. С помощью этого объекта мы можем сохранять и читать последние ID твитов.
Немного про __init__ и атрибуты:
В методе __init__() мы задаём атрибуты для объектов, используя префикс self., чтобы привязать их к конкретному экземпляру класса. Такие атрибуты являются уникальными для каждого экземпляра: изменение значения, например, self.a в одном объекте, не повлияет на значение этого атрибута в другом объекте, даже если они принадлежат одному классу.
Кроме того, существуют классовые атрибуты, которые являются общими для всех экземпляров класса. Ты сможешь увидеть их применение при разборе модуля tweet_id_storage.py
def parse_account(self, username: str) -> None:
Этот метод получает и обрабатывает твиты пользователя.
user_id = self.x_api_handler.fetch_user_id_by_username(username) fetched_tweets = self.x_api_handler.fetch_user_tweets(username, user_id)
Получаем user_id по username, затем загружаем список твитов пользователя. Для этого мы используем объект x_api_handler класса XAPIHandler
if not fetched_tweets.get("data"):
print(f"Нет новых твитов от {username}")
returnЕсли новых твитов нет, программа сообщает об этом и выходит из функции.
for tweet in reversed(fetched_tweets["data"]):
Проходим по списку твитов в обратном порядке (от старых к новым), чтобы обрабатывать их в хронологическом порядке.
if CHECK_RETWEETS and tweet.get("referenced_tweets"):
for ref_tweet in tweet["referenced_tweets"]:
ref_tweet = self.x_api_handler.get_tweet_by_id(ref_tweet["id"])
ref_tweet_data = ref_tweet.get("data", {})
media = ref_tweet.get("includes", {}).get("media", [])
image_urls = []
for media_item in media:
if media_item.get("url"):
image_urls.append(media_item["url"])Если CHECK_RETWEETS включен и твит является ретвитом, мы перебираем все ссылки на оригинальные твиты, получаем их ID и загружаем данные оригинального твита с помощью метода get_tweet_by_id(). Затем мы извлекаем его текст и проверяем, содержит ли он медиафайлы. Если у оригинального твита есть вложенные изображения, мы формируем список ссылок на них и передаём в дальнейшую обработку. Таким образом, если твит является ретвитом, мы анализируем не только сам ретвит, но и его оригинальный источник.
Структуру ответа API мы знаем из официальной документации X, что позволяет нам корректно обрабатывать данные, такие как referenced_tweets, media, text и другие ключи в JSON-ответе.
else:
self.tweet_id_storage.save_last_tweet_id(username, tweet["id"])Если это не ретвит, сохраняем ID твита с помощью TweetIdStorage. Этот класс позволяет нам записывать и читать последние обработанные tweet_id из JSON-файла, который служит своеобразной базой данных. Это предотвращает повторную обработку уже разобранных твитов при последующих запусках.
media = fetched_tweets.get("includes", {}).get("media", [])
media_keys = tweet.get("attachments", {}).get("media_keys", [])
image_urls = []
for media_item in media:
if media_item["media_key"] in media_keys and media_item.get("url"):
image_urls.append(media_item["url"])Из полученного списка media извлекаем все элементы, у которых есть media_key, совпадающий с media_keys твита, и у которых присутствует поле url. Это позволяет нам получить только те изображения, которые действительно принадлежат данному твиту, а не другим вложенным элементам. Таким образом, мы формируем список image_urls, содержащий ссылки на изображения, прикреплённые к твиту.
self._process_tweet(
username,
tweet["id"],
tweet.get("note_tweet", {}).get("text") or tweet.get("text", ""),
image_urls,
)Передаём данные в _process_tweet().
def _process_tweet(
self,
username: str,
tweet_id: str = "",
tweet_text: str = "",
image_urls: List[str] = [],
) -> None:Проверяет, есть ли контрактный адрес в тексте или на изображениях.
if contract := get_contract_address(tweet_text):
self._notify_contract_found(contract, username, tweet_id, "text")
returnКонструкция contract := get_contract_address(tweet_text) использует оператор моржового присваивания (:=), который одновременно присваивает значение переменной contract и проверяет его. Это эквивалентно записи:
contract = get_contract_address(tweet_text)
if contract:
self._notify_contract_found(contract, username, tweet_id, "text")
returnЕсли get_contract_address(tweet_text) вернёт не None, contract получит это значение, и условие выполнится. Это сокращает код и делает его компактнее.
Если контракт найден, вызываем _notify_contract_found() и выходим из функции.
Поиск контракта на изображениях
for image_url in image_urls:
image_text = extract_text_from_image(image_url)
if contract := get_contract_address(image_text):
self._notify_contract_found(contract, username, tweet_id, "image")Если контракт найден на изображении, отправляем уведомление.
5. Метод _notify_contract_found()
def _notify_contract_found(
self, contract: str, username: str, tweet_id: str, source: str
) -> None:Отправляет сообщение пользователю, если найден контракт.
tweet_url = f"https://x.com/{username}/status/{tweet_id}"message = f"Контракт найден в {'тексте' if source == 'text' else 'изображении'} твита: `{contract}`\n\n[Ссылка на твит]({tweet_url})"Формируем сообщение в зависимости от того, где найден контракт. Отметим, что в f строках, помимо переменных, можно использовать и тернарный оператор (не пугайся, это просто if, else)
send_message_to_user(message)
Отправляем уведомление пользователю в телеграм через функцию send_message_to_user() из модуля telegram.py
modules/telegram.py
Очень простой модуль, все что он делает — уведомляет пользователя в телеграм через отправку сообщения ботом о найденом контракте.
import requests
from settings import TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID
def send_message_to_user(text: str = ""):
requests.post(
f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage?chat_id={TELEGRAM_CHAT_ID}&text={text}&parse_mode=Markdown"
)import requests from settings import TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID
requests— библиотеку для выполнения HTTP-запросов.TELEGRAM_BOT_TOKENиTELEGRAM_CHAT_ID— настройки, которые хранят токен бота и ID чата, куда отправляются сообщения.
2. Функция send_message_to_user()
def send_message_to_user(text: str = ""):
Эта функция отвечает за отправку текстового сообщения пользователю через API Telegram.
Отправляем сообщение через requests
requests.post(
f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage?chat_id={TELEGRAM_CHAT_ID}&text={text}&parse_mode=Markdown"
)Для отправки сообщения используется метод requests.post(), который выполняет HTTP-запрос к Telegram API. В этом запросе формируется URL, содержащий токен бота и параметры, необходимые для отправки сообщения:
chat_id={TELEGRAM_CHAT_ID}— указывает, в какой чат должно быть отправлено сообщение. Этот ID привязан к пользователю или группе, в которой работает бот.text={text}— передаваемый текст сообщения, который будет отображаться в чате.parse_mode=Markdown— Telegram поддерживает форматирование Markdown, что позволяет выделять текст жирным (**жирный**), курсивом (_курсив_) или вставлять ссылки ([ссылка](https://example.com)).
Для этого можно было бы использовать библиотеку, но мы решили не перегружать как код, не нужными либами, так и вас.
modules/tweet_id_storage.py
Модуль tweet_id_storage.py отвечает за сохранение и получение идентификаторов последних твитов для каждого пользователя. Он использует локальный файл latest_tweets_id.json для хранения данных. Метод get_last_tweet_id позволяет получить идентификатор последнего твита для конкретного пользователя, а метод save_last_tweet_id сохраняет новый идентификатор твита в файл, обновляя информацию для указанного пользователя. Этот модуль обеспечивает отслеживание последних обработанных твитов, чтобы избежать повторной обработки тех же твитов.
import json
import os
class TweetIdStorage:
LATEST_TWEETS_ID_PATH = "latest_tweets_id.json"
def get_last_tweet_id(self, username: str) -> str:
if os.path.exists(self.LATEST_TWEETS_ID_PATH):
with open(self.LATEST_TWEETS_ID_PATH, "r") as file:
data = json.loads(file.read())
else:
data = {}
return data.get(username, None)
def save_last_tweet_id(self, username: str, tweet_id: str):
if os.path.exists(self.LATEST_TWEETS_ID_PATH):
with open(self.LATEST_TWEETS_ID_PATH, "r") as file:
data = json.loads(file.read())
else:
data = {}
data[username] = tweet_id
with open(self.LATEST_TWEETS_ID_PATH, "w") as file:
file.write(json.dumps(data))import json import os
Этот модуль импортирует два стандартных модуля Python:
json— библиотека для работы с JSON-данными, которая позволяет сериализовать (преобразовывать объекты Python в строковый формат JSON) и десериализовать (из JSON-строки обратно в объект Python). Мы используем её для хранения ID твитов в файле и работы с этим хранилищем как с обычным словарём.os— модуль для работы с операционной системой, который предоставляет функции для взаимодействия с файловой системой. В данном случае он помогает нам проверять, существует ли файлlatest_tweets_id.json, прежде чем пытаться его открыть. Это предотвращает ошибки при обращении к несуществующему файлу.
class TweetIdStorage:
LATEST_TWEETS_ID_PATH = "latest_tweets_id.json"Этот класс предназначен для сохранения и получения последнего ID твита, связанного с пользователем.
LATEST_TWEETS_ID_PATH— это классовый атрибут, который хранит имя JSON-файла, в котором сохраняются данные. Этот файл можно рассматривать как небольшую базу данных.
Мы говорили, что рассмотрим классовые атрибуты при разборе этого модуля. Так вот, LATEST_TWEETS_ID_PATH — это тот самый атрибут. В отличие от обычных атрибутов, классовые атрибуты являются общими для всех объектов класса. Это значит, что если изменить значение атрибута, например, self.a, у одного объекта, оно изменится и у всех остальных объектов этого класса.
3. Функция get_last_tweet_id()
def get_last_tweet_id(self, username: str) -> str:
Эта функция загружает JSON-файл и получает последний сохранённый ID твита для конкретного пользователя.
Проверка файла на существование
if os.path.exists(self.LATEST_TWEETS_ID_PATH):
Сначала проверяется наличие файла latest_tweets_id.json с помощью функции os.path.exists. Эта функция принимает путь к файлу, и так как файл хранится в корне проекта, достаточно указать только его название, которое хранится в self.LATEST_TWEETS_ID_PATH
with open(self.LATEST_TWEETS_ID_PATH, "r") as file:
data = json.loads(file.read())Открываем файл в режиме чтения ("r"), затем с помощью file.read() считываем его содержимое в виде строки. После этого используем json.loads(), чтобы преобразовать JSON-формат в питоновский словарь. Теперь data содержит словарь, где ключи — это имена пользователей, а значения — ID их последних твитов.
else:
data = {}Если файла нет, создаётся пустой словарь data, чтобы избежать ошибок при доступе к несуществующим данным.
return data.get(username, None)
Функция ищет в словаре ключ с именем пользователя (username) и возвращает его значение (ID последнего твита). Если ключ отсутствует, возвращается None.
4. Функция save_last_tweet_id()
def save_last_tweet_id(self, username: str, tweet_id: str):
Эта функция сохраняет последний ID твита пользователя в JSON-файл.
Проверка файла на существование и чтение из него
if os.path.exists(self.LATEST_TWEETS_ID_PATH):
with open(self.LATEST_TWEETS_ID_PATH, "r") as file:
data = json.loads(file.read())Как и в get_last_tweet_id(), сначала проверяется, существует ли JSON-файл. Если он есть, его содержимое загружается в виде словаря.
else:
data = {}Если файла нет, создаётся пустой словарь.
Обновляем ID последнего твита для юзера
data[username] = tweet_id
Далее в словарь добавляется/изменяется запись, где ключ — это имя пользователя, а значение — ID твита.
Записываем результат обратно в файл
with open(self.LATEST_TWEETS_ID_PATH, "w") as file:
file.write(json.dumps(data))Сначала обновлённый словарь с данными преобразуется в JSON-строку с помощью json.dumps(). Затем, используя file.write(), эта строка записывается в файл, открытый в режиме "w" (перезапись файла).
modules/x.py
Модуль x.py обеспечивает взаимодействие с API платформы X (Twitter). Он позволяет выполнять запросы для получения данных о пользователях, их твитах и конкретных твитах по идентификатору. Основные функции включают извлечение идентификатора пользователя по его имени, получение списка твитов пользователя с учетом ранее обработанных, а также получение информации о твите по его ID. Модуль обрабатывает лимиты запросов, поддерживает прокси и интегрируется с модулем tweet_id_storage.py для хранения последних обработанных твитов.
import time
from typing import Any, Dict
import requests
from modules.tweet_id_storage import TweetIdStorage
from settings import BEARER_TOKEN, MAX_RETRIES, PROXY
class XAPIHandler:
BASE_URL = "https://api.x.com/2"
def __init__(self):
self.tweet_id_storage = TweetIdStorage()
def _make_request(
self,
endpoint: str,
method: str,
data: Dict[str, Any] = {},
params: Dict[str, str] = {},
) -> Dict[str, Any]:
headers = {
"Authorization": f"Bearer {BEARER_TOKEN}",
}
retries = 0
while retries < MAX_RETRIES:
try:
response = requests.request(
method=method,
url=self.BASE_URL + endpoint,
headers=headers,
data=data,
params=params,
proxies={"http": PROXY, "https": PROXY}
if PROXY != "scheme://username:password@host:port"
else None,
)
response.raise_for_status()
return response.json()
except requests.exceptions.HTTPError as error:
if error.response.status_code == 429:
reset_time = error.response.headers["X-Rate-Limit-Reset"]
time_to_wait = int(reset_time) - int(time.time())
print(
f"Превышено максимальное количество запросов. Подождите {time_to_wait} секунд"
)
time.sleep(time_to_wait)
else:
print(error.response.text)
retries += 1
raise Exception("Превышено максимальное количество попыток")
def fetch_user_id_by_username(self, username: str) -> str:
user = self._make_request(f"/users/by/username/{username}", method="GET")
return user["data"]["id"]
def fetch_user_tweets(self, username: str, user_id: str) -> Dict[str, Any]:
last_tweet_id = self.tweet_id_storage.get_last_tweet_id(username)
params = {
"expansions": "attachments.media_keys,referenced_tweets.id",
"media.fields": "url,type,media_key",
"tweet.fields": "text,note_tweet",
}
if last_tweet_id:
params["since_id"] = last_tweet_id
params["max_results"] = "100"
else:
params["max_results"] = "5"
return self._make_request(
f"/users/{user_id}/tweets", method="GET", params=params
)
def get_tweet_by_id(self, tweet_id: str) -> Dict[str, Any]:
params = {
"expansions": "attachments.media_keys,referenced_tweets.id",
"media.fields": "url,type,media_key",
"tweet.fields": "text,note_tweet",
}
return self._make_request(f"/tweets/{tweet_id}", method="GET", params=params)Далее разберем работу кода, который взаимодействует с этим API. Вся информация получена из официальной документации. Рекомендуем всегда начинать с изучения документации при работе с любыми API.
import time from typing import Any, Dict import requests from modules.tweet_id_storage import TweetIdStorage from settings import BEARER_TOKEN, MAX_RETRIES, PROXY
time— используется для работы с задержками (sleep).AnyиDictизtyping— используются для аннотации типов.requests— библиотека для отправки HTTP-запросов.TweetIdStorage— класс для чтение и добавления ID последних обработанных твитов.BEARER_TOKEN,MAX_RETRIES,PROXY— настройки, используемые для работы с API X.
class XAPIHandler:
BASE_URL = "https://api.x.com/2"Этот класс предназначен для взаимодействия с API X.
BASE_URL— это классовый атрибут, который доступен всем объектам классаXAPIHandler. Он задаёт базовый URL, к которому добавляются конечные точки API для формирования запросов.
def __init__(self):
self.tweet_id_storage = TweetIdStorage()При создании объекта класса XAPIHandler создаётся объект TweetIdStorage, который будет использоваться для чтения и сохранения ID последних твитов, чтобы избежать повторной обработки одних и тех же данных.
def _make_request(
self,
endpoint: str,
method: str,
data: Dict[str, Any] = {},
params: Dict[str, str] = {},
) -> Dict[str, Any]:Этот метод выполняет HTTP-запросы к API X. Принимает следующие аргументы:
endpoint— конечная точка API, например,"/users/by/username/{username}". То есть, мы ожидаем, что это будет путь к X API, но безhttps://api.x.com/2, так как это у нас уже записано вself.BASE_URLmethod— HTTP-метод (GET,POSTи т. д.)data— данные, передаваемые в теле запроса (по умолчанию пустой словарь)params— параметры запроса, передаваемые в URL (по умолчанию также пустой словарь)
headers = {
"Authorization": f"Bearer {BEARER_TOKEN}",
}Добавляется заголовок Authorization, содержащий токен BEARER_TOKEN, который требуется для работы с API X.
retries = 0 while retries < MAX_RETRIES:
Ограничивается число попыток запроса по MAX_RETRIES, чтобы избежать бесконечных повторных попыток в случае ошибки.
try:
response = requests.request(
method=method,
url=self.BASE_URL + endpoint,
headers=headers,
data=data,
params=params,
proxies={"http": PROXY, "https": PROXY}
if PROXY != "scheme://username:password@host:port"
else None,
)
response.raise_for_status()
return response.json()Отправляем HTTP-запрос по адресу, составленному из BASE_URL и переданного endpoint. Заголовки запроса включают авторизационный токен, который необходим для доступа к API.
Если PROXY указан, то он будет использоваться для маршрутизации запроса.
После отправки запроса вызывается response.raise_for_status(), который проверяет код ответа. Если сервер возвращает ошибку (например, 404 — не найдено или 500 — внутренняя ошибка), то мы вызываем исключение.
Если запрос выполнен успешно, выполняем response.json(), который преобразует ответ API в словарь Python.
Обработка ошибки "Слишком много запросов" (код 429)
except requests.exceptions.HTTPError as error:
if error.response.status_code == 429:
reset_time = error.response.headers["X-Rate-Limit-Reset"]
time_to_wait = int(reset_time) - int(time.time())
print(
f"Превышено максимальное количество запросов. Подождите {time_to_wait} секунд"
)
time.sleep(time_to_wait)X API имеет ограничения на количество запросов для бесплатной версии, поэтому нам нужно обрабатывать данную ситуацию.
Если API X отвечает кодом 429 (слишком много запросов), скрипт получает из заголовков ответа значение X-Rate-Limit-Reset, которое указывает, когда можно отправить следующий запрос.
Далее вычисляется разница между этим значением и текущим временем, чтобы определить, сколько секунд необходимо подождать. После этого выполнение кода приостанавливается на рассчитанный интервал с помощью time.sleep(time_to_wait), а затем запрос повторяется.
else:
print(error.response.text)
retries += 1Если ошибка не связана с лимитом запросов, выводится текст ошибки и увеличивается retries.
После того как retries превысит MAX_RETRIES, скрипт выходит из while цикла и вызывает исключение.
raise Exception("Превышено максимальное количество попыток")4. Метод fetch_user_id_by_username
Этот метод получает ID пользователя по его username.
def fetch_user_id_by_username(self, username: str) -> str:
user = self._make_request(f"/users/by/username/{username}", method="GET")
return user["data"]["id"]Выполняем запрос используя ранее описанный метод self._make_request(). В него мы передаем endpoint на который нужно сделать GET запрос для получения ID пользователя по его username.
Ответом от X API будет словарь, который содержит данные о пользователе. Из него мы вытаскиваем ID пользователя и возвращаем.
def fetch_user_tweets(self, username: str, user_id: str) -> Dict[str, Any]:
Этот метод позволяет получить твиты пользователя. В качестве аргумента принимает username и user_id
Вытаскиваем ID последних постов
last_tweet_id = self.tweet_id_storage.get_last_tweet_id(username)
Сначала берём ID последнего сохранённого твита пользователя. Для этого используем уже знакомый нам self.tweet_id_storage.get_last_tweet_id() и передаем ему username аккаунта.
Указываем параметры запроса к X API
params = {
"expansions": "attachments.media_keys,referenced_tweets.id",
"media.fields": "url,type,media_key",
"tweet.fields": "text,note_tweet",
}Нам нужно получить твиты вместе с ретвитами, приложенными изображениями, а также полный текст. Из документации мы узнали, что для этого необходимо указать следующие параметры:
expansions используется для включения связанных объектов, которые не входят в основной ответ API:
referenced_tweets.id— позволяет получать ретвиты и цитаты.attachments.media_keys— включает идентификаторы медиафайлов, прикрепленных к твиту.
tweet.fields определяет, какие свойства твита должны быть включены в ответ API:
Если твит содержит изображения, видео или GIF, то для получения информации о них используется media.fields:
media_key— уникальный идентификатор медиафайла.type— тип медиа (photo,video,animated_gif).url— ссылка на медиафайл (доступна для изображений).
Модифицируем параметры в зависимости от наличия last_tweet_id
if last_tweet_id:
params["since_id"] = last_tweet_id
params["max_results"] = "100"
else:
params["max_results"] = "5"- Если есть
last_tweet_id, берём твиты только после него (до 100 штук). - Если нет — просто берём последние 5 твитов.
return self._make_request(f"/users/{user_id}/tweets", method="GET", params=params)В качестве результата мы возвращаем ответ от запрос на GET /users/{user_id}/tweets, что вернет нам твиты пользователя
def get_tweet_by_id(self, tweet_id: str) -> Dict[str, Any]:
Получаем информацию о конкретном твите, что пригодится при разборе ретвитов, так как их изображения не хранятся в ответе из fetch_user_tweets
params = {
"expansions": "attachments.media_keys,referenced_tweets.id",
"media.fields": "url,type,media_key",
"tweet.fields": "text,note_tweet",
}Снова указываем параматеры к запросу, они будут такими же как и в предыдущий раз.
return self._make_request(f"/tweets/{tweet_id}", method="GET", params=params)Делаем запрос на GET /tweets/{tweet_id} и возвращаем результат
Заключение
Мы разобрали наш парсер X аккаунтов от а до я. Надеемся вам все было понятно, а если нет, то го в комменты — ответим и поясним за любую строчку в коде.
Данный скрипт мы написали всего за один день. Да, пришлось попотеть с API X, но в целом код, ну очень простой. Поэтому не бойся и просто начинай писать. Никогда не будет всё чётко и гладко, главное быть готовым к трудностям.
📌 Сохраняй статью и делись с друзьями, пускай тоже прошарят за код