followthemoney.types.common

  1from inspect import cleandoc
  2from itertools import product
  3from babel.core import Locale
  4from banal import ensure_list
  5from normality import stringify
  6from typing import Any, Dict, Optional, Sequence, Callable, TYPE_CHECKING, TypedDict
  7
  8from followthemoney.rdf import Literal, Identifier
  9from followthemoney.util import get_locale
 10from followthemoney.util import gettext, sanitize_text
 11
 12if TYPE_CHECKING:
 13    from followthemoney.proxy import EntityProxy
 14
 15EnumValues = Dict[str, str]
 16
 17
 18class PropertyTypeToDict(TypedDict, total=False):
 19    label: str
 20    plural: str
 21    description: Optional[str]
 22    group: Optional[str]
 23    matchable: Optional[bool]
 24    pivot: Optional[bool]
 25    values: Optional[EnumValues]
 26
 27
 28class PropertyType(object):
 29    """Base class for all property types."""
 30
 31    name: str = "any"
 32    """A machine-facing, variable safe name for the given type."""
 33
 34    group: Optional[str] = None
 35    """Groups are used to invert all the properties of an entity that have a
 36    given  type into a single list before indexing them. This way, in Aleph,
 37    you can query for ``countries:gb`` instead of having to make a set of filters
 38    like ``properties.jurisdiction:gb OR properties.country:gb OR ...``."""
 39
 40    label: str = "Any"
 41    """A name for this type to be shown to users."""
 42
 43    plural: str = "Any"
 44    """A plural name for this type which can be used in appropriate places in
 45    a user interface."""
 46
 47    matchable: bool = True
 48    """Matchable types allow properties to be compared with each other in order to
 49    assess entity similarity. While it makes sense to compare names, countries or
 50    phone numbers, the same isn't true for raw JSON blobs or descriptive text
 51    snippets."""
 52
 53    pivot: bool = False
 54    """Pivot property types are like a stronger form of :attr:`~matchable` types:
 55    they will be used when value-based lookups are used to find commonalities
 56    between entities. For example, pivot typed-properties are used to show all the
 57    other entities that mention the same phone number, email address or name as the
 58    one currently seen by the user."""
 59
 60    max_size: Optional[int] = None
 61    """Some types have overall size limitations in place in order to avoid generating
 62    entities that are very large (upstream ElasticSearch has a 100MB document limit).
 63    Once the total size of all properties of this type has exceed the given limit,
 64    an entity will refuse to add further values."""
 65
 66    @property
 67    def docs(self) -> Optional[str]:
 68        if not self.__doc__:
 69            return None
 70
 71        return cleandoc(self.__doc__)
 72
 73    def validate(
 74        self, value: str, fuzzy: bool = False, format: Optional[str] = None
 75    ) -> bool:
 76        """Returns a boolean to indicate if the given value is a valid instance of
 77        the type."""
 78        cleaned = self.clean(value, fuzzy=fuzzy, format=format)
 79        return cleaned is not None
 80
 81    def clean(
 82        self,
 83        raw: Any,
 84        fuzzy: bool = False,
 85        format: Optional[str] = None,
 86        proxy: Optional["EntityProxy"] = None,
 87    ) -> Optional[str]:
 88        """Create a clean version of a value of the type, suitable for storage
 89        in an entity proxy."""
 90        text = sanitize_text(raw)
 91        if text is None:
 92            return None
 93        return self.clean_text(text, fuzzy=fuzzy, format=format, proxy=proxy)
 94
 95    def clean_text(
 96        self,
 97        text: str,
 98        fuzzy: bool = False,
 99        format: Optional[str] = None,
100        proxy: Optional["EntityProxy"] = None,
101    ) -> Optional[str]:
102        """Specific types can apply their own cleaning routines here (this is called
103        by ``clean`` after the value has been converted to a string and null values
104        have been filtered)."""
105        return text
106
107    def join(self, values: Sequence[str]) -> str:
108        """Helper function for converting multi-valued FtM data into formats that
109        allow only a single value per field (e.g. CSV). This is not fully reversible
110        and should be used as a last option."""
111        values = ensure_list(values)
112        return "; ".join(values)
113
114    def _specificity(self, value: str) -> float:
115        return 1.0
116
117    def specificity(self, value: Optional[str]) -> float:
118        """Return a score for how specific the given value is. This can be used as a
119        weighting factor in entity comparisons in order to rate matching property
120        values by how specific they are. For example: a longer address is considered
121        to be more specific than a short one, a full date more specific than just a
122        year number, etc."""
123        if not self.matchable or value is None:
124            return 0.0
125        return self._specificity(value)
126
127    def compare_safe(self, left: Optional[str], right: Optional[str]) -> float:
128        """Compare, but support None values on either side of the comparison."""
129        left = stringify(left)
130        right = stringify(right)
131        if left is None or right is None:
132            return 0.0
133        return self.compare(left, right)
134
135    def compare(self, left: str, right: str) -> float:
136        """Comparisons are a float between 0 and 1. They can assume
137        that the given data is cleaned, but not normalised."""
138        if left.lower() == right.lower():
139            return 1.0 * self.specificity(left)
140        return 0.0
141
142    def compare_sets(
143        self,
144        left: Sequence[str],
145        right: Sequence[str],
146        func: Callable[[Sequence[float]], float] = max,
147    ) -> float:
148        """Compare two sets of values and select the highest-scored result."""
149        results = []
150        for l, r in product(ensure_list(left), ensure_list(right)):
151            results.append(self.compare(l, r))
152        if not len(results):
153            return 0.0
154        return func(results)
155
156    def country_hint(self, value: str) -> Optional[str]:
157        """Determine if the given value allows us to infer a country that it may
158        be related to (e.g. using a country prefix on a phone number or IBAN)."""
159        return None
160
161    def rdf(self, value: str) -> Identifier:
162        """Return an RDF term to represent the given value - either a string
163        literal, or a URI reference."""
164        return Literal(value)
165
166    def pick(self, values: Sequence[str]) -> Optional[str]:
167        """Pick the best value to show to the user."""
168        raise NotImplemented
169
170    def node_id(self, value: str) -> Optional[str]:
171        """Return an ID suitable to identify this entity as a typed node in a
172        graph representation of some FtM data. It's usually the same as the the
173        RDF form."""
174        return str(self.rdf(value))
175
176    def node_id_safe(self, value: Optional[str]) -> Optional[str]:
177        """Wrapper for node_id to handle None values."""
178        if value is None:
179            return None
180        return self.node_id(value)
181
182    def caption(self, value: str) -> Optional[str]:
183        """Return a label for the given property value. This is often the same as the
184        value, but for types like countries or languages, it would return the label,
185        while other values like phone numbers can be formatted to be nicer to read."""
186        return value
187
188    def to_dict(self) -> PropertyTypeToDict:
189        """Return a serialisable description of this data type."""
190        data: PropertyTypeToDict = {
191            "label": gettext(self.label),
192            "plural": gettext(self.plural),
193            "description": gettext(self.docs),
194        }
195        if self.group:
196            data["group"] = self.group
197        if self.matchable:
198            data["matchable"] = True
199        if self.pivot:
200            data["pivot"] = True
201        return data
202
203    def __eq__(self, other: Any) -> bool:
204        if not isinstance(other, PropertyType):
205            return False
206        return self.name == other.name
207
208    def __hash__(self) -> int:
209        return hash(self.name)
210
211    def __str__(self) -> str:
212        return self.name
213
214    def __repr__(self) -> str:
215        return f"<{self.name}>"
216
217
218class EnumType(PropertyType):
219    """Enumerated type properties are used for types which have a defined set
220    of possible values, like languages and countries."""
221
222    def __init__(self) -> None:
223        self._names: Dict[Locale, EnumValues] = {}
224        self.codes = set(self.names.keys())
225
226    def _locale_names(self, locale: Locale) -> EnumValues:
227        return {}
228
229    @property
230    def names(self) -> EnumValues:
231        """Return a mapping from property values to their labels in the current
232        locale."""
233        locale = get_locale()
234        if locale not in self._names:
235            self._names[locale] = self._locale_names(locale)
236        return self._names[locale]
237
238    def validate(
239        self, value: str, fuzzy: bool = False, format: Optional[str] = None
240    ) -> bool:
241        """Make sure that the given code value is one of the supported set."""
242        if value is None:
243            return False
244        return str(value).lower().strip() in self.codes
245
246    def clean_text(
247        self,
248        code: str,
249        fuzzy: bool = False,
250        format: Optional[str] = None,
251        proxy: Optional["EntityProxy"] = None,
252    ) -> Optional[str]:
253        """All code values are cleaned to be lowercase and trailing whitespace is
254        removed."""
255        code = code.lower().strip()
256        if code not in self.codes:
257            return None
258        return code
259
260    def caption(self, value: str) -> str:
261        """Given a code value, return the label that should be shown to a user."""
262        return self.names.get(value, value)
263
264    def to_dict(self) -> PropertyTypeToDict:
265        """When serialising the model to JSON, include all values."""
266        data = super(EnumType, self).to_dict()
267        data["values"] = self.names
268        return data