#!/usr/bin/env python3 """ Repository Sync Manager with REST API Integration Manages automatic syncing between A6-9V/MQL5-Google-Onedrive and L6-N9 repositories """ import os import sys import json import subprocess import requests from datetime import datetime, timezone from pathlib import Path from typing import Dict, Optional, Tuple class RepoSyncManager: """Manages repository synchronization with REST API authentication""" def __init__(self): self.source_repo = "A6-9V/MQL5-Google-Onedrive" self.target_repo = os.getenv("L6_N9_REPO", "A6-9V/L6-N9") self.sync_branch = os.getenv("L6_N9_SYNC_BRANCH", "main") self.rest_api_key = os.getenv("REST_API_KEY", "") self.rest_api_url = os.getenv("REST_API_URL", "") self.sync_token = os.getenv("REPO_SYNC_TOKEN", "") def validate_config(self) -> Tuple[bool, str]: """Validate sync configuration""" if not self.sync_token: return False, "REPO_SYNC_TOKEN not configured" if not self.rest_api_key and self.rest_api_url: return True, "Warning: REST_API_KEY not set but URL is configured" return True, "Configuration valid" def get_git_info(self) -> Dict[str, str]: """Get current git information""" try: commit = subprocess.check_output( ["git", "rev-parse", "HEAD"], text=True ).strip() branch = subprocess.check_output( ["git", "rev-parse", "--abbrev-ref", "HEAD"], text=True ).strip() return { "commit": commit, "branch": branch, "timestamp": datetime.now(timezone.utc).isoformat().replace('+00:00', 'Z') } except subprocess.CalledProcessError as e: print(f"⚠️ Error getting git info: {e}") return {} def notify_rest_api(self, event_data: Dict) -> bool: """Send notification to REST API endpoint""" if not self.rest_api_url or not self.rest_api_key: print("ℹ️ REST API notification skipped (not configured)") return True headers = { "Content-Type": "application/json", "Authorization": f"Bearer {self.rest_api_key}" } try: response = requests.post( self.rest_api_url, headers=headers, json=event_data, timeout=30 ) if response.status_code in [200, 201, 202]: print(f"✅ REST API notification sent successfully") return True else: print(f"⚠️ REST API notification failed: {response.status_code}") return False except requests.RequestException as e: print(f"⚠️ REST API notification error: {e}") return False def sync_directories(self, direction: str = "push") -> bool: """Sync directories between repositories""" git_info = self.get_git_info() # Prepare sync event data event_data = { "event": "repo_sync", "source": self.source_repo, "target": self.target_repo, "direction": direction, "timestamp": git_info.get("timestamp", datetime.now(timezone.utc).isoformat().replace('+00:00', 'Z')), "branch": git_info.get("branch", "unknown"), "commit": git_info.get("commit", "unknown") } print(f"🔄 Syncing repositories: {self.source_repo} → {self.target_repo}") print(f" Direction: {direction}") print(f" Branch: {self.sync_branch}") # Notify REST API before sync self.notify_rest_api({**event_data, "phase": "start"}) try: # Clone target repository target_dir = Path("target-repo") if target_dir.exists(): print("🗑️ Removing existing target directory") subprocess.run(["rm", "-rf", str(target_dir)], check=True) print(f"📦 Cloning {self.target_repo}...") clone_url = f"https://{self.sync_token}@github.com/{self.target_repo}.git" subprocess.run( ["git", "clone", clone_url, str(target_dir)], check=True, capture_output=True ) # Checkout sync branch subprocess.run( ["git", "checkout", self.sync_branch], cwd=target_dir, check=True, capture_output=True ) if direction in ["push", "bidirectional"]: print("📤 Copying files to target repository...") # Copy directories dirs_to_sync = [ ("mt5/MQL5", "mt5/MQL5"), ("scripts", "scripts"), ("config", "config"), ("docs", "docs") ] for src, dest in dirs_to_sync: src_path = Path(src) dest_path = target_dir / dest if src_path.exists(): dest_path.parent.mkdir(parents=True, exist_ok=True) subprocess.run( ["cp", "-r", str(src_path), str(dest_path.parent)], check=True ) print(f" ✅ Copied {src}") # Copy configuration files files_to_sync = ["README.md", "requirements.txt", ".env.example"] for file in files_to_sync: src_file = Path(file) if src_file.exists(): dest_file = target_dir / file subprocess.run( ["cp", str(src_file), str(dest_file)], check=True ) print(f" ✅ Copied {file}") # Check if there are changes to commit result = subprocess.run( ["git", "diff", "--staged", "--quiet"], cwd=target_dir, capture_output=True ) subprocess.run(["git", "add", "-A"], cwd=target_dir, check=True) result = subprocess.run( ["git", "diff", "--staged", "--quiet"], cwd=target_dir, capture_output=True ) if result.returncode != 0: # There are changes to commit commit_msg = f"""Auto-sync from {self.source_repo} Synced at: {git_info.get('timestamp', 'unknown')} Source commit: {git_info.get('commit', 'unknown')} Direction: {direction} """ subprocess.run( ["git", "commit", "-m", commit_msg], cwd=target_dir, check=True ) # Push changes push_url = f"https://{self.sync_token}@github.com/{self.target_repo}" subprocess.run( ["git", "push", push_url, self.sync_branch], cwd=target_dir, check=True ) print(f"✅ Successfully synced to {self.target_repo}") # Notify REST API after successful sync self.notify_rest_api({**event_data, "phase": "complete", "status": "success"}) return True else: print("✅ No changes to sync") return True except subprocess.CalledProcessError as e: print(f"❌ Sync failed: {e}") self.notify_rest_api({**event_data, "phase": "error", "status": "failed"}) return False finally: # Cleanup if target_dir.exists(): subprocess.run(["rm", "-rf", str(target_dir)], check=False) def run(self, direction: str = "push") -> int: """Run the sync process""" print("🚀 Repository Sync Manager") print("=" * 50) # Validate configuration valid, msg = self.validate_config() if not valid: print(f"❌ Configuration error: {msg}") return 1 if "Warning" in msg: print(f"⚠️ {msg}") # Run sync success = self.sync_directories(direction) print("=" * 50) if success: print("✅ Sync completed successfully") return 0 else: print("❌ Sync failed") return 1 def main(): """Main entry point""" import argparse parser = argparse.ArgumentParser( description="Repository Sync Manager with REST API Integration" ) parser.add_argument( "--direction", choices=["push", "pull", "bidirectional"], default="push", help="Sync direction (default: push)" ) parser.add_argument( "--check-config", action="store_true", help="Check configuration and exit" ) args = parser.parse_args() manager = RepoSyncManager() if args.check_config: valid, msg = manager.validate_config() print(msg) return 0 if valid else 1 return manager.run(args.direction) if __name__ == "__main__": sys.exit(main())