| # Copyright (C) 2018 and later: Unicode, Inc. and others. | |
| # License & terms of use: http://www.unicode.org/copyright.html | |
| # Python 2/3 Compatibility (ICU-20299) | |
| # TODO(ICU-20301): Remove this. | |
| from __future__ import print_function | |
| from abc import abstractmethod | |
| from collections import defaultdict | |
| import re | |
| import sys | |
| from . import * | |
| from . import utils | |
| from .request_types import * | |
| # Note: for this to be a proper abstract class, it should extend abc.ABC. | |
| # There is no nice way to do this that works in both Python 2 and 3. | |
| # TODO(ICU-20301): Make this inherit from abc.ABC. | |
| class Filter(object): | |
| @staticmethod | |
| def create_from_json(json_data, io): | |
| assert io != None | |
| if "filterType" in json_data: | |
| filter_type = json_data["filterType"] | |
| else: | |
| filter_type = "file-stem" | |
| if filter_type == "file-stem": | |
| return FileStemFilter(json_data) | |
| elif filter_type == "language": | |
| return LanguageFilter(json_data) | |
| elif filter_type == "regex": | |
| return RegexFilter(json_data) | |
| elif filter_type == "exclude": | |
| return ExclusionFilter() | |
| elif filter_type == "union": | |
| return UnionFilter(json_data, io) | |
| elif filter_type == "locale": | |
| return LocaleFilter(json_data, io) | |
| else: | |
| print("Error: Unknown filterType option: %s" % filter_type, file=sys.stderr) | |
| return None | |
| def filter(self, request): | |
| if not request.apply_file_filter(self): | |
| return [] | |
| for file in request.all_input_files(): | |
| assert self.match(file) | |
| return [request] | |
| @staticmethod | |
| def _file_to_file_stem(file): | |
| start = file.filename.rfind("/") | |
| limit = file.filename.rfind(".") | |
| return file.filename[start+1:limit] | |
| @staticmethod | |
| def _file_to_subdir(file): | |
| limit = file.filename.rfind("/") | |
| if limit == -1: | |
| return None | |
| return file.filename[:limit] | |
| @abstractmethod | |
| def match(self, file): | |
| pass | |
| class InclusionFilter(Filter): | |
| def match(self, file): | |
| return True | |
| class ExclusionFilter(Filter): | |
| def match(self, file): | |
| return False | |
| class IncludeExcludeFilter(Filter): | |
| def __init__(self, json_data): | |
| if "whitelist" in json_data: | |
| self.is_includelist = True | |
| self.includelist = json_data["whitelist"] | |
| elif "includelist" in json_data: | |
| self.is_includelist = True | |
| self.includelist = json_data["includelist"] | |
| elif "blacklist" in json_data: | |
| self.is_includelist = False | |
| self.excludelist = json_data["blacklist"] | |
| elif "excludelist" in json_data: | |
| self.is_includelist = False | |
| self.excludelist = json_data["excludelist"] | |
| else: | |
| raise AssertionError("Need either includelist or excludelist: %s" % str(json_data)) | |
| def match(self, file): | |
| file_stem = self._file_to_file_stem(file) | |
| return self._should_include(file_stem) | |
| @abstractmethod | |
| def _should_include(self, file_stem): | |
| pass | |
| class FileStemFilter(IncludeExcludeFilter): | |
| def _should_include(self, file_stem): | |
| if self.is_includelist: | |
| return file_stem in self.includelist | |
| else: | |
| return file_stem not in self.excludelist | |
| class LanguageFilter(IncludeExcludeFilter): | |
| def _should_include(self, file_stem): | |
| language = file_stem.split("_")[0] | |
| if language == "root": | |
| # Always include root.txt | |
| return True | |
| if self.is_includelist: | |
| return language in self.includelist | |
| else: | |
| return language not in self.excludelist | |
| class RegexFilter(IncludeExcludeFilter): | |
| def __init__(self, *args): | |
| # TODO(ICU-20301): Change this to: super().__init__(*args) | |
| super(RegexFilter, self).__init__(*args) | |
| if self.is_includelist: | |
| self.includelist = [re.compile(pat) for pat in self.includelist] | |
| else: | |
| self.excludelist = [re.compile(pat) for pat in self.excludelist] | |
| def _should_include(self, file_stem): | |
| if self.is_includelist: | |
| for pattern in self.includelist: | |
| if pattern.match(file_stem): | |
| return True | |
| return False | |
| else: | |
| for pattern in self.excludelist: | |
| if pattern.match(file_stem): | |
| return False | |
| return True | |
| class UnionFilter(Filter): | |
| def __init__(self, json_data, io): | |
| # Collect the sub-filters. | |
| self.sub_filters = [] | |
| for filter_json in json_data["unionOf"]: | |
| self.sub_filters.append(Filter.create_from_json(filter_json, io)) | |
| def match(self, file): | |
| """Match iff any of the sub-filters match.""" | |
| for filter in self.sub_filters: | |
| if filter.match(file): | |
| return True | |
| return False | |
| LANGUAGE_SCRIPT_REGEX = re.compile(r"^([a-z]{2,3})_[A-Z][a-z]{3}$") | |
| LANGUAGE_ONLY_REGEX = re.compile(r"^[a-z]{2,3}$") | |
| class LocaleFilter(Filter): | |
| def __init__(self, json_data, io): | |
| if "whitelist" in json_data: | |
| self.locales_requested = list(json_data["whitelist"]) | |
| elif "includelist" in json_data: | |
| self.locales_requested = list(json_data["includelist"]) | |
| else: | |
| raise AssertionError("You must have an includelist in a locale filter") | |
| self.include_children = json_data.get("includeChildren", True) | |
| self.include_scripts = json_data.get("includeScripts", False) | |
| # Load the dependency graph from disk | |
| self.dependency_data_by_tree = { | |
| tree: io.read_locale_deps(tree) | |
| for tree in utils.ALL_TREES | |
| } | |
| def match(self, file): | |
| tree = self._file_to_subdir(file) | |
| assert tree is not None | |
| locale = self._file_to_file_stem(file) | |
| # A locale is *required* if it is *requested* or an ancestor of a | |
| # *requested* locale. | |
| if locale in self._locales_required(tree): | |
| return True | |
| # Resolve include_scripts and include_children. | |
| return self._match_recursive(locale, tree) | |
| def _match_recursive(self, locale, tree): | |
| # Base case: return True if we reached a *requested* locale, | |
| # or False if we ascend out of the locale tree. | |
| if locale is None: | |
| return False | |
| if locale in self.locales_requested: | |
| return True | |
| # Check for alternative scripts. | |
| # This causes sr_Latn to check sr instead of going directly to root. | |
| if self.include_scripts: | |
| match = LANGUAGE_SCRIPT_REGEX.match(locale) | |
| if match and self._match_recursive(match.group(1), tree): | |
| return True | |
| # Check if we are a descendant of a *requested* locale. | |
| if self.include_children: | |
| parent = self._get_parent_locale(locale, tree) | |
| if self._match_recursive(parent, tree): | |
| return True | |
| # No matches. | |
| return False | |
| def _get_parent_locale(self, locale, tree): | |
| """Gets the parent locale in the given tree, according to dependency data.""" | |
| dependency_data = self.dependency_data_by_tree[tree] | |
| if "parents" in dependency_data and locale in dependency_data["parents"]: | |
| return dependency_data["parents"][locale] | |
| if "aliases" in dependency_data and locale in dependency_data["aliases"]: | |
| return dependency_data["aliases"][locale] | |
| if LANGUAGE_ONLY_REGEX.match(locale): | |
| return "root" | |
| i = locale.rfind("_") | |
| if i < 0: | |
| assert locale == "root", "Invalid locale: %s/%s" % (tree, locale) | |
| return None | |
| return locale[:i] | |
| def _locales_required(self, tree): | |
| """Returns a generator of all required locales in the given tree.""" | |
| for locale in self.locales_requested: | |
| while locale is not None: | |
| yield locale | |
| locale = self._get_parent_locale(locale, tree) | |
| def apply_filters(requests, config, io): | |
| """Runs the filters and returns a new list of requests.""" | |
| requests = _apply_file_filters(requests, config, io) | |
| requests = _apply_resource_filters(requests, config, io) | |
| return requests | |
| def _apply_file_filters(old_requests, config, io): | |
| """Filters out entire files.""" | |
| filters = _preprocess_file_filters(old_requests, config, io) | |
| new_requests = [] | |
| for request in old_requests: | |
| category = request.category | |
| if category in filters: | |
| new_requests += filters[category].filter(request) | |
| else: | |
| new_requests.append(request) | |
| return new_requests | |
| def _preprocess_file_filters(requests, config, io): | |
| all_categories = set( | |
| request.category | |
| for request in requests | |
| ) | |
| all_categories.remove(None) | |
| all_categories = list(sorted(all_categories)) | |
| json_data = config.filters_json_data | |
| filters = {} | |
| default_filter_json = "exclude" if config.strategy == "additive" else "include" | |
| for category in all_categories: | |
| filter_json = default_filter_json | |
| # Figure out the correct filter to create | |
| if "featureFilters" in json_data and category in json_data["featureFilters"]: | |
| filter_json = json_data["featureFilters"][category] | |
| if filter_json == "include" and "localeFilter" in json_data and category.endswith("_tree"): | |
| filter_json = json_data["localeFilter"] | |
| # Resolve the filter JSON into a filter object | |
| if filter_json == "exclude": | |
| filters[category] = ExclusionFilter() | |
| elif filter_json == "include": | |
| pass # no-op | |
| else: | |
| filters[category] = Filter.create_from_json(filter_json, io) | |
| if "featureFilters" in json_data: | |
| for category in json_data["featureFilters"]: | |
| if category not in all_categories: | |
| print("Warning: category %s is not known" % category, file=sys.stderr) | |
| return filters | |
| class ResourceFilterInfo(object): | |
| def __init__(self, category, strategy): | |
| self.category = category | |
| self.strategy = strategy | |
| self.filter_tmp_dir = "filters/%s" % category | |
| self.input_files = None | |
| self.filter_files = None | |
| self.rules_by_file = None | |
| def apply_to_requests(self, all_requests): | |
| # Call this method only once per list of requests. | |
| assert self.input_files is None | |
| for request in all_requests: | |
| if request.category != self.category: | |
| continue | |
| if not isinstance(request, AbstractExecutionRequest): | |
| continue | |
| if request.tool != IcuTool("genrb"): | |
| continue | |
| if not request.input_files: | |
| continue | |
| self._set_files(request.input_files) | |
| request.dep_targets += [self.filter_files[:]] | |
| arg_str = "--filterDir {TMP_DIR}/%s" % self.filter_tmp_dir | |
| request.args = "%s %s" % (arg_str, request.args) | |
| # Make sure we found the target request | |
| if self.input_files is None: | |
| print("WARNING: Category not found: %s" % self.category, file=sys.stderr) | |
| self.input_files = [] | |
| self.filter_files = [] | |
| self.rules_by_file = [] | |
| def _set_files(self, files): | |
| # Note: The input files to genrb for a certain category should always | |
| # be the same. For example, there are often two genrb calls: one for | |
| # --writePoolBundle, and the other for --usePoolBundle. They are both | |
| # expected to have the same list of input files. | |
| if self.input_files is not None: | |
| assert self.input_files == files | |
| return | |
| self.input_files = list(files) | |
| self.filter_files = [ | |
| TmpFile("%s/%s" % (self.filter_tmp_dir, basename)) | |
| for basename in ( | |
| file.filename[file.filename.rfind("/")+1:] | |
| for file in files | |
| ) | |
| ] | |
| if self.strategy == "additive": | |
| self.rules_by_file = [ | |
| [r"-/", r"+/%%ALIAS", r"+/%%Parent"] | |
| for _ in range(len(files)) | |
| ] | |
| else: | |
| self.rules_by_file = [ | |
| [r"+/"] | |
| for _ in range(len(files)) | |
| ] | |
| def add_rules(self, file_filter, rules): | |
| for file, rule_list in zip(self.input_files, self.rules_by_file): | |
| if file_filter.match(file): | |
| rule_list += rules | |
| def make_requests(self): | |
| # Map from rule list to filter files with that rule list | |
| unique_rules = defaultdict(list) | |
| for filter_file, rules in zip(self.filter_files, self.rules_by_file): | |
| unique_rules[tuple(rules)].append(filter_file) | |
| new_requests = [] | |
| i = 0 | |
| for rules, filter_files in unique_rules.items(): | |
| base_filter_file = filter_files[0] | |
| new_requests += [ | |
| PrintFileRequest( | |
| name = "%s_print_%d" % (self.category, i), | |
| output_file = base_filter_file, | |
| content = self._generate_resource_filter_txt(rules) | |
| ) | |
| ] | |
| i += 1 | |
| for filter_file in filter_files[1:]: | |
| new_requests += [ | |
| CopyRequest( | |
| name = "%s_copy_%d" % (self.category, i), | |
| input_file = base_filter_file, | |
| output_file = filter_file | |
| ) | |
| ] | |
| i += 1 | |
| return new_requests | |
| @staticmethod | |
| def _generate_resource_filter_txt(rules): | |
| result = "# Caution: This file is automatically generated\n\n" | |
| result += "\n".join(rules) | |
| return result | |
| def _apply_resource_filters(all_requests, config, io): | |
| """Creates filters for looking within resource bundle files.""" | |
| json_data = config.filters_json_data | |
| if "resourceFilters" not in json_data: | |
| return all_requests | |
| collected = {} | |
| for entry in json_data["resourceFilters"]: | |
| if "files" in entry: | |
| file_filter = Filter.create_from_json(entry["files"], io) | |
| else: | |
| file_filter = InclusionFilter() | |
| for category in entry["categories"]: | |
| # not defaultdict because we need to pass arguments to the constructor | |
| if category not in collected: | |
| filter_info = ResourceFilterInfo(category, config.strategy) | |
| filter_info.apply_to_requests(all_requests) | |
| collected[category] = filter_info | |
| else: | |
| filter_info = collected[category] | |
| filter_info.add_rules(file_filter, entry["rules"]) | |
| # Add the filter generation requests to the beginning so that by default | |
| # they are made before genrb gets run (order is required by windirect) | |
| new_requests = [] | |
| for filter_info in collected.values(): | |
| new_requests += filter_info.make_requests() | |
| new_requests += all_requests | |
| return new_requests |