From e0a43d9e66d27b79edfe8659ac6b90e9c14c29dd Mon Sep 17 00:00:00 2001 From: oxmc7769 Date: Wed, 5 Nov 2025 00:45:21 +0000 Subject: [PATCH] Base, more work to do --- .gitignore | 2 + README.md | 129 ++++++++- environ.py | 392 +++++++++++++++++++++++++ repo.yml | 10 + rpivanilla-ensetup.sh | 13 + scripts/better_patch.py | 613 ++++++++++++++++++++++++++++++++++++++++ 6 files changed, 1158 insertions(+), 1 deletion(-) create mode 100644 environ.py create mode 100644 repo.yml create mode 100644 rpivanilla-ensetup.sh create mode 100644 scripts/better_patch.py diff --git a/.gitignore b/.gitignore index 36b13f1..c31e620 100644 --- a/.gitignore +++ b/.gitignore @@ -174,3 +174,5 @@ cython_debug/ # PyPI configuration file .pypirc +# Android source code +AOSP-source/ \ No newline at end of file diff --git a/README.md b/README.md index 0464d52..36075cc 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,130 @@ # PawletOS-Build -A repo that has scripts and helpers to build PawletOS \ No newline at end of file +A comprehensive build system for PawletOS with advanced patch management and Android source synchronization capabilities. + +## Overview + +PawletOS-Build provides automated tools for setting up and managing the PawletOS build environment, featuring: + +- **Intelligent Patch Management**: Advanced patch application with conflict detection and resolution +- **Source Synchronization**: Automated Android source sync using repo +- **Build Environment Setup**: Streamlined environment configuration for Raspberry Pi builds +- **Robust Error Handling**: Graceful fallback mechanisms and detailed error reporting + +## Features + +### 🔧 Build System (`build.py`) +The main build script provides comprehensive Android build environment setup with the following capabilities: + +- **Source Code Synchronization**: Sync Android source code using `repo sync` with configurable parallel jobs +- **Advanced Patch Application**: Intelligent patch system with two modes: + - **Robust Handler**: Git-based patch application with conflict analysis and dependency detection + - **Legacy Handler**: Traditional patch command fallback for compatibility +- **Flexible Configuration**: Customizable source directories, patch levels, and build options +- **Batch Processing**: Apply multiple patches with selective filtering and ignore lists + +### 🛠️ Patch Management (`scripts/better-patch.py`) +Advanced patch handling system featuring: + +- **Conflict Analysis**: Pre-application analysis of potential patch conflicts +- **Dependency Detection**: Automatic detection of patch dependencies and ordering +- **Rollback Capability**: Automatic rollback on failure with cleanup +- **Git Integration**: Leverages Git's powerful merge capabilities for better conflict resolution +- **Detailed Reporting**: Comprehensive patch application reports and statistics + +### 🍓 Raspberry Pi Setup (`rpivanilla-ensetup.sh`) +Environment setup script for Raspberry Pi vanilla builds: + +- Sets up the RaspberryVanilla build environment +- Sources necessary build configuration files +- Provides template for build-specific exports + +## Usage + +### Quick Start + +1. **Basic patch application**: + ```bash + python3 build.py --applyPatches + ``` + +2. **Sync source and apply patches**: + ```bash + python3 build.py --resync --applyPatches + ``` + +3. **Force operations with verbose output**: + ```bash + python3 build.py --forceResync --forceApplyPatches --verbosePatches + ``` + +### Command Line Options + +| Option | Description | +|--------|-------------| +| `--resync` | Sync source code with repo sync | +| `--forceResync` | Force sync source code with repo sync --force-sync | +| `--applyPatches` | Apply patches from patches directory | +| `--forceApplyPatches` | Force apply patches even if some fail | +| `--patchLevel ` | Patch level for -p option (default: 1) | +| `--sourceDir ` | Source directory (default: current directory) | +| `--patchDir ` | Patch directory (default: patches) | +| `--syncJobs ` | Number of parallel jobs for repo sync | +| `--ignorePatches ` | List of patches to ignore | +| `--useLegacyPatch` | Use legacy patch command instead of robust handler | +| `--verbosePatches` | Verbose output for patch application | + +### Advanced Patch Management + +The build system includes an intelligent patch handler that provides: + +- **Pre-flight Analysis**: Analyzes patches before application to detect conflicts +- **Smart Conflict Resolution**: Uses Git's three-way merge for better conflict handling +- **Dependency Management**: Automatically orders patches based on file dependencies +- **Rollback Protection**: Automatic rollback on failure with repository cleanup + +### Environment Setup + +For Raspberry Pi builds, use the provided setup script: + +```bash +chmod +x rpivanilla-ensetup.sh +./rpivanilla-ensetup.sh +``` + +This script will: +- Navigate to the RaspberryVanilla directory +- Source the build environment setup +- Prepare the environment for lunch and build commands + +## Directory Structure + +``` +PawletOS-Build/ +├── build.py # Main build system script +├── rpivanilla-ensetup.sh # Raspberry Pi environment setup +├── scripts/ +│ └── better-patch.py # Advanced patch management system +├── patches/ # Patch files directory (created as needed) +└── README.md # This file +``` + +## Requirements + +- Python 3.6+ +- Git (recommended for advanced patch features) +- Android build tools (repo, patch) +- Bash shell for setup scripts + +## Contributing + +When adding new patches or modifying the build system: + +1. Test patches individually before batch application +2. Use `--verbosePatches` for debugging patch issues +3. Consider patch dependencies when organizing patch files +4. Update documentation for new features or options + +## License + +See the LICENSE file for licensing information. \ No newline at end of file diff --git a/environ.py b/environ.py new file mode 100644 index 0000000..34b06eb --- /dev/null +++ b/environ.py @@ -0,0 +1,392 @@ +import os +import sys +import subprocess +import argparse +from pathlib import Path +import yaml +import requests + +# Import the patch handler +from scripts.better_patch import GitPatchHandler + +# Helper functions +def run_command(command, cwd=None, shell=False): + """Run a shell command in a subprocess.""" + print(f"Running command: {' '.join(command) if isinstance(command, list) else command}") + result = subprocess.run(command, cwd=cwd, capture_output=True, text=True, shell=shell) + if result.returncode != 0: + print(f"Error running command: {' '.join(command) if isinstance(command, list) else command}") + print(result.stderr) + raise subprocess.CalledProcessError(result.returncode, command) + print(result.stdout) + return result + +def run_command_realtime(command, cwd=None, shell=False): + """Run a shell command in a subprocess with real-time output.""" + print(f"Running command: {' '.join(command) if isinstance(command, list) else command}") + + # For repo commands, use direct terminal output to preserve progress bars + if isinstance(command, list) and "repo" in command[0]: + # Direct passthrough to terminal for repo commands + result = subprocess.run(command, cwd=cwd, shell=shell) + if result.returncode != 0: + raise subprocess.CalledProcessError(result.returncode, command) + return result.returncode + else: + # For other commands, use the buffered approach + process = subprocess.Popen( + command, + cwd=cwd, + shell=shell, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + bufsize=1, + universal_newlines=True + ) + + # Print output in real-time + for line in process.stdout: + print(line, end='', flush=True) + + # Wait for process to complete and get return code + return_code = process.wait() + + if return_code != 0: + raise subprocess.CalledProcessError(return_code, command) + + return return_code + +def load_repo_config(config_file="repo.yml"): + """Load repository configuration from YAML file.""" + if not os.path.exists(config_file): + print(f"Warning: Repo config file {config_file} not found") + return None + + with open(config_file, 'r') as f: + config = yaml.safe_load(f) + + return config + +def repo_init(source_dir, config): + """Initialize repo with configuration.""" + if not config or 'repo' not in config: + print("No repo configuration found") + sys.exit(1) + + repo_config = config['repo'] + + # Build repo init command + init_cmd = ["repo", "init"] + + # Add repo URL + if 'url' in repo_config: + init_cmd.extend(["-u", repo_config['url']]) + else: + print("Error: No repo URL specified in config") + return + + # Add branch if specified + if 'branch' in repo_config: + init_cmd.extend(["-b", repo_config['branch']]) + + # Add other options + if 'options' in repo_config: + for option in repo_config['options']: + init_cmd.append(option) + + print("Initializing repo...") + run_command_realtime(init_cmd, cwd=source_dir) + + # Download local manifests if specified + if 'local_manifests' in repo_config: + download_local_manifests(source_dir, repo_config['local_manifests']) + +def download_local_manifests(source_dir, manifests): + """Download local manifest files using requests.""" + manifests_dir = os.path.join(source_dir, ".repo", "local_manifests") + os.makedirs(manifests_dir, exist_ok=True) + + for manifest_name, manifest_url in manifests.items(): + manifest_path = os.path.join(manifests_dir, f"{manifest_name}.xml") + print(f"Downloading local manifest: {manifest_name} from {manifest_url}") + + try: + response = requests.get(manifest_url, timeout=30) + response.raise_for_status() + + with open(manifest_path, 'w') as f: + f.write(response.text) + print(f"Successfully downloaded {manifest_name}.xml") + + except requests.exceptions.RequestException as e: + print(f"Failed to download {manifest_name}.xml: {e}") + +def find_patches(patch_dirs): + """Find all patch files in the patch directories.""" + patches = [] + for patch_dir in patch_dirs: + if not os.path.exists(patch_dir): + print(f"Warning: Patch directory does not exist: {patch_dir}") + continue + patch_files = list(Path(patch_dir).glob("*.patch")) + print(f"Found {len(patch_files)} patches in {patch_dir}") + patches.extend(patch_files) + + # Remove duplicates (in case same patch appears in multiple directories) + unique_patches = [] + seen_names = set() + for patch in patches: + if patch.name not in seen_names: + unique_patches.append(patch) + seen_names.add(patch.name) + + return unique_patches + +def apply_patch_legacy(patch_file, source_dir, patch_level=1, force=False): + """Apply a single patch file using legacy patch command.""" + print(f"Applying patch: {patch_file} to {source_dir} with -p{patch_level}") + try: + run_command(["patch", f"-p{patch_level}", "-i", str(patch_file)], cwd=source_dir) + except subprocess.CalledProcessError as e: + if force: + print(f"Patch failed but forcing continue: {patch_file}") + # Try to reverse any partially applied patch + try: + run_command(["patch", f"-p{patch_level}", "-R", "-i", str(patch_file)], cwd=source_dir) + except: + print(f"Could not reverse failed patch: {patch_file}") + else: + raise + +def apply_patches_robust(patches, source_dir, patch_level=1, force=False, ignore_patches=None, verbose=False): + """Apply patches using the robust patch handler.""" + if ignore_patches is None: + ignore_patches = [] + + # Filter out ignored patches + patches_to_apply = [p for p in patches if p.name not in ignore_patches] + + if not patches_to_apply: + print("No patches to apply after filtering") + return True, [] + + print(f"Using robust patch handler for {len(patches_to_apply)} patches") + + # Initialize the patch handler + handler = GitPatchHandler(repo_path=source_dir, verbose=verbose) + + # Analyze patches first + analysis = handler.analyze_patch_series([str(p) for p in patches_to_apply]) + + print("Patch analysis completed:") + print(f" - Total patches: {analysis['summary']['total_patches']}") + print(f" - Files modified: {analysis['summary']['total_files_modified']}") + print(f" - Potential conflicts: {len(analysis['summary']['file_conflicts_detected'])}") + + # Apply patches + result = handler.apply_patch_series( + [str(p) for p in patches_to_apply], + target_dir=".", # Already in source_dir + stop_on_conflict=not force, + rollback_on_failure=True + ) + + # Print results + print("\nPatch application results:") + print(f" - Overall success: {result['success']}") + print(f" - Patches applied: {len(result['patches_applied'])}/{len(patches_to_apply)}") + + # Print details for failed patches + failed_patches = [] + for patch_result in result['results']: + patch_name = Path(patch_result['patch']).name + if not patch_result['success']: + failed_patches.append(patch_name) + print(f" - FAILED: {patch_name} - {patch_result['message']}") + if patch_result.get('conflicts'): + print(f" Conflicts: {', '.join(patch_result['conflicts'])}") + + return result['success'], failed_patches + +def apply_patches_legacy(patches, source_dir, patch_level=1, force=False, ignore_patches=None): + """Legacy patch application for fallback.""" + if ignore_patches is None: + ignore_patches = [] + + applied_patches = 0 + failed_patches = [] + + for patch_file in patches: + patch_name = patch_file.name + if patch_name in ignore_patches: + print(f"Ignoring patch: {patch_name}") + continue + + try: + apply_patch_legacy(patch_file, source_dir, patch_level, force) + applied_patches += 1 + except Exception as e: + print(f"Failed to apply patch {patch_name}: {e}") + failed_patches.append(patch_name) + if not force: + print("Stopping due to patch failure. Use --forceApplyPatches to continue.") + return False, failed_patches + + print(f"Successfully applied {applied_patches} patches") + return True, failed_patches + +def repo_sync(force=False, jobs=None, current_dir="."): + """Sync Android source using repo with real-time output.""" + print("Syncing Android source...") + + sync_cmd = ["repo", "sync"] + + if force: + sync_cmd.extend(["--force-sync"]) + + if jobs: + sync_cmd.extend([f"-j{jobs}"]) + + # Use real-time output for repo sync + run_command_realtime(sync_cmd, cwd=current_dir) + +def parse_arguments(): + """Parse command line arguments.""" + parser = argparse.ArgumentParser(description="Android build environment setup script") + parser.add_argument("--noSync", action="store_true", + help="Skip repo sync (sync is enabled by default)") + parser.add_argument("--forceResync", action="store_true", + help="Force sync source code with repo sync --force-sync") + parser.add_argument("--applyPatches", action="store_true", + help="Apply patches from patches directory") + parser.add_argument("--forceApplyPatches", action="store_true", + help="Force apply patches even if some fail") + parser.add_argument("--patchLevel", type=int, default=1, + help="Patch level for -p option (default: 1)") + parser.add_argument("--sourceDir", type=str, default="AOSP-source", + help="Source directory (default: AOSP-source)") + parser.add_argument("--patchDir", type=str, default=["patches"], nargs="+", + help="Patch directory (default: patches). Can specify multiple directories.") + parser.add_argument("--syncJobs", type=int, default=None, + help="Number of parallel jobs for repo sync (default: CPU count)") + parser.add_argument("--ignorePatches", type=str, nargs="*", default=[], + help="List of patches to ignore") + parser.add_argument("--useLegacyPatch", action="store_true", + help="Use legacy patch command instead of robust handler") + parser.add_argument("--verbosePatches", action="store_true", + help="Verbose output for patch application") + parser.add_argument("--config", type=str, default="repo.yml", + help="Path to repo configuration file (default: repo.yml)") + + return parser.parse_args() + +def main(): + # Parse command line arguments + args = parse_arguments() + + # Use command line arguments with defaults + source_dir = args.sourceDir + patch_dirs = args.patchDir + patch_level = args.patchLevel + sync_jobs = args.syncJobs if args.syncJobs is not None else os.cpu_count() + ignore_patches = args.ignorePatches + + print("Android Build Environment Setup") + print(f"Source directory: {source_dir}") + print(f"Patch directories: {', '.join(patch_dirs)}") + print(f"Patch level: -p{patch_level}") + print(f"Sync jobs: {sync_jobs}") + if ignore_patches: + print(f"Ignoring patches: {', '.join(ignore_patches)}") + + # Get absolute path to prevent any relative path confusion + current_dir = os.path.abspath(".") + absolute_source_dir = os.path.abspath(source_dir) + + print(f"Current directory: {current_dir}") + print(f"Absolute source directory: {absolute_source_dir}") + + # Check if we're trying to create a nested directory + if absolute_source_dir.startswith(current_dir) and absolute_source_dir != current_dir: + # We're creating a subdirectory, which is fine + working_dir = absolute_source_dir + if not os.path.exists(working_dir): + os.makedirs(working_dir) + print(f"Created source directory: {working_dir}") + else: + print(f"Using existing source directory: {working_dir}") + else: + # Use the specified directory directly + working_dir = absolute_source_dir + if not os.path.exists(working_dir): + os.makedirs(working_dir) + print(f"Created source directory: {working_dir}") + else: + print(f"Using existing source directory: {working_dir}") + + # Load repo configuration + config = load_repo_config(args.config) + + # Initialize repo if config exists and .repo directory doesn't exist + repo_dir = os.path.join(working_dir, ".repo") + if config and not os.path.exists(repo_dir): + repo_init(working_dir, config) + elif config and os.path.exists(repo_dir): + print("Repo already initialized, skipping init") + elif not config: + print("No repo configuration found, skipping repo init") + + # Handle repo sync - always sync unless --noSync is specified + if not args.noSync: + repo_sync(force=args.forceResync, jobs=sync_jobs, current_dir=working_dir) + else: + print("Skipping repo sync as requested") + + # Apply patches + if args.applyPatches or args.forceApplyPatches: + # Handle relative paths - if working_dir is not ".", prepend it to patch directories + processed_patch_dirs = [] + for patch_dir in patch_dirs: + if working_dir != "." and not os.path.isabs(patch_dir): + processed_patch_dirs.append(os.path.join(working_dir, patch_dir)) + else: + processed_patch_dirs.append(patch_dir) + + patches = find_patches(processed_patch_dirs) + + if patches: + if not args.useLegacyPatch: + # Use robust patch handler + success, failed_patches = apply_patches_robust( + patches, + working_dir, + patch_level, + args.forceApplyPatches, + ignore_patches, + args.verbosePatches + ) + else: + # Use legacy patch command + success, failed_patches = apply_patches_legacy( + patches, + working_dir, + patch_level, + args.forceApplyPatches, + ignore_patches + ) + + if not success and not args.forceApplyPatches: + print(f"Patch application failed. Failed patches: {', '.join(failed_patches)}") + sys.exit(1) + elif failed_patches: + print(f"Some patches failed but continuing (force mode): {', '.join(failed_patches)}") + else: + print("All patches applied successfully!") + else: + print(f"No patches found in {', '.join(processed_patch_dirs)}") + + print("Environment setup completed!") + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/repo.yml b/repo.yml new file mode 100644 index 0000000..85c2895 --- /dev/null +++ b/repo.yml @@ -0,0 +1,10 @@ +repo: + url: "https://android.googlesource.com/platform/manifest" + branch: "android-16.0.0_r1" + options: + - "--depth=1" + local_manifests: + manifest_brcm_rpi: "https://git.oxmc.me/PawletOS/android_local_manifest/raw/branch/android-16.0/vendors/brcm_rpi.xml" + manifest_lineageos: "https://git.oxmc.me/PawletOS/android_local_manifest/raw/branch/android-16.0/vendors/lineageos.xml" + manifest_pawletos: "https://git.oxmc.me/PawletOS/android_local_manifest/raw/branch/android-16.0/pawletos.xml" + remove_projects: "https://git.oxmc.me/PawletOS/android_local_manifest/raw/branch/android-16.0/remove_projects.xml" \ No newline at end of file diff --git a/rpivanilla-ensetup.sh b/rpivanilla-ensetup.sh new file mode 100644 index 0000000..f52a3bc --- /dev/null +++ b/rpivanilla-ensetup.sh @@ -0,0 +1,13 @@ +#!/bin/bash + +# Change dir +cd ~/RaspberryVanilla + +# sourch files +source ~/RaspberryVanilla/build/envsetup.sh + +# export +#export SOONG_ALLOW_MISSING_DEPENDENCIES=true # Used for multi disk build +#export OUT_DIR=/mnt/256gb-ssd/aosp_build # Used for multi disk build + +echo "Env setup, lunch than build" diff --git a/scripts/better_patch.py b/scripts/better_patch.py new file mode 100644 index 0000000..432a400 --- /dev/null +++ b/scripts/better_patch.py @@ -0,0 +1,613 @@ +#!/usr/bin/env python3 +""" +Robust Patch Handler +A Python module for applying patch files with intelligent conflict handling. +Can be used as both a CLI tool and an importable module. +""" + +import subprocess +import tempfile +import os +import sys +import re +import shutil +from pathlib import Path +from typing import List, Dict, Optional, Any +import argparse +import json + +class GitPatchHandler: + """ + A robust patch handler that uses Git for patch application with fallbacks + and intelligent conflict handling. + """ + + def __init__(self, repo_path: str = ".", verbose: bool = False): + self.repo_path = Path(repo_path).absolute() + self.verbose = verbose + self.use_git = self._check_git_available() + self._log(f"Initialized patch handler in {self.repo_path}") + self._log(f"Git available: {self.use_git}") + + def _log(self, message: str) -> None: + """Log messages if verbose mode is enabled""" + if self.verbose: + print(f"[PatchHandler] {message}", file=sys.stderr) + + def _check_git_available(self) -> bool: + """Check if git is available in the system""" + try: + subprocess.run(['git', '--version'], check=True, capture_output=True, timeout=5) + return True + except (subprocess.CalledProcessError, FileNotFoundError, subprocess.TimeoutExpired): + return False + + def apply_patch_series(self, patch_files: List[str], target_dir: str = ".", + stop_on_conflict: bool = False, rollback_on_failure: bool = False) -> Dict[str, Any]: + """ + Apply multiple patches with conflict handling and dependency detection. + + Args: + patch_files: List of paths to patch files + target_dir: Directory to apply patches to + stop_on_conflict: Stop applying patches if a conflict occurs + rollback_on_failure: Rollback all patches if any fail + + Returns: + Dictionary with application results + """ + target_path = self.repo_path / target_dir + self._log(f"Applying {len(patch_files)} patches to {target_path}") + + # Analyze patches first + analysis = self.analyze_patch_series(patch_files) + ordered_patches = self._order_patches_by_dependency(analysis) + + results = [] + all_success = True + applied_patches = [] + + for patch_file in ordered_patches: + self._log(f"Applying patch: {patch_file}") + + if rollback_on_failure: + result = self.apply_with_rollback(patch_file, str(target_path)) + else: + result = self.apply_single_patch(patch_file, str(target_path)) + + result_data = { + 'patch': patch_file, + 'success': result['success'], + 'conflicts': result.get('conflicts', []), + 'message': result.get('message', ''), + 'rolled_back': result.get('rolled_back', False) + } + results.append(result_data) + + if result['success']: + applied_patches.append(patch_file) + else: + all_success = False + if stop_on_conflict: + self._log(f"Stopping on conflict in {patch_file}") + break + + return { + 'success': all_success, + 'patches_applied': applied_patches, + 'results': results, + 'analysis': analysis + } + + def apply_single_patch(self, patch_file: str, target_dir: str = ".") -> Dict[str, Any]: + """ + Apply a single patch file using git with fallbacks. + + Args: + patch_file: Path to the patch file + target_dir: Directory to apply the patch to + + Returns: + Dictionary with application results + """ + patch_path = Path(patch_file).absolute() + target_path = self.repo_path / target_dir + + self._log(f"Applying single patch: {patch_path} -> {target_path}") + + if not patch_path.exists(): + return { + 'success': False, + 'fatal': True, + 'message': f'Patch file not found: {patch_file}' + } + + if not self.use_git: + return self._fallback_patch_apply(str(patch_path), str(target_path)) + + try: + # Use git apply with 3-way merge for better conflict resolution + cmd = [ + 'git', 'apply', + '--3way', # Allow 3-way merge if base available + '--allow-overlap', # Allow overlapping patches + '--verbose', # Get detailed output + str(patch_path) + ] + + self._log(f"Running: {' '.join(cmd)}") + result = subprocess.run( + cmd, + cwd=str(target_path), + capture_output=True, + text=True, + timeout=30 # Safety timeout + ) + + if result.returncode == 0: + return { + 'success': True, + 'message': 'Patch applied cleanly', + 'output': result.stdout, + 'method': 'git' + } + else: + return self._handle_git_failure(result, str(patch_path), str(target_path)) + + except subprocess.TimeoutExpired: + return { + 'success': False, + 'fatal': True, + 'message': 'Patch application timed out' + } + except Exception as e: + return { + 'success': False, + 'fatal': True, + 'message': f'Unexpected error: {str(e)}' + } + + def _handle_git_failure(self, result: subprocess.CompletedProcess, patch_file: str, target_dir: str) -> Dict[str, Any]: + """Handle git apply failures intelligently""" + stderr = result.stderr.lower() + stdout = result.stdout + + # Analyze the error type + if 'conflict' in stderr: + conflicts = self._extract_conflicts(stderr, stdout) + return { + 'success': False, + 'message': 'Patch conflicts detected', + 'conflicts': conflicts, + 'output': result.stdout, + 'error': result.stderr, + 'method': 'git' + } + elif 'already exists' in stderr or 'already applied' in stderr: + return { + 'success': True, # Treat as success since change is already present + 'message': 'Patch already applied or redundant', + 'output': result.stdout, + 'method': 'git' + } + elif 'patch does not apply' in stderr: + # Try with fuzz factor + return self._try_fuzzy_apply(patch_file, target_dir, result) + else: + return { + 'success': False, + 'fatal': True, + 'message': f'Patch failed: {result.stderr}', + 'output': result.stdout, + 'error': result.stderr, + 'method': 'git' + } + + def _try_fuzzy_apply(self, patch_file: str, target_dir: str, previous_result: subprocess.CompletedProcess) -> Dict[str, Any]: + """Try applying patch with fuzzy matching""" + self._log("Attempting fuzzy patch application") + try: + cmd = [ + 'git', 'apply', + '--3way', + '--allow-overlap', + '--verbose', + '--reject', # Apply parts that work, reject others + str(patch_file) + ] + + result = subprocess.run( + cmd, + cwd=target_dir, + capture_output=True, + text=True, + timeout=30 + ) + + if result.returncode == 0 or 'applied' in result.stdout.lower(): + return { + 'success': True, + 'message': 'Patch applied with rejects', + 'output': result.stdout, + 'error': result.stderr, + 'method': 'git-fuzzy', + 'had_rejects': 'reject' in result.stdout.lower() + } + else: + return { + 'success': False, + 'fatal': True, + 'message': 'Fuzzy application also failed', + 'output': result.stdout, + 'error': result.stderr, + 'method': 'git-fuzzy' + } + + except Exception as e: + return { + 'success': False, + 'fatal': True, + 'message': f'Fuzzy application error: {str(e)}' + } + + def _extract_conflicts(self, stderr: str, stdout: str) -> List[str]: + """Extract conflict information from git output""" + conflicts = [] + + # Look for file names in conflict messages + conflict_patterns = [ + r'error:\s*([^\n]*conflict[^\n]*\.patch[^\n]*)', + r'conflict\s+in\s+([^\s]+)', + r'([/\w][^\s]*\.[\w]+).*conflict' + ] + + for pattern in conflict_patterns: + matches = re.findall(pattern, stderr + stdout, re.IGNORECASE) + conflicts.extend(matches) + + # Also look for .rej files mentioned + rej_matches = re.findall(r'([^\s]+\.rej)', stderr + stdout) + conflicts.extend(rej_matches) + + return list(set(conflicts)) # Remove duplicates + + def _fallback_patch_apply(self, patch_file: str, target_dir: str) -> Dict[str, Any]: + """Fallback to system patch command if git isn't available""" + self._log("Falling back to system patch command") + try: + # Try with patch command + result = subprocess.run( + ['patch', '-p1', '-i', patch_file, '--verbose', '--dry-run'], + cwd=target_dir, + capture_output=True, + text=True + ) + + if result.returncode == 0: + # Dry run succeeded, apply for real + result = subprocess.run( + ['patch', '-p1', '-i', patch_file], + cwd=target_dir, + capture_output=True, + text=True + ) + + return { + 'success': result.returncode == 0, + 'message': 'Applied via system patch', + 'output': result.stdout, + 'error': result.stderr if result.returncode != 0 else '', + 'method': 'system-patch' + } + except FileNotFoundError: + return { + 'success': False, + 'fatal': True, + 'message': 'Neither git nor patch command available' + } + + def analyze_patch_series(self, patch_files: List[str]) -> Dict[str, Any]: + """ + Analyze patches for potential conflicts before applying. + + Args: + patch_files: List of paths to patch files + + Returns: + Dictionary with patch analysis + """ + self._log(f"Analyzing {len(patch_files)} patches") + analysis = {} + all_modified_files = set() + + for patch_file in patch_files: + patch_path = Path(patch_file) + if not patch_path.exists(): + analysis[patch_file] = {'error': 'File not found'} + continue + + with open(patch_path, 'r', encoding='utf-8', errors='ignore') as f: + patch_content = f.read() + + modified_files = self._extract_modified_files(patch_content) + all_modified_files.update(modified_files) + + analysis[patch_file] = { + 'files_modified': modified_files, + 'patch_size': len(patch_content), + 'lines_changed': self._count_lines_changed(patch_content), + 'conflict_risk': self._assess_conflict_risk(patch_content, modified_files), + 'is_binary': self._is_binary_patch(patch_content) + } + + # Detect file conflicts between patches + file_conflicts = self._detect_file_conflicts(analysis) + + return { + 'patches': analysis, + 'summary': { + 'total_patches': len(patch_files), + 'total_files_modified': len(all_modified_files), + 'file_conflicts_detected': file_conflicts + } + } + + def _extract_modified_files(self, patch_content: str) -> List[str]: + """Extract list of files modified by this patch""" + files = [] + for line in patch_content.split('\n'): + if line.startswith('--- ') or line.startswith('+++ '): + file_path = line.split(' ')[1] + if file_path != '/dev/null' and not file_path.startswith('a/') and not file_path.startswith('b/'): + # Clean up git-style prefixes + clean_path = re.sub(r'^(a|b)/', '', file_path) + files.append(clean_path) + return list(set(files)) + + def _count_lines_changed(self, patch_content: str) -> int: + """Count approximate number of lines changed in patch""" + lines_changed = 0 + for line in patch_content.split('\n'): + if line.startswith('+') and not line.startswith('+++'): + lines_changed += 1 + elif line.startswith('-') and not line.startswith('---'): + lines_changed += 1 + return lines_changed + + def _assess_conflict_risk(self, patch_content: str, modified_files: List[str]) -> str: + """Assess conflict risk level""" + lines_changed = self._count_lines_changed(patch_content) + + if lines_changed > 100: + return 'high' + elif lines_changed > 50: + return 'medium' + else: + return 'low' + + def _is_binary_patch(self, patch_content: str) -> bool: + """Check if patch contains binary files""" + return 'GIT binary patch' in patch_content or 'Binary files' in patch_content + + def _detect_file_conflicts(self, analysis: Dict[str, Any]) -> List[List[str]]: + """Detect which patches modify the same files""" + file_to_patches = {} + conflicts = [] + + for patch_file, info in analysis.items(): + if 'files_modified' not in info: + continue + for file_path in info['files_modified']: + if file_path not in file_to_patches: + file_to_patches[file_path] = [] + file_to_patches[file_path].append(patch_file) + + for file_path, patches in file_to_patches.items(): + if len(patches) > 1: + conflicts.append(patches) + + return conflicts + + def _order_patches_by_dependency(self, analysis: Dict[str, Any]) -> List[str]: + """Order patches to minimize conflicts""" + patches = list(analysis.get('patches', {}).keys()) + + # Simple heuristic: smaller patches first, larger patches later + def get_patch_size(patch_file): + info = analysis['patches'].get(patch_file, {}) + return info.get('patch_size', 0) + + return sorted(patches, key=get_patch_size) + + def apply_with_rollback(self, patch_file: str, target_dir: str) -> Dict[str, Any]: + """ + Apply patch with rollback capability on failure. + + Args: + patch_file: Path to the patch file + target_dir: Directory to apply the patch to + + Returns: + Dictionary with application results + """ + patch_path = Path(patch_file) + target_path = Path(target_dir) + + # Extract files that will be modified + with open(patch_path, 'r') as f: + patch_content = f.read() + modified_files = self._extract_modified_files(patch_content) + + # Create backup of modified files + backup_dir = tempfile.mkdtemp(prefix='patch_backup_') + backed_up_files = [] + + for file_path in modified_files: + full_path = target_path / file_path + if full_path.exists(): + backup_path = Path(backup_dir) / file_path + backup_path.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(full_path, backup_path) + backed_up_files.append(str(full_path)) + + try: + result = self.apply_single_patch(patch_file, target_dir) + if not result['success']: + self._log("Patch failed, restoring backup") + self._restore_backup(backup_dir, target_path) + result['rolled_back'] = True + else: + result['rolled_back'] = False + + result['backed_up_files'] = backed_up_files + return result + + except Exception as e: + self._log(f"Exception during patch: {e}, restoring backup") + self._restore_backup(backup_dir, target_path) + return { + 'success': False, + 'rolled_back': True, + 'message': f'Failed and rolled back: {str(e)}', + 'backed_up_files': backed_up_files + } + finally: + # Clean up backup directory + shutil.rmtree(backup_dir, ignore_errors=True) + + def _restore_backup(self, backup_dir: str, target_path: Path) -> None: + """Restore files from backup""" + backup_path = Path(backup_dir) + for backup_file in backup_path.rglob('*'): + if backup_file.is_file(): + relative_path = backup_file.relative_to(backup_path) + target_file = target_path / relative_path + target_file.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(backup_file, target_file) + + def create_patch(self, original_file: str, modified_file: str, output_patch: str) -> bool: + """ + Create a patch file from original and modified files. + + Args: + original_file: Path to original file + modified_file: Path to modified file + output_patch: Path for output patch file + + Returns: + True if successful + """ + try: + with open(original_file, 'r') as f1, open(modified_file, 'r') as f2: + original_lines = f1.readlines() + modified_lines = f2.readlines() + + diff = difflib.unified_diff( + original_lines, + modified_lines, + fromfile=original_file, + tofile=modified_file, + lineterm='' + ) + + with open(output_patch, 'w') as f: + f.write('\n'.join(diff)) + + self._log(f"Patch created: {output_patch}") + return True + + except Exception as e: + self._log(f"Error creating patch: {e}") + return False + + +def main(): + """Command line interface for the patch handler""" + parser = argparse.ArgumentParser(description='Robust patch file handler') + parser.add_argument('patches', nargs='+', help='Patch files to apply') + parser.add_argument('--target-dir', '-t', default='.', help='Target directory for patches') + parser.add_argument('--repo-path', '-r', default='.', help='Repository root path') + parser.add_argument('--stop-on-conflict', '-s', action='store_true', help='Stop on first conflict') + parser.add_argument('--rollback', '-b', action='store_true', help='Rollback on failure') + parser.add_argument('--analyze-only', '-a', action='store_true', help='Only analyze patches, do not apply') + parser.add_argument('--verbose', '-v', action='store_true', help='Verbose output') + parser.add_argument('--output-format', '-f', choices=['text', 'json'], default='text', help='Output format') + + args = parser.parse_args() + + handler = GitPatchHandler(repo_path=args.repo_path, verbose=args.verbose) + + if args.analyze_only: + analysis = handler.analyze_patch_series(args.patches) + if args.output_format == 'json': + print(json.dumps(analysis, indent=2)) + else: + print_patch_analysis(analysis) + else: + result = handler.apply_patch_series( + args.patches, + target_dir=args.target_dir, + stop_on_conflict=args.stop_on_conflict, + rollback_on_failure=args.rollback + ) + + if args.output_format == 'json': + print(json.dumps(result, indent=2)) + else: + print_patch_results(result) + + sys.exit(0 if result['success'] else 1) + + +def print_patch_analysis(analysis: Dict[str, Any]) -> None: + """Print patch analysis in human-readable format""" + print("Patch Analysis Report") + print("=" * 50) + + summary = analysis['summary'] + print(f"Total patches: {summary['total_patches']}") + print(f"Total files modified: {summary['total_files_modified']}") + print(f"File conflicts detected: {len(summary['file_conflicts_detected'])}") + + if summary['file_conflicts_detected']: + print("\nPotential conflicts:") + for conflict in summary['file_conflicts_detected']: + print(f" - {', '.join(conflict)}") + + print("\nPatch details:") + for patch_file, info in analysis['patches'].items(): + if 'error' in info: + print(f" {patch_file}: ERROR - {info['error']}") + else: + print(f" {patch_file}:") + print(f" - Files: {', '.join(info['files_modified'])}") + print(f" - Size: {info['patch_size']} bytes") + print(f" - Lines changed: {info['lines_changed']}") + print(f" - Risk: {info['conflict_risk']}") + print(f" - Binary: {info['is_binary']}") + + +def print_patch_results(result: Dict[str, Any]) -> None: + """Print patch application results in human-readable format""" + print("Patch Application Report") + print("=" * 50) + print(f"Overall success: {'YES' if result['success'] else 'NO'}") + print(f"Patches applied: {len(result['patches_applied'])}/{len(result['results'])}") + + print("\nDetailed results:") + for patch_result in result['results']: + status = "✓" if patch_result['success'] else "✗" + print(f" {status} {patch_result['patch']}") + if not patch_result['success']: + print(f" Message: {patch_result['message']}") + if patch_result.get('conflicts'): + print(f" Conflicts: {', '.join(patch_result['conflicts'])}") + if patch_result.get('rolled_back'): + print(f" [ROLLED BACK]") + + +# Add difflib for the create_patch method +import difflib + +if __name__ == '__main__': + main() \ No newline at end of file