Skip to content

Improve the stdout with color #601

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
genedna merged 1 commit into from
Sep 15, 2024
Merged

Improve the stdout with color #601

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
102 changes: 59 additions & 43 deletions scripts/crates-sync/crates-sync.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,21 +1,38 @@
import os
import sys
import json
import urllib.request
import tarfile
import subprocess
import shutil
from collections import defaultdict
from packaging import version

# Create a directory if it doesn't exist
import os # For file and directory operations
import sys # For system-specific parameters and functions
import json # For JSON parsing
import urllib.request # For downloading files from URLs
import tarfile # For handling tar archives
import subprocess # For running system commands
import shutil # For high-level file operations
from collections import defaultdict # For creating dictionaries with default values

# ANSI color codes
GREEN = '\033[92m'
BLUE = '\033[94m'
RED = '\033[91m'
RESET = '\033[0m'

def print_green(text):
# Print text in green color
print(f"{GREEN}{text}{RESET}")

def print_blue(text):
# Print text in blue color
print(f"{BLUE}{text}{RESET}")

def print_red(text):
# Print text in red color
print(f"{RED}{text}{RESET}")

def ensure_directory(path):
# Create a directory if it doesn't exist
if not os.path.exists(path):
os.makedirs(path)
print(f"Created directory: {path}")
print_blue(f"Created directory: {path}")

# Construct the filename and path for the crate
def check_and_download_crate(crates_dir, crate_name, crate_version, dl_base_url):
# Construct the filename and path for the crate
crate_filename = f"{crate_name}-{crate_version}.crate"
crate_path = os.path.join(crates_dir, crate_name, crate_filename)

Expand All @@ -24,46 +41,46 @@ def check_and_download_crate(crates_dir, crate_name, crate_version, dl_base_url)
ensure_directory(os.path.dirname(crate_path)) # Ensure the directory exists
download_url = f"{dl_base_url}/{crate_name}/{crate_filename}"
try:
print(f"Downloading: {download_url}")
print_green(f"Downloading: {download_url}")
urllib.request.urlretrieve(download_url, crate_path) # Download the file
print(f"Downloaded: {crate_path}")
print_red(f"Downloaded: {crate_path}")
except Exception as e:
print(f"Error downloading {crate_filename}: {str(e)}")
print_red(f"Error downloading {crate_filename}: {str(e)}")
return crate_path

# Run a git command in the specified repository
def run_git_command(repo_path, command):
# Run a git command in the specified repository
try:
result = subprocess.run(command, cwd=repo_path, check=True, capture_output=True, text=True)
return result.stdout.strip()
except subprocess.CalledProcessError as e:
print(f"Warning: Git command failed: {e}")
print(f"Command output: {e.output}")
print_red(f"Warning: Git command failed: {e}")
print_red(f"Command output: {e.output}")
return None

# Initialize a git repository if it doesn't exist
def init_git_repo(repo_path):
# Initialize a git repository if it doesn't exist
if not os.path.exists(os.path.join(repo_path, '.git')):
run_git_command(repo_path, ['git', 'init', '-b', 'main'])
print(f"Initialized git repository in {repo_path}")

def extract_crate(crate_path, extract_path):
# Check if a path is within a directory (for security)
def is_within_directory(directory, target):
# Check if a path is within a directory (for security)
abs_directory = os.path.abspath(directory)
abs_target = os.path.abspath(target)
prefix = os.path.commonprefix([abs_directory, abs_target])
return prefix == abs_directory

# Safely extract files from a tar archive
def safe_extract(tar, path=".", members=None, *, numeric_owner=False):
# Safely extract files from a tar archive
for member in tar.getmembers():
member_path = os.path.join(path, member.name)
if not is_within_directory(path, member_path):
raise Exception("Attempted Path Traversal in Tar File")

# Filter function to ensure extracted files are within the target directory
def filter_member(tarinfo, filterpath):
# Filter function to ensure extracted files are within the target directory
if is_within_directory(path, os.path.join(filterpath, tarinfo.name)):
return tarinfo
else:
Expand All @@ -74,7 +91,7 @@ def filter_member(tarinfo, filterpath):
try:
with tarfile.open(crate_path, 'r:gz') as tar:
if not tar.getmembers():
print(f"Warning: Empty crate file {crate_path}. Skipping extraction.")
print_red(f"Warning: Empty crate file {crate_path}. Skipping extraction.")
return False

# Create a temporary directory for extraction
Expand All @@ -95,17 +112,17 @@ def filter_member(tarinfo, filterpath):
print(f"Extracted version to {extract_path}")
return True
except tarfile.ReadError:
print(f"Warning: Failed to read crate file {crate_path}. Skipping extraction.")
print_red(f"Warning: Failed to read crate file {crate_path}. Skipping extraction.")
return False

# Process a specific version of a crate
def process_crate_version(crate_name, version, crate_path, git_repos_dir, git_base_url):
# Process a specific version of a crate
repo_path = os.path.join(git_repos_dir, crate_name, version)
ensure_directory(repo_path)

# Extract crate directly to the repo directory
if not extract_crate(crate_path, repo_path):
print(f"Skipping processing for {crate_name} version {version} due to extraction failure.")
print_red(f"Skipping processing for {crate_name} version {version} due to extraction failure.")
return

# Initialize git repo
Expand All @@ -114,7 +131,7 @@ def process_crate_version(crate_name, version, crate_path, git_repos_dir, git_ba
# Add all files to git
run_git_command(repo_path, ['git', 'add', '.'])

# Commit changes with updated message format
# Commit changes with updated message format and additional parameters
commit_message = f"{crate_name} {version}"
run_git_command(repo_path, ['git', 'commit', '-a', '-s', '-S', '-m', commit_message])

Expand All @@ -125,32 +142,32 @@ def process_crate_version(crate_name, version, crate_path, git_repos_dir, git_ba
# Push to remote
push_result = run_git_command(repo_path, ['git', 'push', '-u', 'mega', 'main'])
if push_result is None:
print(f"Warning: Failed to push {crate_name} version {version} to remote repository.")
print_red(f"Warning: Failed to push {crate_name} version {version} to remote repository.")
else:
print(f"Successfully pushed {crate_name} version {version} to remote repository.")
print_green(f"Successfully pushed {crate_name} version {version} to remote repository.")

# Process all versions of a crate
def process_crate(crate_name, versions, crates_dir, git_repos_dir, dl_base_url, git_base_url):
# Process all versions of a crate
for v in versions:
repo_path = os.path.join(git_repos_dir, crate_name, v)
if os.path.exists(repo_path):
print(f"Repository for {crate_name} version {v} already exists. Skipping.")
print_red(f"Repository for {crate_name} version {v} already exists. Skipping.")
continue

crate_path = check_and_download_crate(crates_dir, crate_name, v, dl_base_url)
process_crate_version(crate_name, v, crate_path, git_repos_dir, git_base_url)

print(f"Finished processing {crate_name}")
print_blue(f"Finished processing {crate_name}")

# Scan the crates.io index and process all crates
def scan_and_process_crates(index_path, crates_dir, git_repos_dir, git_base_url):
# Scan the crates.io index and process all crates
crates = defaultdict(set)
dl_base_url = None

# Check if the directories exist
for path in [index_path, crates_dir, git_repos_dir]:
if not os.path.isdir(path):
print(f"Error: The directory {path} does not exist.")
print_red(f"Error: The directory {path} does not exist.")
sys.exit(1)

# Read the config.json to get the dl base URL
Expand All @@ -160,10 +177,10 @@ def scan_and_process_crates(index_path, crates_dir, git_repos_dir, git_base_url)
config = json.load(config_file)
dl_base_url = config.get('dl')
if not dl_base_url:
print("Error: 'dl' key not found in config.json")
print_red("Error: 'dl' key not found in config.json")
sys.exit(1)
except Exception as e:
print(f"Error reading config.json: {str(e)}")
print_red(f"Error reading config.json: {str(e)}")
sys.exit(1)

# Walk through the index directory
Expand All @@ -186,25 +203,24 @@ def scan_and_process_crates(index_path, crates_dir, git_repos_dir, git_base_url)
crate_version = crate_info['vers']
crates[crate_name].add(crate_version)
except Exception as e:
print(f"Error processing file {full_path}: {str(e)}")
print_red(f"Error processing file {full_path}: {str(e)}")

# Process each crate
for crate_name, versions in crates.items():
process_crate(crate_name, versions, crates_dir, git_repos_dir, dl_base_url, git_base_url)

return len(crates)

# Main function to run the script
def main():
# Main function to run the script
if len(sys.argv) != 5:
print("Usage: python script.py <path_to_crates.io-index> <path_to_crates_directory> <path_to_git_repos_directory> <git_base_url>")
print_red("Usage: python script.py <path_to_crates.io-index> <path_to_crates_directory> <path_to_git_repos_directory> <git_base_url>")
sys.exit(1)

index_path, crates_dir, git_repos_dir, git_base_url = sys.argv[1:5]

total_crates = scan_and_process_crates(index_path, crates_dir, git_repos_dir, git_base_url)
print(f"\nTotal number of crates processed: {total_crates}")
print_blue(f"\nTotal number of crates processed: {total_crates}")

# Run the main function if this script is executed directly
if __name__ == "__main__":
main()
main() # Run the main function if this script is executed directly