notebooks/class_5.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "外媒揭秘<span style=\"color:red\">苹果</span>A13<span style=\"color:red\">芯片</span>：比三星华为<span style=\"color:red\">高通</span><span style=\"color:red\">芯片</span>好在哪"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "from IPython.core.display import HTML\n",
    "import jieba\n",
    "\n",
    "\n",
    "class SearcherScan():\n",
    "    def __init__(self, title_file):\n",
    "        with open(title_file, 'r') as f:\n",
    "            titles = f.read()\n",
    "        self.title_list = list(set(titles.split()))\n",
    "\n",
    "    def word_match(self, words, title):\n",
    "        ifmatch = True\n",
    "        keyword_ = ' '.join(jieba.cut(words))\n",
    "        for word in keyword_.split():\n",
    "            if word != ' ' and word not in title:\n",
    "                ifmatch = False\n",
    "        return ifmatch\n",
    "\n",
    "    def conv_query(self, query):\n",
    "        query_new_parts = []\n",
    "        for part in list(jieba.cut(query)):\n",
    "            if part == '(' or part == ')':\n",
    "                query_new_parts.append(part)\n",
    "            elif part in ('and', 'AND', 'or', 'OR', 'NOT', 'not', ' '):\n",
    "                query_new_parts.append(part.lower())\n",
    "            else:\n",
    "                query_new_parts.append(\"self.word_match('{}',title)\".format(part))\n",
    "        query_new = ''.join(query_new_parts)\n",
    "        return query_new\n",
    "\n",
    "    def highlighter(self, doc, word):\n",
    "        for part in list(jieba.cut(word)):\n",
    "            if part not in ('(', ')', 'and', 'AND', 'or', 'OR', 'NOT', 'not', ' '):\n",
    "                doc = doc.replace(part, '<span style=\"color:red\">{}</span>'.format(part))\n",
    "        return doc\n",
    "\n",
    "    def search(self, query):\n",
    "        query_new = self.conv_query(query)\n",
    "        for title in self.title_list:\n",
    "            if eval(query_new):\n",
    "                display(HTML(self.highlighter(title, query)))\n",
    "\n",
    "\n",
    "query = '苹果 and (芯片 or 高通)'\n",
    "searcher = SearcherScan('titles.txt')\n",
    "searcher.search(query)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "from IPython.core.display import HTML\n",
    "import jieba\n",
    "\n",
    "class SearcherIIndex():\n",
    "    \"\"\"倒排索引文本搜索实现类\n",
    "    \n",
    "    用倒排索引\n",
    "    利用Python的集合运算，来实现候选结果集之间交、并运算\n",
    "    \n",
    "    Attributes:\n",
    "        index: 检索使用的倒排索引\n",
    "        max_id: 当前索引的文档最大ID\n",
    "        doc_list: 索引文档原文\n",
    "    \"\"\"\n",
    "    def __init__(self, docs_file): \n",
    "        \"\"\"初始化，用文件中的文本行构建倒排索引\n",
    "        \n",
    "        Args:\n",
    "            docs_file:包含带索引文档(文本)的文件名\n",
    "            \n",
    "        \"\"\"\n",
    "        self.index = dict()    \n",
    "        self.max_id = 0\n",
    "        self.doc_list = [] \n",
    "        \n",
    "        with open(docs_file, 'r') as f:\n",
    "            docs_data = f.read()\n",
    "        \n",
    "        for doc in docs_data.split():\n",
    "            self.add_doc(doc)\n",
    "\n",
    "    def add_doc(self, doc):\n",
    "        \"\"\"向索引中添加新文档\n",
    "        \n",
    "        Args:\n",
    "            doc:待检索的文档(文本)\n",
    "        \n",
    "        Returns:\n",
    "            新增文档ID\n",
    "        \"\"\"\n",
    "        self.doc_list.append(doc)\n",
    "        for term in list(jieba.cut(doc)):\n",
    "            #构建和更新各Term对应的Posting(集合)\n",
    "            if term in self.index: \n",
    "                self.index[term].add(self.max_id)\n",
    "            else:\n",
    "                self.index[term] = set([self.max_id])\n",
    "        self.max_id += 1\n",
    "        return self.max_id - 1\n",
    "    \n",
    "    def word_match(self, word):\n",
    "        \"\"\"从倒排索引中获取包含word的候选文档ID集合\n",
    "        \n",
    "        Args:\n",
    "            word:待检索的词(短语)\n",
    "            \n",
    "        Returns：\n",
    "            包含待检索词(短语)的文档ID集合\n",
    "        \"\"\"\n",
    "        result = None\n",
    "        for term in list(jieba.cut(word)):\n",
    "            if result is None:\n",
    "                result = self.index.get(term, set())\n",
    "            else:\n",
    "                result = result & self.index.get(term, set())\n",
    "        if result is None:\n",
    "            result = set()\n",
    "        return result\n",
    "\n",
    "    def conv_query(self, query):\n",
    "        \"\"\"将用户的查询转换成用eval可运行、返回结果ID集合的代码段\n",
    "        \n",
    "        Args:\n",
    "            query:待转换的原始查询字符串\n",
    "        \n",
    "        Returns:\n",
    "            转换完成可通过eval执行返回ID集合的代码段字符串\n",
    "        \"\"\"\n",
    "        query_new_parts = []\n",
    "        all_parts = list(jieba.cut(query))\n",
    "        idx = 0\n",
    "        cache = '' #缓存变量，用于回收分词过程被切开的短语片段\n",
    "        count_parts = len(all_parts)\n",
    "        while idx < count_parts:\n",
    "            if all_parts[idx] == '(' or all_parts[idx] == ')':\n",
    "                query_new_parts.append(all_parts[idx])\n",
    "            elif all_parts[idx] == ' ':\n",
    "                query_new_parts.append(' ')\n",
    "            elif all_parts[idx] in ('and', 'AND'):\n",
    "                query_new_parts.append('&')\n",
    "            elif all_parts[idx] in ('or', 'OR'):\n",
    "                query_new_parts.append('|')\n",
    "            elif all_parts[idx] in ('not', 'NOT'):\n",
    "                query_new_parts.append('-')\n",
    "            elif idx + 1 < count_parts and all_parts[idx+1] not in (' ', ')'): #被分词切开的短语部分回收至缓存\n",
    "                cache += all_parts[idx]\n",
    "            else:\n",
    "                query_new_parts.append(\"self.word_match('{}')\".format(cache + all_parts[idx]))\n",
    "                cache = '' #合并完成清空缓存\n",
    "            idx += 1\n",
    "        query_new = ''.join(query_new_parts)\n",
    "        return query_new\n",
    "\n",
    "    def highlighter(self, doc, word):\n",
    "        \"\"\"用word对doc进行HTML高亮\n",
    "        \n",
    "        Args:\n",
    "            doc:需要高亮的文档\n",
    "            word:要进行高亮的关键词(查询)\n",
    "            \n",
    "        Returns:\n",
    "            返回对关键词(查询)进行高亮的文档\n",
    "        \"\"\"\n",
    "        for part in list(jieba.cut(word)):\n",
    "            #TODO(CHG):短语高亮需要先分词\n",
    "            if part not in ('(', ')', 'and', 'AND', 'or', 'OR', 'NOT', 'not', ' '):\n",
    "                doc = doc.replace(part, '<span style=\"color:red\">{}</span>'.format(part))\n",
    "        return doc\n",
    "\n",
    "    def search(self, query):\n",
    "        \"\"\"用query进行查询返回结果文档列表\n",
    "        \n",
    "        Args:\n",
    "            query:用户的(复合)布尔查询字符串\n",
    "            \n",
    "        Returns:\n",
    "            复合查询要求的(高亮)文档结果列表\n",
    "        \"\"\"\n",
    "        result = []\n",
    "        query_new = self.conv_query(query)\n",
    "        for did in eval(query_new):\n",
    "            result.append(self.highlighter(self.doc_list[did], query))\n",
    "        return result"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "searcher = SearcherIIndex('titles.txt')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "ename": "SyntaxError",
     "evalue": "invalid syntax (<string>, line 1)",
     "output_type": "error",
     "traceback": [
      "Traceback \u001b[0;36m(most recent call last)\u001b[0m:\n",
      "  File \u001b[1;32m\"/usr/local/lib/python3.7/site-packages/IPython/core/interactiveshell.py\"\u001b[0m, line \u001b[1;32m3326\u001b[0m, in \u001b[1;35mrun_code\u001b[0m\n    exec(code_obj, self.user_global_ns, self.user_ns)\n",
      "  File \u001b[1;32m\"<ipython-input-14-2d58ef1ccd80>\"\u001b[0m, line \u001b[1;32m2\u001b[0m, in \u001b[1;35m<module>\u001b[0m\n    for doc in searcher.search(query):\n",
      "\u001b[0;36m  File \u001b[0;32m\"<ipython-input-11-adf5b7413ee9>\"\u001b[0;36m, line \u001b[0;32m131\u001b[0;36m, in \u001b[0;35msearch\u001b[0;36m\u001b[0m\n\u001b[0;31m    for did in eval(query_new):\u001b[0m\n",
      "\u001b[0;36m  File \u001b[0;32m\"<string>\"\u001b[0;36m, line \u001b[0;32m1\u001b[0m\n\u001b[0;31m    self.word_match('苹果') self.word_match('芯片') & (self.word_match('三星') | self.word_match('华为'))\u001b[0m\n\u001b[0m                             ^\u001b[0m\n\u001b[0;31mSyntaxError\u001b[0m\u001b[0;31m:\u001b[0m invalid syntax\n"
     ]
    }
   ],
   "source": [
    "query = '苹果 芯片 and (三星 or 华为)'\n",
    "for doc in searcher.search(query):\n",
    "    display(HTML(doc))"
   ]
  }
 ],
 "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.7.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}