Open Ownership and Neo4j Setup.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Import Open Ownership data into a Neo4j database\n",
    "\n",
    "Open Ownership provides a regularly-updating register of beneficial company owners from corporate registrars around the world. This notebook provides a rough guide to importing their data (once) into a Neo4j graph database, and setting it up for search.\n",
    "\n",
    "## Installation\n",
    "\n",
    "This is the process for deploying Neo4j as a [Debian installation](https://neo4j.com/docs/operations-manual/current/installation/linux/debian/).\n",
    "\n",
    "Add the repository (check the link for the latest version):\n",
    "\n",
    "    wget -O - https://debian.neo4j.org/neotechnology.gpg.key | sudo apt-key add -\n",
    "    echo 'deb https://debian.neo4j.org/repo stable/' | sudo tee -a /etc/apt/sources.list.d/neo4j.list\n",
    "    sudo apt-get update\n",
    "    \n",
    "Install community edition:\n",
    "\n",
    "    sudo apt-get install neo4j=1:3.5.12\n",
    "    \n",
    "And [start](https://medium.com/@Jessicawlm/installing-neo4j-on-ubuntu-14-04-step-by-step-guide-ed943ec16c56)\n",
    "\n",
    "    sudo service neo4j restart\n",
    "    \n",
    "[File locations](https://neo4j.com/docs/operations-manual/current/configuration/file-locations/)\n",
    "[Configuration](https://neo4j.com/docs/operations-manual/current/configuration/neo4j-conf/)\n",
    "\n",
    "    sudo nano /etc/neo4j/neo4j.conf\n",
    " \n",
    "## Using Neo4j with Python / Django\n",
    "\n",
    "I use Python and Django, but you should find your language of choice supported in Neo4j's docs.\n",
    "\n",
    "[Python drivers](https://neo4j.com/developer/python/)\n",
    "\n",
    "    pip install neo4j\n",
    "    \n",
    "Preferred [Neomodel](https://pypi.python.org/pypi/neomodel)\n",
    "\n",
    "    pip install neomodel\n",
    "    \n",
    "[Neomodel documentation](https://neomodel.readthedocs.io/en/latest/getting_started.html)\n",
    "\n",
    "To setup and automatically [create nodes](https://neomodel.readthedocs.io/en/latest/configuration.html):\n",
    "\n",
    "    from neomodel import install_labels\n",
    "    install_labels(YourClass)\n",
    "\n",
    "Or, in Django:\n",
    "\n",
    "    import yourapp  # make sure your app is loaded\n",
    "    from neomodel import install_all_labels\n",
    "\n",
    "    install_all_labels()\n",
    "\n",
    "Or, from the command line:\n",
    "\n",
    "    neomodel_install_labels yourapp.py someapp.models --db bolt://neo4j:ne05j@localhost:7687\n",
    "    neomodel_remove_labels --db bolt://neo4j:ne05j@localhost:7687\n",
    "    \n",
    "## Building and populating the database"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from neomodel import (config, db, StructuredNode, StructuredRel, \n",
    "                      StringProperty, IntegerProperty, ArrayProperty,\n",
    "                      JSONProperty, BooleanProperty,\n",
    "                      DateProperty, RelationshipTo)\n",
    "\n",
    "NEO4J_USERNAME = \"neo4j\"\n",
    "NEO4J_PASSWORD = \"ne05j\"\n",
    "NEO4J_BOLT_URL = F\"bolt://{NEO4J_USERNAME}:{NEO4J_PASSWORD}@localhost:7687\"\n",
    "\n",
    "config.DATABASE_URL = NEO4J_BOLT_URL  # default\n",
    "config.FORCE_TIMEZONE = True  # default False"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create the Python database model via `neomodel` based on the Open Ownership [data standard / schema](http://standard.openownership.org/en/v0-2-0/schema/index.html):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from neomodel import (config, db, StructuredNode, StructuredRel, \n",
    "                      StringProperty, IntegerProperty, ArrayProperty,\n",
    "                      JSONProperty, BooleanProperty,\n",
    "                      DateProperty, RelationshipTo)\n",
    "\n",
    "config.DATABASE_URL = NEO4J_BOLT_URL\n",
    "\n",
    "class OwnershipOrControl(StructuredRel):\n",
    "    statementID = StringProperty(required=True)\n",
    "    statementType = StringProperty(required=True)\n",
    "    statementDate = DateProperty()\n",
    "    isActive = BooleanProperty(default=True)\n",
    "    # isComponent = BooleanProperty()\n",
    "    # These are the objects of the ownership or control statement\n",
    "    # componentStatementIDs = ArrayProperty(StringProperty())\n",
    "    # This is the subject of the ownership or control statement\n",
    "    # subject = StringProperty()\n",
    "    # This is the object of the ownership or control statement\n",
    "    # interestedParty \n",
    "    interests = ArrayProperty(JSONProperty())\n",
    "    source = JSONProperty()\n",
    "    annotations = ArrayProperty(JSONProperty())\n",
    "    replacesStatements = ArrayProperty(StringProperty())\n",
    "    \n",
    "class BaseEntity(StructuredNode):\n",
    "    statementID = StringProperty(unique_index=True, required=True)\n",
    "    statementType = StringProperty(required=True)\n",
    "    statementDate = DateProperty()\n",
    "    isActive = BooleanProperty(default=True)\n",
    "    publicationDetails = JSONProperty()\n",
    "    source = JSONProperty()\n",
    "    annotations = ArrayProperty(JSONProperty())\n",
    "    replacesStatements = ArrayProperty(StringProperty())\n",
    "    addresses = ArrayProperty(JSONProperty())\n",
    "    postcode = StringProperty()\n",
    "    controls = RelationshipTo(\"BaseEntity\", \"CONTROLS\", model=OwnershipOrControl)\n",
    "\n",
    "class Entity(BaseEntity):\n",
    "    entityType = StringProperty()\n",
    "    unspecifiedEntityDetails = JSONProperty()\n",
    "    name = StringProperty()\n",
    "    name_token = StringProperty()\n",
    "    alternateNames = ArrayProperty(StringProperty())\n",
    "    incorporatedInJurisdiction = JSONProperty()\n",
    "    identifiers = ArrayProperty(JSONProperty())\n",
    "    foundingDate = DateProperty()\n",
    "    dissolutionDate = DateProperty()\n",
    "    uri = StringProperty()\n",
    "    \n",
    "class Person(BaseEntity):\n",
    "    personType = StringProperty()\n",
    "    unspecifiedPersonDetails = JSONProperty()\n",
    "    names = ArrayProperty(JSONProperty())\n",
    "    identifiers = ArrayProperty(JSONProperty())\n",
    "    nationalities = ArrayProperty(JSONProperty())\n",
    "    birthDate = DateProperty()\n",
    "    deathDate = DateProperty()\n",
    "    placeOfBirth = JSONProperty()\n",
    "    placeOfResidence = JSONProperty()\n",
    "    taxResidencies = ArrayProperty(JSONProperty())\n",
    "    hasPepStatus = BooleanProperty()\n",
    "    pepStatusDetails = ArrayProperty(JSONProperty())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 0 ns, sys: 0 ns, total: 0 ns\n",
      "Wall time: 1.15 s\n"
     ]
    }
   ],
   "source": [
    "# https://neo4j.com/developer/kb/large-delete-transaction-best-practices-in-neo4j/\n",
    "#from neomodel import clear_neo4j_database, db\n",
    "#%time clear_neo4j_database(db)  # deletes all nodes and relationships\n",
    "\n",
    "from neomodel import install_labels, remove_all_labels\n",
    "install_labels(OwnershipOrControl)\n",
    "install_labels(Entity)\n",
    "install_labels(Person)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Open Ownership JSONLines to Neo4j\n",
    "\n",
    "This is slightly more than simply importing the data. We are also going to prepare a tokenised search term for use in the Neo4j [full text search](https://neo4j.com/developer/kb/fulltext-search-in-neo4j/). Edit this out if you don't want to use it.\n",
    "\n",
    "Note that you'll need to install NLTK, Unidecode and Cleanco:\n",
    "\n",
    "    pip install nltk unidecode cleanco"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "class TokeniseName:\n",
    "\n",
    "    def split_on_term(self, name, terms, return_second_term=False):\n",
    "        splt = 0\n",
    "        if return_second_term: splt = 1\n",
    "        for term in terms:\n",
    "            if term in name:\n",
    "                name_split = name.split(term)\n",
    "                # Double-check ... if the split choice is blank, pick the other\n",
    "                if name_split[splt].strip(): \n",
    "                    name = name_split[splt].strip()\n",
    "                else:\n",
    "                    name = name_split[splt-1].strip()\n",
    "        return name\n",
    "\n",
    "    def parse_token(self, name):\n",
    "        \"\"\"\n",
    "        Do the following:\n",
    "\n",
    "        - Drop to lower-case;\n",
    "        - Split on trading as (\"t/a\");\n",
    "        - Remove punctuation;\n",
    "        - Standardise whitespace (1-space);\n",
    "        - Unidecode;\n",
    "        - Remove legal control terms (co, ltd, etc);\n",
    "\n",
    "         Create a word frequency dictionary: `token_frequency`\n",
    "\n",
    "         Ref: [Figure out if a business name is very similar to another one]\n",
    "         (https://stackoverflow.com/questions/6400416/figure-out-if-a-business-name-is-very-similar-to-another-one-python)\n",
    "        \"\"\"\n",
    "        # Drop to lower-case and strip out leading / lagging spaces\n",
    "        if not (name and str(name).strip()):\n",
    "            return False\n",
    "        parsed_name = \" \".join(str(name).lower().strip().split())\n",
    "        # Split 2nd term on t/a, ta, trading as\n",
    "        second_term = [\"t/a\", \" ta \", \"trading as\", \"(a trading name)\", \"( a trading name)\", \"a trading name\"]\n",
    "        parsed_name = self.split_on_term(parsed_name, second_term, return_second_term=True)\n",
    "        # Split on 1st term on ltd, limited, c/o, co, note that the space ensures it is a word, or not a start\n",
    "        first_term = [\" ltd\", \" limited\", \",limited\", \"c/o\", \" co \", \" pty\", \" proprietary\", \" sarl\"]\n",
    "        parsed_name = self.split_on_term(parsed_name, first_term)\n",
    "        # Unidecode to normalise alphabet\n",
    "        parsed_name = unidecode(parsed_name)\n",
    "        # Remove punctuation and normalise whitespace\n",
    "        # https://stackoverflow.com/a/34294398 & https://stackoverflow.com/a/1546251\n",
    "        punc = \"!\\\"#$%£()*+,./:;<=>?@[\\\\]^_`{|}~\" # removed -'& from the punctuation list\n",
    "        translator = str.maketrans(\"\", \"\", punc)\n",
    "        parsed_name = parsed_name.translate(translator)\n",
    "        # Strip out legal control terms\n",
    "        parsed_name = cleanco(parsed_name).clean_name()\n",
    "        # Simple deletions\n",
    "        parsed_name = parsed_name.replace(\"?\", \"\") \n",
    "        if parsed_name.strip():\n",
    "            return \" \".join(parsed_name.strip().split())\n",
    "        return False\n",
    "    \n",
    "    def parse_search_token(self, name):\n",
    "        parsed_name = self.parse_token(name)\n",
    "        if not parsed_name:\n",
    "            return False\n",
    "        punc = \"!\\\"#$%£()*+,./:;<=>?@[\\\\]^_`{|}~-'&\"\n",
    "        translator = str.maketrans(\"\", \"\", punc)\n",
    "        parsed_name = parsed_name.translate(translator)\n",
    "        parsed_name = \" \".join([l[0].strip() for l in nltk.pos_tag(parsed_name.split()) if l[1] not in [\"CC\"]])\n",
    "        if parsed_name:\n",
    "            return parsed_name\n",
    "        return False"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Download the latest [jsonlines file](https://register.openownership.org/download) from Open Ownership."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "import jsonlines\n",
    "\n",
    "import nltk\n",
    "from cleanco import cleanco\n",
    "from unidecode import unidecode\n",
    "import re\n",
    "\n",
    "from datetime import date, datetime\n",
    "\n",
    "IMPORT_DATA_DIRECTORY = \"/directory/where/you/saved/oojsonlines\"\n",
    "tn = TokeniseName()\n",
    "\n",
    "config.DATABASE_URL = NEO4J_BOLT_URL\n",
    "\n",
    "def parse_postcode(x):\n",
    "    # Postcodes: https://www.safaribooksonline.com/library/view/regular-expressions-cookbook/9781449327453/ch04s16.html\n",
    "    # https://stackoverflow.com/a/51885364\n",
    "    RE_POST = \"([A-Z][A-HJ-Y]?[0-9][A-Z0-9]? ?[0-9][A-Z]{2})\"\n",
    "    u = re.search(RE_POST, str(x), re.IGNORECASE)\n",
    "    if u == None:\n",
    "        return None\n",
    "    return u.group(0)\n",
    "\n",
    "def generate_base_node(**kwargs):\n",
    "    for dt in [\"birthDate\", \"deathDate\", \"dissolutionDate\", \"foundingDate\", \"statementDate\"]:\n",
    "        if dt in kwargs:\n",
    "            kwargs[dt] = date(*[int(i) for i in kwargs[dt].split(\"-\")])\n",
    "    if kwargs[\"statementType\"] == \"entityStatement\" and kwargs.get(\"name\"):\n",
    "            name_token = tn.parse_search_token(kwargs[\"name\"])\n",
    "            if name_token: kwargs[\"name_token\"] = name_token\n",
    "    if \"addresses\" in kwargs:\n",
    "        # list, including keys: \n",
    "        #   type: placeOfBirth, residence, registered, service, alternative, business\n",
    "        #   address: \n",
    "        #   postCode: \n",
    "        ad_dict = {}\n",
    "        for addr in kwargs[\"addresses\"]:\n",
    "            pc = addr.get(\"postCode\")\n",
    "            if not addr.get(\"type\"): addr[\"type\"] = \"unclear\"\n",
    "            if not pc:\n",
    "                pc = parse_postcode(addr[\"address\"])\n",
    "            if pc:\n",
    "                ad_dict[addr[\"type\"]] = pc\n",
    "        for typ in [\"business\", \"registered\", \"service\", \"alternative\", \"residence\", \"placeOfBirth\", \"unclear\"]:\n",
    "            if typ in ad_dict:\n",
    "                kwargs[\"postcode\"] = ad_dict[typ]\n",
    "                break\n",
    "    if \"statementID\" in kwargs:\n",
    "        kwargs[\"statementID\"] = cf.get_id(kwargs[\"statementID\"])\n",
    "        return kwargs\n",
    "    return None\n",
    "\n",
    "def create_or_update_node(**kwargs):\n",
    "    node = None\n",
    "    if kwargs[\"statementType\"] == \"entityStatement\":\n",
    "        node = Entity.create_or_update\n",
    "        if kwargs.get(\"name\"):\n",
    "            name_token = tn.parse_search_token(kwargs[\"name\"])\n",
    "            if name_token: kwargs[\"name_token\"] = name_token\n",
    "    if kwargs[\"statementType\"] == \"personStatement\":\n",
    "        node = Person.create_or_update\n",
    "    if not node: return None\n",
    "    kwargs = generate_base_node(**kwargs)\n",
    "    obj = None\n",
    "    if kwargs:\n",
    "        obj = node(kwargs)\n",
    "    return obj\n",
    "\n",
    "def get_edge_party(term):\n",
    "    node = None\n",
    "    if type(term) == str:\n",
    "        node = Person.nodes.get_or_none(statementID=term, lazy=True)\n",
    "        if not node:\n",
    "            node = Entity.nodes.get_or_none(statementID=term, lazy=True)\n",
    "    if type(term) == dict:\n",
    "        if term.get(\"describedByPersonStatement\"):\n",
    "            node = Person.nodes.get_or_none(statementID=term[\"describedByPersonStatement\"])\n",
    "        if term.get(\"describedByEntityStatement\"):\n",
    "            node = Entity.nodes.get_or_none(statementID=term[\"describedByEntityStatement\"])\n",
    "    return node\n",
    "\n",
    "def import_open_ownership_graph(source_file):\n",
    "    # Open Ownership source data\n",
    "    d = IMPORT_DATA_DIRECTORY\n",
    "    node_errors = []\n",
    "    edge_errors = []\n",
    "    print(\"Starting ...\")\n",
    "    with jsonlines.open(d + source_file) as reader:\n",
    "        for i, kwargs in enumerate(reader):\n",
    "            if i % 1000 == 0 and i != 0: print(\"row: {}\\r\".format(i), end=\"\")\n",
    "            if kwargs[\"statementType\"] in [\"entityStatement\",\"personStatement\"]:\n",
    "                node = create_or_update_node(**kwargs)\n",
    "                if not node: node_errors.append(kwargs)\n",
    "            if kwargs[\"statementType\"] == \"ownershipOrControlStatement\":\n",
    "                kwargs = generate_base_node(**kwargs)\n",
    "                first_party = get_edge_party(kwargs[\"interestedParty\"])\n",
    "                second_party = get_edge_party(kwargs[\"subject\"])\n",
    "                if first_party and second_party:\n",
    "                    with db.transaction:\n",
    "                        edge = first_party.controls.connect(second_party, kwargs)\n",
    "                        edge.save()\n",
    "                else:\n",
    "                    edge_errors.append(kwargs)\n",
    "    print(\"Completed first run...\")\n",
    "    return edge_errors, node_errors\n",
    "\n",
    "def test_open_ownership_graph(source_file):\n",
    "    # Open Ownership source data\n",
    "    d = IMPORT_DATA_DIRECTORY\n",
    "    node_errors = []\n",
    "    edge_errors = []\n",
    "    print(\"Starting ...\")\n",
    "    with jsonlines.open(d + source_file) as reader:\n",
    "        for i, kwargs in enumerate(reader):\n",
    "            if i % 1 == 0 and i != 0: print(\"row: {}\\r\".format(i), end=\"\")\n",
    "            if kwargs[\"statementType\"] in [\"entityStatement\" ,\"personStatement\", \"ownershipOrControlStatement\"]:\n",
    "                print(kwargs)\n",
    "                kwargs = generate_base_node(**kwargs)\n",
    "                print(kwargs)\n",
    "                if i > 10: break\n",
    "    print(\"Completed test run...\")\n",
    "    return edge_errors, node_errors"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2019-12-30 09:01:03.374281\n",
      "Starting ...\n",
      "Completed first run...\n",
      "CPU times: user 9h 31min 28s, sys: 1h 2min 28s, total: 10h 33min 57s\n",
      "Wall time: 22h 15min 31s\n"
     ]
    }
   ],
   "source": [
    "# import jsonlines from OO\n",
    "print(datetime.now())\n",
    "\n",
    "config.DATABASE_URL = NEO4J_BOLT_URL\n",
    "\n",
    "%time edge_errors, node_errors = import_open_ownership_graph(\"statements.latest.jsonl\")\n",
    "#%time edge_errors, node_errors = test_open_ownership_graph(\"statements.latest.jsonl\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Set up the full-text search"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 15.6 ms, sys: 15.6 ms, total: 31.2 ms\n",
      "Wall time: 45 ms\n",
      "[]\n",
      "CPU times: user 0 ns, sys: 0 ns, total: 0 ns\n",
      "Wall time: 7.31 ms\n",
      "[]\n"
     ]
    }
   ],
   "source": [
    "# Create a custom analyser to keep stopwords \n",
    "# https://graphaware.com/neo4j/2019/01/11/neo4j-full-text-search-deep-dive.html\n",
    "\n",
    "#q = \"CALL db.index.fulltext.drop('entityNames')\"\n",
    "#%time results, meta = db.cypher_query(q)\n",
    "\n",
    "q = \"CALL db.index.fulltext.createNodeIndex('entityNames',['Entity'],['name'])\"\n",
    "%time results, meta = db.cypher_query(q)\n",
    "print(results)\n",
    "\n",
    "q = \"CALL db.index.fulltext.createNodeIndex('entityTokens',['Entity'],['name_token'], {analyzer: 'simple'})\"\n",
    "#q = \"CALL db.index.fulltext.createNodeIndex('entityNames',['Entity'],['name', 'name_token'])\"\n",
    "%time results, meta = db.cypher_query(q)\n",
    "print(results)\n",
    "\n",
    "#q = \"CALL db.index.fulltext.createNodeIndex('postCodes',['Entity'],['postcode'], {analyzer: 'keyword'})\"\n",
    "#%time results, meta = db.cypher_query(q)\n",
    "#print(results)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Test a query\n",
    "\n",
    "Perform a simple query and see if you get anything"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{'statementID': 'CxD9NxIcFcFBdPRkHCTP3VBG', 'statementType': 'entityStatement', 'statementDate': None, 'isActive': True, 'publicationDetails': None, 'source': None, 'annotations': None, 'replacesStatements': None, 'addresses': [{'type': 'registered', 'address': 'The Spectrum, 56-58 Benson Road Birchwood, Warrington, Cheshire, WA3 7PQ', 'country': 'GB'}], 'postcode': 'WA3 7PQ', 'entityType': 'registeredEntity', 'unspecifiedEntityDetails': None, 'name': 'DONE BROTHERS (CASH BETTING) LIMITED', 'name_token': 'done brothers cash betting', 'alternateNames': None, 'incorporatedInJurisdiction': {'name': 'United Kingdom of Great Britain and Northern Ireland', 'code': 'GB'}, 'identifiers': [{'scheme': 'GB-COH', 'id': '01277703'}], 'foundingDate': datetime.date(1976, 9, 17), 'dissolutionDate': None, 'uri': None, 'id': 195220}\n"
     ]
    }
   ],
   "source": [
    "from neomodel import Q, Traversal\n",
    "\n",
    "config.DATABASE_URL = NEO4J_BOLT_URL\n",
    "\n",
    "q = Q(name__startswith='DONE BROTHERS')\n",
    "\n",
    "node = Entity.nodes.filter(q)\n",
    "for n in node:\n",
    "    print(n)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Django Shell-Plus",
   "language": "python",
   "name": "django_extensions"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}