1.2 - Thesaurus - Build entity intermediate normalized files.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# PREAMBLE"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import cfg\n",
    "\n",
    "from collections import OrderedDict\n",
    "import gzip\n",
    "import re\n",
    "\n",
    "import progress_bar as pb\n",
    "import parallel_stream as ps\n",
    "from efficient_query_expansion.normalize_text import normalize_text, normalize_hyphens, normalize_multiword"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "_src_tags_order = (\"name\", \"alias\", \"redir\")\n",
    "_and_replacements = (\" \", \"\", \" and \", \" n \", \"n\")\n",
    "_parenthesis_re = re.compile(\"\\s*\\(.*?\\)\\s*$\")\n",
    "\n",
    "\n",
    "def normalize_aliases_raw(src_tag_to_aliases_raw, ampersand=True, hyphens=True, multiword=True, acronyms=True):\n",
    "    assert isinstance(src_tag_to_aliases_raw, dict)\n",
    "    assert (len(src_tag_to_aliases_raw) - sum(src_tag in src_tag_to_aliases_raw for src_tag in _src_tags_order)) == 0\n",
    "\n",
    "    # ordered list of src_tag - aliases_raw\n",
    "    src_tag__aliases_raw__pairs = tuple(\n",
    "        (src_tag, tuple(alias_raw for alias_raw in src_tag_to_aliases_raw[src_tag] if \":\" not in alias_raw))  # remove aliases containing :\n",
    "        for src_tag in _src_tags_order\n",
    "        if src_tag in src_tag_to_aliases_raw and src_tag_to_aliases_raw[src_tag] is not None\n",
    "    )\n",
    "\n",
    "    # result\n",
    "    aliases = OrderedDict()\n",
    "\n",
    "    # support data structure\n",
    "    aliases_raw_support = OrderedDict()\n",
    "\n",
    "    # following the fixed src_tag order performs the normalization on the aliases_raw\n",
    "    for src_tag, aliases_raw in src_tag__aliases_raw__pairs:\n",
    "        aliases_raw_support.clear()\n",
    "\n",
    "        # raw version\n",
    "        postponed_insertions = []  # ambiguous insertions will be inserted after the others\n",
    "        for alias_raw in aliases_raw:\n",
    "            new_alias_raw = _parenthesis_re.sub(\"\", alias_raw)\n",
    "            if len(new_alias_raw) != len(alias_raw):\n",
    "                postponed_insertions.append(new_alias_raw)  # ambiguous alias\n",
    "            elif alias_raw not in aliases_raw_support:\n",
    "                aliases_raw_support[alias_raw] = tuple([])  # clear alias\n",
    "        # raw version without parenthesis\n",
    "        for alias_raw in postponed_insertions:\n",
    "            if alias_raw not in aliases_raw_support:\n",
    "                aliases_raw_support[alias_raw] = tuple([\"norm()\"])\n",
    "\n",
    "        # & normalized version\n",
    "        if ampersand:\n",
    "            for alias_raw, tags in aliases_raw_support.iteritems():\n",
    "                if \"&\" in alias_raw:\n",
    "                    alias_raw_splitted = alias_raw.strip().split(\"&\")\n",
    "                    for i, and_replacement in enumerate(_and_replacements):\n",
    "                        new_alias_raw = and_replacement.join(alias_raw_splitted)\n",
    "                        if new_alias_raw not in aliases_raw_support:\n",
    "                            aliases_raw_support[new_alias_raw] = (\"norm&\" + str(i),) + tags\n",
    "\n",
    "        # default normalization before all\n",
    "        for alias_raw, tags in aliases_raw_support.iteritems():\n",
    "            alias = normalize_text(alias_raw)\n",
    "            if alias not in aliases:\n",
    "                aliases[alias] = (src_tag,) + tags\n",
    "\n",
    "        # hyphens normalization\n",
    "        if hyphens:\n",
    "            for alias_raw, tags in aliases_raw_support.iteritems():\n",
    "                alias = normalize_hyphens(alias_raw)\n",
    "                if alias not in aliases:\n",
    "                    aliases[alias] = (src_tag, \"norm-\") + tags\n",
    "\n",
    "        # multiword normalization\n",
    "        if hyphens:\n",
    "            for alias_raw, tags in aliases_raw_support.iteritems():\n",
    "                alias = normalize_multiword(alias_raw)\n",
    "                if alias not in aliases:\n",
    "                    aliases[alias] = (src_tag, \"norm|\") + tags\n",
    "\n",
    "    # acronyms normalization\n",
    "    if acronyms:\n",
    "        acronyms_support = dict()\n",
    "        for alias in aliases:\n",
    "            if \" \" not in alias:\n",
    "                continue\n",
    "            # get the initial letters\n",
    "            initials = [\n",
    "                letter\n",
    "                for i, letter in enumerate(alias)\n",
    "                if i==0 or (alias[i-1] == ' ' and letter != ' ')\n",
    "            ]\n",
    "            if len(initials) <= 1:\n",
    "                continue\n",
    "\n",
    "            # create the two acronyms using the initial letters only\n",
    "            acronyms = (\n",
    "                ''.join(initials),\n",
    "                ' '.join(initials)\n",
    "            )\n",
    "\n",
    "            # add the tag acronym to these acronyms\n",
    "            for acronym in acronyms:\n",
    "                if acronym not in aliases:\n",
    "                    continue\n",
    "                if len(aliases[acronym]) == 0 or aliases[acronym][-1] != \"acronym\": # since we are in this loop the tag \"acronym\" can only be the last one\n",
    "                    aliases[acronym] += (\"acronym\", )\n",
    "\n",
    "            # if only one of the two acronyms appear among the aliases I add the other one\n",
    "            if acronyms[0] in aliases:\n",
    "                if acronyms[1] not in aliases:\n",
    "                    # only acronyms[0] is inside\n",
    "                    acronyms_support[acronyms[1]] = aliases[acronyms[0]] + (\"norm.\", )\n",
    "            else:\n",
    "                if acronyms[1] in aliases:\n",
    "                    # only acronyms[1] is inside\n",
    "                    acronyms_support[acronyms[0]] = aliases[acronyms[1]] + (\"norm.\", )\n",
    "\n",
    "        aliases.update(acronyms_support)\n",
    "\n",
    "    # discard empty aliases if any\n",
    "    aliases.pop(\"\", None)\n",
    "\n",
    "    # assertions\n",
    "    assert all((\"  \" not in alias) for alias in aliases)\n",
    "\n",
    "    # return the aliases\n",
    "    return aliases"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# PROCESS WIKIPEDIA RAW INTERMEDIATE FILES"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "path_template_aliases = cfg.processed_dir + \"wikipedia_raw/aliases.part_{{}}_{}.tsv.gz\".format(cfg.wiki_preprocessing_split_into)\n",
    "path_template_redirects = cfg.processed_dir + \"wikipedia_raw/redirects.part_{{}}_{}.tsv.gz\".format(cfg.wiki_preprocessing_split_into)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# get from https://en.wikipedia.org/wiki/Help:Category\n",
    "exclude_category_list = [\n",
    "    \"User\", \"Wikipedia\", \"File\", \"MediaWiki\", \"Template\", \"Help\", \"Category\", \"Portal\", \"Book\", \"Draft\", \"Education Program\", \"TimedText\", \"Module\", \"Gadget\", \"Gadget definition\"\n",
    "]\n",
    "\n",
    "def filter_page_title_raw(page_title_raw):\n",
    "    p = page_title_raw.find(\":\")\n",
    "    if p > 0:\n",
    "        category = page_title_raw[:p].lower()\n",
    "        return category in exclude_category_list\n",
    "    return page_title_raw.startswith(\"List of \")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## READ REDIRECTS"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "%%time\n",
    "page_title_to_in_link_page_title_list = dict()\n",
    "\n",
    "pbar = pb.ProgressBar()\n",
    "for part in xrange(1, cfg.wiki_preprocessing_split_into+1):\n",
    "    with gzip.open(path_template_redirects.format(part)) as redirects_file:\n",
    "        for line in redirects_file:\n",
    "            pbar.increase()\n",
    "\n",
    "            page_id, page_title_src, page_title_dest = line[:-1].decode('utf-8').split(\"\\t\")\n",
    "\n",
    "            if filter_page_title_raw(page_title_src) or filter_page_title_raw(page_title_dest):\n",
    "                continue\n",
    "\n",
    "            # put the redirect into the dictionary\n",
    "            in_link_page_title_list = page_title_to_in_link_page_title_list.get(page_title_dest, None)\n",
    "            if in_link_page_title_list is None:\n",
    "                page_title_to_in_link_page_title_list[page_title_dest] = in_link_page_title_list = [page_title_src]\n",
    "            else:\n",
    "                in_link_page_title_list.append(page_title_src)\n",
    "pbar.stop(True)\n",
    "# it lasts 1min 40s"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## READ ALIASES"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "%%time\n",
    "page_title_to_page_id_aliases_raw = dict()\n",
    "\n",
    "pbar = pb.ProgressBar()\n",
    "for part in xrange(1, cfg.wiki_preprocessing_split_into+1):\n",
    "    with gzip.open(path_template_aliases.format(part)) as aliases_file:\n",
    "        for line in aliases_file:\n",
    "            pbar.increase()\n",
    "\n",
    "            # get the 3 main components\n",
    "            page_id, page_title, aliases_raw = line[:-1].decode('utf-8').split(\"\\t\", 2)\n",
    "\n",
    "            if filter_page_title_raw(page_title):\n",
    "                continue\n",
    "\n",
    "            # split the last component that contains the aliases\n",
    "            aliases_raw = set(aliases_raw.split(\"\\t\") if aliases_raw else [])\n",
    "\n",
    "            # warning about an entry that is already inside the dictionary\n",
    "            if page_title in page_title_to_page_id_aliases_raw:\n",
    "                print u\"'{}' is already in\".format(page_title)\n",
    "                # include the previous aliases\n",
    "                aliases_raw.update(page_title_to_page_id_aliases_raw[page_title][1])\n",
    "\n",
    "            # discard the page title\n",
    "            aliases_raw.discard(page_title)\n",
    "\n",
    "            # put the aliases into the dictionary\n",
    "            page_title_to_page_id_aliases_raw[page_title] = (page_id, tuple(aliases_raw))\n",
    "pbar.stop(True)\n",
    "# it lasts 2min"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## WRITE ON DISK"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### WRITE THE ALIASES"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "page_title_set = set(page_title_to_in_link_page_title_list) | set(page_title_to_page_id_aliases_raw)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def _emitter(outqueue):\n",
    "    for page_title in page_title_set:\n",
    "        outqueue.put(page_title)\n",
    "\n",
    "def _worker(worker_id, inqueue, outqueue):\n",
    "    src_tag_to_aliases_raw = {\"name\": None, \"alias\": None, \"redir\": None}\n",
    "\n",
    "    for page_title in inqueue:\n",
    "        # set the name\n",
    "        src_tag_to_aliases_raw[\"name\"] = (page_title,)\n",
    "\n",
    "        # get page id and raw aliases\n",
    "        if page_title in page_title_to_page_id_aliases_raw:\n",
    "            page_id, aliases_raw = page_title_to_page_id_aliases_raw[page_title]\n",
    "        else:\n",
    "            page_id, aliases_raw = \"-\", None\n",
    "\n",
    "        # set the aliases\n",
    "        src_tag_to_aliases_raw[\"alias\"] = aliases_raw\n",
    "\n",
    "        # get the raw in-redirects\n",
    "        in_link_page_title_list = page_title_to_in_link_page_title_list.get(page_title, None)\n",
    "        \n",
    "        # set the redirects\n",
    "        src_tag_to_aliases_raw[\"redir\"] = None if in_link_page_title_list is None else in_link_page_title_list\n",
    "\n",
    "        # normalize all the information\n",
    "        aliases_to_tags = normalize_aliases_raw(src_tag_to_aliases_raw)\n",
    "        \n",
    "        # format the output line\n",
    "        if len(aliases_to_tags):\n",
    "            line = \"{}\\t{}\\n\".format(page_id, \"\\t\".join(\"{}:{}\".format(alias, \",\".join(tags)) for alias, tags in aliases_to_tags.iteritems()))\n",
    "            outqueue.put(line)\n",
    "\n",
    "        del aliases_to_tags\n",
    "\n",
    "def _collector(inqueue):\n",
    "    with gzip.open(cfg.processed_dir + \"wikipedia.aliases.tsv.gz\", \"w\") as outfile:\n",
    "        for line in pb.iter_progress(inqueue):\n",
    "            outfile.write(line)\n",
    "\n",
    "ps.parallel_stream(\n",
    "    _emitter,\n",
    "    _worker,\n",
    "    _collector,\n",
    "    emitter_output_chunk_size=100,\n",
    "    worker_output_chunk_size=100,\n",
    "    emitter_queue_size=100,\n",
    "    collector_queue_size=100,\n",
    "    fork_collector=False,\n",
    "    n_jobs=-1\n",
    ")\n",
    "# it lasts 3min 50s"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# PROCESS WIKIDATA RAW INTERMEDIATE FILE"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def _emitter(outqueue):\n",
    "    with gzip.open(cfg.processed_dir + \"wikidata_raw/aliases.tsv.gz\", \"r\") as infile:\n",
    "        for line in infile:\n",
    "            outqueue.put(line)\n",
    "\n",
    "def _worker(worker_id, inqueue, outqueue):    \n",
    "    src_tag_to_aliases_raw = {\"name\": None, \"alias\": None}\n",
    "\n",
    "    for line in inqueue:\n",
    "        # split the line\n",
    "        entity_id, aliases_raw = line[:-1].decode('utf-8').split(\"\\t\", 1)\n",
    "        aliases_raw = aliases_raw.split(\"\\t\")\n",
    "        \n",
    "        # set name and aliases\n",
    "        src_tag_to_aliases_raw[\"name\"] = aliases_raw[:1]\n",
    "        src_tag_to_aliases_raw[\"alias\"] = aliases_raw[1:]\n",
    "\n",
    "        # normalization\n",
    "        aliases_to_tags = normalize_aliases_raw(src_tag_to_aliases_raw)\n",
    "\n",
    "        # format the output line\n",
    "        if len(aliases_to_tags):\n",
    "            line = \"{}\\t{}\\n\".format(entity_id, \"\\t\".join(\"{}:{}\".format(alias, \",\".join(tags)) for alias, tags in aliases_to_tags.iteritems()))\n",
    "            outqueue.put(line)\n",
    "\n",
    "        # free the memory\n",
    "        del aliases_to_tags\n",
    "\n",
    "def _collector(inqueue):\n",
    "    with gzip.open(cfg.processed_dir + \"wikidata.aliases.tsv.gz\", \"w\") as outfile:\n",
    "        for line in pb.iter_progress(inqueue):\n",
    "            outfile.write(line)\n",
    "\n",
    "ps.parallel_stream(\n",
    "    _emitter,\n",
    "    _worker,\n",
    "    _collector,\n",
    "    emitter_output_chunk_size=100,\n",
    "    worker_output_chunk_size=100,\n",
    "    emitter_queue_size=100,\n",
    "    collector_queue_size=100,\n",
    "    fork_collector=False,\n",
    "    n_jobs=-1\n",
    ")\n",
    "# it lasts 6min 40s"
   ]
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}