git-merge-versiontxt.py

#!/usr/bin/python3
__author__ = 'vigurskiy'

import sys
import difflib
import re
import os

version_template_arcus = '\A\d+\.\d+\.\d+\.\d+ \[\d{1,2}\.\d{1,2}\.\d{2,4}\]\s{1}'
version_template = '\A\d+\.\d+\.\d+\.\d+ \[\d{1,4}\.\d{1,2}\.\d{1,2}\]\s{1}'
encoding = "cp1251"
nl = "\n"

differ = difflib.Differ()
base_file_lines = []

class Block:

    def __init__(self, start, pos_in_base, **kwargs):
        self.start = self.end = start
        self.pos = pos_in_base
        self.conflicted = False
        self.duplicate = False
        for key in kwargs.keys():
            if key == 'commit':
                self.commit = kwargs.get(key)

    def equals(self, block):
        list1 = self.commit.file_lines[self.start: self.end+1]
        list2 = block.commit.file_lines[block.start: block.end+1]
        if ((list1 > list2) - (list1 < list2)) == 0:
            return True
        else:
            return False

class Commit:

    def __init__(self, file_path, commit_type):
        self.file_path = file_path
        self.commit_type = commit_type
        self.mutual_lines = []
        self.added_blocks = []
        with open(self.file_path, 'r', encoding=encoding) as file:
            self.file_lines = file.readlines()

    def form_added_lines(self):
        diff = differ.compare(base_file_lines, self.file_lines)
        line_num = 0
        base_line_num = 0
        added_lines = []
        for line in diff:
            # print(line,end='')
            code = line[:2]
            if code == "  ":
                self.mutual_lines.append(base_line_num)
                line_num += 1
                base_line_num += 1
            elif code in "+ ":
                added_lines.append((line_num, base_line_num))
                line_num += 1
            elif code in "- ":
                base_line_num += 1
        return added_lines

    def form_added_blocks(self):
        added_lines = self.form_added_lines()
        if len(added_lines) == 0:
            return
        block = Block(added_lines[0][0], added_lines[0][1], commit=self)
        if len(added_lines) == 1:
            self.added_blocks.append(block)
            return
        for line_num, pos_tmp in added_lines[1:]:
            if block.end + 1 != line_num:
                self.added_blocks.append(block)
                block = Block(line_num, pos_tmp, commit=self)
            block.end = line_num
        self.added_blocks.append(block)

    def mark_conflicted_blocks(self, commit, ver_pos):
        for block1 in commit.added_blocks:
            for block2 in self.added_blocks:
                if block1.pos == block2.pos:
                    if block1.equals(block2):
                        block2.duplicate = True
                    elif block1.pos > ver_pos:
                        block1.conflicted = block2.conflicted = True

                # if block1.pos > ver_pos and block2.pos > ver_pos:
                #     if block1.pos == block2.pos and not block1.equals(block2):
                #         block1.conflicted = block2.conflicted = True


def get_version_position(base_lines):
    pattern = re.compile(version_template)
    for i, string in enumerate(base_lines):
        match = pattern.fullmatch(string)
        if match:
            return i
    return -1

def form_mutual_blocks(self, commit1, commit2):
    mutual_blocks = []
    intersection = set(commit1.mutual_lines).intersection(commit2.mutual_lines)
    for line_num in intersection:
        mutual_blocks.append(Block(line_num, line_num, commit=self))  # c is for base file
    return mutual_blocks

def do_three_way_merge(commit_base, commit_local, commit_remote):
    merged_blocks = []
    res = 0
    pattern = re.compile(version_template)
    merged_blocks.extend(commit_local.added_blocks)
    merged_blocks.extend(commit_remote.added_blocks)
    merged_blocks.extend(commit_base.added_blocks)

    merged_blocks.sort(key=lambda block: (block.pos, block.commit.commit_type))

    with open(local, 'w', encoding=encoding, newline=nl) as merged_file:
        for block in merged_blocks:
            start = block.start
            end = block.end + 1
            if block.duplicate:
                continue
            if block.conflicted:
                merged_file.write("#-#-#-#-#\n")
                res = 1
            for line in block.commit.file_lines[start:end]:
                if block.pos <= version_pos:
                    if line == '\n':
                        continue
                    elif pattern.fullmatch(line):
                        merged_file.write('\n')
                merged_file.write(line)
            # merged_file.writelines(block.commit.file_lines[start:end])
            if block.conflicted:
                merged_file.write("#-#-#-#-#\n")

    return res

base = sys.argv[1]
local = sys.argv[2]
remote = sys.argv[3]

#if os.name == 'posix':
nl = '\r\n'
# while True:
#     sf = 0
print("Using custom version.txt merge driver")

with open(base, 'r', encoding=encoding) as base_file:
    base_file_lines = base_file.readlines()


local_commit = Commit(local, 'a')
remote_commit = Commit(remote, 'b')
base_commit = Commit(base, 'c')

version_pos = get_version_position(base_file_lines)
if version_pos == -1:
    print("Sorry, version position not found")
    exit(1)


local_commit.form_added_blocks()
remote_commit.form_added_blocks()
local_commit.mark_conflicted_blocks(remote_commit, version_pos)
base_commit.added_blocks = form_mutual_blocks(base_commit, local_commit, remote_commit)

res = do_three_way_merge(base_commit, local_commit, remote_commit)
if res == 1:
    print("Sorry, some conflicts detected")
    exit(1)
else:
    print("Successful merge")
    exit(0)

 

Custom merge script for revision file. Used to avoid conflicts while pull requests are merged into develop.