# Copyright (c) ONNX Project Contributors # SPDX-License-Identifier: Apache-2.0 from __future__ import annotations import locale as pylocale import unicodedata import warnings import numpy as np from onnx.reference.op_run import OpRun, RuntimeTypeError class StringNormalizer(OpRun): """The operator is not really threadsafe as python cannot play with two locales at the same time. stop words should not be implemented here as the tokenization usually happens after this steps. """ def _run( self, x, case_change_action=None, is_case_sensitive=None, locale=None, stopwords=None, ): slocale = locale if stopwords is None: raw_stops = set() stops = set() else: raw_stops = set(stopwords) if case_change_action == "LOWER": stops = {w.lower() for w in stopwords} elif case_change_action == "UPPER": stops = {w.upper() for w in stopwords} else: stops = set(stopwords) res = np.empty(x.shape, dtype=x.dtype) if len(x.shape) == 2: for i in range(x.shape[1]): self._run_column( x[:, i], res[:, i], slocale=slocale, stops=stops, raw_stops=raw_stops, is_case_sensitive=is_case_sensitive, case_change_action=case_change_action, ) elif len(x.shape) == 1: self._run_column( x, res, slocale=slocale, stops=stops, raw_stops=raw_stops, is_case_sensitive=is_case_sensitive, case_change_action=case_change_action, ) else: raise RuntimeTypeError("x must be a matrix or a vector.") if len(res.shape) == 2 and res.shape[0] == 1: res = np.array([[w for w in res.tolist()[0] if len(w) > 0]]) if res.shape[1] == 0: res = np.array([[""]]) elif len(res.shape) == 1: res = np.array([w for w in res.tolist() if len(w) > 0]) if len(res) == 0: res = np.array([""]) return (res,) @staticmethod def _run_column( cin, cout, slocale=None, stops=None, raw_stops=None, is_case_sensitive=None, case_change_action=None, ): if pylocale.getlocale() != slocale: try: pylocale.setlocale(pylocale.LC_ALL, slocale) except pylocale.Error as e: warnings.warn( f"Unknown local setting {slocale!r} (current: {pylocale.getlocale()!r}) - {e!r}.", stacklevel=1, ) cout[:] = cin[:] for i in range(cin.shape[0]): if isinstance(cout[i], float): # nan cout[i] = "" else: cout[i] = StringNormalizer.strip_accents_unicode(cout[i]) if is_case_sensitive and len(stops) > 0: for i in range(cin.shape[0]): cout[i] = StringNormalizer._remove_stopwords(cout[i], raw_stops) if case_change_action == "LOWER": for i in range(cin.shape[0]): cout[i] = cout[i].lower() elif case_change_action == "UPPER": for i in range(cin.shape[0]): cout[i] = cout[i].upper() elif case_change_action != "NONE": raise RuntimeError( f"Unknown option for case_change_action: {case_change_action!r}." ) if not is_case_sensitive and len(stops) > 0: for i in range(cin.shape[0]): cout[i] = StringNormalizer._remove_stopwords(cout[i], stops) return cout @staticmethod def _remove_stopwords(text, stops): spl = text.split(" ") return " ".join(filter(lambda s: s not in stops, spl)) @staticmethod def strip_accents_unicode(s): """Transforms accentuated unicode symbols into their simple counterpart. Source: `sklearn/feature_extraction/text.py `_. Args: s: string The string to strip Returns: the cleaned string """ try: # If `s` is ASCII-compatible, then it does not contain any accented # characters and we can avoid an expensive list comprehension s.encode("ASCII", errors="strict") return s # noqa: TRY300 except UnicodeEncodeError: normalized = unicodedata.normalize("NFKD", s) return "".join([c for c in normalized if not unicodedata.combining(c)])