grao_tables_parsing.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "Untitled3.ipynb",
      "provenance": [],
      "collapsed_sections": [],
      "toc_visible": true,
      "authorship_tag": "ABX9TyOnPMvHd/9WyhksEeTmL6ox",
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/rumen-cholakov/SemanticWeb/blob/master/grao_tables_parsing.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "7q84-rt4geto",
        "colab_type": "text"
      },
      "source": [
        "## Imports"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "j5qHpKx8krm2",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "import pandas as pd\n",
        "import requests\n",
        "import pickle\n",
        "import regex\n",
        "import enum\n",
        "import os\n",
        "\n",
        "from typing import TypeVar, Callable, Sequence, List, Optional, Tuple\n",
        "from collections import namedtuple\n",
        "from bs4 import BeautifulSoup\n",
        "from functools import reduce"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "UFWn3xGIgiFz",
        "colab_type": "text"
      },
      "source": [
        "## Type Declarations"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "0KI4p5vSr6pt",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "class HeaderEnum(enum.Enum):\n",
        "  Old = 0\n",
        "  New = 1\n",
        "\n",
        "class TableTypeEnum(enum.Enum):\n",
        "  Qarterly = 0\n",
        "  Yearly = 1\n",
        "\n",
        "DataTuple = namedtuple('DataTuple', 'data header_type table_type')\n",
        "MunicipalityIdentifier = namedtuple('MunicipalityIdentifier', 'region municipality')\n",
        "SettlementInfo = namedtuple('SettlementInfo', 'name permanent_residents current_residents')\n",
        "FullSettlementInfo = namedtuple('FullSettlementInfo', 'region municipality settlement permanent_residents current_residents')\n",
        "PopulationInfo = namedtuple('PopulationInfo', 'permanent current')\n",
        "ParsedLines = namedtuple('ParsedLines', 'municipality_ids settlements_info')\n",
        "\n",
        "T = TypeVar('T')"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "eURPckPOsM44",
        "colab_type": "text"
      },
      "source": [
        "## Definition of Data Source"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "GRNvEjslrAWZ",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "data_source: List[DataTuple] = [\n",
        "  DataTuple(\"https://www.grao.bg/tna/t41nm-15-03-2020_2.txt\", HeaderEnum.New, TableTypeEnum.Qarterly), \n",
        "  DataTuple(\"https://www.grao.bg/tna/tadr2019.txt\", HeaderEnum.New, TableTypeEnum.Yearly), \n",
        "  DataTuple(\"https://www.grao.bg/tna/tadr2018.txt\", HeaderEnum.New, TableTypeEnum.Yearly), \n",
        "  DataTuple(\"https://www.grao.bg/tna/tadr2017.txt\", HeaderEnum.New, TableTypeEnum.Yearly), \n",
        "  DataTuple(\"https://www.grao.bg/tna/tadr-2016.txt\", HeaderEnum.New, TableTypeEnum.Yearly), \n",
        "  DataTuple(\"https://www.grao.bg/tna/tadr-2015.txt\", HeaderEnum.New, TableTypeEnum.Yearly), \n",
        "  DataTuple(\"https://www.grao.bg/tna/tadr-2014.txt\", HeaderEnum.New, TableTypeEnum.Yearly), \n",
        "  DataTuple(\"https://www.grao.bg/tna/tadr-2013.txt\", HeaderEnum.New, TableTypeEnum.Yearly), \n",
        "  DataTuple(\"https://www.grao.bg/tna/tadr-2012.txt\", HeaderEnum.New, TableTypeEnum.Yearly), \n",
        "  DataTuple(\"https://www.grao.bg/tna/tadr-2011.txt\", HeaderEnum.New, TableTypeEnum.Yearly), \n",
        "  DataTuple(\"https://www.grao.bg/tna/tadr-2010.txt\", HeaderEnum.New, TableTypeEnum.Yearly), \n",
        "  DataTuple(\"https://www.grao.bg/tna/tadr-2009.txt\", HeaderEnum.New, TableTypeEnum.Yearly), \n",
        "  DataTuple(\"https://www.grao.bg/tna/tadr-2008.txt\", HeaderEnum.New, TableTypeEnum.Yearly), \n",
        "  DataTuple(\"https://www.grao.bg/tna/tadr-2007.txt\", HeaderEnum.New, TableTypeEnum.Yearly), \n",
        "  DataTuple(\"https://www.grao.bg/tna/tadr-2006.txt\", HeaderEnum.New, TableTypeEnum.Yearly), \n",
        "  DataTuple(\"https://www.grao.bg/tna/tadr-2005.txt\", HeaderEnum.Old, TableTypeEnum.Yearly), \n",
        "  DataTuple(\"https://www.grao.bg/tna/tadr-2004.txt\", HeaderEnum.Old, TableTypeEnum.Yearly), \n",
        "  DataTuple(\"https://www.grao.bg/tna/tadr-2003.txt\", HeaderEnum.Old, TableTypeEnum.Yearly), \n",
        "  DataTuple(\"https://www.grao.bg/tna/tadr-2002.txt\", HeaderEnum.Old, TableTypeEnum.Yearly), \n",
        "  DataTuple(\"https://www.grao.bg/tna/tadr-2001.txt\", HeaderEnum.Old, TableTypeEnum.Yearly), \n",
        "  DataTuple(\"https://www.grao.bg/tna/tadr-2000.txt\", HeaderEnum.Old, TableTypeEnum.Yearly), \n",
        "  DataTuple(\"https://www.grao.bg/tna/tadr-1999.txt\", HeaderEnum.Old, TableTypeEnum.Yearly), \n",
        "  DataTuple(\"https://www.grao.bg/tna/tadr-1998.txt\", HeaderEnum.Old, TableTypeEnum.Yearly), \n",
        "]"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "96QwY-r2gnB8",
        "colab_type": "text"
      },
      "source": [
        "## Regular Expresions Construction"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "ZP3ba3zzZ9cv",
        "colab_type": "code",
        "outputId": "0aa5d1f0-c6d8-44c5-ef6c-485e78489403",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 107
        }
      },
      "source": [
        "# Building regex strings\n",
        "cap_letter = '\\p{Lu}'\n",
        "low_letter = '\\p{Ll}'\n",
        "separator = '[\\||\\!]\\s*'\n",
        "number = '\\d+'\n",
        "\n",
        "year_group = '(\\d{4})'\n",
        "\n",
        "name_part = f'\\s*{cap_letter}*'\n",
        "name_part_old = f'\\s{cap_letter}*'\n",
        "type_abbr = f'{cap_letter}+\\.'\n",
        "name = f'{cap_letter}+{name_part * 3}'\n",
        "name_old = f'{cap_letter}+{name_part_old * 3}'\n",
        "word = f'{low_letter}+'\n",
        "number_group = f'{separator}({number})\\s*'\n",
        "\n",
        "old_reg = f'ОБЛАСТ:({name_old})'\n",
        "print(old_reg)\n",
        "old_mun = f'ОБЩИНА:({name_old})'\n",
        "print(old_mun)\n",
        "\n",
        "region_name_new_re = f'{word} ({name}) {word} ({name})'\n",
        "print(region_name_new_re)\n",
        "# Quaterly\n",
        "settlement_info_quarterly_re = f'({type_abbr}{name})\\s*{number_group * 3}'\n",
        "print(settlement_info_quarterly_re)\n",
        "# Yearly\n",
        "settlement_info_yearly_re = f'({type_abbr}{name})\\s*{number_group * 6}'\n",
        "print(settlement_info_yearly_re)"
      ],
      "execution_count": 4,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "ОБЛАСТ:(\\p{Lu}+\\s\\p{Lu}*\\s\\p{Lu}*\\s\\p{Lu}*)\n",
            "ОБЩИНА:(\\p{Lu}+\\s\\p{Lu}*\\s\\p{Lu}*\\s\\p{Lu}*)\n",
            "\\p{Ll}+ (\\p{Lu}+\\s*\\p{Lu}*\\s*\\p{Lu}*\\s*\\p{Lu}*) \\p{Ll}+ (\\p{Lu}+\\s*\\p{Lu}*\\s*\\p{Lu}*\\s*\\p{Lu}*)\n",
            "(\\p{Lu}+\\.\\p{Lu}+\\s*\\p{Lu}*\\s*\\p{Lu}*\\s*\\p{Lu}*)\\s*[\\||\\!]\\s*(\\d+)\\s*[\\||\\!]\\s*(\\d+)\\s*[\\||\\!]\\s*(\\d+)\\s*\n",
            "(\\p{Lu}+\\.\\p{Lu}+\\s*\\p{Lu}*\\s*\\p{Lu}*\\s*\\p{Lu}*)\\s*[\\||\\!]\\s*(\\d+)\\s*[\\||\\!]\\s*(\\d+)\\s*[\\||\\!]\\s*(\\d+)\\s*[\\||\\!]\\s*(\\d+)\\s*[\\||\\!]\\s*(\\d+)\\s*[\\||\\!]\\s*(\\d+)\\s*\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "EJZO2jcNguZi",
        "colab_type": "text"
      },
      "source": [
        "## Helper Functions"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "uUKEPfkTYkcd",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "def pipeline(\n",
        "        value: T,\n",
        "        function_pipeline: Sequence[Callable[[T], T]],\n",
        ") -> T:\n",
        "    '''A generic Unix-like pipeline\n",
        "\n",
        "    :param value: the value you want to pass through a pipeline\n",
        "    :param function_pipeline: an ordered list of functions that\n",
        "        comprise your pipeline\n",
        "    '''\n",
        "    return reduce(lambda v, f: f(v), function_pipeline, value)\n",
        "\n",
        "def build_pipline(functions: Sequence[Callable[[T], T]]) -> Callable[[T], T]:\n",
        "  return (lambda value: pipeline(value, function_pipeline=functions))\n",
        "\n",
        "def execute_pipeline(value: T, pipeline: Callable[[T], T]) -> T:\n",
        "  return pipeline(value)\n",
        "\n",
        "\n",
        "\n",
        "def static_vars_funktion(**kwargs):\n",
        "  def decorate(func):\n",
        "      for k in kwargs:\n",
        "          setattr(func, k, kwargs[k])\n",
        "      return func\n",
        "  return decorate"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "l_KliFzwgz4o",
        "colab_type": "text"
      },
      "source": [
        "## Parsing Pipeline Definitions"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "R1Fk28RElQBB",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "def fetch_raw_table(data_tuple: DataTuple) -> DataTuple:\n",
        "  headers = requests.utils.default_headers()\n",
        "  headers.update({ \n",
        "      'User-Agent': 'Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:52.0) Gecko/20100101 Firefox/52.0'\n",
        "  })\n",
        "\n",
        "  url = data_tuple.data\n",
        "  req = requests.get(url, headers)\n",
        "  req.encoding = 'windows-1251'\n",
        "\n",
        "  return DataTuple(req, data_tuple.header_type, data_tuple.table_type)\n",
        "\n",
        "def raw_table_to_lines(data_tuple: DataTuple) -> DataTuple:\n",
        "  req = data_tuple.data\n",
        "  soup = BeautifulSoup(req.text, 'lxml').prettify()\n",
        "  split = soup.split('\\r\\n')\n",
        "\n",
        "  return DataTuple(split, data_tuple.header_type, data_tuple.table_type)\n",
        "\n",
        "def parse_lines(data_tuple: DataTuple) -> DataTuple:\n",
        "\n",
        "  def parse_data_line(line: str, table_type: TableTypeEnum) -> Optional[SettlementInfo]:\n",
        "    settlement_info_re = ''\n",
        "    permanent_population_position = -1\n",
        "    current_population_position = -1\n",
        "\n",
        "    if table_type == TableTypeEnum.Qarterly:\n",
        "      settlement_info_re = settlement_info_quarterly_re\n",
        "      permanent_population_position = 2\n",
        "      current_population_position = 3\n",
        "    elif table_type == TableTypeEnum.Yearly:\n",
        "      settlement_info_re = settlement_info_yearly_re\n",
        "      permanent_population_position = 2\n",
        "      current_population_position = 6\n",
        "\n",
        "    settlement_info = regex.search(settlement_info_re, line)\n",
        "\n",
        "    if settlement_info:\n",
        "      name, permanent, current = settlement_info.group(1, \n",
        "                                                       permanent_population_position, \n",
        "                                                       current_population_position)\n",
        "      settlement_info = SettlementInfo(name.strip(), permanent, current)\n",
        "\n",
        "    return settlement_info\n",
        "\n",
        "  @static_vars_funktion(region=None)\n",
        "  def parse_header_line(line: str, header_type: HeaderEnum) -> Optional[MunicipalityIdentifier]:\n",
        "    region_name = None\n",
        "\n",
        "    if header_type == HeaderEnum.New:\n",
        "      region_name_re = region_name_new_re\n",
        "      region_gr = regex.search(region_name_re, line)\n",
        "\n",
        "      if region_gr:\n",
        "        region, municipality = region_gr.group(1, 2)\n",
        "        region_name = MunicipalityIdentifier(region.strip(), municipality.strip())\n",
        "\n",
        "    elif header_type == HeaderEnum.Old:\n",
        "      if not parse_header_line.region:\n",
        "        parse_header_line.region = regex.search(old_reg, line)\n",
        "        region_name = None\n",
        "      else:\n",
        "        mun_gr = regex.search(old_mun, line)\n",
        "\n",
        "        if mun_gr:\n",
        "          region, municipality = parse_header_line.region.group(1), mun_gr.group(1)\n",
        "          region_name = MunicipalityIdentifier(region.strip(), municipality.strip())\n",
        "\n",
        "          parse_header_line.region = None\n",
        "\n",
        "    return region_name\n",
        "\n",
        "  municipality_ids = {}\n",
        "  settlements_info = {}  \n",
        "\n",
        "  for line_num, line in enumerate(data_tuple.data):\n",
        "    municipality_id = parse_header_line(line, data_tuple.header_type)\n",
        "    if municipality_id:\n",
        "      municipality_ids[line_num] = municipality_id\n",
        "\n",
        "    settlement_info = parse_data_line(line, data_tuple.table_type)\n",
        "    if settlement_info:\n",
        "      settlements_info[line_num] = settlement_info\n",
        "\n",
        "  return DataTuple(ParsedLines(municipality_ids, settlements_info), data_tuple.header_type, data_tuple.table_type)\n",
        "\n",
        "def parssed_lines_to_full_info_list(data_tuple: DataTuple) -> DataTuple:\n",
        "  regions = data_tuple.data.municipality_ids\n",
        "  settlements_info = data_tuple.data.settlements_info\n",
        "\n",
        "  reg_keys = list(regions.keys())\n",
        "  settlement_keys = list(settlements_info.keys())\n",
        "  \n",
        "  reg_keys_pairs = zip(reg_keys[:-1], reg_keys[1:])\n",
        "\n",
        "  sk_index = 0\n",
        "  full_name_settlement_infos = []\n",
        "\n",
        "  for current_mun, next_mun in reg_keys_pairs:\n",
        "    while current_mun < settlement_keys[sk_index] < next_mun:\n",
        "      reg = regions[current_mun]\n",
        "      set_info = settlements_info[settlement_keys[sk_index]]\n",
        "      fnsi = FullSettlementInfo(reg.region,\n",
        "                                reg.municipality,\n",
        "                                set_info.name,\n",
        "                                set_info.permanent_residents,\n",
        "                                set_info.current_residents)\n",
        "      full_name_settlement_infos.append(fnsi)\n",
        "\n",
        "      sk_index += 1\n",
        "\n",
        "  return DataTuple(full_name_settlement_infos, data_tuple.header_type, data_tuple.table_type)\n",
        "\n",
        "def full_info_list_to_data_frame(data_tuple: DataTuple) -> DataTuple:\n",
        "  df = pd.DataFrame(data_tuple.data)\n",
        "  df.set_index(['region', 'municipality', 'settlement'], drop=True, inplace=True)\n",
        "\n",
        "  return DataTuple(df, data_tuple.header_type, data_tuple.table_type)\n",
        "\n",
        "parsing_pipeline = build_pipline(functions=(\n",
        "  fetch_raw_table,\n",
        "  raw_table_to_lines,\n",
        "  parse_lines,\n",
        "  parssed_lines_to_full_info_list,\n",
        "  full_info_list_to_data_frame\n",
        "))"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "_t_uPK2W7XFx",
        "colab_type": "text"
      },
      "source": [
        "## Data Processing Pipeline"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Ms2PHPCxRkjI",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "def process_data(data_source: List[DataTuple]) -> List[DataTuple]:\n",
        "  parsed_data = None\n",
        "  data_frame_list = []\n",
        "\n",
        "  for data_tuple in data_source:\n",
        "    year = regex.search(year_group, data_tuple.data).group(1)\n",
        "    data_frame = execute_pipeline(data_tuple, parsing_pipeline).data\n",
        "    data_frame = data_frame.rename(columns={'permanent_residents':f'permanent_{year}', \n",
        "                                            'current_residents':f'current_{year}'})\n",
        "    \n",
        "    data_frame_list.append(data_frame)\n",
        "    if isinstance(parsed_data, pd.DataFrame):\n",
        "      parsed_data = parsed_data.merge(data_frame, sort=False, how='right', left_index=True, right_index=True)\n",
        "    else:\n",
        "      parsed_data = data_frame\n",
        "\n",
        "  return [DataTuple(parsed_data,0,0), DataTuple(data_frame_list,0,0)]\n",
        "\n",
        "def store_data(processed_data: List[DataTuple]) -> List[DataTuple]:\n",
        "  directory = './grao'\n",
        "  if not os.path.exists(directory):\n",
        "    os.makedirs(directory)\n",
        "\n",
        "  combined_data = processed_data[0].data\n",
        "  combined_data.to_csv(f'{directory}/combined_data.csv')\n",
        "  combined_data.to_pickle(f'{directory}/combined_data.pkl')\n",
        "\n",
        "  data_list = processed_data[1].data\n",
        "  with open(f'{directory}/data_frames_list.pkl', 'wb') as f:\n",
        "    pickle.dump(data_list, f)\n",
        "\n",
        "  return processed_data\n",
        "\n",
        "processing_pipeline = build_pipline(functions=(\n",
        "    process_data,\n",
        "    store_data\n",
        "))"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "pLNB7d5jNqqD",
        "colab_type": "text"
      },
      "source": [
        "## Data Processing"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "3TynB8XaM5cC",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "processed_data = execute_pipeline(data_source, processing_pipeline)\n",
        "processed_data"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "TzrtsiVNSMrD",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "!sed 's/,,/, ,/g;s/,,/, ,/g' ./grao/combined_data.csv | column -s, -t"
      ],
      "execution_count": 0,
      "outputs": []
    }
  ]
}