followthemoney.graph

  1"""
  2Converting FtM data to a property graph data model.
  3
  4This module provides an abstract data object that represents a property
  5graph. This is used by the exporter modules to convert data
  6to a specific output format, like Cypher or NetworkX.
  7"""
  8import logging
  9from typing import Any, Dict, Generator, Iterable, List, Optional
 10
 11from followthemoney.types import registry
 12from followthemoney.types.common import PropertyType
 13from followthemoney.schema import Schema
 14from followthemoney.proxy import EntityProxy
 15from followthemoney.property import Property
 16from followthemoney.exc import InvalidModel
 17
 18log = logging.getLogger(__name__)
 19
 20
 21class Node(object):
 22    """A node represents either an entity that can be rendered as a
 23    node in a graph, or as a re-ified value, like a name, email
 24    address or phone number."""
 25
 26    __slots__ = ["type", "value", "id", "proxy", "schema"]
 27
 28    def __init__(
 29        self,
 30        type_: PropertyType,
 31        value: str,
 32        proxy: Optional[EntityProxy] = None,
 33        schema: Optional[Schema] = None,
 34    ) -> None:
 35        self.type = type_
 36        self.value = value
 37        # _id = type_.node_id_safe(value)
 38        # if _id is None:
 39        #     raise InvalidData("No ID for node")
 40        self.id = type_.node_id_safe(value)
 41        self.proxy = proxy
 42        self.schema = schema if proxy is None else proxy.schema
 43
 44    @property
 45    def is_entity(self) -> bool:
 46        """Check to see if the node represents an entity. If this is false, the
 47        node represents a non-entity property value that has been reified, like
 48        a phone number or a name."""
 49        return self.type == registry.entity
 50
 51    @property
 52    def caption(self) -> str:
 53        """A user-facing label for the current node."""
 54        if self.type == registry.entity and self.proxy is not None:
 55            return self.proxy.caption
 56        caption = self.type.caption(self.value)
 57        return caption or self.value
 58
 59    def to_dict(self) -> Dict[str, Any]:
 60        """Return a simple dictionary to reflect this graph node."""
 61        return {
 62            "id": self.id,
 63            "type": self.type.name,
 64            "value": self.value,
 65            "caption": self.caption,
 66        }
 67
 68    @classmethod
 69    def from_proxy(cls, proxy: EntityProxy) -> "Node":
 70        """For a given :class:`~followthemoney.proxy.EntityProxy`, return a node
 71        based on the entity."""
 72        return cls(registry.entity, proxy.id, proxy=proxy)
 73
 74    def __str__(self) -> str:
 75        return self.caption
 76
 77    def __repr__(self) -> str:
 78        return "<Node(%r, %r, %r)>" % (self.id, self.type, self.caption)
 79
 80    def __hash__(self) -> int:
 81        return hash(self.id)
 82
 83    def __eq__(self, other: Any) -> bool:
 84        return bool(self.id == other.id)
 85
 86
 87class Edge(object):
 88    """A link between two nodes."""
 89
 90    __slots__ = [
 91        "id",
 92        "weight",
 93        "source_id",
 94        "target_id",
 95        "prop",
 96        "proxy",
 97        "schema",
 98        "graph",
 99    ]
100
101    def __init__(
102        self,
103        graph: "Graph",
104        source: Node,
105        target: Node,
106        proxy: Optional[EntityProxy] = None,
107        prop: Optional[Property] = None,
108        value: Optional[str] = None,
109    ):
110        self.graph = graph
111        self.id = f"{source.id}<>{target.id}"
112        self.source_id = source.id
113        self.target_id = target.id
114        self.weight = 1.0
115        self.prop = prop
116        self.proxy = proxy
117        self.schema: Optional[Schema] = None
118        if prop is not None and value is not None:
119            self.weight = prop.specificity(value)
120        if proxy is not None:
121            self.id = f"{source.id}<{proxy.id}>{target.id}"
122            self.schema = proxy.schema
123
124    @property
125    def source(self) -> Optional[Node]:
126        """The graph node from which the edge originates."""
127        if self.source_id is None:
128            return None
129        return self.graph.nodes.get(self.source_id)
130
131    @property
132    def source_prop(self) -> Property:
133        """Get the entity property originating this edge."""
134        if self.schema is not None and self.schema.source_prop is not None:
135            if self.schema.source_prop.reverse is not None:
136                return self.schema.source_prop.reverse
137        if self.prop is None:
138            raise InvalidModel("Contradiction: %r" % self)
139        return self.prop
140
141    @property
142    def target(self) -> Optional[Node]:
143        """The graph node to which the edge points."""
144        if self.target_id is None:
145            return None
146        return self.graph.nodes.get(self.target_id)
147
148    @property
149    def target_prop(self) -> Optional[Property]:
150        """Get the entity property originating this edge."""
151        if self.schema is not None and self.schema.target_prop is not None:
152            return self.schema.target_prop.reverse
153        if self.prop is not None:
154            return self.prop.reverse
155        # NOTE: this edge points at a value node.
156        return None
157
158    @property
159    def type_name(self) -> str:
160        """Return a machine-readable description of the type of the edge.
161        This is either a property name or a schema name."""
162        if self.schema is not None:
163            return self.schema.name
164        if self.prop is None:
165            raise InvalidModel("Invalid edge: %r" % self)
166        return self.prop.name
167
168    def to_dict(self) -> Dict[str, Optional[str]]:
169        return {
170            "id": self.id,
171            "source_id": self.source_id,
172            "target_id": self.target_id,
173            "type_name": self.type_name,
174        }
175
176    def __repr__(self) -> str:
177        return "<Edge(%r)>" % self.id
178
179    def __hash__(self) -> int:
180        return hash(self.id)
181
182    def __eq__(self, other: Any) -> bool:
183        return bool(self.id == other.id)
184
185
186class Graph(object):
187    """A set of nodes and edges, derived from entities and their properties.
188    This represents an alternative interpretation of FtM data as a property
189    graph.
190
191    This class is meant to be extensible in order to support additional
192    backends, like Aleph.
193    """
194
195    def __init__(self, edge_types: Iterable[PropertyType] = registry.pivots) -> None:
196        types = registry.get_types(edge_types)
197        self.edge_types = [t for t in types if t.matchable]
198        self.flush()
199
200    def flush(self) -> None:
201        """Remove all nodes, edges and proxies from the graph."""
202        self.edges: Dict[str, Edge] = {}
203        self.nodes: Dict[str, Node] = {}
204        self.proxies: Dict[str, Optional[EntityProxy]] = {}
205
206    def queue(self, id_: str, proxy: Optional[EntityProxy] = None) -> None:
207        """Register a reference to an entity in the graph."""
208        if id_ not in self.proxies or proxy is not None:
209            self.proxies[id_] = proxy
210
211    @property
212    def queued(self) -> List[str]:
213        """Return a list of all the entities which are referenced from the graph
214        but that haven't been loaded yet. This can be used to get a list of
215        entities that should be included to expand the whole graph by one degree.
216        """
217        return [i for (i, p) in self.proxies.items() if p is None]
218
219    def _get_node_stub(self, prop: Property, value: str) -> Node:
220        if prop.type == registry.entity:
221            self.queue(value)
222        node = Node(prop.type, value, schema=prop.range)
223        if node.id is None:
224            return node
225        if node.id not in self.nodes:
226            self.nodes[node.id] = node
227        return self.nodes[node.id]
228
229    def _add_edge(self, proxy: EntityProxy, source: str, target: str) -> None:
230        if proxy.schema.source_prop is None:
231            raise InvalidModel("Invalid edge entity: %r" % proxy)
232        source_node = self._get_node_stub(proxy.schema.source_prop, source)
233        if proxy.schema.target_prop is None:
234            raise InvalidModel("Invalid edge entity: %r" % proxy)
235        target_node = self._get_node_stub(proxy.schema.target_prop, target)
236        if source_node.id is not None and target_node.id is not None:
237            edge = Edge(self, source_node, target_node, proxy=proxy)
238            self.edges[edge.id] = edge
239
240    def _add_node(self, proxy: EntityProxy) -> None:
241        """Derive a node and its value edges from the given proxy."""
242        entity = Node.from_proxy(proxy)
243        if entity.id is not None:
244            self.nodes[entity.id] = entity
245        for prop, value in proxy.itervalues():
246            if prop.type not in self.edge_types:
247                continue
248            node = self._get_node_stub(prop, value)
249            if node.id is None:
250                continue
251            edge = Edge(self, entity, node, prop=prop, value=value)
252            if edge.weight > 0:
253                self.edges[edge.id] = edge
254
255    def add(self, proxy: EntityProxy) -> None:
256        """Add an :class:`~followthemoney.proxy.EntityProxy` to the graph and make
257        it either a :class:`~followthemoney.graph.Node` or an
258        :class:`~followthemoney.graph.Edge`."""
259        if proxy is None:
260            return
261        self.queue(proxy.id, proxy)
262        if proxy.schema.edge:
263            for (source, target) in proxy.edgepairs():
264                self._add_edge(proxy, source, target)
265        else:
266            self._add_node(proxy)
267
268    def iternodes(self) -> Iterable[Node]:
269        """Iterate all :class:`nodes <followthemoney.graph.Node>` in the graph."""
270        return self.nodes.values()
271
272    def iteredges(self) -> Iterable[Edge]:
273        """Iterate all :class:`edges <followthemoney.graph.Edge>` in the graph."""
274        return self.edges.values()
275
276    def get_outbound(
277        self, node: Node, prop: Optional[Property] = None
278    ) -> Generator[Edge, None, None]:
279        """Get all edges pointed out from the given node."""
280        for edge in self.iteredges():
281            if edge.source == node:
282                if prop and edge.source_prop != prop:
283                    continue
284                yield edge
285
286    def get_inbound(
287        self, node: Node, prop: Optional[Property] = None
288    ) -> Generator[Edge, None, None]:
289        """Get all edges pointed at the given node."""
290        for edge in self.iteredges():
291            if edge.target == node:
292                if prop and edge.target_prop != prop:
293                    continue
294                yield edge
295
296    def get_adjacent(
297        self, node: Node, prop: Optional[Property] = None
298    ) -> Generator[Edge, None, None]:
299        "Get all adjacent edges of the given node."
300        yield from self.get_outbound(node, prop=prop)
301        yield from self.get_inbound(node, prop=prop)
302
303    def to_dict(self) -> Dict[str, Any]:
304        """Return a dictionary with the graph nodes and edges."""
305        return {
306            "nodes": [n.to_dict() for n in self.iternodes()],
307            "edges": [e.to_dict() for e in self.iteredges()],
308        }