Show a diff when a file will be or gets modified

Author: jgelens

pyinfra/operations/files.py

@@ -10,10 +10,11 @@
 import traceback
 from datetime import timedelta
 from fnmatch import fnmatch
-from io import StringIO
+from io import BytesIO, StringIO
 from pathlib import Path
 from typing import IO, Any, Union
 
+import click
 from jinja2 import TemplateRuntimeError, TemplateSyntaxError, UndefinedError
 
 from pyinfra import host, logger, state
@@ -59,6 +60,7 @@
 from .util.files import (
     adjust_regex,
     ensure_mode_int,
+    generate_color_diff,
     get_timestamp,
     sed_delete,
     sed_replace,
@@ -915,6 +917,20 @@ def put(
 
         # Check sha1sum, upload if needed
         if local_sum != remote_sum:
+            current_contents = BytesIO()
+
+            # Generate diff when contents change
+            host.get_file(dest, current_contents)
+            current_lines = current_contents.getvalue().decode("utf-8").splitlines(keepends=True)
+            logger.info(f"\n    Will modify {click.style(dest, bold=True)}")
+
+            with get_file_io(src, "r") as f:
+                desired_lines = f.readlines()
+
+            for line in generate_color_diff(current_lines, desired_lines):
+                logger.info(f"  {line}")
+            logger.info("")
+
             yield FileUploadCommand(
                 local_file,
                 dest,

pyinfra/operations/util/files.py

@@ -1,8 +1,11 @@
 from __future__ import annotations
 
+import difflib
 import re
 from datetime import datetime
-from typing import Callable
+from typing import Callable, Generator
+
+import click
 
 from pyinfra.api import QuoteString, StringCommand
 
@@ -173,3 +176,34 @@ def adjust_regex(line: str, escape_regex_characters: bool) -> str:
         match_line = "{0}.*$".format(match_line)
 
     return match_line
+
+
+def generate_color_diff(
+    current_lines: list[str], desired_lines: list[str]
+) -> Generator[str, None, None]:
+    def _format_range_unified(start: int, stop: int) -> str:
+        beginning = start + 1  # lines start numbering with one
+        length = stop - start
+        if length == 1:
+            return "{}".format(beginning)
+        if not length:
+            beginning -= 1  # empty ranges begin at line just before the range
+        return "{},{}".format(beginning, length)
+
+    for group in difflib.SequenceMatcher(None, current_lines, desired_lines).get_grouped_opcodes(2):
+        first, last = group[0], group[-1]
+        file1_range = _format_range_unified(first[1], last[2])
+        file2_range = _format_range_unified(first[3], last[4])
+        yield "@@ -{} +{} @@".format(file1_range, file2_range)
+
+        for tag, i1, i2, j1, j2 in group:
+            if tag == "equal":
+                for line in current_lines[i1:i2]:
+                    yield "  " + line.rstrip()
+                continue
+            if tag in {"replace", "delete"}:
+                for line in current_lines[i1:i2]:
+                    yield click.style("- " + line.rstrip(), "red")
+            if tag in {"replace", "insert"}:
+                for line in desired_lines[j1:j2]:
+                    yield click.style("+ " + line.rstrip(), "green")

tests/util.py

@@ -187,6 +187,16 @@ def noop(self, description):
     def get_temp_filename(*args, **kwargs):
         return "_tempfile_"
 
+    def get_file(
+        self,
+        remote_filename,
+        filename_or_io,
+        remote_temp_filename=None,
+        print_output=False,
+        *arguments,
+    ):
+        return True
+
     @staticmethod
     def _get_fact_key(fact_cls):
         return "{0}.{1}".format(fact_cls.__module__.split(".")[-1], fact_cls.__name__)