import
requests
from
bs4
import
BeautifulSoup
import
json
import
traceback
# раздела форума
url
=
"https://www.blast.hk/forums/110/"
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"
}
DATA_FILE
=
"forum_topics.json"
class
ParseTokenError
(
Exception
)
:
pass
class
Forum
:
def
__init__
(
self
,
login
,
password
)
:
self
.
session
=
requests
.
Session
(
)
self
.
login
=
login
        self
.
password
=
password
        self
.
token
=
""
def
update_token
(
self
)
:
try
:
r
=
self
.
session
.
get
(
"https://www.blast.hk/"
,
headers
=
headers
)
r
.
raise_for_status
(
)
soup
=
BeautifulSoup
(
r
.
text
,
"html.parser"
)
token
=
soup
.
find
(
"input"
,
{
"name"
:
"_xfToken"
}
)
if
token
is
None
:
raise
ParseTokenError
(
"Не удалось найти токен."
)
self
.
token
=
token
[
"value"
]
except
Exception
as
e
:
print
(
f"Ошибка при обновлении токена:{e}"
)
traceback
.
print_exc
(
)
def
authenticate
(
self
)
:
try
:
self
.
update_token
(
)
r
=
self
.
session
.
post
(
"https://www.blast.hk/login/login"
,
data
=
{
"login"
:
self
.
login
,
"password"
:
self
.
password
,
"remember"
:
1
,
"_xfRedirect"
:
"https://www.blast.hk/"
,
"_xfToken"
:
self
.
token
}
)
if
"Некорректный пароль"
in
r
.
text
:
print
(
"Ошибка: Некорректный пароль."
)
return
False
self
.
id
=
self
.
session
.
cookies
.
get
(
"xf_user"
,
""
)
.
split
(
"%"
)
[
0
]
print
(
"Авторизация успешна."
)
return
True
except
Exception
as
e
:
print
(
f"Ошибка авторизации:{e}"
)
traceback
.
print_exc
(
)
return
False
def
post_reply
(
self
,
thread_url
,
message
)
:
try
:
self
.
update_token
(
)
# Обновление токена перед запросом
thread_id
=
thread_url
.
split
(
"/"
)
[
-
2
]
# Извлекаем ID темы из URL
reply_url
=
f"https://www.blast.hk/threads/{thread_id}/add-reply"
data
=
{
"message_html"
:
message
,
"_xfToken"
:
self
.
token
,
"_xfResponseType"
:
"json"
,
}
r
=
self
.
session
.
post
(
reply_url
,
data
=
data
,
headers
=
headers
)
if
r
.
status_code
==
200
:
print
(
f"Сообщение успешно отправлено в тему:{thread_url}"
)
else
:
print
(
f"Ошибка отправки сообщения:{r.status_code}"
)
print
(
"Ответ сервера:"
,
r
.
text
)
# Отладочная информация
except
Exception
as
e
:
print
(
f"Ошибка при добавлении ответа:{e}"
)
def
fetch_forum_topics
(
session
,
url
)
:
try
:
response
=
session
.
get
(
url
,
headers
=
headers
)
response
.
raise_for_status
(
)
soup
=
BeautifulSoup
(
response
.
text
,
"html.parser"
)
# Поиск тем на странице
topics
=
soup
.
find_all
(
"div"
,
class_
=
"structItem"
)
data
=
[
]
for
topic
in
topics
:
# Пропуск закреплённых тем
if
topic
.
find
(
"i"
,
class_
=
"structItem-status--sticky"
)
:
continue
title_element
=
topic
.
find
(
"div"
,
class_
=
"structItem-title"
)
.
find
(
"a"
)
if
not
title_element
:
continue
title
=
title_element
.
get_text
(
strip
=
True
)
# Название темы
link
=
"https://www.blast.hk"
+
title_element
[
"href"
]
# Ссылка на тему
# Проверка, чтобы ссылка вела на тему, а не на фильтр
if
"/threads/"
not
in
link
:
continue
author
=
topic
.
find
(
"a"
,
class_
=
"username"
)
.
get_text
(
strip
=
True
)
# Автор темы
data
.
append
(
{
"title"
:
title
,
"link"
:
link
,
"author"
:
author
}
)
return
data
except
Exception
as
e
:
print
(
f"Ошибка при парсинге:{e}"
)
return
[
]
def
load_previous_topics
(
file
)
:
try
:
with
open
(
file
,
"r"
,
encoding
=
"utf-8"
)
as
f
:
return
json
.
load
(
f
)
except
FileNotFoundError
:
return
[
]
except
json
.
JSONDecodeError
:
return
[
]
def
save_topics
(
file
,
topics
)
:
with
open
(
file
,
"w"
,
encoding
=
"utf-8"
)
as
f
:
json
.
dump
(
topics
,
f
,
ensure_ascii
=
False
,
indent
=
4
)
def
find_new_topics
(
old_topics
,
current_topics
)
:
old_links
=
{
topic
[
"link"
]
for
topic
in
old_topics
}
new_topics
=
[
topic
for
topic
in
current_topics
if
topic
[
"link"
]
not
in
old_links
]
return
new_topics
def
main
(
)
:
login
=
input
(
"Введите логин: "
)
.
strip
(
)
password
=
input
(
"Введите пароль: "
)
.
strip
(
)
forum
=
Forum
(
login
,
password
)
if
not
forum
.
authenticate
(
)
:
print
(
"Не удалось авторизоваться."
)
return
previous_topics
=
load_previous_topics
(
DATA_FILE
)
current_topics
=
fetch_forum_topics
(
forum
.
session
,
url
)
print
(
"Все текущие темы:"
)
for
topic
in
current_topics
:
print
(
f"Название:{topic['title']}| Автор:{topic['author']}| Ссылка:{topic['link']}"
)
new_topics
=
find_new_topics
(
previous_topics
,
current_topics
)
save_topics
(
DATA_FILE
,
current_topics
)
if
new_topics
:
print
(
"\nНайдены новые темы:"
)
for
topic
in
new_topics
:
print
(
f"Название:{topic['title']}| Автор:{topic['author']}| Ссылка:{topic['link']}"
)
reply
=
input
(
f"Хотите отправить ответ в тему \"{topic['title']}\"? (y/n): "
)
.
strip
(
)
.
lower
(
)
if
reply
==
"y"
:
message
=
input
(
"Введите текст сообщения: "
)
.
strip
(
)
try
:
forum
.
post_reply
(
topic
[
"link"
]
,
message
)
except
Exception
as
e
:
print
(
f"Ошибка при отправке ответа:{e}"
)
else
:
print
(
"\nНовых тем нет."
)
if
__name__
==
"__main__"
:
main
(
)
=