RISE_subindicators.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import requests\n",
    "from bs4 import BeautifulSoup\n",
    "import pandas as pd\n",
    "from data.data_preparation import ELECTRIFICATION_OPTIONS, RISE_SUB_INDICATOR_STRUCTURE"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "raw_data = pd.read_csv('data/raw_data.csv')\n",
    "ndc_countries = raw_data['country_wb'].to_list()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*Source*: https://rise.worldbank.org/country"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# without headers the website rejects the request (look at your Network Monitor to see which one to use)\n",
    "HDR = {\n",
    "    'User-Agent': 'Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:69.0) Gecko/20100101 Firefox/69.0'\n",
    "}\n",
    "\n",
    "def assign_value_to_answer(answer, indicator, idx):\n",
    "    \"\"\"Assign the portion of the RISE score depending on the answer\"\"\"\n",
    "    if answer == 'Yes':\n",
    "        return RISE_SUB_INDICATOR_STRUCTURE[indicator][idx]\n",
    "    else:\n",
    "        return 0  \n",
    "\n",
    "\n",
    "def get_country_info(country, country_iso=None):\n",
    "    \"\"\"Extract RISE sub-indicators of a country on the World Bank website\"\"\"\n",
    "    \n",
    "    page = requests.get('https://rise.worldbank.org/country/{}'.format(country), headers=HDR)\n",
    "    \n",
    "    soup = BeautifulSoup(page.text, 'html.parser')\n",
    "    \n",
    "    results = []\n",
    "    \n",
    "    for opt_name, opt_num in zip(ELECTRIFICATION_OPTIONS, [3, 4, 5]):\n",
    "        # find the panel for each electrification options\n",
    "        opt = soup.find('div',class_=f'indicator-{opt_num}')\n",
    "        rows = opt.find_all('div', class_='row')\n",
    "\n",
    "        idx = -1\n",
    "        for row in list(rows)[1:]:\n",
    "            is_subtitle = row.find('span', class_='ans-hyphen') is not None\n",
    "            if is_subtitle:\n",
    "                # find the subgroup of the RISE sub-indicator\n",
    "                subtitle = row.find('div', class_='title-wrap').get_text()\n",
    "                idx = idx + 1\n",
    "            else:\n",
    "                # find the question of the RISE sub-indicator\n",
    "                question = row.find('div', class_='title-wrap').get_text().replace('\\n', ' ')\n",
    "                if question not in list_q:\n",
    "                    print(question)\n",
    "                else:\n",
    "                    # find the answer to the RISE sub-indicator's question\n",
    "                    answer = row.find('span', class_='ans').get_text()\n",
    "                    value = assign_value_to_answer(answer, opt_name, idx)\n",
    "                    results.append(['rise_{}'.format(opt_name), subtitle, question, value])\n",
    "    \n",
    "    results = np.vstack(results)\n",
    "    column_names = ['indicator', 'sub_indicator_group', 'sub_indicator_text', 'value']\n",
    "    results = pd.DataFrame.from_records(results, columns=column_names)    \n",
    "    results['country'] = country\n",
    "    if country_iso is not None:\n",
    "        results['country_iso'] = country_iso\n",
    "    \n",
    "    return results"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Get list of countries of World Bank database"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "page = requests.get('https://rise.worldbank.org/countries', headers=HDR)\n",
    "soup = BeautifulSoup(page.text, 'html.parser')\n",
    "\n",
    "countries = soup.find_all('div', class_='country-name')\n",
    "countries_names = []\n",
    "for i in range(len(countries)):\n",
    "    country_name = list(countries[i].children)[0]['href'].split('/')[2]\n",
    "    countries_names.append(country_name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Prepare the list of the NDC-OG countries matching the WB database"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "raw_data = pd.read_csv('data/raw_data.csv')\n",
    "ndc_countries = raw_data['country_wb'].to_list()\n",
    "raw_data = raw_data.set_index('country_wb')\n",
    "\n",
    "missing = []\n",
    "matching = []\n",
    "\n",
    "for c in ndc_countries:\n",
    "    if c in countries_names:\n",
    "        matching.append((c, raw_data.loc[c].country_iso))\n",
    "    else:\n",
    "        missing.append((c, raw_data.loc[c].country_iso))\n",
    "\n",
    "# peru has apparently a rise score of 100 and no \"yes/no\" answers\n",
    "# list of countries which RISE score are not provided by the World Bank\n",
    "\n",
    "NAN_COUNTRIES = missing + [matching.pop(matching.index(('peru','PER')))]  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Download the RISE subindicators for the NDC-OG countries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Download the info of the matching countries\n",
    "df = get_country_info(*matching[0])\n",
    "for c in matching[1:]:\n",
    "    df = df.append(get_country_info(*c), ignore_index=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Fill the non matching countries with rise scores of 0 with all questions' answers set to no"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "RISE_SUB_INDICATORS = pd.read_csv('data/RISE_indicators.csv')\n",
    "\n",
    "for c in missing:\n",
    "    nan_df = RISE_SUB_INDICATORS.copy()\n",
    "    nan_df = nan_df.drop(['score_count_yes'], axis=1)\n",
    "    nan_df['country'] = c[0]\n",
    "    nan_df['country_iso'] = c[1]\n",
    "    nan_df = nan_df.rename(columns={\"score_count_no\": \"value\"})\n",
    "    nan_df\n",
    "    save_df = df.append(nan_df, ignore_index=True)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_df = pd.read_csv('data/RISE_subindicators_country.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#save_df.to_csv('data/RISE_subindicators_country.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sub_group = 'Funding support for grid electrification'\n",
    "sub_group_df = RISE_SUB_INDICATORS.loc['rise_grid']\n",
    "sub_group_df = sub_group_df.loc[sub_group_df.sub_indicator_group == sub_group]\n",
    "sub_group_df.count()"
   ]
  }
 ],
 "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.6.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}