Parse the numbers of Russian casualties in Ukraine from a news website, save them to CSV files, and plot them

russian_casualty_parser.py

# -*- coding: utf-8 -*-
# Parse the numbers of Russian casualties in Ukraine from a news website, save them to CSV files, and plot them.
#
# Prerequisites:
# $ pip install beautifulsoup4==4.10.0 matplotlib==3.4.3 requests==2.26.0 pandas==1.3.4
#
# Usage:
# $ python3 russian_casualty_parser.py
#
# Output:
# - casualties_cumulative.csv  : daily casualties (accumulated since the beginning)
# - casualties_delta.csv       : daily casualties (change from the previous day)
# - casualties_delta.svg       : daily casualties plot by category

import logging
import re
from datetime import datetime
from dateutil.relativedelta import relativedelta

from bs4 import BeautifulSoup
import matplotlib as mpl
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import requests


SOURCE_URL = "https://index.minfin.com.ua/ua/russian-invading/casualties/month.php?month={month_string}"
OUTPUT_CUMULATIVE_DATA_FILENAME = "casualties_cumulative.csv"
OUTPUT_DELTA_DATA_FILENAME = "casualties_delta.csv"
OUTPUT_DELTA_PLOT_FILENAME = "casualties_delta.svg"


# Normalize inconsistent headers from the earlier days in the dataset.
CATEGORY_MERGE_MAP = {
    "РСЗВ Град": "РСЗВ",
    "ЗРК БУК": "Засоби ППО",
    "Автомобілі": "Автомобілі та автоцистерни",
    "Пускові установки ОТРК": "Крилаті ракети",
    "Цистерни з ППМ": "Автомобілі та автоцистерни",
}


logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)8s | %(message)s")
LOGGER = logging.getLogger(__name__)


def download_monthly_data(month_string):
    url = SOURCE_URL.format(month_string=month_string)
    LOGGER.info(f"Retrieving {url}")
    response = requests.get(url)
    response.raise_for_status()
    return response.text


def parse_daily_data(day_element):
    date = day_element.find("span", {"class": "black"}).text
    date = datetime.strptime(date, "%d.%m.%Y").strftime("%Y-%m-%d")

    result = {"Дата": date}

    categories = day_element.find("div", {"class": "casualties"}).find_all("li")
    for category_element in categories:
        title, count = category_element.text.split("—")[:2]
        title = title.strip()

        count = re.findall(r"\d+", count)[0]
        result[title] = int(count)

    return result


def parse_monthly_data(month_string):
    downloaded_data = download_monthly_data(month_string)
    soup = BeautifulSoup(downloaded_data, "html.parser")

    result = []
    for day_element in soup.find_all("li", {"class": "gold"}):
        result.append(parse_daily_data(day_element))

    return result


def build_all_month_strings():
    result = []
    now = datetime.now()

    start_month = datetime(now.year, now.month, 1)
    end_month = datetime(2022, 2, 1)
    current_month_start = start_month
    while current_month_start >= end_month:
        result.append(f"{current_month_start.year:04d}-{current_month_start.month:02d}")
        current_month_start = current_month_start + relativedelta(months=-1)

    return result


def plot_daily_data(df_delta):
    df = df_delta.copy()
    df.index = pd.to_datetime(df.index)

    date_locator = mpl.dates.MonthLocator()
    date_formatter = mpl.dates.DateFormatter('%b %Y')

    fake_series_name = ""
    if len(df.columns) % 2:
        # Add a fake series with 0 values so that all subplots are filled.
        df[fake_series_name] = 0

    plot_series_names = np.split(df.columns, 2)

    plt.rcParams["font.family"] = "Helvetica"
    colors = plt.rcParams["axes.prop_cycle"]()

    fig, axes = plt.subplots(7, 2, figsize=(35, 15), sharex="col", dpi=300)

    for plot_col_idx, plot_series_list in enumerate(plot_series_names):
        for plot_row_idx, plot_series_name in enumerate(plot_series_list):
            plot_data = df[plot_series_name]

            ax = axes[plot_row_idx, plot_col_idx]
            ax.bar(x=plot_data.index, height=plot_data, color=next(colors)["color"], zorder=3)

            ax.tick_params(axis="x", labelsize=8, labelrotation=90)
            ax.tick_params(axis="y", labelsize=8)

            ax.xaxis.set_major_locator(date_locator)
            ax.xaxis.set_major_formatter(date_formatter)
            ax.set_ylim(bottom=0)

            if plot_series_name != fake_series_name:
                ax.grid(visible=True, axis="y", which="major", color="#F0F0F0", zorder=0)
                ax.set_title(f"{plot_series_name} ({plot_data.sum()})")
            else:
                # If it's the last ("filler") plot, just show an empty plot and a date axis.
                ax.get_yaxis().set_visible(False)

    fig.tight_layout(pad=1.0)


def main():
    all_data = []
    month_strings = build_all_month_strings()
    for month_string in month_strings:
        all_data.extend(parse_monthly_data(month_string))

    LOGGER.info("Generating the dataset")
    df_absolute = pd.DataFrame.from_records(all_data)
    df_absolute = df_absolute.set_index("Дата").sort_values(by="Дата").fillna(0).astype(int)

    # Merge old categories into new ones.
    for merge_from, merge_to in CATEGORY_MERGE_MAP.items():
        df_absolute[merge_to] += df_absolute[merge_from]
        df_absolute.drop(columns=[merge_from], inplace=True)

    df_delta = df_absolute.diff(periods=1, axis=0)
    # Restore the first row in the diffs.
    df_delta.iloc[0] = df_absolute.iloc[0]
    df_delta = df_delta.fillna(0).astype(int)

    LOGGER.info(f"Saving cumulative data to {OUTPUT_CUMULATIVE_DATA_FILENAME}")
    df_absolute.to_csv(OUTPUT_CUMULATIVE_DATA_FILENAME, header=True)

    LOGGER.info(f"Saving daily data to {OUTPUT_DELTA_DATA_FILENAME}")
    df_delta.to_csv(OUTPUT_DELTA_DATA_FILENAME, header=True)

    LOGGER.info(f"Plotting daily data to {OUTPUT_DELTA_PLOT_FILENAME}")
    plot_daily_data(df_delta)
    plt.savefig(OUTPUT_DELTA_PLOT_FILENAME, dpi=300)

    LOGGER.info("Done!")


if __name__ == "__main__":
    main()

z_casualties_cumulative.csv

z_casualties_delta.csv

z_casualties_delta.svg