""" Планировщик публикаций для VK """ import os import logging import schedule import time import threading from datetime import datetime, timedelta from typing import Dict, Any, Callable, Optional from .post_manager import PostManager from .config import SchedulerConfig, AppConfig # Настраиваем логирование logging.basicConfig( level=logging.DEBUG if AppConfig.DEBUG else logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(f'{AppConfig.LOG_DIR}/scheduler.log', encoding='utf-8'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) class Scheduler: """ Планировщик публикаций VK """ def __init__(self, post_manager: Optional[PostManager] = None): """ Инициализация планировщика Args: post_manager: Менеджер публикаций (если не указан, создается новый) """ self.post_manager = post_manager or PostManager() self.running = False self.thread = None self.jobs = [] # Настраиваем расписание self._setup_schedule() logger.info("Scheduler инициализирован") def _setup_schedule(self): """ Настройка расписания """ # Проверяем готовые к публикации записи каждые N минут schedule.every(SchedulerConfig.CHECK_INTERVAL).minutes.do(self._check_scheduled_posts) # Ежедневная очистка старых черновиков schedule.every().day.at("02:00").do(self._cleanup_old_drafts) logger.info(f"Расписание настроено: проверка каждые {SchedulerConfig.CHECK_INTERVAL} минут") def _check_scheduled_posts(self): """ Проверка запланированных публикаций """ try: logger.info("Проверка запланированных публикаций...") # Получаем запланированные публикации scheduled_posts = self.post_manager.get_scheduled_posts() if not scheduled_posts: logger.debug("Нет запланированных публикаций") return logger.info(f"Найдено {len(scheduled_posts)} запланированных публикаций") # Сортируем по дате scheduled_posts.sort(key=lambda x: x['date']) # Публикуем готовые записи published_count = 0 for post in scheduled_posts[:SchedulerConfig.MAX_POSTS_PER_CHECK]: if self._should_publish(post): self._publish_scheduled_post(post) published_count += 1 # Ограничиваем количество публикаций за одну проверку if published_count >= SchedulerConfig.MAX_POSTS_PER_CHECK: break logger.info(f"Проверка завершена: опубликовано {published_count} записей") except Exception as e: logger.error(f"Ошибка проверки запланированных публикаций: {e}") def _should_publish(self, post: Dict[str, Any]) -> bool: """ Проверка, нужно ли публиковать запись Args: post: Данные публикации Returns: Нужно ли публиковать """ now = datetime.now() publish_time = post['date'] # Публикуем, если время пришло return publish_time <= now def _publish_scheduled_post(self, post: Dict[str, Any]): """ Публикация запланированной записи Args: post: Данные публикации """ try: logger.info(f"Публикация запланированной записи: {post['id']} от {post['date']}") # Отменяем запланированную публикацию cancel_result = self.post_manager.cancel_scheduled_post(post['id']) if not cancel_result['success']: logger.error(f"Не удалось отменить запланированную публикацию {post['id']}: {cancel_result.get('error')}") return # Публикуем немедленно # В реальной реализации нужно использовать данные из черновика publish_result = self.post_manager.publish_post({ 'owner_id': post['owner_id'], 'message': post['text'], 'attachments': ','.join([self._format_attachment(att) for att in post['attachments']]) if post['attachments'] else '' }) if publish_result['success']: logger.info(f"Запланированная публикация успешно опубликована: {post['id']} -> {publish_result['post_id']}") else: logger.error(f"Не удалось опубликовать запланированную публикацию {post['id']}: {publish_result.get('error')}") except Exception as e: logger.error(f"Ошибка публикации запланированной записи {post['id']}: {e}") def _format_attachment(self, attachment: Dict[str, Any]) -> str: """ Форматирование вложения для публикации Args: attachment: Данные вложения Returns: Строка с ID вложения """ atype = attachment['type'] item = attachment[atype] if atype == 'photo': return f"photo{item['owner_id']}_{item['id']}" elif atype == 'video': return f"video{item['owner_id']}_{item['id']}" elif atype == 'doc': return f"doc{item['owner_id']}_{item['id']}" elif atype == 'link': return f"link{item['url']}" return '' def _cleanup_old_drafts(self): """ Очистка старых черновиков """ try: logger.info("Очистка старых черновиков...") drafts_dir = os.path.join(AppConfig.MEDIA_DIR, 'posts') if not os.path.exists(drafts_dir): logger.debug(f"Директория черновиков не существует: {drafts_dir}") return # Удаляем черновики старше 30 дней cutoff_date = datetime.now() - timedelta(days=30) deleted_count = 0 for filename in os.listdir(drafts_dir): if filename.endswith('.json'): filepath = os.path.join(drafts_dir, filename) # Получаем время последнего изменения file_time = datetime.fromtimestamp(os.path.getmtime(filepath)) if file_time < cutoff_date: try: os.remove(filepath) deleted_count += 1 logger.debug(f"Удален старый черновик: {filename}") except Exception as e: logger.error(f"Не удалось удалить файл {filepath}: {e}") logger.info(f"Очистка завершена: удалено {deleted_count} старых черновиков") except Exception as e: logger.error(f"Ошибка очистки старых черновиков: {e}") def start(self): """ Запуск планировщика """ if self.running: logger.warning("Планировщик уже запущен") return self.running = True # Запускаем в отдельном потоке self.thread = threading.Thread(target=self._run_schedule, daemon=True) self.thread.start() logger.info("Планировщик запущен") def _run_schedule(self): """ Запуск цикла планировщика """ while self.running: try: schedule.run_pending() time.sleep(1) except Exception as e: logger.error(f"Ошибка в цикле планировщика: {e}") time.sleep(5) # Задержка при ошибке def stop(self): """ Остановка планировщика """ if not self.running: logger.warning("Планировщик не запущен") return self.running = False if self.thread: self.thread.join(timeout=5) # Очищаем расписание schedule.clear() logger.info("Планировщик остановлен") def add_job(self, func: Callable, *args, **kwargs) -> None: """ Добавление задачи в расписание Args: func: Функция для выполнения *args: Аргументы функции **kwargs: Именованные аргументы функции """ # Пример: schedule.every(10).minutes.do(job_function) # Пока просто логируем logger.info(f"Добавлена задача: {func.__name__}") def get_jobs(self) -> list: """ Получение списка задач Returns: Список задач """ return schedule.get_jobs()