""" Create sample posts with metrics for testing the bot """ import sys import os from datetime import datetime, timedelta from dotenv import load_dotenv # Fix encoding for Windows if sys.platform == 'win32': os.environ['PYTHONIOENCODING'] = 'utf-8' if hasattr(sys.stdout, 'reconfigure'): sys.stdout.reconfigure(encoding='utf-8') sys.stderr.reconfigure(encoding='utf-8') load_dotenv() from src.database.db import init_db, session_scope from src.database.models import Source, Post, PostMetrics from src.metrics.calculator import MetricsCalculator def create_sample_posts(): """Create sample posts with metrics for testing""" print("="*60) print("CREATING SAMPLE POSTS WITH METRICS") print("="*60) # Initialize database if not init_db(): print("❌ Error initializing database") return False with session_scope() as session: # Get or create X.com source x_source = session.query(Source).filter_by( platform='x', username='Zillow' ).first() # Real Zillow has ~380k followers, not 2M real_followers = 380000 if not x_source: x_source = Source( platform='x', username='Zillow', identifier='Zillow', subscriber_count=real_followers, followers_count=real_followers, is_active=1 ) session.add(x_source) session.flush() print(f"✓ Created X.com source: @Zillow") else: # Update to real count x_source.subscriber_count = real_followers x_source.followers_count = real_followers print(f"✓ Updated X.com source: @Zillow (followers: {real_followers:,})") # Calculate timestamps # "Yesterday" generally means previous calendar day. # To be safe for /top_yesterday, we use 30 hours ago (creates posts in yesterday's window) yesterday_ts = datetime.now() - timedelta(hours=30) # Create sample X.com posts x_posts = [ { 'external_id': '1702796773071475004', # Real-ish ID format 'content': 'Представляем: Компас Один. Сегодняшний день знаменует собой новую главу для Compass. С момента нашего основания мы инвестировали 1,7 миллиарда долларов в создание первой комплексной технологической платформы для агентов.', 'url': 'https://x.com/Zillow/status/1702796773071475004', 'views': 125000, 'likes': 4500, 'comments': 320, 'shares': 890, 'created_at': yesterday_ts + timedelta(minutes=15) }, { 'external_id': '1702796773071475005', 'content': 'Новое исследование показывает рост цен на недвижимость в крупных городах на 15% за последний год. Эксперты прогнозируют дальнейший рост.', 'url': 'https://x.com/Zillow/status/1702796773071475005', 'views': 89000, 'likes': 2100, 'comments': 180, 'shares': 450, 'created_at': yesterday_ts + timedelta(hours=2) }, { 'external_id': '1702796773071475006', 'content': 'Топ-5 советов для первого покупателя жилья: 1) Проверьте кредитный рейтинг 2) Накопите на первоначальный взнос 3) Изучите рынок 4) Найдите хорошего агента 5) Будьте терпеливы', 'url': 'https://x.com/Zillow/status/1702796773071475006', 'views': 156000, 'likes': 5600, 'comments': 420, 'shares': 1200, 'created_at': yesterday_ts + timedelta(hours=5) } ] print(f"\nCreating X.com posts...") for post_data in x_posts: # Check if exists existing = session.query(Post).filter_by( source_id=x_source.id, external_id=post_data['external_id'] ).first() if not existing: post = Post( source_id=x_source.id, external_id=post_data['external_id'], content=post_data['content'], url=post_data['url'], views=post_data['views'], likes=post_data['likes'], comments=post_data['comments'], shares=post_data['shares'], reactions=0, created_at=post_data['created_at'], parsed_at=datetime.utcnow() ) session.add(post) session.flush() # Calculate metrics metrics = MetricsCalculator.calculate_post_metrics(post, x_source) session.add(metrics) print(f" ✓ Created post {post_data['external_id']}: views={post.views:,}, likes={post.likes:,}") else: print(f" ↻ Post {post_data['external_id']} already exists, updating...") existing.created_at = post_data['created_at'] existing.url = post_data['url'] existing.views = post_data['views'] existing.likes = post_data['likes'] existing.shares = post_data['shares'] session.add(existing) # Get Instagram sources instagram_sources = session.query(Source).filter_by(platform='instagram').limit(3).all() if instagram_sources: print(f"\nCreating Instagram posts...") for i, ig_source in enumerate(instagram_sources): # Create sample Instagram post post_data = { 'external_id': f'C{i}xYz{i}AbCd', # Real-ish shortcode 'content': f'Потрясающий дом в {["Майами", "Лос-Анджелесе", "Нью-Йорке"][i]}! Современный дизайн, панорамные виды, премиум-локация. #недвижимость #luxury #realestate', 'url': f'https://www.instagram.com/p/C{i}xYz{i}AbCd/', 'views': [95000, 120000, 78000][i], 'likes': [3200, 4500, 2800][i], 'comments': [180, 250, 150][i], 'shares': 0, # Instagram doesn't have shares 'created_at': yesterday_ts + timedelta(hours=8+i*2) } # Check if exists existing = session.query(Post).filter_by( source_id=ig_source.id, external_id=post_data['external_id'] ).first() if not existing: post = Post( source_id=ig_source.id, external_id=post_data['external_id'], content=post_data['content'], url=post_data['url'], views=post_data['views'], likes=post_data['likes'], comments=post_data['comments'], shares=post_data['shares'], reactions=0, created_at=post_data['created_at'], parsed_at=datetime.utcnow() ) session.add(post) session.flush() # Calculate metrics metrics = MetricsCalculator.calculate_post_metrics(post, ig_source) session.add(metrics) print(f" ✓ Created post for @{ig_source.username}: views={post.views:,}, likes={post.likes:,}") else: print(f" ↻ Post for @{ig_source.username} already exists, updating...") existing.created_at = post_data['created_at'] existing.url = post_data['url'] session.add(existing) print(f"\n✓ Sample posts created successfully!") # Show summary total_posts = session.query(Post).count() x_posts_count = session.query(Post).join(Source).filter(Source.platform == 'x').count() ig_posts_count = session.query(Post).join(Source).filter(Source.platform == 'instagram').count() print(f"\n📊 Database summary:") print(f" Total posts: {total_posts}") print(f" X.com posts: {x_posts_count}") print(f" Instagram posts: {ig_posts_count}") return True if __name__ == '__main__': success = create_sample_posts() sys.exit(0 if success else 1)