shared_utils.py

import subprocess
import sys
from pathlib import Path

def format_size(size_bytes: int) -> str:
    """Converts a size in bytes to a human-readable string (KB, MB, GB)."""
    if size_bytes == 0:
        return "0 B"
    power = 1024
    n = 0
    power_labels = {0: 'B', 1: 'KB', 2: 'MB', 3: 'GB', 4: 'TB'}
    while size_bytes >= power and n < len(power_labels) - 1:
        size_bytes /= power
        n += 1
    return f"{size_bytes:.2f} {power_labels[n]}" if n > 0 else f"{int(size_bytes)} {power_labels[n]}"

def run_command(command, cwd=None):
    """Executes a shell command safely and returns its output."""
    try:
        result = subprocess.run(
            command,
            shell=True,
            check=True,
            capture_output=True,
            text=True,
            cwd=cwd
        )
        return result.stdout.strip()
    except subprocess.CalledProcessError as e:
        print(f"Error executing command: {e.cmd}", file=sys.stderr)
        err_msg = f"{e.stderr.strip()}"
        print(f"{err_msg}", file=sys.stderr)
        return err_msg
    except FileNotFoundError:
        print("Error: Command not found.", file=sys.stderr)
        sys.exit(1)

def get_git_state(target_dir="."):
    """Safely attempts to get the current Git repository state for a given directory."""
    try:
        target_path = str(target_dir)
        # Check if we are inside a git repository
        run_command('git rev-parse --is-inside-work-tree', cwd=target_path)
        
        # Get short commit ID
        short_id = run_command('git rev-parse --short HEAD', cwd=target_path)
        
        # Get the commit message
        commit_msg = run_command('git log -1 --pretty=%s', cwd=target_path)
        
        # Check for untracked or uncommitted changes
        status = run_command('git status --porcelain', cwd=target_path)
        is_dirty = len(status) > 0
        
        return short_id, commit_msg, is_dirty
    except Exception:
        # Fails silently if Git isn't installed or we aren't in a repo
        return None, None, None

def summary(command_args, files_with_sizes, commit_id=None, target_dir="."):
    """
    Generates a formatted summary string of the execution.
    files_with_sizes: list of tuples (file_path_string, size_in_bytes)
    """
    total_size_bytes = sum(size for _, size in files_with_sizes)
    sorted_files = sorted(files_with_sizes, key=lambda x: x[1], reverse=True)
    
    lines = [
        "=== EXECUTION SUMMARY ===",
        f"Command Arguments   : {' '.join(command_args)}"
    ]
    
    # Add target commit parameter (mostly used by save_commits.py)
    if commit_id:
        lines.append(f"Target Commit(s)    : {commit_id}")
        
    # Dynamically inject current workspace Git context using the target directory
    short_id, commit_msg, is_dirty = get_git_state(target_dir)
    if short_id:
        lines.append(f"Current Repo HEAD   : {short_id} {commit_msg}")
        lines.append(f"Uncommitted Changes : {'Yes' if is_dirty else 'No'}")
        
    lines.append(f"Total Size          : {format_size(total_size_bytes)}")
    lines.append(f"Total Files         : {len(sorted_files)}")
    lines.append("Files Included      :")
    
    if not sorted_files:
        lines.append("  (No files)")
    else:
        for f_path, f_size in sorted_files:
            lines.append(f"  - [{format_size(f_size).rjust(10)}] {f_path}")
            
    lines.append("=" * 80 + "\n")
    return "\n".join(lines)

def write_concatenated_artifact(output_file, file_items, summary_text):
    """
    Standardized writer for concatenated scripts.
    file_items should be a list of tuples: (display_path, absolute_file_path)
    """
    print(f"\n📝 Writing files to '{output_file}'...")
    try:
        with open(output_file, 'w', encoding='utf-8', errors='ignore') as outfile:
            outfile.write(f"--- START OF FILE {output_file} ---\n\n")
            outfile.write("# This file was generated by concatenating specified files and folders.\n")
            outfile.write(f"# Total files included: {len(file_items)}\n\n")
            
            for display_path, file_path in file_items:
                header = "=" * 80 + "\n"
                header += f"cat {display_path}\n"
                header += "=" * 80 + "\n\n"
                outfile.write(header)

                try:
                    with open(file_path, 'r', encoding='utf-8') as infile:
                        outfile.write(infile.read())
                except Exception as e:
                    error_message = f"--> [Error] Could not read file '{display_path}': {e}\n"
                    print(error_message, end='')
                    outfile.write(error_message)
                
                outfile.write("\n\n")
            
            outfile.write(summary_text)
        print(f"✅ Success! Content concatenated into '{output_file}'.")
    except IOError as e:
        print(f"❌ Error: Could not write to output file '{output_file}': {e}")
        sys.exit(1)