204 lines
9.7 KiB
Python
204 lines
9.7 KiB
Python
import logging
|
|
import asyncio
|
|
|
|
from LOLZTEAM.Client import Forum
|
|
from typing import Dict, Any, Optional, Union, List
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class LolzAPIError(Exception):
|
|
"""Base exception for Lolz API errors"""
|
|
pass
|
|
|
|
|
|
class LolzNetworkError(LolzAPIError):
|
|
"""Network-related errors"""
|
|
pass
|
|
|
|
|
|
class LolzResponseError(LolzAPIError):
|
|
"""API returned invalid or empty response"""
|
|
pass
|
|
|
|
|
|
class Lolz:
|
|
def __init__(self, token: str):
|
|
if not token:
|
|
raise ValueError("LOLZ_TOKEN обязателен")
|
|
|
|
try:
|
|
self.client = Forum(token=token, timeout=15)
|
|
|
|
# ! Не рекомендуется включить этот логгер,
|
|
# ! ибо логи занимают очень много места в экране при этом логгере
|
|
# self.client.settings.logger.enable()
|
|
|
|
except Exception as e:
|
|
logger.error(f"Не удалось инициализировать Lolz клиент: {e}")
|
|
raise LolzAPIError(f"Не удалось инициализировать Lolz клиент: {e}") from e
|
|
|
|
async def get_thread(self, thread_id: Union[str, int]) -> Dict[str, Any]:
|
|
logger.debug(f"Запрашиваю данные темы {thread_id}...")
|
|
try:
|
|
response = await self.client.threads.get(thread_id=thread_id)
|
|
thread = (response.json()).get("thread", {})
|
|
|
|
if not thread:
|
|
raise LolzResponseError(f"Тема {thread_id} не найдена или вернула пустой ответ")
|
|
|
|
return thread
|
|
|
|
except asyncio.TimeoutError as e:
|
|
logger.error(f"Превышено время ожидания при получении темы {thread_id}: {e}")
|
|
raise LolzNetworkError(f"Превышено время ожидания при получении темы {thread_id}") from e
|
|
except Exception as e:
|
|
logger.error(f"Ошибка при получении темы {thread_id}: {e}")
|
|
raise LolzAPIError(f"Не удалось получить тему {thread_id}") from e
|
|
|
|
async def can_bump(
|
|
self, thread_id: Optional[int] = None, thread: Dict[str, Any] = None
|
|
) -> bool:
|
|
if not thread:
|
|
if not thread_id:
|
|
raise ValueError("Необходимо указать thread_id или thread")
|
|
thread = await self.get_thread(thread_id=thread_id)
|
|
|
|
try:
|
|
return thread.get("permissions", {}).get("bump", {}).get("can", False)
|
|
except (KeyError, AttributeError, TypeError) as e:
|
|
logger.error(f"Неверная структура темы для проверки поднятия: {e}")
|
|
return False
|
|
|
|
async def bump_thread(self, thread_id: Union[str, int]) -> bool:
|
|
logger.debug(f"Поднимаю тему {thread_id}...")
|
|
try:
|
|
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}. Статус: {response.status_code}")
|
|
return False
|
|
|
|
except asyncio.TimeoutError as e:
|
|
logger.error(f"Превышено время ожидания при поднятии темы {thread_id}: {e}")
|
|
raise LolzNetworkError(f"Превышено время ожидания при поднятии темы {thread_id}") from e
|
|
except Exception as e:
|
|
logger.error(f"Ошибка при поднятии темы {thread_id}: {e}")
|
|
raise LolzAPIError(f"Не удалось поднять тему {thread_id}") from e
|
|
|
|
async def get_post(self, post_id: Union[str, int]) -> Dict[str, Any]:
|
|
logger.debug(f"Запрашиваю данные поста {post_id}...")
|
|
try:
|
|
response = await self.client.posts.get(post_id=post_id)
|
|
post = (response.json()).get("post", {})
|
|
|
|
if not post:
|
|
raise LolzResponseError(f"Пост {post_id} не найден или вернул пустой ответ")
|
|
|
|
return post
|
|
|
|
except asyncio.TimeoutError as e:
|
|
logger.error(f"Превышено время ожидания при получении поста {post_id}: {e}")
|
|
raise LolzNetworkError(f"Превышено время ожидания при получении поста {post_id}") from e
|
|
except Exception as e:
|
|
logger.error(f"Ошибка при получении поста {post_id}: {e}")
|
|
raise LolzAPIError(f"Не удалось получить пост {post_id}") from e
|
|
|
|
async def get_thread_posts(
|
|
self, thread_id: Union[str, int], start_page: int = 1
|
|
) -> tuple[List[Dict[str, Any]], int]:
|
|
"""Fetch all pages starting from start_page.
|
|
|
|
Returns (posts, last_page) so the caller can resume from last_page
|
|
on the next cycle instead of re-fetching from page 1 every time.
|
|
"""
|
|
all_posts = []
|
|
page = start_page
|
|
last_page = start_page
|
|
|
|
while True:
|
|
try:
|
|
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)
|
|
last_page = page
|
|
logger.debug(
|
|
f"Получено {len(posts)} постов из темы {thread_id} на странице {page}."
|
|
)
|
|
|
|
page += 1
|
|
await asyncio.sleep(1)
|
|
|
|
except asyncio.TimeoutError as e:
|
|
logger.error(f"Превышено время ожидания при получении постов из темы {thread_id}, страница {page}: {e}")
|
|
raise LolzNetworkError(f"Превышено время ожидания при получении постов из темы {thread_id}") from e
|
|
except Exception as e:
|
|
logger.error(f"Ошибка при получении постов из темы {thread_id}, страница {page}: {e}")
|
|
raise LolzAPIError(f"Не удалось получить посты из темы {thread_id}") from e
|
|
|
|
if all_posts:
|
|
logger.info(f"Всего получено {len(all_posts)} постов из темы {thread_id}.")
|
|
else:
|
|
logger.info(f"Постов в теме {thread_id} не найдено.")
|
|
|
|
return all_posts, last_page
|
|
|
|
async def get_all_thread_posts(
|
|
self, thread_id: Union[str, int]
|
|
) -> List[Dict[str, Any]]:
|
|
posts, _ = await self.get_thread_posts(thread_id=thread_id, start_page=1)
|
|
return posts
|
|
|
|
async def get_post_comments(self, post_id: int) -> List[Dict[str, Any]]:
|
|
logger.debug(f"Запрашиваю комментарии к посту {post_id}...")
|
|
try:
|
|
response = await self.client.posts.comments.list(post_id=post_id)
|
|
comments = (response.json()).get("comments", [])
|
|
logger.debug(f"Найдено {len(comments)} комментариев к посту {post_id}.")
|
|
return comments
|
|
|
|
except asyncio.TimeoutError as e:
|
|
logger.error(f"Превышено время ожидания при получении комментариев к посту {post_id}: {e}")
|
|
raise LolzNetworkError(f"Превышено время ожидания при получении комментариев к посту {post_id}") from e
|
|
except Exception as e:
|
|
logger.error(f"Ошибка при получении комментариев к посту {post_id}: {e}")
|
|
raise LolzAPIError(f"Не удалось получить комментарии к посту {post_id}") from e
|
|
|
|
async def has_comments(
|
|
self, post_id: Optional[int] = None, post: Dict[str, Any] = None
|
|
) -> bool:
|
|
if not post:
|
|
if not post_id:
|
|
raise ValueError("Необходимо указать post_id или post")
|
|
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}...")
|
|
try:
|
|
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}. Статус: {response.status_code}")
|
|
return False
|
|
|
|
except asyncio.TimeoutError as e:
|
|
logger.error(f"Превышено время ожидания при создании комментария к посту {post_id}: {e}")
|
|
raise LolzNetworkError(f"Превышено время ожидания при создании комментария к посту {post_id}") from e
|
|
except Exception as e:
|
|
logger.error(f"Ошибка при создании комментария к посту {post_id}: {e}")
|
|
raise LolzAPIError(f"Не удалось создать комментарий к посту {post_id}") from e
|