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