""" Парсинг одного канала - 3 поста за вчера Простой и быстрый вариант для тестирования """ import asyncio from datetime import datetime, timedelta from src.parsers import XParser, InstagramParser from src.database.db import session_scope from src.database.models import Source, Post, PostMetrics from src.metrics.calculator import MetricsCalculator async def parse_one_channel(): """Парсинг одного канала - взять 3 поста за вчера""" print("=" * 60) print("PARSING ONE CHANNEL - 3 POSTS FROM YESTERDAY") print("=" * 60) # Выбираем канал для парсинга # Вариант 1: X.com (более стабилен) channel = "Zillow" platform = "x" followers = 2000000 # Вариант 2: Instagram (если X.com не работает - раскомментируйте) # channel = "zillow" # platform = "instagram" # followers = 1500000 print(f"\nChannel: {platform}/@{channel}") print(f"Followers: {followers:,}") print(f"Target: 3 posts from yesterday") print("-" * 60) # Создаём парсер if platform == "x": parser = XParser(channel, followers) else: parser = InstagramParser(channel, followers) # Определяем временной диапазон - ВЧЕРА today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) yesterday_start = today - timedelta(days=1) yesterday_end = today print(f"\nYesterday range:") print(f" Start: {yesterday_start.strftime('%Y-%m-%d %H:%M')}") print(f" End: {yesterday_end.strftime('%Y-%m-%d %H:%M')}") print() # Парсим посты за последние 2 дня (чтобы точно захватить вчера) try: print("Parsing posts...") all_posts = await parser.parse_posts(days=2) if not all_posts: print("\n! No posts found from parser") print("\nPossible reasons:") print(" - Platform is blocking requests") print(" - Account is private") print(" - No posts in this period") print("\nTry:") print(" 1. Change channel in this script") print(" 2. Use Instagram instead of X.com (or vice versa)") print(" 3. Use demo posts: python create_demo_posts.py") return print(f"Found {len(all_posts)} posts total") # Фильтруем только посты за ВЧЕРА yesterday_posts = [ post for post in all_posts if yesterday_start <= post.created_at < yesterday_end ] print(f"Posts from yesterday: {len(yesterday_posts)}") if not yesterday_posts: print("\n! No posts found for yesterday") print(f"\nAll posts dates:") for i, post in enumerate(all_posts[:5], 1): print(f" {i}. {post.created_at.strftime('%Y-%m-%d %H:%M')}") print("\nTry:") print(" - Increase days in script (parse_posts(days=7))") print(" - Use demo posts: python create_demo_posts.py") return # Берём только 3 поста selected_posts = yesterday_posts[:3] print(f"\nSelected {len(selected_posts)} posts for saving") print("-" * 60) # Сохраняем в базу данных with session_scope() as session: # Получить или создать источник source = session.query(Source).filter_by( platform=platform, username=channel ).first() if not source: source = Source( platform=platform, username=channel, identifier=channel, subscriber_count=followers, followers_count=followers, is_active=1 ) session.add(source) session.flush() print(f"\n+ Created source: {platform}/@{channel}") else: print(f"\n= Source exists: {platform}/@{channel}") # Сохранить посты saved_count = 0 for i, parsed_post in enumerate(selected_posts, 1): # Проверить, не существует ли уже existing = session.query(Post).filter_by( source_id=source.id, external_id=parsed_post.external_id ).first() if existing: print(f"\n{i}. Post already exists: {parsed_post.external_id}") # Обновить метрики если они изменились if parsed_post.views > existing.views: existing.views = parsed_post.views existing.likes = parsed_post.likes existing.comments = parsed_post.comments existing.shares = parsed_post.shares print(f" Updated metrics") post = existing else: # Создать новый пост post = Post( source_id=source.id, external_id=parsed_post.external_id, content=parsed_post.content, url=parsed_post.url, views=parsed_post.views, reactions=parsed_post.reactions, likes=parsed_post.likes, comments=parsed_post.comments, shares=parsed_post.shares, created_at=parsed_post.created_at, parsed_at=datetime.now() ) session.add(post) session.flush() print(f"\n{i}. + New post saved: {parsed_post.external_id}") saved_count += 1 # Показать информацию о посте content_preview = post.content[:60] + "..." if len(post.content) > 60 else post.content print(f" Content: {content_preview}") print(f" Created: {post.created_at.strftime('%Y-%m-%d %H:%M')}") print(f" Metrics: views={post.views:,}, likes={post.likes:,}, comments={post.comments:,}") print(f" URL: {post.url}") # Рассчитать метрики try: existing_metrics = session.query(PostMetrics).filter_by(post_id=post.id).first() if not existing_metrics: metrics = MetricsCalculator.calculate_post_metrics(post, source) metrics.post_id = post.id session.add(metrics) session.flush() print(f" Score: {metrics.normalized_score:.2f}, Engagement: {metrics.engagement_rate:.2%}") else: # Обновить существующие метрики metrics = MetricsCalculator.calculate_post_metrics(post, source) existing_metrics.engagement_rate = metrics.engagement_rate existing_metrics.normalized_score = metrics.normalized_score existing_metrics.subscriber_ratio = metrics.subscriber_ratio existing_metrics.total_engagement = metrics.total_engagement existing_metrics.calculated_at = datetime.now() print(f" Score: {metrics.normalized_score:.2f}, Engagement: {metrics.engagement_rate:.2%}") except Exception as e: print(f" ! Error calculating metrics: {e}") print("\n" + "=" * 60) print(f"SUCCESS! Saved {saved_count} new posts") print("=" * 60) print("\nNext steps:") print(" 1. Start bot: python main.py") print(" 2. In Telegram: /top_yesterday") print(" 3. Check: python show_top_posts.py") print("=" * 60) except Exception as e: print(f"\n! ERROR: {e}") import traceback traceback.print_exc() print("\nTry:") print(" 1. Use demo posts: python create_demo_posts.py") print(" 2. Change channel in this script") print(" 3. Try Instagram instead of X.com") if __name__ == "__main__": asyncio.run(parse_one_channel())