Improve comments check (#2)

* Improve comment detection
* Added --limit and --only-force options
* Refactored tests

Author: cdce8p

.github/workflows/ci.yaml

@@ -142,7 +142,7 @@ jobs:
         run: |
           . venv/bin/activate
           pip install -e .
-          pylint ${{ env.LIB_FOLDER }}
+          pylint ${{ env.LIB_FOLDER }} tests
 
   mypy:
     name: Check mypy
@@ -171,7 +171,7 @@ jobs:
       - name: Run mypy
         run: |
           . venv/bin/activate
-          mypy ${{ env.LIB_FOLDER }}
+          mypy ${{ env.LIB_FOLDER }} tests
 
 
   prepare-tests-linux:

README.md

@@ -62,6 +62,10 @@ pre-commit run --hook-stage manual python-typing-update --all-files
 **`--verbose`**  
 Always print verbose logging.
 
+**`--limit`**  
+Max number of files that should be changed. No performance improvements,
+since the limit is only applied **after** all files have been processed.
+
 **`--full-reorder`**  
 Use additional options from [python-reorder-imports][pri] to rewrite
 - `--py38-plus` (default): Imports from `mypy_extensions` and `typing_extensions` when possible.
@@ -77,6 +81,10 @@ Check if files would be modified. Return with exitcode `1` or `0` if not. Useful
 Don't revert changes if a modified comment is detected.
 Check `git diff` before committing!
 
+**`--only-force`**  
+Only update files which are likely to require extra work.
+Check `git diff` before committing!
+
 **`--py39-plus`**  
 Set the minimum Python syntax to **3.9**. (Default: **3.8**)
 

python_typing_update/__main__.py

@@ -26,6 +26,10 @@ async def async_main(argv: list[str] | None = None) -> int:
         'filenames',
         nargs='+',
     )
+    parser.add_argument(
+        '--limit', type=int, default=0,
+        help="Max number of files that should be changed. No performance improvement!",
+    )
     parser.add_argument(
         '--full-reorder',
         action='store_true',
@@ -54,6 +58,11 @@ async def async_main(argv: list[str] | None = None) -> int:
         action='store_true',
         help="Update all files. Double check changes afterwards!",
     )
+    group1.add_argument(
+        '--only-force',
+        action='store_true',
+        help="Only update files which are likely to require extra work",
+    )
 
     group2 = parser.add_mutually_exclusive_group()
     group2.add_argument(

python_typing_update/const.py

@@ -1,10 +1,17 @@
 # ---------------------------------------------------------------------------
 # Licensed under the MIT License. See LICENSE file for license information.
 # ---------------------------------------------------------------------------
+from enum import Flag, auto
 
 version = (0, 2, 0)
 dev_version = 1
 
 version_str = '.'.join(map(str, version))
 if dev_version is not None:
     version_str += f'-dev{dev_version}'
+
+
+class FileStatus(Flag):
+    CLEAR = 0
+    COMMENT = auto()
+    COMMENT_TYPING = auto()

python_typing_update/main.py

@@ -7,23 +7,27 @@
 import asyncio
 import builtins
 import logging
-import os
 from functools import partial
 from io import StringIO
-from pathlib import Path
 
 import reorder_python_imports
 from autoflake import _main as autoflake_main
 from isort.main import main as isort_main
 from pyupgrade._main import main as pyupgrade_main
 
+from .const import FileStatus
+from .utils import (
+    async_check_uncommitted_changes, async_restore_files,
+    check_comment_between_imports, check_files_exist)
+
 logger = logging.getLogger("typing-update")
 
 
 async def typing_update(
     loop: asyncio.AbstractEventLoop,
     filename: str,
     args: argparse.Namespace,
+    file_status: FileStatus,
 ) -> tuple[int, str]:
     """Update typing syntax.
 
@@ -67,7 +71,7 @@ async def typing_update(
             None, autoflake_partial,
             [None, '-c', filename],
         )
-        if not args.full_reorder:
+        if not args.full_reorder and FileStatus.COMMENT_TYPING not in file_status:
             # -> No unused imports, revert changes
             return 2, filename
     except SystemExit:
@@ -103,56 +107,6 @@ async def typing_update(
     return 0, filename
 
 
-async def async_restore_files(file_list: list[str]) -> None:
-    if not file_list:
-        return
-    process = await asyncio.create_subprocess_shell(
-        f"git restore -- {' '.join(file_list)}",
-    )
-    await process.communicate()
-
-
-def check_files_exist(file_list: list[str]) -> list[str]:
-    """Check if all files exist. Return False if not."""
-    file_errors: list[str] = []
-    cwd = Path(os.getcwd())
-    for file_ in file_list:
-        if cwd.joinpath(file_).is_file() is False:
-            file_errors.append(file_)
-    return sorted(file_errors)
-
-
-async def async_check_uncommitted_changes(file_list: list[str]) -> bool:
-    """Check for uncommitted changes.
-
-    Returns:
-        False: if changes still need to be committed
-    """
-    process = await asyncio.create_subprocess_shell(
-        "git diff-index --name-only HEAD --",
-        stdout=asyncio.subprocess.PIPE,
-    )
-    stdout, _ = await process.communicate()
-    files_uncommitted: set[str] = {file_ for item in stdout.decode().split('\n')
-                                   if (file_ := item.strip())}
-    return not any(True for file_ in file_list if file_ in files_uncommitted)
-
-
-async def async_check_changes(file_list: list[str]) -> list[str]:
-    """Check if comment was change in files.
-
-    That is a sing we can't update it automatically.
-    """
-    if not file_list:
-        return []
-    process = await asyncio.create_subprocess_shell(
-        f"git diff -G\"^#|^from.*#|^import.*#\" --name-only -- {' '.join(file_list)}",
-        stdout=asyncio.subprocess.PIPE,
-    )
-    stdout, _ = await process.communicate()
-    return sorted([file_ for file_ in stdout.decode().strip().split('\n') if file_])
-
-
 async def async_run(args: argparse.Namespace) -> int:
     """Update Python typing syntax.
 
@@ -175,17 +129,26 @@ async def async_run(args: argparse.Namespace) -> int:
         print("Abort! Commit all changes to '.py' files before running again.")
         return 11
 
+    filenames: dict[str, FileStatus] = {}
+    for filename in args.filenames:
+        with open(filename) as fp:
+            result = check_comment_between_imports(fp)
+            filenames[filename] = result
+
+    if args.only_force:
+        filenames = {filename: file_status for filename, file_status in filenames.items()
+                     if file_status != FileStatus.CLEAR}
+
     loop = asyncio.get_running_loop()
     files_updated: list[str] = []
     files_no_changes: list[str] = []
-    files_with_comments: list[str] = []
 
     # Mock builtin print to omit output
     original_print = builtins.print
     builtins.print = lambda *args, **kwargs: None
 
     return_values = await asyncio.gather(
-        *[typing_update(loop, filename, args) for filename in args.filenames])
+        *[typing_update(loop, filename, args, file_status) for filename, file_status in filenames.items()])
     for status, filename in return_values:
         if status == 0:
             files_updated.append(filename)
@@ -195,6 +158,13 @@ async def async_run(args: argparse.Namespace) -> int:
     builtins.print = original_print
     await async_restore_files(files_no_changes)
 
+    if args.limit > 0 and len(files_updated) > args.limit:
+        print(
+            f"Limit applied! Only updated the first {args.limit} "
+            f"of {len(files_updated)} files")
+        async_restore_files(files_updated[args.limit:])
+        files_updated = files_updated[:args.limit]
+
     if args.check is True:
         await async_restore_files(files_updated)
         if files_updated:
@@ -204,9 +174,13 @@ async def async_run(args: argparse.Namespace) -> int:
             return 1
         return 0
 
-    files_with_comments = await async_check_changes(files_updated)
+    files_updated_set: set[str] = set(files_updated)
+    files_with_comments = sorted([
+        filename for filename, file_status in filenames.items()
+        if FileStatus.COMMENT in file_status and filename in files_updated_set
+    ])
     if files_with_comments:
-        if args.force:
+        if args.force or args.only_force:
             print("Force mode selected!")
             print("Make sure to double check:")
             for file_ in files_with_comments:
@@ -218,12 +192,17 @@ async def async_run(args: argparse.Namespace) -> int:
             await async_restore_files(files_with_comments)
 
     print("---")
-    print(f"All files: {len(args.filenames)}")
+    print(f"All files: {len(filenames)}")
     print(f"No changes: {len(files_no_changes)}")
     print(f"Files updated: {len(files_updated) - len(files_with_comments)}")
     print(f"Files (no automatic update): {len(files_with_comments)}")
 
-    if not files_with_comments and not args.force and args.verbose == 0:
+    if (
+        not files_with_comments
+        and not args.force
+        and not args.only_force
+        and args.verbose == 0
+    ):
         return 0
     if files_with_comments:
         return 2

python_typing_update/utils.py

@@ -0,0 +1,115 @@
+# ---------------------------------------------------------------------------
+# Licensed under the MIT License. See LICENSE file for license information.
+# ---------------------------------------------------------------------------
+from __future__ import annotations
+
+import asyncio
+import os
+import token
+import tokenize
+from collections.abc import Iterable
+from pathlib import Path
+from typing import TextIO
+
+from .const import FileStatus
+
+
+def check_files_exist(file_list: Iterable[str]) -> list[str]:
+    """Check if all files exist. Return False if not."""
+    file_errors: list[str] = []
+    cwd = Path(os.getcwd())
+    for file_ in file_list:
+        if cwd.joinpath(file_).is_file() is False:
+            file_errors.append(file_)
+    return sorted(file_errors)
+
+
+async def async_restore_files(file_list: Iterable[str]) -> None:
+    if not file_list:
+        return
+    process = await asyncio.create_subprocess_shell(
+        f"git restore -- {' '.join(file_list)}",
+    )
+    await process.communicate()
+
+
+async def async_check_uncommitted_changes(file_list: Iterable[str]) -> bool:
+    """Check for uncommitted changes.
+
+    Returns:
+        False: if changes still need to be committed
+    """
+    process = await asyncio.create_subprocess_shell(
+        "git diff-index --name-only HEAD --",
+        stdout=asyncio.subprocess.PIPE,
+    )
+    stdout, _ = await process.communicate()
+    files_uncommitted: set[str] = {file_ for item in stdout.decode().split('\n')
+                                   if (file_ := item.strip())}
+    return not any(True for file_ in file_list if file_ in files_uncommitted)
+
+
+def check_comment_between_imports(fp: TextIO) -> FileStatus:
+    """Return True if comment is found between imports.
+
+    Sign that the file can't be updated automatically.
+    """
+    flag_in_import_block: bool = False
+    flag_multiple_imports: bool = False
+    flag_typing_import: bool = False
+    token_name_count: int = 0
+    line_first_import: int | None = None
+    line_last_import: int = 0
+    line_comments: list[tuple[int, int]] = []
+    return_value: FileStatus = FileStatus.CLEAR
+
+    tokens = tokenize.generate_tokens(fp.readline)
+    while True:
+        try:
+            t = next(tokens)
+            if flag_in_import_block is True:
+                if t.type == token.NAME and token_name_count == 0:
+                    token_name_count += 1
+                    if t.string == 'typing':
+                        flag_typing_import = True
+                elif t.type == token.NEWLINE:
+                    flag_in_import_block = False
+                    flag_multiple_imports = False
+                    flag_typing_import = False
+                    token_name_count = 0
+                elif t.type == token.OP and t.string != '.':
+                    flag_multiple_imports = True
+                elif t.type == token.COMMENT:
+                    if flag_typing_import is True:
+                        return_value = return_value | FileStatus.COMMENT | FileStatus.COMMENT_TYPING
+                    elif flag_multiple_imports is True:
+                        # Comment in same line as import statement
+                        return_value = return_value | FileStatus.COMMENT
+                continue
+            if t.type == token.NAME:
+                if t.string in ('import', 'from'):
+                    flag_in_import_block = True
+                    if line_first_import is None:
+                        line_first_import = t.start[0]
+                    line_last_import = t.start[0]
+                else:
+                    # Any other code block,
+                    # not in main import block anymore
+                    break
+            elif t.type in (token.COMMENT, token.STRING):
+                line_comments.append((t.type, t.start[0]))
+        except StopIteration:
+            break
+
+    if return_value != FileStatus.CLEAR:
+        # If inline comment was detected, stop here
+        return return_value
+
+    for _, line_number in line_comments:
+        if line_first_import is None:
+            # No import block detected
+            return FileStatus.CLEAR
+        if (line_first_import < line_number < line_last_import):
+            # Report all comments in the main import block
+            return FileStatus.COMMENT
+    return FileStatus.CLEAR

setup.cfg

@@ -7,9 +7,14 @@ license = MIT
 
 [flake8]
 ignore = E203,E231,E701,W503,W504
-max-line-length = 119
 # E203,E231,E701 -> walrus
 # W503/504 -> line-break-before/after-binary-operator
+max-line-length = 119
+extend-exclude =
+    .github,
+    .vscode,
+    tests/fixtures,
+    venv*,
 
 [tool:pytest]
 testpaths = tests
@@ -32,6 +37,7 @@ combine_as_imports = true
 python_version = 3.8
 ignore_missing_imports = true
 follow_imports = normal
+exclude = tests/fixtures/.+\.py
 # Untyped definitions and calls
 disallow_untyped_calls = true
 disallow_untyped_defs = true