#!/usr/bin/env python3 import re import time import logging import requests from config import URL_POST, URL_NEWS, URL_CHANGELOG, HEADERS_SITE, SITE_CONFIG class SiteAPI: def __init__(self): self.logger = logging.getLogger(__name__) self.config = SITE_CONFIG def make_request(self, url, headers=None, params=None, max_retries=10, retry_delay=10): """Универсальный метод для HTTP запросов с повторными попытками""" for attempt in range(max_retries): try: if params: response = requests.get(url, params=params) elif headers: response = requests.get(url, headers=headers) else: response = requests.get(url) if response.status_code == 200: return response else: self.logger.warning(f"HTTP {response.status_code} для {url}, попытка {attempt + 1}/{max_retries}") except requests.RequestException as err: self.logger.warning(f"Ошибка HTTP запроса к {url} (попытка {attempt + 1}/{max_retries}): {err}") if attempt < max_retries - 1: self.logger.info(f"Повторная попытка через {retry_delay} секунд...") time.sleep(retry_delay) self.logger.error(f"Все {max_retries} попыток запроса к {url} неудачны") return None def get_changelog(self): """Получение changelog с GitLab""" self.logger.debug("Получаем changelog") response = self.make_request(URL_CHANGELOG, HEADERS_SITE) if response and response.status_code == 200: matches = re.findall(r'###Scripts version (\d+)### / stable', response.text) self.logger.debug(f"Найдены версии в истории изменений: {matches}") if matches: last_version = int(max(matches)) self.logger.info(f"Последняя стабильная версия в changelog: {last_version}") return matches, last_version, response self.logger.error(f'Ошибка при запросе changelog: {response.status_code if response else "No Response"}') return None, None, None def get_news(self): """Получение списка новостей с сайта""" self.logger.debug("Получаем список новостей") response = self.make_request(URL_NEWS, HEADERS_SITE) if response and response.status_code == 200: data = response.json() topics = data['topic_list']['topics'] # Фильтруем новости, исключая описание категории filtered_topics = [ (topic['id'], str(topic['title'])) for topic in topics if topic['title'] != 'Описание категории «Новости»' ] # Фильтруем по ID темы filtered_topics = [ (topic_id, title) for topic_id, title in filtered_topics if topic_id >= self.config['start_topic_id'] ] self.logger.debug(f"Получено {len(filtered_topics)} новостей") return filtered_topics else: self.logger.error(f"Ошибка при запросе новостей: {response.status_code if response else 'Нет доступа к сайту'}") return [] def get_news_content(self, post_id, content_processor): """Получение содержимого конкретной новости""" self.logger.debug(f"Получаем содержимое поста с ID: {post_id}") url = f"https://linux-gaming.ru/t/{post_id}.json" response = self.make_request(url, HEADERS_SITE) if response and response.status_code == 200: topic_data = response.json() posts = topic_data.get('post_stream', {}).get('posts', []) # Ищем первый пост for post in posts: if post.get('post_number') == 1: html_content = post.get('cooked', 'Нет содержимого') text_data = content_processor.html_to_text(html_content) return text_data self.logger.error(f"Первый пост не найден в теме с ID: {post_id}") return None else: self.logger.error(f"Не удалось получить содержимое поста с ID: {post_id}") return None def post_to_site(self, post_data): """Публикация новости на сайт""" title = post_data.get('title', 'Без названия') while True: try: response = requests.post(url=URL_POST, headers=HEADERS_SITE, json=post_data) if response.status_code == 200: self.logger.info("Новость опубликована на сайте!") return response elif response.status_code == 422: self.logger.warning(f'Новость "{title}" уже опубликована: {response.status_code}') return response else: self.logger.error(f'Ошибка при отправке новости "{title}" на сайт: {response.status_code}') except requests.RequestException as error: self.logger.error(f'Ошибка при отправке новости "{title}" на сайт: {error}') time.sleep(900) # Ждем 15 минут перед повторной попыткой def create_script_update_post(self, script_ver, next_version, changelog_response, content_processor): """Создание поста для обновления скрипта""" self.logger.debug(f"Создаем пост для обновления скрипта версии {script_ver}") post_text = content_processor.create_script_content(script_ver, next_version, changelog_response) if post_text: site_text = f"[center][img]/uploads/default/original/1X/5cfa59077a5275971401fab0114e56f3ffdd0ec4.png[/img][/center]\n{post_text}" post_data = { "title": f"Кумулятивное обновление скриптов {script_ver}", "raw": site_text, "category": self.config['category_num'] } self.logger.debug(f"Данные поста: {post_data}") return post_text, post_data return None, None def check_script_versions(self, content_processor): """Проверка и публикация новых версий скриптов""" self.logger.info("Проверяем новые версии скриптов") # Получаем changelog matches_changelog, last_changelog, resp_changelog = self.get_changelog() if not matches_changelog or not last_changelog: return # Получаем существующие новости news_list = self.get_news() # Извлекаем номера версий из заголовков новостей pattern = re.compile(r'Кумулятивное обновление скриптов (\d+)') def extract_number(title): match = pattern.search(title) return int(match.group(1)) if match else None numbers = [extract_number(title) for _, title in news_list if extract_number(title) is not None] if numbers: last_topics_script = max(numbers) self.logger.info(f"Последняя новость на сайте о версии: {last_topics_script}") if last_topics_script >= last_changelog: self.logger.warning("Нет новых версий скриптов PortProton") return else: self.logger.warning("На сайте нет новостей о скриптах") # Публикуем новые версии self._publish_new_script_versions(matches_changelog, resp_changelog, content_processor) def _publish_new_script_versions(self, matches_changelog, resp_changelog, content_processor): """Публикация новых версий скриптов""" for script_ver, next_version in zip(reversed(matches_changelog[:-1]), reversed(matches_changelog[1:])): self.logger.info(f"Найдена новая версия скрипта {script_ver}") post_text, post_data = self.create_script_update_post( script_ver, next_version, resp_changelog, content_processor ) if post_data: self.logger.debug(f"Публикуем {post_data}") self.post_to_site(post_data)