| # Copyright (C) 2020 Red Hat Inc. |
| # |
| # Authors: |
| # Eduardo Habkost <ehabkost@redhat.com> |
| # |
| # This work is licensed under the terms of the GNU GPL, version 2. See |
| # the COPYING file in the top-level directory. |
| from typing import IO, Match, NamedTuple, Optional, Literal, Iterable, Type, Dict, List, Any, TypeVar, NewType, Tuple, Union |
| from pathlib import Path |
| from itertools import chain |
| from tempfile import NamedTemporaryFile |
| import os |
| import re |
| import subprocess |
| from io import StringIO |
| |
| import logging |
| logger = logging.getLogger(__name__) |
| DBG = logger.debug |
| INFO = logger.info |
| WARN = logger.warning |
| ERROR = logger.error |
| |
| from .utils import * |
| |
| T = TypeVar('T') |
| |
| class Patch(NamedTuple): |
| # start inside file.original_content |
| start: int |
| # end position inside file.original_content |
| end: int |
| # replacement string for file.original_content[start:end] |
| replacement: str |
| |
| IdentifierType = Literal['type', 'symbol', 'include', 'constant'] |
| class RequiredIdentifier(NamedTuple): |
| type: IdentifierType |
| name: str |
| |
| class FileMatch: |
| """Base class for regex matches |
| |
| Subclasses just need to set the `regexp` class attribute |
| """ |
| regexp: Optional[str] = None |
| |
| def __init__(self, f: 'FileInfo', m: Match) -> None: |
| self.file: 'FileInfo' = f |
| self.match: Match[str] = m |
| |
| @property |
| def name(self) -> str: |
| if 'name' not in self.match.groupdict(): |
| return '[no name]' |
| return self.group('name') |
| |
| @classmethod |
| def compiled_re(klass): |
| return re.compile(klass.regexp, re.MULTILINE) |
| |
| def start(self) -> int: |
| return self.match.start() |
| |
| def end(self) -> int: |
| return self.match.end() |
| |
| def line_col(self) -> LineAndColumn: |
| return self.file.line_col(self.start()) |
| |
| def group(self, group: Union[int, str]) -> str: |
| return self.match.group(group) |
| |
| def getgroup(self, group: str) -> Optional[str]: |
| if group not in self.match.groupdict(): |
| return None |
| return self.match.group(group) |
| |
| def log(self, level, fmt, *args) -> None: |
| pos = self.line_col() |
| logger.log(level, '%s:%d:%d: '+fmt, self.file.filename, pos.line, pos.col, *args) |
| |
| def debug(self, fmt, *args) -> None: |
| self.log(logging.DEBUG, fmt, *args) |
| |
| def info(self, fmt, *args) -> None: |
| self.log(logging.INFO, fmt, *args) |
| |
| def warn(self, fmt, *args) -> None: |
| self.log(logging.WARNING, fmt, *args) |
| |
| def error(self, fmt, *args) -> None: |
| self.log(logging.ERROR, fmt, *args) |
| |
| def sub(self, original: str, replacement: str) -> str: |
| """Replace content |
| |
| XXX: this won't use the match position, but will just |
| replace all strings that look like the original match. |
| This should be enough for all the patterns used in this |
| script. |
| """ |
| return original.replace(self.group(0), replacement) |
| |
| def sanity_check(self) -> None: |
| """Sanity check match, and print warnings if necessary""" |
| pass |
| |
| def replacement(self) -> Optional[str]: |
| """Return replacement text for pattern, to use new code conventions""" |
| return None |
| |
| def make_patch(self, replacement: str) -> 'Patch': |
| """Make patch replacing the content of this match""" |
| return Patch(self.start(), self.end(), replacement) |
| |
| def make_subpatch(self, start: int, end: int, replacement: str) -> 'Patch': |
| return Patch(self.start() + start, self.start() + end, replacement) |
| |
| def make_removal_patch(self) -> 'Patch': |
| """Make patch removing contents of match completely""" |
| return self.make_patch('') |
| |
| def append(self, s: str) -> 'Patch': |
| """Make patch appending string after this match""" |
| return Patch(self.end(), self.end(), s) |
| |
| def prepend(self, s: str) -> 'Patch': |
| """Make patch prepending string before this match""" |
| return Patch(self.start(), self.start(), s) |
| |
| def gen_patches(self) -> Iterable['Patch']: |
| """Patch source code contents to use new code patterns""" |
| replacement = self.replacement() |
| if replacement is not None: |
| yield self.make_patch(replacement) |
| |
| @classmethod |
| def has_replacement_rule(klass) -> bool: |
| return (klass.gen_patches is not FileMatch.gen_patches |
| or klass.replacement is not FileMatch.replacement) |
| |
| def contains(self, other: 'FileMatch') -> bool: |
| return other.start() >= self.start() and other.end() <= self.end() |
| |
| def __repr__(self) -> str: |
| start = self.file.line_col(self.start()) |
| end = self.file.line_col(self.end() - 1) |
| return '<%s %s at %d:%d-%d:%d: %r>' % (self.__class__.__name__, |
| self.name, |
| start.line, start.col, |
| end.line, end.col, self.group(0)[:100]) |
| |
| def required_identifiers(self) -> Iterable[RequiredIdentifier]: |
| """Can be implemented by subclasses to keep track of identifier references |
| |
| This method will be used by the code that moves declarations around the file, |
| to make sure we find the right spot for them. |
| """ |
| raise NotImplementedError() |
| |
| def provided_identifiers(self) -> Iterable[RequiredIdentifier]: |
| """Can be implemented by subclasses to keep track of identifier references |
| |
| This method will be used by the code that moves declarations around the file, |
| to make sure we find the right spot for them. |
| """ |
| raise NotImplementedError() |
| |
| @classmethod |
| def finditer(klass, content: str, pos=0, endpos=-1) -> Iterable[Match]: |
| """Helper for re.finditer()""" |
| if endpos >= 0: |
| content = content[:endpos] |
| return klass.compiled_re().finditer(content, pos) |
| |
| @classmethod |
| def domatch(klass, content: str, pos=0, endpos=-1) -> Optional[Match]: |
| """Helper for re.match()""" |
| if endpos >= 0: |
| content = content[:endpos] |
| return klass.compiled_re().match(content, pos) |
| |
| def group_finditer(self, klass: Type['FileMatch'], group: Union[str, int]) -> Iterable['FileMatch']: |
| assert self.file.original_content |
| return (klass(self.file, m) |
| for m in klass.finditer(self.file.original_content, |
| self.match.start(group), |
| self.match.end(group))) |
| |
| def try_group_match(self, klass: Type['FileMatch'], group: Union[str, int]) -> Optional['FileMatch']: |
| assert self.file.original_content |
| m = klass.domatch(self.file.original_content, |
| self.match.start(group), |
| self.match.end(group)) |
| if not m: |
| return None |
| else: |
| return klass(self.file, m) |
| |
| def group_match(self, group: Union[str, int]) -> 'FileMatch': |
| m = self.try_group_match(FullMatch, group) |
| assert m |
| return m |
| |
| @property |
| def allfiles(self) -> 'FileList': |
| return self.file.allfiles |
| |
| class FullMatch(FileMatch): |
| """Regexp that will match all contents of string |
| Useful when used with group_match() |
| """ |
| regexp = r'(?s).*' # (?s) is re.DOTALL |
| |
| def all_subclasses(c: Type[FileMatch]) -> Iterable[Type[FileMatch]]: |
| for sc in c.__subclasses__(): |
| yield sc |
| yield from all_subclasses(sc) |
| |
| def match_class_dict() -> Dict[str, Type[FileMatch]]: |
| d = dict((t.__name__, t) for t in all_subclasses(FileMatch)) |
| return d |
| |
| def names(matches: Iterable[FileMatch]) -> Iterable[str]: |
| return [m.name for m in matches] |
| |
| class PatchingError(Exception): |
| pass |
| |
| class OverLappingPatchesError(PatchingError): |
| pass |
| |
| def apply_patches(s: str, patches: Iterable[Patch]) -> str: |
| """Apply a sequence of patches to string |
| |
| >>> apply_patches('abcdefg', [Patch(2,2,'xxx'), Patch(0, 1, 'yy')]) |
| 'yybxxxcdefg' |
| """ |
| r = StringIO() |
| last = 0 |
| def patch_sort_key(item: Tuple[int, Patch]) -> Tuple[int, int, int]: |
| """Patches are sorted by byte position, |
| patches at the same byte position are applied in the order |
| they were generated. |
| """ |
| i,p = item |
| return (p.start, p.end, i) |
| |
| for i,p in sorted(enumerate(patches), key=patch_sort_key): |
| DBG("Applying patch at position %d (%s) - %d (%s): %r", |
| p.start, line_col(s, p.start), |
| p.end, line_col(s, p.end), |
| p.replacement) |
| if last > p.start: |
| raise OverLappingPatchesError("Overlapping patch at position %d (%s), last patch at %d (%s)" % \ |
| (p.start, line_col(s, p.start), last, line_col(s, last))) |
| r.write(s[last:p.start]) |
| r.write(p.replacement) |
| last = p.end |
| r.write(s[last:]) |
| return r.getvalue() |
| |
| class RegexpScanner: |
| def __init__(self) -> None: |
| self.match_index: Dict[Type[Any], List[FileMatch]] = {} |
| self.match_name_index: Dict[Tuple[Type[Any], str, str], Optional[FileMatch]] = {} |
| |
| def _matches_of_type(self, klass: Type[Any]) -> Iterable[FileMatch]: |
| raise NotImplementedError() |
| |
| def matches_of_type(self, t: Type[T]) -> List[T]: |
| if t not in self.match_index: |
| self.match_index[t] = list(self._matches_of_type(t)) |
| return self.match_index[t] # type: ignore |
| |
| def find_matches(self, t: Type[T], name: str, group: str='name') -> List[T]: |
| indexkey = (t, name, group) |
| if indexkey in self.match_name_index: |
| return self.match_name_index[indexkey] # type: ignore |
| r: List[T] = [] |
| for m in self.matches_of_type(t): |
| assert isinstance(m, FileMatch) |
| if m.getgroup(group) == name: |
| r.append(m) # type: ignore |
| self.match_name_index[indexkey] = r # type: ignore |
| return r |
| |
| def find_match(self, t: Type[T], name: str, group: str='name') -> Optional[T]: |
| l = self.find_matches(t, name, group) |
| if not l: |
| return None |
| if len(l) > 1: |
| logger.warn("multiple matches found for %r (%s=%r)", t, group, name) |
| return None |
| return l[0] |
| |
| def reset_index(self) -> None: |
| self.match_index.clear() |
| self.match_name_index.clear() |
| |
| class FileInfo(RegexpScanner): |
| filename: Path |
| original_content: Optional[str] = None |
| |
| def __init__(self, files: 'FileList', filename: os.PathLike, force:bool=False) -> None: |
| super().__init__() |
| self.allfiles = files |
| self.filename = Path(filename) |
| self.patches: List[Patch] = [] |
| self.force = force |
| |
| def __repr__(self) -> str: |
| return f'<FileInfo {repr(self.filename)}>' |
| |
| def filename_matches(self, name: str) -> bool: |
| nameparts = Path(name).parts |
| return self.filename.parts[-len(nameparts):] == nameparts |
| |
| def line_col(self, start: int) -> LineAndColumn: |
| """Return line and column for a match object inside original_content""" |
| return line_col(self.original_content, start) |
| |
| def _matches_of_type(self, klass: Type[Any]) -> List[FileMatch]: |
| """Build FileMatch objects for each match of regexp""" |
| if not hasattr(klass, 'regexp') or klass.regexp is None: |
| return [] |
| assert hasattr(klass, 'regexp') |
| DBG("%s: scanning for %s", self.filename, klass.__name__) |
| DBG("regexp: %s", klass.regexp) |
| matches = [klass(self, m) for m in klass.finditer(self.original_content)] |
| DBG('%s: %d matches found for %s: %s', self.filename, len(matches), |
| klass.__name__,' '.join(names(matches))) |
| return matches |
| |
| def find_match(self, t: Type[T], name: str, group: str='name') -> Optional[T]: |
| for m in self.matches_of_type(t): |
| assert isinstance(m, FileMatch) |
| if m.getgroup(group) == name: |
| return m # type: ignore |
| return None |
| |
| def reset_content(self, s:str): |
| self.original_content = s |
| self.patches.clear() |
| self.reset_index() |
| self.allfiles.reset_index() |
| |
| def load(self) -> None: |
| if self.original_content is not None: |
| return |
| with open(self.filename, 'rt') as f: |
| self.reset_content(f.read()) |
| |
| @property |
| def all_matches(self) -> Iterable[FileMatch]: |
| lists = list(self.match_index.values()) |
| return (m for l in lists |
| for m in l) |
| |
| def gen_patches(self, matches: List[FileMatch]) -> None: |
| for m in matches: |
| DBG("Generating patches for %r", m) |
| for i,p in enumerate(m.gen_patches()): |
| DBG("patch %d generated by %r:", i, m) |
| DBG("replace contents at %s-%s with %r", |
| self.line_col(p.start), self.line_col(p.end), p.replacement) |
| self.patches.append(p) |
| |
| def scan_for_matches(self, class_names: Optional[List[str]]=None) -> Iterable[FileMatch]: |
| DBG("class names: %r", class_names) |
| class_dict = match_class_dict() |
| if class_names is None: |
| DBG("default class names") |
| class_names = list(name for name,klass in class_dict.items() |
| if klass.has_replacement_rule()) |
| DBG("class_names: %r", class_names) |
| for cn in class_names: |
| matches = self.matches_of_type(class_dict[cn]) |
| DBG('%d matches found for %s: %s', |
| len(matches), cn, ' '.join(names(matches))) |
| yield from matches |
| |
| def apply_patches(self) -> None: |
| """Replace self.original_content after applying patches from self.patches""" |
| self.reset_content(self.get_patched_content()) |
| |
| def get_patched_content(self) -> str: |
| assert self.original_content is not None |
| return apply_patches(self.original_content, self.patches) |
| |
| def write_to_file(self, f: IO[str]) -> None: |
| f.write(self.get_patched_content()) |
| |
| def write_to_filename(self, filename: os.PathLike) -> None: |
| with open(filename, 'wt') as of: |
| self.write_to_file(of) |
| |
| def patch_inplace(self) -> None: |
| newfile = self.filename.with_suffix('.changed') |
| self.write_to_filename(newfile) |
| os.rename(newfile, self.filename) |
| |
| def show_diff(self) -> None: |
| with NamedTemporaryFile('wt') as f: |
| self.write_to_file(f) |
| f.flush() |
| subprocess.call(['diff', '-u', self.filename, f.name]) |
| |
| def ref(self): |
| return TypeInfoReference |
| |
| class FileList(RegexpScanner): |
| def __init__(self): |
| super().__init__() |
| self.files: List[FileInfo] = [] |
| |
| def extend(self, *args, **kwargs): |
| self.files.extend(*args, **kwargs) |
| |
| def __iter__(self): |
| return iter(self.files) |
| |
| def _matches_of_type(self, klass: Type[Any]) -> Iterable[FileMatch]: |
| return chain(*(f._matches_of_type(klass) for f in self.files)) |
| |
| def find_file(self, name: str) -> Optional[FileInfo]: |
| """Get file with path ending with @name""" |
| for f in self.files: |
| if f.filename_matches(name): |
| return f |
| else: |
| return None |
| |
| def one_pass(self, class_names: List[str]) -> int: |
| total_patches = 0 |
| for f in self.files: |
| INFO("Scanning file %s", f.filename) |
| matches = list(f.scan_for_matches(class_names)) |
| INFO("Generating patches for file %s", f.filename) |
| f.gen_patches(matches) |
| total_patches += len(f.patches) |
| if total_patches: |
| for f in self.files: |
| try: |
| f.apply_patches() |
| except PatchingError: |
| logger.exception("%s: failed to patch file", f.filename) |
| return total_patches |
| |
| def patch_content(self, max_passes, class_names: List[str]) -> None: |
| """Multi-pass content patching loop |
| |
| We run multiple passes because there are rules that will |
| delete init functions once they become empty. |
| """ |
| passes = 0 |
| total_patches = 0 |
| DBG("max_passes: %r", max_passes) |
| while not max_passes or max_passes <= 0 or passes < max_passes: |
| passes += 1 |
| INFO("Running pass: %d", passes) |
| count = self.one_pass(class_names) |
| DBG("patch content: pass %d: %d patches generated", passes, count) |
| total_patches += count |
| DBG("%d patches applied total in %d passes", total_patches, passes) |
