"""TextChain.""" import difflib import log class TextChain(object): """Text chain to handle various between each commit. Attributes: _save_filename: Name of the file to stores the content of the buffer. _commits: A list of 2-tuple which likes: first element: The commit id. second element: The instance of _TextCommit. _last_commit: An instance of _TextCommit, cache the last commit for updating the cursor position after commiting. """ def __init__(self, save_filename): """Constructor. Args: save_filename: Name of the file to save the lastest commit text. """ self._save_filename = save_filename content = '' try: with open(save_filename, 'r') as f: content = f.read() except IOError: log.info('Cannot load the default text.') self._commits = [ (0, _TextCommit('', '')), (1, _TextCommit('', content))] self._last_commit = None def commit(self, orig_id, new_text, cursors): """Commits a update. Args: orig_id: Original commit id. new_text: Updated text. cursors: Cursors to rebase at the same time. Return: A 3-tuple for new commit id, new text and the rebased cursors. """ old_index = self._get_commit_index(orig_id) commit = _TextCommit(self._commits[old_index][1].text, new_text) cursors_info = [commit.get_cursor_info(cur) for cur in cursors] commit.apply_commits([cm[1] for cm in self._commits[old_index + 1 :]]) self._last_commit = commit.copy() new_id = self._commits[-1][0] + 1 for info in cursors_info: info.apply_commits([cm[1] for cm in self._commits[old_index + 1 :]]) new_cursors = [cursor_info.position for cursor_info in cursors_info] self._commits += [(new_id, commit), (new_id + 1, _TextCommit(commit.text, commit.text))] self.delete(orig_id) self.delete(self._commits[-3][0]) self._save() return new_id, commit.text, new_cursors def update_cursors(self, cursors): """Updates the cursors by the last commit. Args: cursors: List of cursor position. Return: List of updated cursor position. """ cursors_info = [_CursorInfo_OnOrigText(cursor) for cursor in cursors] for cursor_info in cursors_info: cursor_info.apply_commits([self._last_commit]) return [cursor_info.position for cursor_info in cursors_info] def new(self): """Creates an empty commit. Return: The commit id of the new commit. """ commit_id = self._commits[0][0] self._commits.insert(0, (commit_id - 1, _TextCommit('', ''))) return commit_id def delete(self, commit_id): """Deletes a commit. Args: commit_id: The id of the commit to be delete. """ index = self._get_commit_index(commit_id) if index + 1 < len(self._commits): pre_text = self._commits[index - 1][1].text nxt_text = self._commits[index + 1][1].text self._commits[index + 1] = (self._commits[index + 1][0], _TextCommit(pre_text, nxt_text)) del self._commits[index] def get_text(self, commit_id): """Gets the text of a specified commit. Args: commit_id: Id of that commit. Return: The text. """ return self._commits[self._get_commit_index(commit_id)][1].text def _get_commit_index(self, commit_id): """Gets the index of the commits from gived commit id. Args: commit_id: Commit id. Returns: Index of the corrosponding commit. """ for index in range(len(self._commits)): if self._commits[index][0] == commit_id: return index def _save(self): """Saves the last text to the file.""" try: with open(self._save_filename, 'w') as f: f.write(self._commits[-1][1].text) except IOError: log.info('Cannot save the text to the file.') def _opers_apply_opers(orig_opers, opers_tobe_applied): """Let a list of operations apply another list of operations. Args: orig_opers: List of instance of _ChgTextOper. opers_tobe_applied: List of instance of _ChgTextOper. Return: A list of instance of _ChgTextOper, which are the ones applied the opers_tobe_applied from the orig_opers. """ ret = orig_opers for oper_tobe_applied in opers_tobe_applied: # The operation might split into multiple operations after rebasing, # So here we needs to use another list to stores the new operations. updated_opers = [] for orig_oper in ret: updated_opers += orig_oper.apply_oper(oper_tobe_applied) ret = updated_opers return ret class _TextCommit(object): """Stores a text commit. It includes a final text after commited and a sequence of _ChgTextOper for changing the original string to the new one. Attributes: _text: The final text. _opers: List of operations for changing the original string to the new one. """ def __init__(self, old_text, new_text): """Constructor. Args: old_text: The original text. new_text: The final text after commited. """ self._text = new_text self._opers = [] diff = difflib.SequenceMatcher(a=old_text, b=new_text) for tag, begin, end, begin2, end2 in diff.get_opcodes(): if tag in ('replace', 'delete', 'insert'): self._opers.append( _ChgTextOper(begin, end, new_text[begin2 : end2])) @property def text(self): """Gets the final text after this commit.""" return self._text @property def opers(self): """Gets the operations of this commit.""" return self._opers @property def increased_length(self): """Gets the increased length of this commit.""" return sum(o.increased_length for o in self._opers) def copy(self): """Returns a copy of myself. Return: An instance of _TextCommit. """ ret = _TextCommit('', '') ret._text = self.text ret._opers = [_ChgTextOper(oper.begin, oper.end, oper.new_text) for oper in self._opers] return ret def apply_commits(self, commits): """Applies a list of commits before this occured. Args: commits: A list of instance of _TextCommit. """ if commits: for commit in commits: self._opers = _opers_apply_opers(self._opers, commit._opers) self._rebase_text(commits[-1].text) def get_cursor_info(self, cursor_pos): """Gets the cursor information by gived cursor position. If the cursor position is on character which is inserted at this commit, it will return _CursorInfo_OnNewCommit; Otherwise it will return _CursorInfo_OnOrigText. Args: cursor_pos: Position of the cursor. Return: A instance of _CursorInfo_OnNewCommit or _CursorInfo_OnOrigText. """ offset = 0 for oper in self._opers: delta = cursor_pos - (oper.begin + offset) if delta <= len(oper.new_text): return _CursorInfo_OnNewCommit(oper, delta) offset += oper.increased_length return _CursorInfo_OnOrigText(cursor_pos) def _rebase_text(self, new_orig_text): """Rebase the original text to another text. Args: new_orig_text: The new text. """ end_index = 0 self._text = '' for oper in self._opers: self._text += new_orig_text[end_index : oper.begin] self._text += oper.new_text end_index = oper.end self._text += new_orig_text[end_index : ] class _CursorInfo_OnNewCommit(object): """About the cursor position who is at the place changed in the new commit. Attributes: _opers: The duplicated operation of the original operation. _delta: The offset between the cursor position and the begin of the operation's range. """ def __init__(self, oper, delta): """Constructor. Args: oper: The operation which this cursor position is in. delta: The offset the cursor position and the begin of the operation. """ # We need to store it in a list because after applying other commits, it # might split into multiple operations. self._opers = [_ChgTextOper(oper.begin, oper.end, oper.new_text)] self._delta = delta def apply_commits(self, commits): """Applies commits. Does something very similar in commit, because we needs to applies each operations to other commits too. Args: commits: List of commits to be applied. """ for commit in commits: self._opers = _opers_apply_opers(self._opers, commit.opers) @property def position(self): """Calculates and returns the final cursor position.""" dt = self._delta offset = 0 for oper in self._opers: if dt <= len(oper.new_text): return oper.begin + offset + dt dt -= len(oper.new_text) offset += oper.increased_length class _CursorInfo_OnOrigText(object): """About the cursor position who is at the place based on the original text. Attributes: _position: The position of the cursor. """ def __init__(self, position): """Constructor. Args: position: The cursor position. """ self._position = position def apply_commits(self, commits): """Applies the commit's change on it. Args: commits: List of commits to be applied. """ for commit in commits: for oper in commit.opers: if self._position <= oper.begin: # Remain changeless when the operation is after the cursor # position. pass elif oper.end - 1 <= self._position: # Just offset to the right place if the operation occures # totally at the left side of the cursor. self._position += oper.increased_length else: # Moves the position to the begin of this operation when the # cursor is inside the operation. self._position = oper.begin @property def position(self): """Returns the final cursor position.""" return self._position class _ChgTextOper(object): """An operation of changing a text to a new one. Here we define changing a text to a new one contains a lot of operations. Each operation will replace a substring in the original text to another text. In this class, we will handles that if we want to apply an operation to a text before another operation happened, how to merge and prevent the confliction. Attributes: _begin: The begin of the range of the substring in the original string. _end: The end of the range of the substring in the original string. _new_text: The string to replace on. Notes: 1. The range is an open range [_begin, _end) """ def __init__(self, beg, end, new_text): self._begin = beg self._end = end self._new_text = new_text def apply_oper(self, oper): # pylint: disable=R0911,R0912 """Applies an operation before me. Args: oper: An instance of _ChgTextOper. Return: A list of instance of _ChgTextOper which contains the equivalent operations after applying the gived operation before it. The reason that it returns a list instead of just an element is that it may be split into multiple operations. """ if oper.begin < self._begin: if oper.end <= self._begin: return self._apply_left_seperate(oper) elif oper.end < self._end: return self._apply_left_intersection(oper) elif oper.end == self._end: return self._apply_left_exact_cover(oper) else: return self._apply_total_cover(oper) elif oper.begin == self._begin: if oper.end < self._end: return self._apply_left_exact_inside(oper) elif oper.end == self._end: return self._apply_exact_same(oper) else: return self._apply_right_exact_cover(oper) elif oper.begin < self._end: if oper.end < self._end: return self._apply_exact_inside(oper) elif oper.end == self._end: return self._apply_right_exact_inside(oper) else: return self._apply_right_intersection(oper) else: return self._apply_right_seperate(oper) @property def begin(self): """Gets the begin of the range.""" return self._begin @property def end(self): """Gets the end of the range.""" return self._end @property def new_text(self): """Gets the string to replace.""" return self._new_text @property def increased_length(self): """Gets the increased length after done this operation.""" return len(self._new_text) - (self._end - self._begin) def _apply_left_seperate(self, oper): """Applies the case that, The operation: [ ) Me: [ ) Description: Just offset to the right position, because after that operation done, the length of the new string might be changed. Args: oper: Instance of the _ChgTextOper. Return: A list of instance of _ChgTextOper. """ offset = oper.increased_length return [_ChgTextOper(self._begin + offset, self._end + offset, self._new_text)] def _apply_left_intersection(self, oper): """Applies the case that, The operation: [ ) Me: [ ) Result: [ ) Method: Offset the begin of my operaiton to the end of the that operaiont's end. Args: oper: Instance of the _ChgTextOper. Return: A list of instance of _ChgTextOper. """ offset = oper.increased_length return [_ChgTextOper(oper.end + offset, self._end + offset, self._new_text)] def _apply_left_exact_cover(self, oper): """Applies the case that, The operation: [ ) Me: [ ) Result: | (Here "|" means that [ and ) are at the same place) Method: Offset the begin of my operaiton to the end of the that operaiont's end. Args: oper: Instance of the _ChgTextOper. Return: A list of instance of _ChgTextOper. """ end_pos = oper.end + oper.increased_length return [_ChgTextOper(end_pos, end_pos, self._new_text)] def _apply_total_cover(self, oper): """Applies the case that, The operation: [ ) Me: [ ) Result: | (Here "|" means that [ and ) are at the same place) Method: Offset the begin of my operaiton to the end of the that operaiont's end. Args: oper: Instance of the _ChgTextOper. Return: A list of instance of _ChgTextOper. """ end_pos = oper.end + oper.increased_length return [_ChgTextOper(end_pos, end_pos, self._new_text)] def _apply_left_exact_inside(self, oper): """Applies the case that, The operation: [ ) Me: [ ) Result: [ ) Method: Offset the begin of my operaiton to the end of the that operaiont's end. Args: oper: Instance of the _ChgTextOper. Return: A list of instance of _ChgTextOper. """ offset = oper.increased_length return [_ChgTextOper(oper.end + offset, self._end + offset, self._new_text)] def _apply_exact_same(self, oper): """Applies the case that, The operation: [ ) Me: [ ) Result: | (Here "|" means that [ and ) are at the same place) Method: Offset the begin of my operaiton to the end of the that operaiont's end. Args: oper: Instance of the _ChgTextOper. Return: A list of instance of _ChgTextOper. """ offset = oper.increased_length return [_ChgTextOper(oper.end + offset, self._end + offset, self._new_text)] def _apply_right_exact_cover(self, oper): """Applies the case that, The operation: [ ) Me: [ ) Result: | (Here "|" means that [ and ) are at the same place) Method: Offset the begin/end of my operation to the end of the that operaiont's end. Args: oper: Instance of the _ChgTextOper. Return: A list of instance of _ChgTextOper. """ pos = oper.end + oper.increased_length return [_ChgTextOper(pos, pos, self._new_text)] def _apply_exact_inside(self, oper): """Applies the case that, The operation: [ ) Me: [ ) Result: [ ) [xx) (Here "[xx)" means that it an operation to replace a range of string with an empty string.) Method: Splits me into two part, the left side remain the original text to replace and the right part will just delete a substring. Args: oper: Instance of the _ChgTextOper. Return: A list of instance of _ChgTextOper. """ offset = oper.increased_length return [_ChgTextOper(self._begin, oper.begin, self._new_text), _ChgTextOper(oper.end + offset, self._end + offset, '')] def _apply_right_exact_inside(self, oper): """Applies the case that, The operation: [ ) Me: [ ) Result: [ ) Method: Modifies the end of my operation to the begin of that operation. Args: oper: Instance of the _ChgTextOper. Return: A list of instance of _ChgTextOper. """ return [_ChgTextOper(self._begin, oper._begin, self._new_text)] def _apply_right_intersection(self, oper): """Applies the case that, The operation: [ ) Me: [ ) Result: [ ) Method: Modifies the end of my operation to the begin of that operation. Args: oper: Instance of the _ChgTextOper. Return: A list of instance of _ChgTextOper. """ return [_ChgTextOper(self._begin, oper._begin, self._new_text)] def _apply_right_seperate(self, oper): """Applies the case that, The operation: [ ) Me: [ ) Result: [ ) Method: Remain unchanged. Args: oper: Instance of the _ChgTextOper. Return: A list of instance of _ChgTextOper. """ return [_ChgTextOper(self._begin, self._end, self._new_text)]