Первая: я использую библиотеку телеботов для разработки ботов Telegram. Когда я нажимаю кнопку «Опубликовать» на встроенной клавиатуре, должен сработать обработчик запроса обратного вызова (post_news), но он не работает должным образом. Обратный вызов не вызывается при нажатии кнопки.
Второе: я использую Selenium для извлечения новостей с веб-сайта, но он не работает.
Ошибки:< /p>
DevTools listening on ws://127.0.0.1:60076/devtools/browser/f64616b4-7e52-49b3-b36e-6e1f166803af
[17928
[17928
GL_CLOSE_PATH_NV, High): GPU stall due to ReadPixels
[17928
GL_CLOSE_PATH_NV, High): GPU stall due to ReadPixels
[17928
GL_CLOSE_PATH_NV, High): GPU stall due to ReadPixels
[17928
GL_CLOSE_PATH_NV, High): GPU stall due to ReadPixels (this message will no longer repeat)
[7184
[7184
[7184
[7184
[7184
[7184
[17928
Мой код:
class RegionalNewsBot:
def __init__(self):
self.bot = TeleBot(os.getenv("TELEGRAM_BOT_TOKEN"))
self.channel_id = os.getenv("TELEGRAM_CHANNEL_ID")
self.chrome_options = self._configure_chrome_options()
self._initialize_database()
self.register_handlers()
self.current_region = None
self.current_source = None
self.current_news = None
def _configure_chrome_options(self):
options = Options()
options.add_argument("--headless")
options.add_argument("--disable-gpu")
options.add_argument("--no-sandbox")
return options
def _initialize_database(self):
self.db_connection = sqlite3.connect("sources.db", check_same_thread=False)
cursor = self.db_connection.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS sources (
id INTEGER PRIMARY KEY AUTOINCREMENT,
region TEXT NOT NULL,
source_name TEXT NOT NULL,
url TEXT NOT NULL,
keywords TEXT NOT NULL
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS published_news (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
description TEXT,
link TEXT NOT NULL,
region TEXT NOT NULL
)
""")
self.db_connection.commit()
def get_sources_by_region(self, region):
cursor = self.db_connection.cursor()
cursor.execute("SELECT source_name, url, keywords FROM sources WHERE region = ?", (region,))
return cursor.fetchall()
def fetch_severinfo_news(self, url, keywords):
response = requests.get(url)
soup = BeautifulSoup(response.text, 'html.parser')
news_items = []
articles = soup.select("div", class_="main_content container")
seen_links = set() # множество для отслеживания уникальных ссылок
print(f"Articles found: {len(articles)}")
for article in articles:
title_tag = article.find("h5")
title = title_tag.get_text(strip=True) if title_tag else None
description_tag = article.find("p")
description = description_tag.get_text(strip=True) if description_tag else None
link_tag = article.find('a', href=True)
link = link_tag['href'] if link_tag else None
if link in seen_links:
continue
seen_links.add(link)
image_tag = article.find_previous("img")
image_url = image_tag["src"] if image_tag else None
if not title or not link or not description:
continue
if keywords:
if any(keyword.lower() in (title or '').lower() or keyword.lower() in (description or '').lower() for keyword in keywords):
description_with_link = self.embed_link_in_first_verb(description, link) if description else None
news_items.append({
"title": title,
"description": description_with_link,
"link": link,
"image": image_url,
})
else:
description_with_link = self.embed_link_in_first_verb(description, link) if description else None
news_items.append({
"title": title,
"description": description_with_link,
"link": link,
"image": image_url,
})
return news_items
def fetch_dzen_news(self, url, keywords):
driver = webdriver.Chrome(service=Service(ChromeDriverManager().install()), options=self.chrome_options)
driver.get(url)
soup = BeautifulSoup(driver.page_source, 'html.parser')
driver.quit()
news_items = []
articles = soup.select('article.news-card__inner-card')
links = soup.select('a.news-card__link[href]')
for article, link_tag in zip(articles, links):
title_tag = article.find('span', class_='news-card__title')
title = title_tag.get_text(strip=True) if title_tag else None
description_tag = article.find('span', class_='news-card__annotation')
description = description_tag.get_text(strip=True) if description_tag else None
link = link_tag['href']
if link and not link.startswith("http"):
link = f"https://dzen.ru{link}"
if not keywords or any(keyword.lower() in (title or '').lower() or keyword.lower() in (description or '').lower() for keyword in keywords):
description_with_link = self.embed_link_in_first_verb(description, link) if description else None
news_items.append({
"title": title.rstrip('.'),
"description": f"{description_with_link}.",
"link": link,
})
return news_items
def preview_news(self, news_items, chat_id):
print(f"Previewing news: {news_items}")
self.current_news = news_items
for idx, news in enumerate(news_items):
text = f"*{news['title']}*\n\n{news['description']}"
markup = types.InlineKeyboardMarkup()
post_button = types.InlineKeyboardButton(f"
markup.add(post_button)
self.bot.send_message(chat_id, text, reply_markup=markup, parse_mode="Markdown")
def register_handlers(self):
@self.bot.message_handler(func=lambda message: message.text in [source[0] for source in self.get_sources_by_region(self.current_region)])
def select_source(message):
self.current_source = message.text
source_info = next((source for source in self.get_sources_by_region(self.current_region) if source[0] == self.current_source), None)
if source_info:
source_name, url, keywords = source_info
keywords_list = keywords.split(',')
self.bot.send_message(message.chat.id, f"Парсинг новостей из {source_name}...")
if source_name == "Северинфо":
news_items = self.fetch_severinfo_news(url, keywords_list)
elif source_name == "Дзен":
news_items = self.fetch_dzen_news(url, keywords_list)
else:
news_items = []
if news_items:
self.preview_news(news_items, message.chat.id)
else:
self.bot.send_message(message.chat.id, "Нет подходящих новостей.")
else:
self.bot.send_message(message.chat.id, "Ошибка: Источник не найден.")
@self.bot.callback_query_handler(func=lambda call: call.data.startswith(f"post"))
def post_news(call):
print("ping")
try:
idx = int(call.data.split("_")[1])
news_item = self.current_news[idx]
self.bot.send_message(
self.channel_id,
f"*{news_item['title']}*\n\n{news_item['description']}",
parse_mode="Markdown"
)
self.bot.answer_callback_query(call.id, "Новость опубликована!")
except Exception as e:
self.bot.answer_callback_query(call.id, f"Ошибка: {e}")
print(f"Ошибка при публикации новости: {e}")
@self.bot.message_handler(func=lambda message: message.text == "↩ Назад")
def go_back(message):
start_command(message)
def run(self):
print("Бот запущен!")
self.bot.polling()
if __name__ == "__main__":
bot = RegionalNewsBot()
bot.run()
Подробнее здесь: https://stackoverflow.com/questions/793 ... python-bot