LSI.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from abc import ABC, abstractmethod\n",
    "from typing import List\n",
    "from typing import List, Type, Tuple, Union\n",
    "\n",
    "import numpy as np\n",
    "import string\n",
    "import json"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Models"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "class AbstractModel(ABC):\n",
    "    def __init__(self, words: List[str], docs: List[List[str]]) -> None:\n",
    "        self.words = words\n",
    "        self.docs = docs\n",
    "\n",
    "    @abstractmethod\n",
    "    def build(self) -> np.ndarray:\n",
    "        pass\n",
    "\n",
    "\n",
    "class TermCountModel(AbstractModel):\n",
    "    def build(self):\n",
    "        model = np.zeros((len(self.words), len(self.docs)), dtype=int)\n",
    "\n",
    "        for i, word in enumerate(self.words):\n",
    "            for j, doc in enumerate(self.docs):\n",
    "                model[i, j] = doc.count(word)\n",
    "\n",
    "        return model\n",
    "\n",
    "\n",
    "class TFIDFModel(TermCountModel):\n",
    "    def build(self):\n",
    "        term_count_model = super().build()\n",
    "        model = np.zeros((len(self.words), len(self.docs)), dtype=float)\n",
    "\n",
    "        for i, word in enumerate(self.words):\n",
    "            for j, doc in enumerate(self.docs):\n",
    "                tf = term_count_model[i, j] / len(doc)\n",
    "                idf = np.log(sum(term_count_model[i] > 0))\n",
    "                model[i, j] = tf * idf\n",
    "\n",
    "        return model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## LSI"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "class LSI:\n",
    "    \"\"\"Latent Semantic Indexing.\n",
    "    \"\"\"\n",
    "\n",
    "    def __init__(self, docs: List[str], query: str, model: Type[AbstractModel] = TFIDFModel,\n",
    "                 rank_approximation: int = 2, stopwords: List[str] = None,\n",
    "                 ignore_chars=string.punctuation) -> None:\n",
    "        if stopwords is None:\n",
    "            stopwords = []\n",
    "        self.stopwords = stopwords\n",
    "        self.ignore_chars = ignore_chars\n",
    "        self.docs = list(map(self._parse, docs))\n",
    "        self.words = self._get_words()\n",
    "        self.query = self._parse_query(query)\n",
    "        self.model = model\n",
    "        self.rank_approximation = rank_approximation\n",
    "        self.term_doc_matrix = self._build_term_doc_matrix()\n",
    "\n",
    "    def _parse(self, text: str) -> List[str]:\n",
    "        translator = str.maketrans(self.ignore_chars, ' ' * len(self.ignore_chars))\n",
    "        return list(map(str.lower,\n",
    "                        filter(lambda w: w not in self.stopwords,\n",
    "                               text.translate(translator).split())))\n",
    "\n",
    "    def _parse_query(self, query: str) -> np.ndarray:\n",
    "        result = np.zeros(len(self.words))\n",
    "\n",
    "        i = 0\n",
    "        for word in sorted(self._parse(query)):\n",
    "            while word > self.words[i]:\n",
    "                i += 1\n",
    "            if word == self.words[i]:\n",
    "                result[i] += 1\n",
    "\n",
    "        return result\n",
    "\n",
    "    def _get_words(self) -> List[str]:\n",
    "        words = set()\n",
    "\n",
    "        for doc in self.docs:\n",
    "            words = words | set(doc)\n",
    "\n",
    "        return sorted(words)\n",
    "\n",
    "    def _build_term_doc_matrix(self) -> np.ndarray:\n",
    "        model = self.model(self.words, self.docs)\n",
    "        np.savetxt('matrxi_A.txt', model.build(), delimiter=',', fmt='%f')\n",
    "        return model.build()\n",
    "\n",
    "    def _svd_with_dimensionality_reduction(self) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:\n",
    "        u, s, v = np.linalg.svd(self.term_doc_matrix)\n",
    "        s = np.diag(s)\n",
    "        k = self.rank_approximation\n",
    "        return u[:, :k], s[:k, :k], v[:, :k]\n",
    "\n",
    "    def process(self) -> Tuple[np.ndarray, np.ndarray]:\n",
    "        u_k, s_k, v_k = self._svd_with_dimensionality_reduction()\n",
    "\n",
    "        q = self.query.T @ u_k @ np.linalg.pinv(s_k)\n",
    "        d = self.term_doc_matrix.T @ u_k @ np.linalg.pinv(s_k)\n",
    "\n",
    "        res = np.apply_along_axis(lambda row: self._sim(q, row), axis=1, arr=d)\n",
    "        ranking = np.argsort(-res) + 1\n",
    "        return ranking, res\n",
    "\n",
    "    @staticmethod\n",
    "    def _sim(x: np.ndarray, y: np.ndarray):\n",
    "        return (x @ y) / (np.linalg.norm(x) * np.linalg.norm(y))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "articles_data = json.load(open('articles_porter_mystem.json'))['issue']['articles']\n",
    "documents = []\n",
    "documents_names = []\n",
    "for article in articles_data:\n",
    "    documents_names.append(article['link'])\n",
    "    documents.append(article['annotation']['porter'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def rank(query):\n",
    "    lsi = LSI(documents, query)\n",
    "    ranks = lsi.process()\n",
    "    res = list(zip(ranks[1], documents_names))\n",
    "    res.sort(reverse=True)\n",
    "    return res"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "query = 'сследова динамик множеств критическ точек'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(0.98694977435528541, 'http://www.mathnet.ru/rus/uzku1348'),\n",
       " (0.93780932622811231, 'http://www.mathnet.ru/rus/uzku1354'),\n",
       " (0.48310905675650379, 'http://www.mathnet.ru/rus/uzku1356'),\n",
       " (0.31724583956343499, 'http://www.mathnet.ru/rus/uzku1350'),\n",
       " (0.2137728675540487, 'http://www.mathnet.ru/rus/uzku1352'),\n",
       " (0.20875414777511714, 'http://www.mathnet.ru/rus/uzku1357'),\n",
       " (0.0036627067108410357, 'http://www.mathnet.ru/rus/uzku1358'),\n",
       " (-0.28309867572666014, 'http://www.mathnet.ru/rus/uzku1355'),\n",
       " (-0.2873375777621997, 'http://www.mathnet.ru/rus/uzku1349'),\n",
       " (-0.65817914190981652, 'http://www.mathnet.ru/rus/uzku1353')]"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rank(query)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "query = 'реш нестационарн задач для тонк упруг'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(0.97199683499449663, 'http://www.mathnet.ru/rus/uzku1353'),\n",
       " (0.78743242374442557, 'http://www.mathnet.ru/rus/uzku1349'),\n",
       " (0.78469863059935585, 'http://www.mathnet.ru/rus/uzku1355'),\n",
       " (0.57411566265432312, 'http://www.mathnet.ru/rus/uzku1358'),\n",
       " (0.39391353117858835, 'http://www.mathnet.ru/rus/uzku1357'),\n",
       " (0.3891888872469681, 'http://www.mathnet.ru/rus/uzku1352'),\n",
       " (0.28821526498002059, 'http://www.mathnet.ru/rus/uzku1350'),\n",
       " (0.1107564954954537, 'http://www.mathnet.ru/rus/uzku1356'),\n",
       " (-0.56553245036212807, 'http://www.mathnet.ru/rus/uzku1354'),\n",
       " (-0.71307701120388933, 'http://www.mathnet.ru/rus/uzku1348')]"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rank(query)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "query = 'стабилизац решен дифференциальн уравнен в гильбертов пространств'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(0.99103685277467801, 'http://www.mathnet.ru/rus/uzku1356'),\n",
       " (0.950682422308185, 'http://www.mathnet.ru/rus/uzku1350'),\n",
       " (0.91196167935246331, 'http://www.mathnet.ru/rus/uzku1352'),\n",
       " (0.90984305928194509, 'http://www.mathnet.ru/rus/uzku1357'),\n",
       " (0.83751681295668767, 'http://www.mathnet.ru/rus/uzku1354'),\n",
       " (0.80535111955725702, 'http://www.mathnet.ru/rus/uzku1358'),\n",
       " (0.71730317889866424, 'http://www.mathnet.ru/rus/uzku1348'),\n",
       " (0.60166293446289965, 'http://www.mathnet.ru/rus/uzku1355'),\n",
       " (0.59812450174363663, 'http://www.mathnet.ru/rus/uzku1349'),\n",
       " (0.21257282568923108, 'http://www.mathnet.ru/rus/uzku1353')]"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rank(query)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "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.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}