"""Bidirectional data sync between local and production. Usage: # Full bidirectional sync python scripts/remote_sync.py # Push local computed data to production only python scripts/remote_sync.py --direction push # Pull user data from production only python scripts/remote_sync.py --direction pull # Sync specific table python scripts/remote_sync.py --table ohlcv_data --direction push # Force full sync (ignore last sync timestamp) python scripts/remote_sync.py --full # Dry run (show what would sync, don't execute) python scripts/remote_sync.py --dry-run # Check remote server status python scripts/remote_sync.py --status Cron examples: # Push computed data every 30 minutes */30 * * * * cd /path/to/project && .venv/bin/python scripts/remote_sync.py --direction push >> logs/sync.log 2>&1 # Pull user data every 15 minutes */15 * * * * cd /path/to/project && .venv/bin/python scripts/remote_sync.py --direction pull >> logs/sync.log 2>&1 """ from __future__ import annotations import sys import os import argparse import logging import json from datetime import datetime # Ensure project root is in path sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from app import create_app def setup_logging(verbose: bool = False): """Configure logging.""" level = logging.DEBUG if verbose else logging.INFO fmt = '%(asctime)s [%(levelname)s] %(message)s' logging.basicConfig(level=level, format=fmt, datefmt='%Y-%m-%d %H:%M:%S') # Suppress noisy libraries logging.getLogger('urllib3').setLevel(logging.WARNING) logging.getLogger('sqlalchemy.engine').setLevel(logging.WARNING) def check_status(app): """Check remote server sync status.""" with app.app_context(): from app.services.remote_sync.sync_client import RemoteSyncClient remote_url = app.config.get('SYNC_REMOTE_URL', '') sync_key = app.config.get('SYNC_API_KEY', '') if not remote_url or not sync_key: print('[ERROR] SYNC_REMOTE_URL and SYNC_API_KEY must be set in .env') return client = RemoteSyncClient(remote_url, sync_key) print(f'Checking remote: {remote_url}') status = client.check_status() if status: print(f'[OK] Remote server is reachable') print(f' Sync enabled: {status.get("sync_enabled")}') print(f' Server time: {status.get("server_time")}') print(f' Tables: {len(status.get("tables", []))}') counts = status.get('table_counts', {}) if counts: print(f'\n Table record counts:') for t, c in sorted(counts.items()): print(f' {t:30s} {c:>10,}') last = status.get('last_sync', {}) if last: print(f'\n Last sync timestamps:') for k, v in sorted(last.items()): print(f' {k:45s} {v}') else: print('[ERROR] Could not reach remote server') def dry_run(app, direction: str, table: str | None): """Show what would be synced without executing.""" with app.app_context(): from app.services.remote_sync.serializers import ( get_table_registry, get_push_tables, get_pull_tables, ) from app.models.settings import AppSettings registry = get_table_registry() print(f'=== DRY RUN (direction: {direction}) ===\n') if direction in ('push', 'both'): tables = [table] if table else get_push_tables() print('PUSH (local → production):') for t in tables: if t not in registry: print(f' [SKIP] {t} — not in registry') continue cfg = registry[t] model = cfg['model'] ts_col = cfg['timestamp_col'] # Check last sync key = f'sync_last_push_{t}' since = AppSettings.get(key) query = model.query if since: try: since_dt = datetime.fromisoformat(since) col = getattr(model, ts_col, None) if col is not None: query = query.filter(col >= since_dt) except ValueError: pass count = query.count() total = model.query.count() print(f' {t:30s} {count:>8,} new / {total:>10,} total (since: {since or "never"})') if direction in ('pull', 'both'): tables = [table] if table else get_pull_tables() print(f'\nPULL (production → local):') for t in tables: if t not in registry: print(f' [SKIP] {t} — not in registry') continue cfg = registry[t] model = cfg['model'] key = f'sync_last_pull_{t}' since = AppSettings.get(key) local_count = model.query.count() print(f' {t:30s} local: {local_count:>10,} (since: {since or "never"})') print(f'\nRemote: {app.config.get("SYNC_REMOTE_URL", "NOT SET")}') print(f'Sync enabled: {app.config.get("SYNC_ENABLED", False)}') def run_sync(app, direction: str, table: str | None, full: bool): """Execute the actual sync.""" with app.app_context(): from app.services.remote_sync.sync_orchestrator import SyncOrchestrator orchestrator = SyncOrchestrator() if full: print('[INFO] Full sync mode — ignoring last sync timestamps') since_override = datetime.min if full else None if table: print(f'[SYNC] Table: {table}, Direction: {direction}') result = orchestrator.sync_table(table, direction, since=since_override) _print_table_result(table, direction, result) else: print(f'[SYNC] Direction: {direction}') result = orchestrator.sync_all(direction) _print_full_result(result) def _print_table_result(table: str, direction: str, result: dict): """Print result for a single table sync.""" status = result.get('status', 'unknown') count = result.get('count', result.get('upserted', 0)) print(f'\n {table}: {status} — {count} records') if result.get('errors'): for e in result['errors'][:5]: print(f' [ERROR] {e}') def _print_full_result(result: dict): """Print result for full sync.""" duration = result.get('duration_seconds', 0) print(f'\n{"=" * 60}') print(f'Sync completed in {duration:.1f}s') print(f'{"=" * 60}') for direction in ('push', 'pull'): data = result.get(direction, {}) if not data: continue label = 'PUSH (local → production)' if direction == 'push' else 'PULL (production → local)' print(f'\n{label}:') for table, r in data.items(): status = r.get('status', 'unknown') count = r.get('count', r.get('upserted', 0)) icon = '✓' if status == 'ok' else '⚠' if status == 'partial' else '✗' print(f' {icon} {table:30s} {count:>8,} records [{status}]') errors = result.get('errors', []) if errors: print(f'\nErrors ({len(errors)}):') for e in errors: print(f' - {e}') def main(): parser = argparse.ArgumentParser( description='Bidirectional data sync between local and production', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__, ) parser.add_argument('--direction', choices=['push', 'pull', 'both'], default='both', help='Sync direction (default: both)') parser.add_argument('--table', help='Sync specific table only') parser.add_argument('--full', action='store_true', help='Full sync (ignore last sync timestamp)') parser.add_argument('--dry-run', action='store_true', help='Show what would sync without executing') parser.add_argument('--status', action='store_true', help='Check remote server status') parser.add_argument('--verbose', '-v', action='store_true', help='Enable debug logging') args = parser.parse_args() setup_logging(args.verbose) app = create_app() if args.status: check_status(app) return if args.dry_run: dry_run(app, args.direction, args.table) return # Validate config with app.app_context(): if not app.config.get('SYNC_REMOTE_URL'): print('[ERROR] SYNC_REMOTE_URL not set in .env') sys.exit(1) if not app.config.get('SYNC_API_KEY'): print('[ERROR] SYNC_API_KEY not set in .env') sys.exit(1) if not app.config.get('SYNC_ENABLED'): print('[ERROR] SYNC_ENABLED is not true in .env') sys.exit(1) run_sync(app, args.direction, args.table, args.full) if __name__ == '__main__': main()