Internet_Archive.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": [],
      "authorship_tag": "ABX9TyPHXaJlaTkB7nKgH5MkcCh8",
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/ieg-dhr/Notebooks4Historical_Newspapers/blob/main/Internet_Archive.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Accessing Historical Sources from the Internet Archive using the internetarchive.api\n",
        "\n",
        "Created by Sarah Oberbichler [![ORCID](https://info.orcid.org/wp-content/uploads/2019/11/orcid_16x16.png)](https://orcid.org/0000-0002-1031-2759)\n",
        "\n",
        "Internet Archive is a non-profit library of millions of free books, movies, software, music, websites, newspapers, and more.\n",
        "\n",
        "Internetarchive is a python interface to archive.org. This notebook uses code provided by the internet archive (copyright 2012-2019 by Internet Archive; License AGPL 3). A documentation on how to use this python interface is availble here: https://archive.org/developers/internetarchive/internetarchive.html#module-internetarchive.api\n",
        "\n",
        "###This notebook shows how:\n",
        "\n",
        "\n",
        "*   to get the identifiers of a document (with or without in combination with a search keyword)\n",
        "*   to download the documents with their metadata\n",
        "*   to create a DataFrame bringing metadata and fulltext together\n",
        "*   to reduce the full text to a specific context window\n",
        "*   to exort the DataFrame as an Excel file for further processing\n",
        "\n",
        "\n",
        "\n",
        "\n"
      ],
      "metadata": {
        "id": "Pr7odOTxVJNQ"
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Setting up the requirements for the internetarchive.api"
      ],
      "metadata": {
        "id": "ZS-ISLgCQwPx"
      }
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "198iYNGjvuwC"
      },
      "outputs": [],
      "source": [
        "pip install internetarchive"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "pip install --upgrade internetarchive"
      ],
      "metadata": {
        "id": "a94ug5fmv4Rh"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# @markdown ####  Get the identifiers of all documents in a collection from a specific creator. Write the name of the newspaper (e.g., La Stampa) and the collection (e.g., newspapers). You can also choose a date range by adding --> AND date:[1913-10-17 TO 1944-11-17]:\n",
        "# Initialize an ArchiveSession\n",
        "import pandas as pd\n",
        "from internetarchive.session import ArchiveSession\n",
        "from internetarchive.search import Search\n",
        "import os\n",
        "import pandas as pd\n",
        "import xml.etree.ElementTree as ET\n",
        "from internetarchive import get_item\n",
        "s = ArchiveSession()\n",
        "add = \"title:(La patria del friuli) AND collection:(newspapers) AND date:[1909-01-01 TO 1912-01-01]\" # @param {type:\"string\"}\n",
        "\n",
        "# Perform the search\n",
        "search = Search(s, add)\n",
        "\n",
        "# Collect search results into a list of dictionaries\n",
        "collection_identifiers = []\n",
        "for result in search:\n",
        "    collection_identifiers.append(result['identifier'])\n",
        "\n",
        "print(collection_identifiers)\n",
        "print(len(collection_identifiers))"
      ],
      "metadata": {
        "id": "sRt33awuGBw9"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "\n",
        "# @markdown #### Or get the identifiers of documents that contain a specific keyword.\n",
        "# @markdown #####  Write the text of the first part of the identifier (e.g., lastampa):\n",
        "\n",
        "Collection = \"LaPatriadelFriuli\" # @param {type:\"string\"}\n",
        "# @markdown #####  Search the full text with a keyword:\n",
        "Keyword = \"terremoto\" # @param {type:\"string\"}\n",
        "\n",
        "# Initialize an ArchiveSession\n",
        "s = ArchiveSession()\n",
        "\n",
        "# Perform the search\n",
        "search = Search(s, Keyword, full_text_search = True)\n",
        "\n",
        "# Collect search results into a list of dictionaries\n",
        "results_list = []\n",
        "for result in search:\n",
        "    results_list.append(result)\n",
        "\n",
        "keyword_identifiers = []\n",
        "identifier_collection = Collection\n",
        "\n",
        "# Iterate over the search results and extract identifiers\n",
        "for result in results_list:\n",
        "    identifier = result['_id'].split('|')[0]\n",
        "    if identifier_collection in identifier:\n",
        "        keyword_identifiers.append(identifier)\n",
        "\n",
        "print(keyword_identifiers)\n",
        "print(len(keyword_identifiers))\n"
      ],
      "metadata": {
        "id": "weyoZa3z7ejk"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "import re\n",
        "from datetime import datetime\n",
        "# @markdown #### You can also select a date range for your keyword search\n",
        "# @markdown #### Insert the Start Date:\n",
        "\n",
        "Start_year = 1908 # @param {type:\"integer\"}\n",
        "# @markdown #####  Insert the End Date:\n",
        "End_year = 1915 # @param {type:\"integer\"}\n",
        "\n",
        "\n",
        "# Define the date range\n",
        "start_year = Start_year\n",
        "end_year = End_year\n",
        "\n",
        "import re\n",
        "from datetime import datetime\n",
        "\n",
        "# Function to extract the year from the identifier\n",
        "def extract_year(identifier):\n",
        "    # Use regex to find a 4-digit number that represents a year\n",
        "    match = re.search(r'_(\\d{4})_', identifier)\n",
        "    if match:\n",
        "        year_str = match.group(1)\n",
        "        try:\n",
        "            # Convert the year string to a datetime object\n",
        "            return datetime.strptime(year_str, '%Y')\n",
        "        except ValueError:\n",
        "            print(f\"Warning: Invalid date format in identifier: {identifier}\")\n",
        "            return None\n",
        "    else:\n",
        "        print(f\"Warning: Could not find a valid year in identifier: {identifier}\")\n",
        "        return None\n",
        "\n",
        "\n",
        "\n",
        "# Filter the identifiers based on the date range\n",
        "keyword_date_identifiers = []\n",
        "for identifier in keyword_identifiers:\n",
        "    year = extract_year(identifier)\n",
        "    if year is not None and start_year <= year.year <= end_year:\n",
        "        keyword_date_identifiers.append(identifier)\n",
        "\n",
        "print(\"Filtered Identifiers:\", keyword_date_identifiers)\n",
        "print(\"Count:\", len(keyword_date_identifiers))"
      ],
      "metadata": {
        "id": "VtHZyf-fcpQz"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Download the documents with their metadata and create a DataFrame"
      ],
      "metadata": {
        "id": "mpinYK5MKWg8"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "# @markdown #####  The documents can be downloaded as text files, xml, json, image files, pdf's, etc. The advanced search page of the internet archive gives an overview of the possible formats https://archive.org/\n",
        "# @markdown #####  Select the list of indentifiers here:\n",
        "Indentifier_List = collection_identifiers # @param [\"collection_identifiers\", \"keyword_identifiers\", \"keyword_date_identifiers\"] {type:\"raw\"}\n",
        "\n",
        "\n",
        "# @markdown #####  Select the format here:\n",
        "Format = \"djvu.txt\" # @param {type:\"string\"}\n",
        "\n",
        "# Create an empty list to store data\n",
        "data = []\n",
        "\n",
        "\n",
        "# Define a function to extract metadata from XML files\n",
        "def extract_metadata(xml_file):\n",
        "    metadata = {}\n",
        "    tree = ET.parse(xml_file)\n",
        "    root = tree.getroot()\n",
        "    for child in root:\n",
        "        metadata[child.tag] = child.text\n",
        "    return metadata\n",
        "\n",
        "\n",
        "# Process each identifier\n",
        "for identifier in Indentifier_List:\n",
        "    try:\n",
        "        # Get the item from Internet Archive\n",
        "        item = get_item(identifier)\n",
        "        # Get all files associated with the item\n",
        "        files = item.get_files()\n",
        "        # Process each file\n",
        "        metadata = None  # Initialize metadata\n",
        "        for file in files:\n",
        "            try:\n",
        "                # Check if the file name contains the specific extension\n",
        "                if file.name.lower().endswith(Format):\n",
        "                    print(f\"Downloading text file: {file.name}\")\n",
        "                    # Download the text file\n",
        "                    file.download(f'./downloads/{file.name}', verbose=True)\n",
        "                    with open(f\"./downloads/{file.name}\", 'r', encoding='utf-8') as f:\n",
        "                        text = f.read()\n",
        "                # Check if the file name contains the specific extension \".meta.xml\"\n",
        "                if file.name.lower().endswith('meta.xml'):\n",
        "                    print(f\"Downloading metadata file: {file.name}\")\n",
        "                    # Download the metadata file\n",
        "                    file.download(f'./downloads/{file.name}', verbose=True)\n",
        "                    # Extract metadata from XML file\n",
        "                    metadata = extract_metadata(f'./downloads/{file.name}')\n",
        "            except Exception as e:\n",
        "                print(f\"Error processing file {file.name} for identifier {identifier}: {e}\")\n",
        "        # Add the content and metadata to the data list\n",
        "        data.append({ **metadata, 'content': text,})\n",
        "    except Exception as e:\n",
        "        print(f\"Error processing identifier {identifier}: {e}\")\n",
        "\n",
        "# Create a DataFrame from the collected data\n",
        "df = pd.DataFrame(data)\n",
        "\n",
        "# Display the DataFrame\n",
        "df['content'].replace(\"\\n\", \"\")\n",
        "df\n"
      ],
      "metadata": {
        "id": "parpcE41C9-f"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# @markdown #### For the full text we can narrow down the text surrounding the keyword in order to reduce the input tokens for the model. Choose the size of the context window here:\n",
        "context_window = 8000 # @param {type:\"number\"}\n",
        "import pandas as pd\n",
        "from typing import List, Set, Tuple\n",
        "\n",
        "def extract_unique_contexts(df: pd.DataFrame, keywords: List[str], window_size: int = 8000) -> pd.DataFrame:\n",
        "    \"\"\"\n",
        "    Extract context windows for multiple keywords while preventing duplicates.\n",
        "\n",
        "    Args:\n",
        "        df: DataFrame containing the text column\n",
        "        keywords: List of keywords to search for\n",
        "        window_size: Size of the context window (default: 2000)\n",
        "\n",
        "    Returns:\n",
        "        DataFrame with new context column\n",
        "    \"\"\"\n",
        "\n",
        "    def get_context_bounds(text: str, keyword: str, window_size: int) -> Tuple[int, int]:\n",
        "        \"\"\"Get start and end indices for context window.\"\"\"\n",
        "        index = text.find(keyword)\n",
        "        if index == -1:\n",
        "            return (-1, -1)\n",
        "\n",
        "        start_index = max(0, index - window_size)\n",
        "        end_index = min(len(text), index + len(keyword) + window_size)\n",
        "        return (start_index, end_index)\n",
        "\n",
        "    def check_overlap(bounds: Tuple[int, int], existing_bounds: Set[Tuple[int, int]]) -> bool:\n",
        "        \"\"\"Check if new bounds overlap with any existing bounds.\"\"\"\n",
        "        start, end = bounds\n",
        "        for existing_start, existing_end in existing_bounds:\n",
        "            if not (end < existing_start or start > existing_end):\n",
        "                return True\n",
        "        return False\n",
        "\n",
        "    # Create a copy of the DataFrame to avoid modifying the original\n",
        "    result_df = df.copy()\n",
        "\n",
        "    # Initialize columns for contexts\n",
        "    result_df['context'] = ''\n",
        "    result_df['keyword_found'] = ''\n",
        "\n",
        "    # Process each row\n",
        "    for idx, row in result_df.iterrows():\n",
        "        text = row['content']\n",
        "        if not isinstance(text, str):\n",
        "            continue\n",
        "\n",
        "        used_bounds: Set[Tuple[int, int]] = set()\n",
        "        contexts = []\n",
        "        keywords_found = []\n",
        "\n",
        "        # Process each keyword\n",
        "        for keyword in keywords:\n",
        "            bounds = get_context_bounds(text, keyword, window_size)\n",
        "\n",
        "            if bounds[0] == -1:  # Keyword not found\n",
        "                continue\n",
        "\n",
        "            # Check for overlap with existing contexts\n",
        "            if not check_overlap(bounds, used_bounds):\n",
        "                used_bounds.add(bounds)\n",
        "                context = text[bounds[0]:bounds[1]]\n",
        "                contexts.append(context)\n",
        "                keywords_found.append(keyword)\n",
        "\n",
        "        # Join all unique contexts and keywords\n",
        "        result_df.at[idx, 'context'] = ' ||| '.join(contexts) if contexts else 'No unique context found'\n",
        "        result_df.at[idx, 'keyword_found'] = ', '.join(keywords_found) if keywords_found else 'No keywords found'\n",
        "\n",
        "    return result_df\n",
        "\n",
        "# Example usage:\n",
        "keywords = [\"terremoto\", \"erremoto\", \"seismo\", \"terremoti\", \"erremoti\"]  # Add your keywords\n",
        "\n",
        "\n",
        "# Assuming your DataFrame is called 'df' and has a 'text' column\n",
        "processed_df = extract_unique_contexts(df, keywords, context_window)"
      ],
      "metadata": {
        "id": "_0Im8MPX0CWd"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "processed_df"
      ],
      "metadata": {
        "id": "C_128UGTUtso"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "processed_df.to_excel('name.xlsx', index=False)"
      ],
      "metadata": {
        "id": "cy2EYUHdzjQa"
      },
      "execution_count": null,
      "outputs": []
    }
  ]
}