FastPatch.py

FastPatch.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

#
# Author: 黄龑(huangyan13@baidu.com)
# Created Time: 2016/08/18 14:23
# File Name: fast_patch.py
# Description: 
#
# Copyright (c) 2015 Baidu.com, Inc. All Rights Reserved
#

import os
import sys
import md5
import codecs
import subprocess
import shutil

from unidiff import PatchSet


def fast_patch(patch_file, src_dir):
    FastPatch(patch_file, src_dir).run()


class FastPatch(object):
    @staticmethod
    def md5(data):
        m = md5.new()
        m.update(data)
        return m.hexdigest()

    def __init__(self, patch_file, dst_dir):
        self.patch = PatchSet.from_filename(patch_file, encoding='utf-8')
        self.dest_dir = os.path.abspath(dst_dir)
        self.patched_files = []
        self.splited_patches = []
        self.available_patches = []

    def filter_files(self):
        for patched_file in self.patch:
            abs_path = os.path.join(self.dest_dir, patched_file.path)
            if os.path.isfile(abs_path):
                self.patched_files.append(patched_file)

    def split_hunks(self):
        for patched_file in self.patched_files:
            source = "--- %s\n" % patched_file.source_file
            target = "+++ %s\n" % patched_file.target_file
            for hunk in patched_file:
                content = source + target + unicode(hunk) + '\n'
                self.splited_patches.append(content)

    def check_hunks(self):
        os.chdir(self.dest_dir)
        fid = open('error.log', 'w')
        patch_dir = os.path.join(self.dest_dir, 'succeed')
        failed_dir = os.path.join(self.dest_dir, 'failed')
        if os.path.exists(patch_dir):
            shutil.rmtree(patch_dir)
        if os.path.exists(failed_dir):
            shutil.rmtree(failed_dir)
        os.makedirs(patch_dir)
        os.makedirs(failed_dir)

        def check_func(hunk):
            hunk = hunk.encode('utf-8')
            filename = os.path.join(patch_dir, self.md5(hunk) + '.diff')
            open(filename, 'w').write(hunk)
            cmd = 'git apply --check %s' % filename
            p = subprocess.Popen(cmd, shell=True,
                                 stdout=subprocess.PIPE,
                                 stderr=subprocess.PIPE)
            p.wait()
            fid.write(p.stdout.read())
            fid.write(p.stderr.read())
            if p.returncode == 0:
                return True
            else:
                shutil.move(filename, failed_dir)
                return False
        total = len(self.splited_patches)
        failed = 0
        for idx, hunk in enumerate(self.splited_patches):
            if idx % 100 == 0:
                print "[%.2f %%] Checked %d patches, failed %d." \
                      % (float(idx) / total * 100., idx, failed)
            if check_func(hunk):
                self.available_patches.append(hunk)
            else:
                failed += 1
        fid.close()

    def run(self):
        self.filter_files()
        self.split_hunks()
        self.check_hunks()
        print 'Available patches: %d' % len(self.available_patches)
        print 'All patches: %d' % len(self.splited_patches)

if __name__ == '__main__':
    if len(sys.argv) != 3:
        print 'Usage: python fast_patch.py <patch-file> <dest-dir>'
        sys.exit(-1)
    fast_patch(sys.argv[1], sys.argv[2])