dense-v2-Copy2.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3b55c849",
   "metadata": {},
   "outputs": [],
   "source": [
    "import spacy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a4a0eb40",
   "metadata": {},
   "outputs": [],
   "source": [
    "import logging\n",
    "\n",
    "logging.basicConfig(format=\"%(levelname)s - %(name)s -  %(message)s\", level=logging.WARNING)\n",
    "logging.getLogger(\"haystack\").setLevel(logging.INFO)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f03ce85d",
   "metadata": {},
   "outputs": [],
   "source": [
    "import PyPDF2\n",
    "import os"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a22de541",
   "metadata": {},
   "outputs": [],
   "source": [
    "with open('stop.txt',encoding='utf-8') as stopwords_file:\n",
    "    stopwords = set(map(lambda x: x[:-1],stopwords_file.readlines()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1fd524f1",
   "metadata": {},
   "outputs": [],
   "source": [
    "class LemmaMaker:\n",
    "    \n",
    "    def __init__(self):\n",
    "        self.spacy = spacy.load('pl_core_news_lg')\n",
    "    \n",
    "    def lemmatise(self, word):\n",
    "        return self.spacy(word)[0].lemma_"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "daf76078",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "lemmatiser = LemmaMaker() "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "13a62da2",
   "metadata": {},
   "outputs": [],
   "source": [
    "from haystack.document_stores import InMemoryDocumentStore\n",
    "\n",
    "document_store_pages = InMemoryDocumentStore(use_bm25=True, embedding_dim=1024)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "150887bb",
   "metadata": {},
   "outputs": [],
   "source": [
    "processed_directory_pages = 'processed_documents_pages_intfloat'\n",
    "os.makedirs(processed_directory_pages, exist_ok=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a195485a",
   "metadata": {},
   "outputs": [],
   "source": [
    "def pdf_txt_pages(directory_path):\n",
    "    for filename in os.listdir(directory_path):\n",
    "        reader = PyPDF2.PdfReader(os.path.join(directory_path,filename))\n",
    "        new_file_name = filename.replace(\".pdf\",\"_\")\n",
    "        for index,page in enumerate(reader.pages):\n",
    "            with open(os.path.join(processed_directory_pages,f\"{new_file_name}{index}.txt\"),'w',encoding='utf-8') as output_file:\n",
    "                text = page.extract_text()\n",
    "                text = \" \".join(word for word in map(lambda x: lemmatiser.lemmatise(x), text.split()) if word not in stopwords)\n",
    "                text = \"passage: \" + text\n",
    "                output_file.write(text.replace('\\n',''))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "393e01c9",
   "metadata": {},
   "outputs": [],
   "source": [
    "pdf_txt_pages('input')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ff790b71",
   "metadata": {},
   "outputs": [],
   "source": [
    "from haystack.pipelines.standard_pipelines import TextIndexingPipeline\n",
    "from haystack.nodes import PreProcessor\n",
    "\n",
    "preprocessor_pages = PreProcessor(\n",
    "    clean_empty_lines=True,\n",
    "    clean_whitespace=True,\n",
    "    clean_header_footer=True,\n",
    "    remove_substrings=None,\n",
    "    split_by=\"word\",\n",
    "    split_length=1000,\n",
    "    split_respect_sentence_boundary=True,\n",
    "    split_overlap=0,\n",
    "    max_chars_check=10_000\n",
    ")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "dd41c68c",
   "metadata": {},
   "outputs": [],
   "source": [
    "files_to_index_pages = [os.path.join(processed_directory_pages,f) for f in os.listdir(processed_directory_pages)]\n",
    "indexing_pipeline_pages = TextIndexingPipeline(document_store_pages, preprocessor=preprocessor_pages)\n",
    "return_val = indexing_pipeline_pages.run_batch(file_paths=files_to_index_pages)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2c2d7583",
   "metadata": {},
   "outputs": [],
   "source": [
    "id_dict = {k:v for k,v in zip(map(lambda x: x.id,return_val['documents']),return_val['file_paths'])}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8f86b1b7",
   "metadata": {},
   "outputs": [],
   "source": [
    "from haystack.nodes import EmbeddingRetriever\n",
    "\n",
    "retriever_pages = EmbeddingRetriever(\n",
    "    document_store=document_store_pages,\n",
    "    embedding_model=\"intfloat/multilingual-e5-large\",\n",
    ")\n",
    "document_store_pages.update_embeddings(retriever_pages)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "46244dc8",
   "metadata": {},
   "outputs": [],
   "source": [
    "from haystack import Pipeline\n",
    "\n",
    "\n",
    "p_p = Pipeline()\n",
    "p_p.add_node(component=retriever_pages, name=\"ESRetriever1\", inputs=[\"Query\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1903ca38",
   "metadata": {},
   "outputs": [],
   "source": [
    "cutoff_t = 0.9984"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7c4960a1",
   "metadata": {},
   "outputs": [],
   "source": [
    "accept_t = 0.9"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3190d3cb",
   "metadata": {},
   "outputs": [],
   "source": [
    "docs = p_p.run(query=\"ROLAP\")[\"documents\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2e9b78c9",
   "metadata": {},
   "outputs": [],
   "source": [
    "def process_query(pipeline, query, cutoff_t = 0.9984, accept_t = 0.9984):\n",
    "    \n",
    "    docs = pipeline.run(query=query)[\"documents\"]\n",
    "    \n",
    "    selected_docs = list(filter(lambda x: x.score > cutoff_t, docs))\n",
    "    docs = selected_docs if len(selected_docs) > 0 else docs\n",
    "    \n",
    "    split_pattern = r'( |\\.|,)i |(oraz)|,|\\.|\\(|\\)'\n",
    "    phrase_parts = re.split(split_pattern, query)\n",
    "    print(phrase_parts)\n",
    "    combined_text = ' '.join(map(lambda x: x.content, docs))\n",
    "    \n",
    "    for phrase_part in phrase_parts:\n",
    "        if phrase_part and phrase_part not in combined_text:\n",
    "            new_docs = pipeline.run(query=phrase_part)[\"documents\"]\n",
    "            selected_docs = list(filter(lambda x: x.score > accept_t, new_docs))\n",
    "            new_docs = selected_docs if len(selected_docs) > 0 else new_docs\n",
    "            docs.extend(new_docs)\n",
    "            \n",
    "    return docs\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8a54a1fe",
   "metadata": {},
   "outputs": [],
   "source": [
    "import re\n",
    "\n",
    "def run_queries(pipeline, query_dir, top_answers=1):\n",
    "    query_files = [os.path.join(query_dir,f) for f in os.listdir(query_dir)]\n",
    "    res = {}\n",
    "    for file_path in query_files:\n",
    "        with open(file_path,\"r\",encoding=\"utf-8\") as queries:\n",
    "            for query_line in queries:\n",
    "                query = re.sub(r'\\s*[0-9]+\\.\\s*','',query_line)\n",
    "                modified_query = \"query: \" + (\" \".join(map(lambda x: lemmatiser.lemmatise(x), query.split()))).lower()\n",
    "                docs = process_query(pipeline, modified_query)\n",
    "                res[query] = set(map(lambda x: id_dict[x.id], docs))\n",
    "            \n",
    "                break\n",
    "    \n",
    "    return res"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e1c24058",
   "metadata": {},
   "outputs": [],
   "source": [
    "def print_dict(dictionary):\n",
    "    for key in dictionary.keys():\n",
    "        print(key)\n",
    "        print(dictionary[key])\n",
    "        print()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1c02d893",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "print_dict(run_queries(p_p,'queries'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6d773722",
   "metadata": {},
   "outputs": [],
   "source": [
    "import PyPDF2\n",
    "from PyPDF2 import PdfReader, PdfWriter\n",
    "from PyPDF2.generic import AnnotationBuilder"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ab983094",
   "metadata": {},
   "outputs": [],
   "source": [
    "answers = run_queries(p_p,'queries')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f902a739",
   "metadata": {},
   "outputs": [],
   "source": [
    "from reportlab.lib.pagesizes import landscape\n",
    "from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer\n",
    "from reportlab.lib.styles import getSampleStyleSheet\n",
    "from reportlab.lib.units import inch\n",
    "\n",
    "# Create a function to generate the title page\n",
    "def create_title_page(pdf_file, text):\n",
    "    doc = SimpleDocTemplate(pdf_file, pagesize=landscape((10 * inch, 7.5 * inch)))\n",
    "\n",
    "    # Create a list to hold the elements of the title page\n",
    "    elements = []\n",
    "\n",
    "    # Define the title style\n",
    "    styles = getSampleStyleSheet()\n",
    "    title_style = styles[\"Title\"]\n",
    "\n",
    "    # Create the title and add it to the elements list\n",
    "    title = Paragraph(text, title_style)\n",
    "    elements.append(title)\n",
    "\n",
    "    # Add some space below the title\n",
    "    elements.append(Spacer(1, 0.5 * inch))\n",
    "\n",
    "    # Add any other content you want on the title page\n",
    "    # For example:\n",
    "    # author = Paragraph(\"Author: Your Name\", styles[\"Normal\"])\n",
    "    # elements.append(author)\n",
    "\n",
    "    # Build the title page\n",
    "    doc.build(elements)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9cf6654d",
   "metadata": {},
   "outputs": [],
   "source": [
    "def create_page(output_pdf_writer, welcome_text):\n",
    "    page = PyPDF2.PageObject.create_blank_page(width=500, height=500)\n",
    "    page.mergeScaledTranslatedPage(welcome_text, 0.5, 100, 200)\n",
    "    output_pdf_writer.addPage(page)\n",
    "\n",
    "def get_files(file_set):\n",
    "    input_directory = \"input\"\n",
    "    file_set = map(lambda x: x.replace(processed_directory_pages,input_directory), file_set)\n",
    "    file_set = map(lambda x: (\"_\".join((x.split(\"_\")[:-1]))  + \".pdf\", int(x.split(\"_\")[-1].split(\".\")[0])) , file_set)\n",
    "    return file_set\n",
    "    \n",
    "def gen_pdf(queries, output_pdf_name):\n",
    "    output_pdf_writer = PyPDF2.PdfWriter()\n",
    "    curr_page = 0\n",
    "    for query in queries.keys():\n",
    "        \n",
    "        create_title_page(\"temp.pdf\",query)\n",
    "        with open(\"temp.pdf\", 'rb') as temp_page_pdf_file:\n",
    "            temp_page_pdf_reader = PyPDF2.PdfReader(temp_page_pdf_file)\n",
    "            page = temp_page_pdf_reader.pages[0]\n",
    "            output_pdf_writer.add_page(page)\n",
    "\n",
    "        curr_page += 1\n",
    "        \n",
    "        for input_pdf, page in get_files(queries[query]):\n",
    "            print(input_pdf, page)\n",
    "            \n",
    "            with open(input_pdf, 'rb') as pdf_file:\n",
    "                pdf_reader = PyPDF2.PdfReader(pdf_file)\n",
    "                out_page = pdf_reader.pages[page]\n",
    "                output_pdf_writer.add_page(out_page)\n",
    "                curr_page += 1\n",
    "\n",
    "    with open(output_pdf_name, 'wb') as output_file:\n",
    "        output_pdf_writer.write(output_file)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "61cfbd49",
   "metadata": {},
   "outputs": [],
   "source": [
    "gen_pdf(answers, \"res.pdf\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "01c67b4c",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.11.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}