# -*- coding: utf-8 -*- import os import sys import time import random import argparse from datetime import datetime import config import settings_manager class LoggerOutput: def __init__(self, stream): self.stream = stream def write(self, text): self.stream.write(text) text_lower = text.lower() if "[error]" in text_lower or "[exception]" in text_lower: timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") try: with open(config.ERROR_LOG_PATH, "a", encoding="utf-8") as f: f.write(f"[{timestamp}] {text}") except Exception: pass def flush(self): self.stream.flush() sys.stdout = LoggerOutput(sys.stdout) sys.stderr = LoggerOutput(sys.stderr) PREVIEW_MODE = "--preview" in sys.argv from content_generator import ( fetch_latest_news_and_trends, generate_single_post, generate_carousel_post, smart_truncate ) from renderer import create_card_image from safety import filter_and_correct_content, evaluate_image_with_vlm from content_generator import predict_post_success, verify_and_clean_urls from tracker import add_post_to_tracker, get_top_performing_posts, refresh_metrics from threads_api import ( publish_text_post, publish_image_post, publish_carousel_post, fetch_recent_posts ) def log_error(error_message): """ Записывает ошибку в error.log с временной меткой. """ timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") log_line = f"[{timestamp}] {error_message}\n" with open(config.ERROR_LOG_PATH, "a", encoding="utf-8") as f: f.write(log_line) print(f"[ERROR LOGGED] {error_message}") def execute_with_backoff(func, *args, **kwargs): """ Обертка для выполнения функций с экспоненциальным бэккоффом при ошибках. При сбое ждет 1 мин -> 5 мин -> 15 мин, затем повторяет. """ backoff_intervals = [60, 300, 900] for attempt, delay in enumerate(backoff_intervals): try: result = func(*args, **kwargs) if result: return result raise Exception("Функция вернула пустой результат или ошибку") except Exception as e: err_msg = f"Попытка {attempt+1} завершилась ошибкой для {func.__name__}: {e}. Ожидание {delay} сек перед повтором..." log_error(err_msg) time.sleep(delay) # Если за 3 попытки не вышло log_error(f"Критический сбой {func.__name__} после всех попыток повтора.") return None # --- ТЕСТОВЫЕ РЕЖИМЫ (CLI) --- def run_test_single(): """ Тестовый запуск: генерация, рендеринг и публикация одного одиночного поста. """ print("\n=== [ТЕСТ] ЗАПУСК ГЕНЕРАЦИИ ОДИНОЧНОГО ПОСТА ===") recent_posts = fetch_recent_posts(limit=10) print(f"[Scheduler] Загружено {len(recent_posts)} последних постов для исключения дублирования.") news = fetch_latest_news_and_trends(exclude_posts=recent_posts) if not news: print("[ТЕСТ Error] Не удалось получить новости.") return top_posts = get_top_performing_posts(limit=3) post_data = None safe_desc = "" best_score = 0 for attempt in range(1, 4): print(f"[Predictor] Генерация текста, попытка {attempt}...") temp_data = generate_single_post(news, exclude_posts=recent_posts) if not temp_data: continue temp_desc = filter_and_correct_content(temp_data['description']) temp_desc = verify_and_clean_urls(temp_desc) temp_desc = smart_truncate(temp_desc, 490) prediction = predict_post_success(temp_desc, top_posts) score = prediction.get('score', 0) criticism = prediction.get('criticism', '') print(f"[Predictor] Оценка: {score}/10. {criticism}") if score >= 7: post_data = temp_data safe_desc = temp_desc best_score = score print("[Predictor] Текст одобрен!") break else: print("[Predictor] Текст забракован, переписываем...") # В идеале нужно передать criticism обратно в промпт, но для упрощения просто генерируем заново if attempt == 3: post_data = temp_data safe_desc = temp_desc best_score = score print("[Predictor] Лимит попыток исчерпан, берем последний вариант.") if not post_data: print("[ТЕСТ Error] Не удалось сгенерировать JSON поста.") return print(f"[ТЕСТ] Проверенное описание: {safe_desc}") # Рендерим изображение (формат 1080x608 для одиночного поста) с автоулучшением через VLM bg_prompt = post_data['bg_prompt'] img_path = None for attempt in range(1, 4): filename = f"single_{int(time.time())}_att{attempt}.png" img_path = create_card_image( category=post_data['category'], bg_prompt=bg_prompt, title=post_data['card_title'], bullets=post_data['card_bullets'], output_filename=filename, is_carousel=False ) if not img_path: break # Мультимодальная оценка сгенерированного изображения verdict = evaluate_image_with_vlm(img_path, text_context=safe_desc) if verdict and not verdict.get("is_perfect", True): improvement = verdict.get("bg_prompt_improvement", "") if improvement: print(f"[Auto-Improvement] VLM выявил недостатки на попытке {attempt}. Улучшаем промпт фона: '{bg_prompt}' + '{improvement}'") bg_prompt = f"{bg_prompt}, {improvement}" if attempt < 3: try: os.remove(img_path) except Exception: pass continue # Если всё идеально или нет рекомендаций print(f"[Auto-Improvement] Изображение прошло VLM-аудит на попытке {attempt}!") break # Публикация if PREVIEW_MODE: print(f"[ТЕСТ PREVIEW] Режим предпросмотра. Пост НЕ опубликован. Картинка: {img_path}") return True media_id = publish_image_post(img_path, text=safe_desc) if media_id: print(f"[ТЕСТ SUCCESS] Пост успешно опубликован! ID: {media_id}") add_post_to_tracker(media_id, safe_desc, post_data["category"], best_score) return True else: print("[ТЕСТ Error] Ошибка публикации поста.") return False def run_test_carousel(): """ Тестовый запуск: генерация, рендеринг и публикация карусели. """ print("\n=== [ТЕСТ] ЗАПУСК ГЕНЕРАЦИИ КАРУСЕЛИ ===") recent_posts = fetch_recent_posts(limit=10) print(f"[Scheduler] Загружено {len(recent_posts)} последних постов для исключения дублирования.") news = fetch_latest_news_and_trends(exclude_posts=recent_posts) if not news: print("[ТЕСТ Error] Не удалось получить новости.") return carousel_data = generate_carousel_post(news, exclude_posts=recent_posts) if not carousel_data: print("[ТЕСТ Error] Не удалось сгенерировать JSON карусели.") return print(f"[ТЕСТ] Данные карусели:") print(f"Категория: {carousel_data['category']}") print(f"Описание: {carousel_data['description']}") safe_desc = filter_and_correct_content(carousel_data['description']) safe_desc = verify_and_clean_urls(safe_desc) safe_desc = smart_truncate(safe_desc, 490) # Генерируем слайды (квадратный формат 1080x1080 для карусели) с автоулучшением через VLM slide_paths = [] total_slides = len(carousel_data['slides']) for idx, slide in enumerate(carousel_data['slides']): bg_prompt = slide.get('bg_prompt', "background only, dark aesthetics") path = None for attempt in range(1, 4): filename = f"carousel_slide_{idx+1}_{int(time.time())}_att{attempt}.png" path = create_card_image( category=carousel_data['category'], bg_prompt=bg_prompt, title=slide['title'], bullets=slide['bullets'], output_filename=filename, is_carousel=True, slide_index=idx+1, total_slides=total_slides ) if not path: break # Мультимодальная оценка слайда карусели verdict = evaluate_image_with_vlm(path, text_context=f"{slide['title']} - {', '.join(slide['bullets'])}") if verdict and not verdict.get("is_perfect", True): improvement = verdict.get("bg_prompt_improvement", "") if improvement: print(f"[Auto-Improvement] VLM выявил недостатки на слайде {idx+1} (попытка {attempt}). Улучшаем промпт фона: '{bg_prompt}' + '{improvement}'") bg_prompt = f"{bg_prompt}, {improvement}" if attempt < 3: try: os.remove(path) except Exception: pass continue # Всё отлично print(f"[Auto-Improvement] Слайд {idx+1} успешно прошел VLM-аудит на попытке {attempt}!") break if path: slide_paths.append(path) # Публикация карусели if PREVIEW_MODE: print(f"[ТЕСТ PREVIEW] Режим предпросмотра. Карусель НЕ опубликована. Картинки: {slide_paths}") return True media_id = publish_carousel_post(slide_paths, text=safe_desc) if media_id: print(f"[ТЕСТ SUCCESS] Карусель успешно опубликована! ID: {media_id}") return True else: print("[ТЕСТ Error] Не удалось опубликовать карусель.") return False # --- ОСНОВНОЙ РЕЖИМ ПЛАНИРОВАНИЯ --- def calculate_next_prime_time(settings): """ Рассчитывает время следующей публикации с учетом настроек из settings.json. """ from zoneinfo import ZoneInfo tz = ZoneInfo("Europe/Kyiv") now = datetime.now(tz) def parse_time(time_str): h, m = map(int, time_str.split(':')) return now.replace(hour=h, minute=m, second=0, microsecond=0) # Определяем утреннее окно на сегодня morning_start = parse_time(settings["MORNING_SLOT_START"]) morning_end = parse_time(settings["MORNING_SLOT_END"]) # Определяем вечернее окно на сегодня evening_start = parse_time(settings["EVENING_SLOT_START"]) evening_end = parse_time(settings["EVENING_SLOT_END"]) if now < morning_end: # Планируем на сегодняшнее утро target_dt = morning_start.replace(minute=random.randint(0, 59)) if target_dt < now: return now.timestamp() + 300 # Если время уже немного в прошлом, запускаем через 5 минут return target_dt.timestamp() elif now < evening_end: # Планируем на сегодняшний вечер target_dt = evening_start.replace(minute=random.randint(0, 59)) if target_dt < now: return now.timestamp() + 300 return target_dt.timestamp() else: # Планируем на завтрашнее утро tomorrow = now.timestamp() + 24 * 3600 tomorrow_dt = datetime.fromtimestamp(tomorrow, tz=tz) h, m = map(int, settings["MORNING_SLOT_START"].split(':')) target_dt = tomorrow_dt.replace(hour=h, minute=random.randint(0, 59), second=0, microsecond=0) return target_dt.timestamp() def main_loop(): """ Основной бесконечный цикл планировщика публикаций и активности. """ print("[Hermes Agent] Планировщик запущен. Нажмите Ctrl+C для выхода.") from zoneinfo import ZoneInfo tz = ZoneInfo("Europe/Kyiv") carousels_today = 0 single_posts_today = 0 comments_today = 0 last_post_time = 0 last_reset_day = datetime.now(tz).day settings = settings_manager.load_settings() # Рассчитываем время следующего поста next_post_time = calculate_next_prime_time(settings) wait_minutes = (next_post_time - time.time()) / 60 next_post_dt_str = datetime.fromtimestamp(next_post_time, tz=tz).strftime("%Y-%m-%d %H:%M:%S") print(f"[Scheduler] Следующая публикация запланирована на {next_post_dt_str} (Киев) (через {wait_minutes:.1f} минут).") while True: try: current_time = time.time() now_tz = datetime.now(tz) settings = settings_manager.load_settings() # Динамически обновляем настройки # Сброс суточных лимитов в полночь if now_tz.day != last_reset_day: carousels_today = 0 single_posts_today = 0 comments_today = 0 last_reset_day = now_tz.day print("[Scheduler] Новый день начался. Сброс суточных лимитов и счетчиков.") # Проверка: пришло ли время делать пост? if current_time >= next_post_time: # Проверяем, не тихий ли сегодня день (Воскресенье = 6) is_quiet_day = now_tz.weekday() == 6 if is_quiet_day: print("[Scheduler] Сегодня воскресенье — тихий день. Пропускаем публикации, оставляем только комментирование (вовлечение).") else: # Проверяем, есть ли доступные лимиты if carousels_today < settings["DAILY_CAROUSELS_LIMIT"] or single_posts_today < settings["DAILY_SINGLE_POSTS_LIMIT"]: # Выбираем тип публикации на основе лимитов if carousels_today < settings["DAILY_CAROUSELS_LIMIT"] and single_posts_today < settings["DAILY_SINGLE_POSTS_LIMIT"]: # Если оба лимита не превышены, выбираем с весом total_limit = settings["DAILY_CAROUSELS_LIMIT"] + settings["DAILY_SINGLE_POSTS_LIMIT"] is_carousel = random.random() < (settings["DAILY_CAROUSELS_LIMIT"] / total_limit) elif carousels_today < settings["DAILY_CAROUSELS_LIMIT"]: is_carousel = True else: is_carousel = False # Выполняем генерацию и публикацию с бэккоффом if is_carousel: print("[Scheduler] Время публикации поста-карусели.") success = execute_with_backoff(run_test_carousel) if success: carousels_today += 1 else: print("[Scheduler] Время публикации одиночного поста.") success = execute_with_backoff(run_test_single) if success: single_posts_today += 1 last_post_time = current_time else: print(f"[Scheduler] Все суточные лимиты публикаций исчерпаны на сегодня ({settings['DAILY_CAROUSELS_LIMIT']} каруселей и {settings['DAILY_SINGLE_POSTS_LIMIT']} постов).") # Планируем следующий пост next_post_time = calculate_next_prime_time(settings) wait_minutes = (next_post_time - time.time()) / 60 next_post_dt_str = datetime.fromtimestamp(next_post_time, tz=tz).strftime("%Y-%m-%d %H:%M:%S") print(f"[Scheduler] Пост обработан. Следующий запланирован на {next_post_dt_str} (Киев) (через {wait_minutes:.1f} минут).") # Проверка активности комментирования доноров (Отключено по просьбе пользователя) pass # Спим 60 секунд между проверками планировщика time.sleep(60) except KeyboardInterrupt: print("\n[Hermes Agent] Работа планировщика остановлена вручную.") sys.exit(0) except Exception as e: log_error(f"Непредвиденное исключение в основном цикле планировщика: {e}") time.sleep(60) if __name__ == "__main__": parser = argparse.ArgumentParser(description="Hermes Agent - Threads Automation Agent") parser.add_argument("--test-single", action="store_true", help="Запустить тест генерации и отправки одного поста") parser.add_argument("--test-carousel", action="store_true", help="Запустить тест генерации и отправки одной карусели") parser.add_argument("--loop", action="store_true", help="Запустить планировщик в бесконечном цикле") parser.add_argument("--preview", action="store_true", help="Режим отладки (без отправки в Threads)") args = parser.parse_args() if args.test_single: run_test_single() elif args.test_carousel: run_test_carousel() elif args.loop: main_loop() else: parser.print_help()