Merge pull request #48 from saritasa-nest/feature/git-copy

Add git blame copy command

CHANGELOG.md

@@ -6,7 +6,11 @@ We follow [Semantic Versions](https://semver.org/).
 
 - Add configuration for django `manage.py` file path
 - Improve `pre-commit.run-hooks` command with `params`
+<<<<<<< HEAD
+- Add `git.blame-copy` command
+=======
 - Add fallback for `poetry.update` and `poetry.update_to_latest`
+>>>>>>> main
 
 ## 0.9.1
 

README.md

@@ -257,6 +257,65 @@ Settings:
 
 Clone repo or pull latest changes to specified repo
 
+#### git.blame-copy
+
+Command for creating copies of a file with git blame history saving.
+
+Original script written in bash:
+https://dev.to/deckstar/how-to-git-copy-copying-files-while-keeping-git-history-1c9j
+
+Usage:
+```shell
+  inv git.blame-copy <path to original file> <path to copy>,<path to copy>...
+```
+
+If `<path to copy>` is file, then data will be copied in it.
+
+If `<path to copy>` is directory, then data will be copied in provided
+directory with original name.
+
+Algorithm:
+
+1) Remember current HEAD state
+2) For each copy path:
+    move file to copy path, restore file using `checkout`,
+    remember result commits
+3) Restore state of branch
+4) Move file to temp file
+5) Merge copy commits to branch
+6) Move file to it's original path from temp file
+
+Settings:
+
+* `copy_commit_template` template for commits created during command workflow
+* `copy_init_message_template` template for init message printed at command start
+
+Template variables:
+
+* `action` - The copy algorithm consists of several intermediate actions
+(creating temporary files, merging commits, etc.)
+The `action` variable stores the header of the intermediate action.
+* `original_path` - Contains value of first argument of the command
+(path of original file that will be copied)
+* `destination_paths` - Sequence of paths to which the original file will be copied
+* `project_task` - project task that will be parsed from current git branch.
+If no task found in branch, then will be empty
+
+Default values for templates:
+* `copy_commit_template`:
+```python
+  "[automated-commit]: {action}\n\n"
+  "copy: {original_path}\n"
+  "to:\n* {destination_paths}\n\n"
+  "{project_task}"
+```
+* `copy_init_message_template`:
+```python
+  "Copy {original_path} to:\n"
+  "* {destination_paths}\n\n"
+  "Count of created commits: {commits_count}"
+```
+
 ### pre-commit
 
 #### pre-commit.install

saritasa_invocations/_config.py

@@ -35,6 +35,17 @@ class GitSettings:
 
     merge_ff: str = "false"
     pull_ff: str = "only"
+    copy_commit_template: str = (
+        "[automated-commit]: {action}\n\n"
+        "copy: {original_path}\n"
+        "to:\n* {destination_paths}\n\n"
+        "{project_task}"
+    )
+    copy_init_message_template: str = (
+        "Copy {original_path} to:\n"
+        "* {destination_paths}\n\n"
+        "Count of created commits: {commits_count}"
+    )
 
 
 @dataclasses.dataclass

saritasa_invocations/git.py

@@ -1,4 +1,6 @@
+import collections.abc
 import pathlib
+import re
 
 import invoke
 
@@ -48,3 +50,258 @@ def clone_repo(
         printing.print_success(f"Pulling changes for {repo_link}...")
         with context.cd(repo_path):
             context.run("git pull")
+
+
+@invoke.task
+def blame_copy(
+    context: invoke.Context,
+    original_path: str,
+    destination_paths: str,
+) -> None:
+    """Copy file from original path to destination paths with saving blame.
+
+    If destination path is file, then data will be copied in it.
+    If destination path is directory, then data will be copied in provided
+    directory with original name.
+    How script works:
+        1) Remember current HEAD state
+        2) For each copy path:
+            move file to copy path, restore file using `checkout`,
+            remember result commits
+        3) Restore state of branch
+        4) Move file to temp file
+        5) Merge copy commits to branch
+        6) Move file to it's original path from temp file
+    Count of created commits:
+        N + 3, where N - is count of copies,
+        3 - 1 commit to put original file to temp file
+        1 commit to merge commits with creation of copy files
+        1 commit to put data from temp file to original file back.
+
+    """
+    config = _config.Config.from_context(context)
+    destination_paths_list = _split_destination_paths(destination_paths)
+    printing.print_success("Validating provided paths")
+    _validate_paths(original_path, destination_paths_list)
+    printing.print_success(
+        config.git.copy_init_message_template.format(
+            original_path=original_path,
+            destination_paths="\n* ".join(destination_paths_list),
+            commits_count=len(destination_paths_list) + 3,
+        ),
+    )
+    _display_continue_prompt()
+    # formatted commit template with only space for action
+    printing.print_success("Build formatted commit")
+    formatted_commit_template = config.git.copy_commit_template.format(
+        action="{action}",
+        original_path=original_path,
+        destination_paths="\n* ".join(destination_paths_list),
+        project_task=_build_task_string(context=context),
+    )
+
+    # temp file to save original file
+    printing.print_success("Create temp file")
+    temp_file = _get_command_output(
+        context=context,
+        command=f"mktemp ./{destination_paths_list[0]}.XXXXXX",
+    )
+
+    # current HEAD state
+    printing.print_success("Get current HEAD sha")
+    root_commit = _get_command_output(
+        context=context,
+        command="git rev-parse HEAD",
+    )
+
+    # create copies with blame of original
+    printing.print_success("Create copies with blame of original file")
+    copy_commits = _copy_files(
+        context=context,
+        original_path=original_path,
+        destination_paths=destination_paths_list,
+        root_commit=root_commit,
+        commit_template=formatted_commit_template,
+    )
+
+    # put original file to temp copy
+    printing.print_success("Restore branch to original state")
+    context.run("git reset --hard HEAD^")
+    printing.print_success(f"Move {original_path} to temp file")
+    _move_file(
+        context=context,
+        from_path=original_path,
+        to_path=temp_file,
+        options=["-f"],
+        message=formatted_commit_template.format(
+            action=f"put {original_path} to temp file",
+        ),
+    )
+
+    # merge copy commits
+    printing.print_success("Merge copy commits")
+    _merge_commits(
+        context=context,
+        commits=copy_commits,
+        message=formatted_commit_template.format(action="merge"),
+    )
+
+    # move original file back
+    printing.print_success(f"Move data from temp file to {original_path}")
+    _move_file(
+        context=context,
+        from_path=temp_file,
+        to_path=original_path,
+        message=formatted_commit_template.format(
+            action=f"put temp file data to {original_path}",
+        ),
+    )
+    printing.print_success("Success")
+
+
+def _split_destination_paths(
+    destination_paths: str,
+) -> list[str]:
+    """Split destination path to sequence and strip trailing symbols."""
+    return [path.strip() for path in destination_paths.split(",")]
+
+
+def _validate_paths(
+    original_path: str,
+    destination_paths: collections.abc.Sequence[str],
+) -> None:
+    """Validate provided paths exists."""
+    error_messages = []
+    if not pathlib.Path(original_path).exists():
+        error_messages.append(
+            f"{original_path} not found.",
+        )
+    for destination in destination_paths:
+        dirname = pathlib.Path(destination).parent
+        if dirname and not pathlib.Path(dirname).exists():
+            error_messages.append(f"{dirname} not found.")
+    if error_messages:
+        printing.print_error(
+            "\n".join(error_messages),
+        )
+        raise invoke.Exit(
+            message="Failed to validate provided paths.",
+            code=1,
+        )
+
+
+def _merge_commits(
+    context: invoke.Context,
+    commits: collections.abc.Sequence[str],
+    message: str,
+) -> None:
+    """Merge passed commits."""
+    context.run(f"git merge {' '.join(commits)} -m '{message}'")
+
+
+def _move_file(
+    context: invoke.Context,
+    from_path: str,
+    to_path: str,
+    message: str,
+    options: collections.abc.Sequence[str] = tuple(),
+) -> None:
+    """Move `first_file `to `second_file` path using git."""
+    context.run(f"git mv {' '.join(options)} {from_path} {to_path}")
+    context.run(f"git commit --no-verify -n -m '{message}'")
+
+
+def _copy_files(
+    context: invoke.Context,
+    original_path: str,
+    destination_paths: collections.abc.Sequence[str],
+    root_commit: str,
+    commit_template: str,
+) -> list[str]:
+    """Copy file from `original_path` to `destination_paths` using git.
+
+    Return commits related to each copy.
+
+    """
+    commits = []
+    for path in destination_paths:
+        context.run(f"git reset --soft {root_commit}")
+        context.run(f"git checkout {root_commit} {original_path}")
+        context.run(f"git mv -f {original_path} {path}")
+
+        commit_message = commit_template.format(
+            action=f"create {path}",
+        )
+        context.run(f"git commit --no-verify -n -m '{commit_message}'")
+        new_commit = _get_command_output(
+            context=context,
+            command="git rev-parse HEAD",
+        )
+        commits.append(new_commit)
+    return commits
+
+
+def _build_task_string(
+    context: invoke.Context,
+) -> str:
+    """Build task string.
+
+    Build string with following format: Task: <project-task-id>
+    If current git branch has no task id, then empty string will return.
+
+    """
+    task_id = _get_project_task_from_current_branch(context=context)
+    if not task_id:
+        return ""
+    return f"Task: {task_id}"
+
+
+def _get_project_task_from_current_branch(
+    context: invoke.Context,
+) -> str:
+    """Get project task from current branch.
+
+    If branch has no task, then empty string will return.
+
+    """
+    current_branch = _get_command_output(
+        context=context,
+        command="git branch --show-current",
+    )
+    match = re.search(r"\w+\/(\w+-\d+)", current_branch)
+    if match is None:
+        return ""
+    task = match.group(1)
+    return task
+
+
+def _display_continue_prompt():
+    """Display continue message.
+
+    If `n` entered, then exit script.
+
+    """
+    if input("Continue? [Enter/N]: ").lower() == "n":
+        raise invoke.Exit("Exit from script", code=1)
+
+
+def _get_command_output(
+    context: invoke.Context,
+    command: str,
+) -> str:
+    """Get command output.
+
+    Try to run command using context.
+    If no result returned then cancel command execution.
+
+    """
+    command_result = context.run(command)
+    if command_result is None:
+        raise invoke.Exit(
+            code=1,
+            message=(
+                "Something went wrong.\n"
+                "Make sure you have enough system permissions."
+            ),
+        )
+    return command_result.stdout.rstrip()