Хочу написать телеграм бота на питон который будет отправлять запросы на сайт "какой то сайт" искать там нужного пользователя и кликать лайки по его статьям.
1# В общих чертах мне понятно что нужно отправить боту адрес например https://dzen.ru/xz24n и дальше он парсит эту страницу находит заголовки статей создает их список.
2# Переходит по ним на страницу статьи, парсит ее , находит "палец вверх" и кликает его.
Вот до второго пункта это мне знакомо, а второй пункт уже вызывает вопросы - как?
Идея то не нова, наверняка ее измочалили уже вдоль и поперек, но я не нашел инфы в интернете.
Кто знает подскажите
1# В общих чертах мне понятно что нужно отправить боту адрес например https://dzen.ru/xz24n и дальше он парсит эту страницу находит заголовки статей создает их список.
2# Переходит по ним на страницу статьи, парсит ее , находит "палец вверх" и кликает его.
Вот до второго пункта это мне знакомо, а второй пункт уже вызывает вопросы - как?
Идея то не нова, наверняка ее измочалили уже вдоль и поперек, но я не нашел инфы в интернете.
Кто знает подскажите
Python:
import asyncio
import aiohttp
import os
import urllib.parse
from aiogram import Bot, Dispatcher
from aiogram.filters import Command
from aiogram.types import Message
from bs4 import BeautifulSoup
from dotenv import load_dotenv
import logging
# Настройка логирования
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
handlers=[
logging.FileHandler("bot.log"), # Логирование в файл
logging.StreamHandler(), # Логирование в консоль (опционально)
],
)
logger = logging.getLogger(__name__)
# Токен вашего бота
load_dotenv()
TOKEN = os.getenv("TOKEN")
# Инициализация бота и диспетчера
bot = Bot(token=TOKEN)
dp = Dispatcher()
# Список поисковых систем
search_engines = {
"Google": "https://www.google.com/search?q=",
"Yandex": "https://yandex.ru/search/?text=",
}
# User-Agent для имитации браузера
# Универсальные заголовки для Google и Yandex
HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
"Accept-Language": "ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7", # Языковые предпочтения
"Accept-Encoding": "gzip, deflate, br", # Поддержка сжатия
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8",
}
# Функция для извлечения ссылок из HTML
def extract_links(html, search_engine):
soup = BeautifulSoup(html, 'html.parser')
links = []
try:
if search_engine == "Google":
# Парсинг ссылок из Google
for item in soup.find_all('div', class_='tF2Cxc'):
link = item.find('a')['href']
links.append(link)
elif search_engine == "Yandex":
# Парсинг ссылок из Yandex
for item in soup.find_all('a', class_='Link Link_theme_normal OrganicTitle-Link'):
link = item['href']
links.append(link)
except Exception as e:
logger.error(f"Error extracting links from {search_engine}: {e}")
return links
# Функция для отправки запроса на поисковую систему и обработки результатов
async def send_search_request(query):
# Кодируем запрос с кириллицей
encoded_query = urllib.parse.quote(query)
async with aiohttp.ClientSession() as session:
for engine, url in search_engines.items():
# Динамически изменяем заголовки для разных поисковых систем
headers = HEADERS.copy() # Копируем базовые заголовки
if engine == "Yandex":
headers["Referer"] = "https://www.yandex.ru/" # Добавляем реферер для Yandex
elif engine == "Google":
headers["Referer"] = "https://www.google.com/" # Добавляем реферер для Google
full_url = url + encoded_query
try:
async with session.get(full_url, headers=headers) as response:
if response.status == 200:
logger.info(f"Request sent to {engine}: {full_url}")
html = await response.text()
links = extract_links(html, engine)
logger.info(f"Found {len(links)} links from {engine}")
# Фильтрация ссылок
target_links = [link for link in links if "https://dzen.ru/xz24n" in link]
logger.info(f"Found {len(target_links)} target links from {engine}")
# Инициация перехода по найденным ссылкам
for target_link in target_links:
logger.info(f"Visiting {target_link}")
try:
async with session.get(target_link, headers=headers) as visit_response:
logger.info(f"Visited {target_link}, status: {visit_response.status}")
except Exception as e:
logger.error(f"Error visiting {target_link}: {e}")
else:
logger.error(f"Failed to fetch {full_url}, status: {response.status}")
except Exception as e:
logger.error(f"Error sending request to {engine}: {e}")
'''
'''
# Функция, которая будет запускаться каждые 30 минут
async def scheduled_task():
query = "Дзен Джун xz24n"
while True:
await send_search_request(query)
await asyncio.sleep(1800) # 1800 секунд = 30 минут
# Команда /start
@dp.message(Command("start"))
async def send_welcome(message: Message):
await message.reply("Бот запущен. Запросы будут отправляться каждые 30 минут.")
logger.info("Бот запущен пользователем.")
# Запуск бота и планировщика
async def main():
# Запуск планировщика
asyncio.create_task(scheduled_task())
logger.info("Бот и планировщик запущены.")
try:
# Запуск бота
await dp.start_polling(bot)
except asyncio.CancelledError:
# Обработка отмены (например, при завершении программы)
logger.info("Бот остановлен.")
if __name__ == '__main__':
asyncio.run(main())
Последнее редактирование:
