jhu-covid19-scraper.py

from chalice import Chalice, BadRequestError
import json
from datetime import date, datetime
import csv
import requests
import io

app = Chalice(app_name='jhu-data-scraper')


def timeNow():
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")


def csvRespToListOfDict(responseText):
    reader = csv.DictReader(io.StringIO(responseText))
    list_of_records = list(reader)
    list_of_dict = [json.loads(json.dumps(record))
                    for record in list_of_records]
    return list_of_dict


@app.route('/')
def index():
    return {
        'links': {
            'uidIsoFipsLookupTable': '/uid-iso-fips-lookup-table',
            'timeSeriesGlobalConfirmed': '/time-series-global/confirmed',
            'timeSeriesGlobalDeaths': '/time-series-global/deaths',
            'timeSeriesGlobalRecovered': '/time-series-global/recovered',
            'dailyReport': '/daily-report'
        }
    }


@app.route('/uid-iso-fips-lookup-table')
def getUidIsoFipsLookupTable():
    csv_url = "https://raw.githubusercontent.com/CSSEGISandData/COVID-19/master/csse_covid_19_data/UID_ISO_FIPS_LookUp_Table.csv"
    response = requests.get(csv_url)
    response_text = response.text
    list_of_dict = csvRespToListOfDict(response_text)
    return {
        'updateTime': timeNow(),
        'uidIsoFipsLookupTable': list_of_dict
    }


@app.route('/time-series-global/{caseType}')
def getTimeSeriesGlobal(caseType):
    data_names = ['confirmed', 'deaths', 'recovered']
    covid_19_ts_dict = {}
    for dn in data_names:
        csv_url = "https://raw.githubusercontent.com/CSSEGISandData/COVID-19/master/csse_covid_19_data/csse_covid_19_time_series/time_series_covid19_{}_global.csv".format(
            dn)
        response = requests.get(csv_url)
        response_text = response.text
        list_of_dict = csvRespToListOfDict(response_text)
        covid_19_ts_dict[dn] = list_of_dict
    try:
        time_series_global = {
            'updateTime': timeNow(),
            'timeSeriesGlobal': covid_19_ts_dict[caseType]
        }
        return time_series_global
    except KeyError:
        valid_keys = covid_19_ts_dict.keys()
        valid_keys_str = ', '.join(valid_keys)
        raise BadRequestError(
            "Unknown case type: {}, valid choices are {}".format(caseType, valid_keys_str))


@app.route('/daily-report')
def getDailyReport():
    csv_url = "https://raw.githubusercontent.com/CSSEGISandData/COVID-19/master/csse_covid_19_data/csse_covid_19_time_series/time_series_covid19_confirmed_global.csv"
    response = requests.get(csv_url)
    response_text = response.text
    list_of_dict = csvRespToListOfDict(response_text)
    columns = list(list_of_dict[0].keys())
    last_column = columns[-1]
    month, day, year = last_column.split("/")
    month = int(month)
    day = int(day)
    year = int(year) + 2000
    latest_date = date(year, month, day)
    latest_date_str = latest_date.strftime('%m-%d-%Y')
    csv_url = "https://raw.githubusercontent.com/CSSEGISandData/COVID-19/master/csse_covid_19_data/csse_covid_19_daily_reports/{}.csv".format(
        latest_date_str)
    response = requests.get(csv_url)
    response_text = response.text
    list_of_dict = csvRespToListOfDict(response_text)
    return {
        'updateTime': timeNow(),
        'dailyReport': list_of_dict
    }