forked from LengKundee/MQL5-Google-Onedrive
283 lines
9.6 KiB
Python
Executable file
283 lines
9.6 KiB
Python
Executable file
#!/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())
|