""" MTG Database Monitor Monitors PostgreSQL database metrics for both mtgonline and mtgdata databases. Tracks: - Database size and growth - Table sizes - Index sizes - Query performance - Weekly refresh metrics - Image URL statistics - Refresh success/failure rates """ import asyncpg import asyncio from datetime import datetime from typing import Dict, List, Tuple import json from pathlib import Path from app.core.settings import get_settings settings = get_settings() # Database configuration from settings COCKATRICE_DB = settings.DATABASE_URL MTG_DB = settings.MTG_DATABASE_URL class MtgMonitor: def __init__(self): self.mtg_conn = None self.mtgonline_conn = None self.metrics = {} async def connect(self): """Establish connections to both databases.""" try: self.mtg_conn = await asyncpg.connect(MTG_DB) self.mtgonline_conn = await asyncpg.connect(COCKATRICE_DB) return True except Exception as e: print(f"Connection error: {e}") return False async def disconnect(self): """Close database connections.""" if self.mtg_conn: await self.mtg_conn.close() if self.mtgonline_conn: await self.mtgonline_conn.close() async def get_database_size(self) -> Dict[str, float]: """Get size of databases in GB.""" try: # MTG database size mtg_size = await self.mtg_conn.fetchval(""" SELECT pg_database_size(current_database()) as size """) # MTG Online database size mtgonline_size = await self.mtgonline_conn.fetchval(""" SELECT pg_database_size(current_database()) as size """) return { "mtgdata": mtg_size / (1024**3), # Convert to GB "mtgonline": mtgonline_size / (1024**3) } except Exception as e: print(f"Error getting database sizes: {e}") return {} async def get_table_sizes(self) -> Dict[str, float]: """Get sizes of all tables in MB.""" try: result = await self.mtg_conn.fetch(""" SELECT schemaname || '.' || tablename as table_name, pg_size_pretty(pg_total_relation_size(schemaname || '.' || tablename)) as size, pg_total_relation_size(schemaname || '.' || tablename) as size_bytes FROM pg_tables WHERE schemaname = 'public' ORDER BY pg_total_relation_size(schemaname || '.' || tablename) DESC """) return { row[0]: row[2] / (1024**2) # Convert to MB for row in result } except Exception as e: print(f"Error getting table sizes: {e}") return {} async def get_index_sizes(self) -> Dict[str, float]: """Get sizes of all indexes in MB.""" try: result = await self.mtg_conn.fetch(""" SELECT indexname as index_name, pg_size_pretty(pg_relation_size(indexname::regclass)) as size, pg_relation_size(indexname::regclass) as size_bytes FROM pg_indexes WHERE schemaname = 'public' ORDER BY pg_relation_size(indexname::regclass) DESC """) return { row[0]: row[2] / (1024**2) # Convert to MB for row in result } except Exception as e: print(f"Error getting index sizes: {e}") return {} async def get_table_row_counts(self) -> Dict[str, int]: """Get row counts for all tables.""" try: result = await self.mtg_conn.fetch(""" SELECT schemaname || '.' || tablename as table_name, n_live_tup as row_count FROM pg_stat_user_tables WHERE schemaname = 'public' ORDER BY n_live_tup DESC """) return { row[0]: row[1] for row in result } except Exception as e: print(f"Error getting row counts: {e}") return {} async def get_image_url_stats(self) -> Dict[str, any]: """Get statistics about image URLs in cards table.""" try: # Count cards with images cards_with_images = await self.mtg_conn.fetchval(""" SELECT COUNT(*) FROM mtg_cards WHERE images IS NOT NULL AND images != '{}' """) # Total unique image URLs unique_images = await self.mtg_conn.fetchval(""" SELECT COUNT(DISTINCT jsonb_array_elements_text(images)) FROM mtg_cards WHERE images IS NOT NULL AND images != '{}' """) # Most common image resolutions resolutions = await self.mtg_conn.fetch(""" SELECT jsonb_object_keys(images) as resolution, COUNT(*) as count FROM mtg_cards WHERE images IS NOT NULL AND images != '{}' GROUP BY jsonb_object_keys(images) ORDER BY count DESC """) # Image URL patterns (domains) domains = await self.mtg_conn.fetch(""" SELECT regexp_replace(images::text, '.*("normal":"[^"]*").*', '\\1') as domain FROM mtg_cards WHERE images IS NOT NULL AND images != '{}' LIMIT 1000 """) return { "cards_with_images": cards_with_images, "unique_image_urls": unique_images, "resolutions": {row[0]: row[1] for row in resolutions}, "sample_domains": [str(d[0]) for d in domains[:5]] } except Exception as e: print(f"Error getting image stats: {e}") return {} async def get_refresh_metrics(self) -> Dict[str, any]: """Get refresh statistics from mtg_refresh_log.""" try: # Total refreshes total_refreshes = await self.mtg_conn.fetchval(""" SELECT COUNT(*) FROM mtg_refresh_log """) # Success vs failure rates status_counts = await self.mtg_conn.fetch(""" SELECT status, COUNT(*) as count FROM mtg_refresh_log GROUP BY status ORDER BY count DESC """) # Average duration avg_duration = await self.mtg_conn.fetchval(""" SELECT AVG(duration_seconds) FROM mtg_refresh_log """) # Last refresh last_refresh = await self.mtg_conn.fetch(""" SELECT * FROM mtg_refresh_log ORDER BY refresh_date DESC LIMIT 1 """) # Cards updated per refresh (average) avg_cards = await self.mtg_conn.fetchval(""" SELECT AVG(cards_updated) FROM mtg_refresh_log WHERE status = 'SUCCESS' """) return { "total_refreshes": total_refreshes, "status_counts": {row[0]: row[1] for row in status_counts}, "avg_duration_seconds": avg_duration, "avg_cards_per_refresh": avg_cards, "last_refresh": last_refresh[0] if last_refresh else None } except Exception as e: print(f"Error getting refresh metrics: {e}") return {} async def collect_metrics(self) -> Dict[str, any]: """Collect all metrics.""" if not await self.connect(): return {"error": "Failed to connect to databases"} try: metrics = { "timestamp": datetime.now().isoformat(), "database_sizes": await self.get_database_size(), "table_sizes": await self.get_table_sizes(), "index_sizes": await self.get_index_sizes(), "row_counts": await self.get_table_row_counts(), "image_stats": await self.get_image_url_stats(), "refresh_metrics": await self.get_refresh_metrics() } self.metrics = metrics return metrics finally: await self.disconnect() def generate_report(self, metrics: Dict[str, any]) -> str: """Generate a human-readable report.""" report = [] report.append("=" * 60) report.append("MTG Database Monitor Report") report.append("=" * 60) report.append(f"Generated: {metrics['timestamp']}") report.append("") # Database sizes report.append("DATABASE SIZES") report.append("-" * 40) for db, size in metrics.get('database_sizes', {}).items(): report.append(f" {db}: {size:.2f} GB") report.append("") # Table sizes report.append("TABLE SIZES") report.append("-" * 40) for table, size in metrics.get('table_sizes', {}).items(): report.append(f" {table}: {size:.2f} MB") report.append("") # Row counts report.append("ROW COUNTS") report.append("-" * 40) for table, count in metrics.get('row_counts', {}).items(): report.append(f" {table}: {count:,} rows") report.append("") # Image stats report.append("IMAGE STATISTICS") report.append("-" * 40) image_stats = metrics.get('image_stats', {}) report.append(f" Cards with images: {image_stats.get('cards_with_images', 0):,}") report.append(f" Unique image URLs: {image_stats.get('unique_image_urls', 0):,}") if image_stats.get('resolutions'): report.append(" Image resolutions:") for res, count in image_stats['resolutions'].items(): report.append(f" {res}: {count:,} cards") if image_stats.get('sample_domains'): report.append(" Sample domains:") for domain in image_stats['sample_domains']: report.append(f" {domain}") report.append("") # Refresh metrics report.append("REFRESH METRICS") report.append("-" * 40) refresh = metrics.get('refresh_metrics', {}) report.append(f" Total refreshes: {refresh.get('total_refreshes', 0)}") if refresh.get('status_counts'): report.append(" Status counts:") for status, count in refresh['status_counts'].items(): report.append(f" {status}: {count}") report.append(f" Average duration: {refresh.get('avg_duration_seconds', 0):.1f} seconds") report.append(f" Average cards per refresh: {refresh.get('avg_cards_per_refresh', 0):,}") if refresh.get('last_refresh'): report.append(f" Last refresh: {refresh['last_refresh'].get('refresh_date', 'N/A')}") report.append(f" Status: {refresh['last_refresh'].get('status', 'N/A')}") report.append(f" Cards updated: {refresh['last_refresh'].get('cards_updated', 0):,}") report.append("") return "\n".join(report) async def main(): """Run monitoring and generate report.""" monitor = MtgMonitor() metrics = await monitor.collect_metrics() if metrics.get('error'): print(f"Error: {metrics['error']}") return report = monitor.generate_report(metrics) print(report) # Save to file with open("/app/mtg-monitor-report.txt", "w") as f: f.write(report) # Save metrics as JSON for programmatic use with open("/app/mtg-monitor-metrics.json", "w") as f: json.dump(metrics, f, indent=2, default=str) print("Report saved to /app/mtg-monitor-report.txt") print("Metrics saved to /app/mtg-monitor-metrics.json") if __name__ == "__main__": asyncio.run(main())