85 lines
2.8 KiB
Python
85 lines
2.8 KiB
Python
import logging
|
|
import asyncio
|
|
|
|
from LOLZTEAM.Client import Forum
|
|
from typing import Dict, Any, Optional, Union, List
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
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_post(self, post_id: Union[str, int]) -> Dict[str, Any]:
|
|
response = await self.client.posts.get(post_id=post_id)
|
|
return (response.json()).get("post", {})
|
|
|
|
async def get_thread_posts(
|
|
self, thread_id: Union[str, int], start_page: int = 1
|
|
) -> List[Dict[str, Any]]:
|
|
all_posts = []
|
|
page = start_page
|
|
|
|
while True:
|
|
response = await self.client.posts.list(thread_id=thread_id, page=page)
|
|
posts = (response.json()).get("posts", [])
|
|
|
|
if len(posts) == 0:
|
|
break
|
|
|
|
all_posts.extend(posts)
|
|
logger.info(
|
|
f"Получено {len(posts)} постов из темы {thread_id} на странице {page}."
|
|
)
|
|
|
|
page += 1
|
|
await asyncio.sleep(1)
|
|
|
|
if all_posts:
|
|
logger.info(f"Всего получено {len(all_posts)} постов из темы {thread_id}.")
|
|
else:
|
|
logger.info(f"Постов в теме {thread_id} не найдено.")
|
|
|
|
return all_posts
|
|
|
|
async def get_all_thread_posts(
|
|
self, thread_id: Union[str, int]
|
|
) -> List[Dict[str, Any]]:
|
|
return await self.get_thread_posts(thread_id=thread_id, start_page=1)
|
|
|
|
async def get_post_comments(self, post_id: int) -> List[Dict[str, Any]]:
|
|
response = await self.client.posts.comments.list(post_id=post_id)
|
|
comments = (response.json()).get("comments", [])
|
|
|
|
return comments
|
|
|
|
async def has_comments(
|
|
self, post_id: Optional[int], post: Dict[str, Any] = None
|
|
) -> bool:
|
|
if not post:
|
|
if not post_id:
|
|
post = {}
|
|
else:
|
|
post = await self.get_post(post_id=post_id)
|
|
|
|
return post.get("post_comment_count", 0) > 0
|
|
|
|
async def create_comment(self, post_id: int, comment_body: str) -> bool:
|
|
logger.info(f"Публикую комментарий к посту {post_id}...")
|
|
response = await self.client.posts.comments.create(
|
|
post_id=post_id, comment_body=comment_body
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
logger.info(f"Комментарий к посту {post_id} успешно опубликован.")
|
|
return True
|
|
else:
|
|
logger.error(f"Не удалось опубликовать комментарий к посту {post_id}.")
|
|
return False
|