PawletOS/PawletOS-Build
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Base, more work to do

Browse Source
This commit is contained in:
oxmc7769
2025-11-05 00:45:21 +00:00
parent 680a521169
commit e0a43d9e66
6 changed files with 1158 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+2
View File
@@ -174,3 +174,5 @@ cython_debug/
# PyPI configuration file
.pypirc
# Android source code
AOSP-source/
README.md
+128 -1
View File
@@ -1,3 +1,130 @@
# PawletOS-Build
A repo that has scripts and helpers to build PawletOS
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 <n>` | Patch level for -p option (default: 1) |
| `--sourceDir <path>` | Source directory (default: current directory) |
| `--patchDir <path>` | Patch directory (default: patches) |
| `--syncJobs <n>` | Number of parallel jobs for repo sync |
| `--ignorePatches <files>` | 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.
environ.py
+392
View File
@@ -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()
repo.yml
+10
View File
@@ -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"
rpivanilla-ensetup.sh
+13
View File
@@ -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"
scripts/better_patch.py
+613
View File
@@ -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()