ProtoCode_Content_Extraction.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "02f26c35",
   "metadata": {},
   "source": [
    "# Content Extraction\n",
    "script to obtain protocols from article url\n",
    "   \n",
    "## input files\n",
    "- **config file** \n",
    "\n",
    "    - In cofig file, please specify the following three items:\n",
    "\n",
    "        - **input_link** : Link of the website (BioRxiv) \n",
    "        - **output_path** (string) : path for the obtained protocols **\". /protocols/\"** \n",
    "        - **num_keywords** (int) : Number of keywords in each protocol - Higher numer = Higher quality \n",
    "        \n",
    "## output files\n",
    "\n",
    "The obtained protocols are stored in the **\". /protocols/\"** folder.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "eba7fe2b",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import requests\n",
    "import re\n",
    "import configparser\n",
    "import errno\n",
    "import logging\n",
    "import ast\n",
    "import csv\n",
    "\n",
    "from bs4 import BeautifulSoup"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "3031959d",
   "metadata": {},
   "outputs": [],
   "source": [
    "config_path = \"./content_config.ini\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7e0a028d",
   "metadata": {},
   "source": [
    "## There are two different type of functions to extract content based on different website\n",
    "### e.g. If get_text is not working, please use the get_text2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "12ae3daf",
   "metadata": {},
   "outputs": [],
   "source": [
    "def modify_biorxiv_url(url, logger):\n",
    "    if 'biorxiv' in url:\n",
    "        if '.full' not in url:\n",
    "            # If biorxiv.org is present and .full is not in the URL, add .full\n",
    "            modified_url = url + '.full'\n",
    "        else:\n",
    "            # If .full is already in the URL, keep it unchanged\n",
    "            modified_url = url\n",
    "        return modified_url\n",
    "    else:\n",
    "        logger.error(f\"The input link is not from biorxiv!\")\n",
    "        return url\n",
    "    \n",
    "def get_text(url, exclude_sections):\n",
    "#     full_url = modify_biorxiv_url(url, logger)\n",
    "    response = requests.get(url)\n",
    "    soup = BeautifulSoup(response.content, \"html.parser\")\n",
    "    sections = soup.find_all(\"div\", class_=\"section\")\n",
    "\n",
    "    section_titles = []\n",
    "    section_texts = []\n",
    "    subsections = {}\n",
    "    sub_subsections = {}\n",
    "    \n",
    "    paragraph = \"\"  # String variable to store the paragraph\n",
    "    \n",
    "    # Find and filter headings\n",
    "    for section in sections:\n",
    "        section_title = section.find(\"h2\")\n",
    "        if section_title:\n",
    "            section_title = section_title.text.strip()\n",
    "            section_stripped = re.sub(r'^[\\d.]+\\s*', '', section_title)\n",
    "            if any(exclude_section.lower() in section_stripped.lower() for exclude_section in exclude_sections):\n",
    "                continue\n",
    "            else:\n",
    "                section_titles.append(section_title)\n",
    "#                 print(section_title)\n",
    "                section_text = section.get_text()\n",
    "                section_text = section_text.replace(section_title, \"\").strip()\n",
    "                section_texts.append(section_text)\n",
    "#                 print(section_text)\n",
    "                paragraph += f\"{section_title}\\n\"\n",
    "                paragraph += f\"{section_text}\\n\"\n",
    "                \n",
    "                # subtitles under titles\n",
    "                subtitles = section.find_all(\"h3\")\n",
    "                for subtitle in subtitles:\n",
    "                    subtitle_text = subtitle.text.strip()\n",
    "                    subtitle_content = subtitle.find_next(\"p\").get_text()\n",
    "                    if subtitle_text in subsections:\n",
    "                        subsections[subtitle_text].append(subtitle_content)\n",
    "                    else:\n",
    "                        subsections[subtitle_text] = [subtitle_content]\n",
    "                \n",
    "                # titles under subtitles\n",
    "                sub_subtitles = section.find_all(\"h4\")\n",
    "                if len(sub_subtitles) == 0:\n",
    "                    continue\n",
    "                else:\n",
    "                    for sub_subtitle in sub_subtitles:\n",
    "                        sub_subtitle_text = sub_subtitle.text.strip()\n",
    "                        sub_subtitle_content = sub_subtitle.find_next(\"p\").get_text()\n",
    "                        if sub_subtitle_text in sub_subsections:\n",
    "                            sub_subsections[sub_subtitle_text].append(sub_subtitle_content)\n",
    "                        else:\n",
    "                            sub_subsections[sub_subtitle_text] = [sub_subtitle_content]\n",
    "                \n",
    "    return paragraph, section_titles, subsections, sub_subsections\n",
    "\n",
    "def get_text2(url, exclude_sections):\n",
    "#     full_url = modify_biorxiv_url(url, logger)\n",
    "    response = requests.get(url)\n",
    "    soup = BeautifulSoup(response.content, \"html.parser\")\n",
    "    sections = soup.find_all(\"section\")\n",
    "    \n",
    "    section_titles = []\n",
    "    section_texts = []\n",
    "    subsections = {}\n",
    "    sub_subsections = {}\n",
    "    paragraph = \"\"  # String variable to store the paragraph\n",
    "    \n",
    "    # Find and filter headings\n",
    "    for section in sections:\n",
    "        section_title_element = section.find(\"h2\", class_=\"c-article-section__title\")\n",
    "        if section_title_element:\n",
    "            section_title = section_title_element.text.strip()\n",
    "            section_stripped = re.sub(r'^[\\d.]+\\s*', '', section_title)\n",
    "            if any(exclude_section.lower() in section_stripped.lower() for exclude_section in exclude_sections):\n",
    "                continue\n",
    "            else:\n",
    "                section_titles.append(section_title)\n",
    "                #print(section_title)\n",
    "                section_text = section.get_text()\n",
    "                section_text = section_text.replace(section_title, \"\").strip()\n",
    "                section_texts.append(section_text)\n",
    "                #print(section_text)\n",
    "                paragraph += f\"{section_title}\\n\"\n",
    "                paragraph += f\"{section_text}\\n\"\n",
    "                \n",
    "                subtitles = section.find_all(\"h3\")\n",
    "                # subtitles under titles\n",
    "                for subtitle in subtitles:\n",
    "                    subtitle_text = subtitle.text.strip()\n",
    "                    subtitle_content = subtitle.find_next(\"p\").get_text()\n",
    "                    if subtitle_text in subsections:\n",
    "                        subsections[subtitle_text].append(subtitle_content)\n",
    "                    else:\n",
    "                        subsections[subtitle_text] = subtitle_content\n",
    "                \n",
    "                # titles under subtitles\n",
    "                sub_subtitles = section.find_all(\"h4\")\n",
    "                if len(sub_subtitles) == 0:\n",
    "                    continue\n",
    "                else:\n",
    "                    for sub_subtitle in sub_subtitles:\n",
    "                        sub_subtitle_text = sub_subtitle.text.strip()\n",
    "                        sub_subtitle_content = sub_subtitle.find_next(\"p\").get_text()\n",
    "                        if sub_subtitle_text in sub_subsections:\n",
    "                            sub_subsections[sub_subtitle_text].append(sub_subtitle_content)\n",
    "                        else:\n",
    "                            sub_subsections[sub_subtitle_text] = sub_subtitle_content\n",
    "                \n",
    "    return paragraph, section_titles, subsections, sub_subsections\n",
    "\n",
    "def prt_selection(sections, protocol_title, subsections_exclude):\n",
    "    # Initiate infomation sections and protocol sections\n",
    "    info_sections = {}\n",
    "    protocol_sections = {}\n",
    "    for heading, c in sections.items():\n",
    "        content = c[0]\n",
    "        \n",
    "        # Classify infomation section and protocol section\n",
    "        if any(exclude_section.lower() in heading.lower() for exclude_section in subsections_exclude):\n",
    "            continue\n",
    "        else:\n",
    "            if any(include_section.lower() in heading.lower() for include_section in protocol_title):\n",
    "                if heading in protocol_sections:\n",
    "                    protocol_sections[heading].append(content)\n",
    "                else:\n",
    "                    protocol_sections[heading] = content\n",
    "            else:\n",
    "                if heading in info_sections:\n",
    "                    info_sections[heading].append(content)\n",
    "                else:\n",
    "                    info_sections[heading] = content\n",
    "                    \n",
    "    return info_sections, protocol_sections\n",
    "\n",
    "def pcr_decider(sections, keywords, units, num): \n",
    "    protocol_sections = {}\n",
    "    k = {}\n",
    "    # Check if the input is a dictionary\n",
    "    if isinstance(sections, dict):\n",
    "        for heading, content in sections.items():\n",
    "            # count how many keywords in content\n",
    "            count_k = 0\n",
    "            count_u = 0\n",
    "            count = 0\n",
    "            k[heading] = []  # Initialize an empty list for each heading\n",
    "            for i in keywords:\n",
    "                keyword = i.lower()\n",
    "                if keyword in content.lower():\n",
    "                    count_k +=1\n",
    "                    k[heading].append(keyword)\n",
    "            for j in units:\n",
    "                unit = j.lower()\n",
    "                 # Pattern with optional space in front\n",
    "                pattern = r'(\\b|\\d)' + re.escape(unit) + r'\\b'\n",
    "                if re.search(pattern, content, re.IGNORECASE):\n",
    "                    count_u +=1\n",
    "                    k[heading].append(unit)\n",
    "\n",
    "            count = count_k + count_u   # Avoid situation that only units exist\n",
    "            \n",
    "            if count_k != 0:\n",
    "                if count >= num:\n",
    "                    if heading in protocol_sections:\n",
    "                        protocol_sections[heading].append(content)\n",
    "                    else:\n",
    "                        protocol_sections[heading] = content\n",
    "                        \n",
    "    # Cases if only one section    \n",
    "    else:    \n",
    "        count_k = 0\n",
    "        count_u = 0\n",
    "        count = 0\n",
    "        k = []\n",
    "        for i in keywords:\n",
    "            keyword = i.lower()\n",
    "            if keyword in sections.lower():\n",
    "                count_k +=1\n",
    "                k.append(keyword)\n",
    "        for j in units:\n",
    "            unit = j.lower()\n",
    "             # Pattern with optional space in front\n",
    "            pattern = r'(\\b|\\d)' + re.escape(unit) + r'\\b'\n",
    "            if re.search(pattern, sections, re.IGNORECASE):\n",
    "                count_u +=1\n",
    "                k.append(unit)\n",
    "        count = count_k + count_u   # Avoid situation that only units exist\n",
    "        \n",
    "        if count_k != 0:\n",
    "            if count >= num:\n",
    "                protocol_sections = sections\n",
    "\n",
    "    return protocol_sections, k\n",
    "\n",
    "def store_pcr(subsections, protocol_title, subsections_exclude, keywords, units, num):\n",
    "    protocols = []\n",
    "    new_protocol_sections = {}\n",
    "    k = []\n",
    "    info_sections = {}\n",
    "    protocol_sections = {}\n",
    "    # Process subsection\n",
    "    if subsections:\n",
    "        info_sections, protocol_sections = prt_selection(subsections, protocol_title, subsections_exclude)\n",
    "        if protocol_sections:\n",
    "            check_protocol_sections, k = pcr_decider(protocol_sections, keywords, units, num)\n",
    "            # Check information section if there is no protocol in filtered protocol section\n",
    "            if not check_protocol_sections:\n",
    "                temp_sec, k = pcr_decider(info_sections, keywords, units, num)\n",
    "                if temp_sec:\n",
    "                    new_protocol_sections = temp_sec\n",
    "            else:\n",
    "                new_protocol_sections = check_protocol_sections\n",
    "                \n",
    "    # Combines information sections and each protocol\n",
    "    if new_protocol_sections:\n",
    "        if not check_protocol_sections:\n",
    "            for h, c in new_protocol_sections.items():\n",
    "                temp_ = {}\n",
    "                temp_.update({h: c})\n",
    "                protocols.append(temp_)\n",
    "        else:\n",
    "            for h, c in new_protocol_sections.items():\n",
    "                temp_ = dict(info_sections)  # Create a new dictionary based on info_sections\n",
    "                temp_.update({h: c})\n",
    "                protocols.append(temp_)\n",
    "    return protocols, k, info_sections, protocol_sections"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "b287c6fb",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "def main(url, sections_exclude, num_keywords, subsections_exclude, keywords, units, protocol_title, results_save_path, logger):\n",
    "    \n",
    "    protocol = None\n",
    "    headers = [\"protocol index\", \"heading name\", \"content\"]\n",
    "    \n",
    "    try:\n",
    "        num_keywords = int(num_keywords)\n",
    "    except Exception as e:\n",
    "        logger.error(f\"Error occured in number of keywords selections:{str(e)}\")\n",
    "        raise ValueError(f\"Error occured in number of keywords selections:{str(e)}\")\n",
    "        \n",
    "    # Check whether the link is from BioRXiv or not\n",
    "    full_url = modify_biorxiv_url(url, logger)\n",
    "    \n",
    "    try:\n",
    "        # get text from filtered headings\n",
    "        spec_paragraph, titles, subsections, sub_subsections = get_text(full_url, sections_exclude)\n",
    "        if not spec_paragraph:\n",
    "            spec_paragraph, titles, subsections, sub_subsections = get_text2(full_url, sections_exclude)\n",
    "        \n",
    "    except Exception as e:\n",
    "        logger.error(f\"Error occured when filtering headings and extractint text:{str(e)}\")\n",
    "        raise ValueError(f\"Error occured in filtered headings and extractint text:{str(e)}\")\n",
    "        \n",
    "    if not spec_paragraph:\n",
    "        logger.error(f\"There is no text extracted from the link you've provided!\")\n",
    "        raise ValueError(\"There is no text extracted from the link you've provided!\")\n",
    "        \n",
    "    if not subsections and not sub_subsections:\n",
    "        try:\n",
    "            protocol, k = pcr_decider(spec_paragraph, keywords, units, num_keywords)\n",
    "        except Exception as e:\n",
    "            logger.error(f\"Error occured in obtain text from main headings:{str(e)}\")\n",
    "            raise ValueError(f\"Error occured in obtain text from main headings:{str(e)}\")\n",
    "    else:\n",
    "        # subsections\n",
    "        try:\n",
    "            subsection_protocol, k1, info, p = store_pcr(subsections, protocol_title, subsections_exclude, keywords, units, num_keywords)\n",
    "        except Exception as e:\n",
    "            logger.error(f\"Error occured in obtain text from subheadings:{str(e)}\")\n",
    "            raise ValueError(f\"Error occured in obtain text from subheadings:{str(e)}\")\n",
    "            \n",
    "        # sub-subsections\n",
    "        try:\n",
    "            sub_subsection_protocol, k2, _, _ = store_pcr(sub_subsections, protocol_title, subsections_exclude, keywords, units, num_keywords)\n",
    "        except Exception as e:\n",
    "            logger.error(f\"Error occured in obtain text from sub-subheadings:{str(e)}\")\n",
    "            raise ValueError(f\"Error occured in obtain text from sub-subheadings:{str(e)}\")\n",
    "            \n",
    "    #  Check whether the article mentions PCR or not     \n",
    "    if not protocol and not subsection_protocol and not sub_subsection_protocol:\n",
    "        logger.error(f\"There is no PCR mentioned in the article!\")\n",
    "        raise ValueError(\"There is no PCR mentioned in the article!\")\n",
    "\n",
    "    else:\n",
    "        # save content into csv file\n",
    "        with open(results_save_path, mode='w', newline='', encoding=\"utf-8\") as file:\n",
    "            writer = csv.writer(file)\n",
    "            # Write the headers to the CSV file\n",
    "            writer.writerow(headers)\n",
    "            index = 0\n",
    "\n",
    "            # Heading content\n",
    "            if protocol:\n",
    "                # Write each data row with protocol index, heading name, and content\n",
    "                writer.writerow([index, \"\", protocol])\n",
    "                index += 1  \n",
    "\n",
    "            # Subheading content        \n",
    "            if subsection_protocol:\n",
    "                for protocol_index, data_dict in enumerate(subsection_protocol, start=index):\n",
    "                    for subheading_name, subcontent in data_dict.items():\n",
    "                        # Write each data row with protocol index, heading name, and content\n",
    "                        writer.writerow([protocol_index, subheading_name, subcontent])\n",
    "                    index = protocol_index+1\n",
    "\n",
    "            # Sub-subheading content        \n",
    "            if sub_subsection_protocol:\n",
    "                for protocol_index, data_dict in enumerate(sub_subsection_protocol, start=index):\n",
    "                    for subsubheading_name, subsubcontent in data_dict.items():\n",
    "                        # Write each data row with protocol index, heading name, and content\n",
    "                        writer.writerow([protocol_index, subsubheading_name, subsubcontent])\n",
    "                    index = protocol_index+1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "c51c6cbd",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "ename": "ValueError",
     "evalue": "There is no text extracted from the link you've provided!",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mValueError\u001b[0m                                Traceback (most recent call last)",
      "Cell \u001b[1;32mIn[5], line 56\u001b[0m\n\u001b[0;32m     53\u001b[0m \u001b[38;5;28;01mexcept\u001b[39;00m \u001b[38;5;167;01mException\u001b[39;00m \u001b[38;5;28;01mas\u001b[39;00m e:\n\u001b[0;32m     54\u001b[0m     \u001b[38;5;28mprint\u001b[39m(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mError occured in reading setting.ini:\u001b[39m\u001b[38;5;124m\"\u001b[39m, e)\n\u001b[1;32m---> 56\u001b[0m \u001b[43mmain\u001b[49m\u001b[43m(\u001b[49m\u001b[43murl\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43msections_exclude\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mnum_keywords\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43msubsections_exclude\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mkeywords\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43munits\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mprotocol_title\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mresults_save_path\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mlogger\u001b[49m\u001b[43m)\u001b[49m\n",
      "Cell \u001b[1;32mIn[4], line 27\u001b[0m, in \u001b[0;36mmain\u001b[1;34m(url, sections_exclude, num_keywords, subsections_exclude, keywords, units, protocol_title, results_save_path, logger)\u001b[0m\n\u001b[0;32m     25\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;129;01mnot\u001b[39;00m spec_paragraph:\n\u001b[0;32m     26\u001b[0m     logger\u001b[38;5;241m.\u001b[39merror(\u001b[38;5;124mf\u001b[39m\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mThere is no text extracted from the link you\u001b[39m\u001b[38;5;124m'\u001b[39m\u001b[38;5;124mve provided!\u001b[39m\u001b[38;5;124m\"\u001b[39m)\n\u001b[1;32m---> 27\u001b[0m     \u001b[38;5;28;01mraise\u001b[39;00m \u001b[38;5;167;01mValueError\u001b[39;00m(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mThere is no text extracted from the link you\u001b[39m\u001b[38;5;124m'\u001b[39m\u001b[38;5;124mve provided!\u001b[39m\u001b[38;5;124m\"\u001b[39m)\n\u001b[0;32m     29\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;129;01mnot\u001b[39;00m subsections \u001b[38;5;129;01mand\u001b[39;00m \u001b[38;5;129;01mnot\u001b[39;00m sub_subsections:\n\u001b[0;32m     30\u001b[0m     \u001b[38;5;28;01mtry\u001b[39;00m:\n",
      "\u001b[1;31mValueError\u001b[0m: There is no text extracted from the link you've provided!"
     ]
    }
   ],
   "source": [
    "### read ini file\n",
    "config = configparser.ConfigParser()\n",
    "\n",
    "if not os.path.exists(config_path):\n",
    "    raise FileNotFoundError(errno.ENOENT, os.strerror(errno.ENOENT), config_path)\n",
    "\n",
    "try:\n",
    "    config.read(config_path)\n",
    "\n",
    "    ## input link\n",
    "    url = config.get(\"input_link\", \"url\")\n",
    "    \n",
    "    ## output dir\n",
    "    results_save_path = config.get(\"output_path\", \"out_path\")\n",
    "    \n",
    "    ## number of keywords\n",
    "    num_keywords = config.get(\"num_keywords\", \"num_keywords\")\n",
    "    \n",
    "    # Set the file path for the log file\n",
    "    log_file = \"./error.log\"\n",
    "\n",
    "    # Configure the logging settings\n",
    "    logger = logging.getLogger()\n",
    "    logger.setLevel(logging.ERROR)\n",
    "    logger.setLevel(logging.INFO)\n",
    "\n",
    "    # Create a FileHandler to save logs to the specified file path\n",
    "    file_handler = logging.FileHandler(log_file)\n",
    "    file_handler.setLevel(logging.ERROR)\n",
    "    file_handler.setLevel(logging.INFO)\n",
    "\n",
    "    # Create a formatter to customize the log message format\n",
    "    formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')\n",
    "    file_handler.setFormatter(formatter)\n",
    "    # Add the FileHandler to the logger\n",
    "    logger.addHandler(file_handler)\n",
    "\n",
    "    ## get sections\n",
    "    # Get the sections_exclude value as a string\n",
    "    sections_exclude = config.get(\"sections\", \"sections_exclude\")\n",
    "    subsections_exclude = config.get(\"sections\", \"subsections_exclude\")\n",
    "    keywords = config.get(\"sections\", \"keywords\")\n",
    "    units = config.get(\"sections\", \"units\")\n",
    "    protocol_title = config.get(\"sections\", \"protocol_title\") \n",
    "                                     \n",
    "    # Parse the string representation of the list into an actual list\n",
    "    sections_exclude = ast.literal_eval(sections_exclude)\n",
    "    subsections_exclude = ast.literal_eval(subsections_exclude)\n",
    "    keywords = ast.literal_eval(keywords)\n",
    "    units = ast.literal_eval(units)\n",
    "    protocol_title = ast.literal_eval(protocol_title)\n",
    "\n",
    "except Exception as e:\n",
    "    print(\"Error occured in reading setting.ini:\", e)\n",
    "\n",
    "main(url, sections_exclude, num_keywords, subsections_exclude, keywords, units, protocol_title, results_save_path, logger)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "230f6820",
   "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.9.17"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}