feat(bump): add auto thread bumping
This commit is contained in:
@@ -1,3 +1,4 @@
|
||||
from concurrent.futures import thread
|
||||
import logging
|
||||
import asyncio
|
||||
|
||||
@@ -11,11 +12,46 @@ class Lolz:
|
||||
def __init__(self, token: str):
|
||||
try:
|
||||
self.client = Forum(token=token, timeout=15)
|
||||
|
||||
# ! Не рекомендуется включить этот логгер,
|
||||
# ! ибо логи занимают очень много места в экране при этом логгере
|
||||
# self.client.settings.logger.enable()
|
||||
|
||||
except:
|
||||
raise
|
||||
|
||||
async def get_thread(self, thread_id: Union[str, int]) -> Dict[str, Any]:
|
||||
logger.debug(f"Запрашиваю данные темы {thread_id}...")
|
||||
response = await self.client.threads.get(thread_id=thread_id)
|
||||
thread = (response.json()).get("thread", {})
|
||||
|
||||
if thread == {}:
|
||||
logger.warning(f"Тема {thread_id} не найдена или вернула пустой ответ.")
|
||||
|
||||
return thread
|
||||
|
||||
async def can_bump(
|
||||
self, thread_id: Optional[int] = None, thread: Dict[str, Any] = None
|
||||
) -> bool:
|
||||
if not thread:
|
||||
if not thread_id:
|
||||
thread = {}
|
||||
else:
|
||||
thread = await self.get_thread(thread_id=thread_id)
|
||||
|
||||
return thread["permissions"]["bump"]["can"]
|
||||
|
||||
async def bump_thread(self, thread_id: Union[str, int]) -> bool:
|
||||
logger.debug(f"Поднимаю тему {thread_id}...")
|
||||
response = await self.client.threads.bump(thread_id=thread_id)
|
||||
|
||||
if response.status_code == 200:
|
||||
logger.info(f"Тема {thread_id} успешно поднята.")
|
||||
return True
|
||||
else:
|
||||
logger.error(f"Не удалось поднять тему {thread_id}.")
|
||||
return False
|
||||
|
||||
async def get_post(self, post_id: Union[str, int]) -> Dict[str, Any]:
|
||||
logger.debug(f"Запрашиваю данные поста {post_id}...")
|
||||
response = await self.client.posts.get(post_id=post_id)
|
||||
|
||||
Reference in New Issue
Block a user