Study of Amazon’s Best Selling & Most Read Book Charts Since 2017

from datetime import datetime, timedelta

from bs4 import BeautifulSoup
import dataset
import requests

def _parse_weeks(text):
    '''For parsing weeks in text to an integer value
    '''
    text = text.upper().split(" WEEK")[0]
    return 1 if text=="FIRST" else int(text)

# connect to SQLite DB
db = dataset.connect("sqlite:///data.db")
# Creates a new one if data.db does not exist

# iterate over all sundays starting from 2017-05-14
date = datetime(2017, 5, 14)
end_date = datetime(2024, 6, 17)

while date<=end_date:
    date_str = date.isoformat().split("T")[0]

    # 4 possible chart urls for each date
    chart_names = [
        "mostread/fiction",
        "mostread/nonfiction",
        "mostsold/fiction",
        "mostsold/nonfiction",
    ]

    # get chart entries for each URL
    for chart_name in chart_names:
        url = f'https://www.amazon.com/charts/{date_str}/{chart_name}'
        print(url)

        # check if URL was scraped before
        last_key = chart_name + "-" + date_str + "-20"
        exists = db["chart_entries"].find_one(key=last_key)
        if exists:
            # if URL was already scraped, books were added
            # so, skip this URL
            print("SKIPPING: ALREADY PROCESSED")
            continue

        # if URL wasn't scraped before, go ahead
        # send HTTP request to URL
        r = requests.get(url)
        # convert response into a Soup
        soup = BeautifulSoup(r.text, features="html5lib")

        # each chart entry is within a div of a particular class
        # use CSS selectors to select these divs
        cards = soup.select("div.kc-horizontal-rank-card")
        assert len(cards)==20

        for card in cards:
            # run loop over each div containing a chart entry
            chart_position = int(card.select_one("div.kc-rank-card-rank").text.strip())

            # construct a unique key to avoid duplicate entries into db
            key = chart_name + "-" + date_str + "-" + str(chart_position)

            exists = db["chart_entries"].find_one(key=key)
            if exists:
                print("SKIPPING:", key, "- ALREADY EXISTS")
                continue

            # scrape necessary fields using CSS selectors
            author_el = card.select_one("div.kc-rank-card-author")
            chart_entry = dict(
                key=key,
                chart_name=chart_name,
                date=date_str,
                chart_position=chart_position,
                book_name=card.select_one("div.kc-rank-card-title").text.strip(),
                author_name=author_el.get("title") if author_el else "N/A",
                # use previously defined function to get weeks as int
                weeks_on_chart=_parse_weeks(card.select_one("div.kc-wol,div.kc-wol-first").text),
                link=card.select_one("a.kc-cover-link").get("href"),
                cover_image=card.select_one("a.kc-cover-link img").get("src"),
            )

            # make sure no fields are blank
            for dict_key in chart_entry:
                assert chart_entry[dict_key]

            # add the chart entry to db
            db["chart_entries"].insert(chart_entry)

    # repeat for subsequent sunday
    date = date + timedelta(days=7)

from contextlib import contextmanager
from datetime import datetime
import html
import signal
import time
import traceback

from bs4 import BeautifulSoup
import dataset
import requests
from scrapingbee import ScrapingBeeClient

# connect to SQLite DB
db = dataset.connect("sqlite:///data.db")

# initialize ScrapingBeeClient using the ScrapingBee Python Package
spb_client = ScrapingBeeClient(api_key="{{REDACTED}}")

# for parsing date strings
MONTHS = {
    "january": 1, "february": 2, "march": 3, "april": 4,
    "may": 5, "june": 6, "july": 7, "august": 8,
    "september": 9, "october": 10, "november": 11, "december": 12,
}
def _parse_date(raw_date):
    chunks = raw_date.lower().replace(",","").split(" ")
    month = int(MONTHS[chunks[0]])
    day = int(chunks[1])
    year = int(chunks[2])
    return datetime(year, month, day).isoformat().split("T")[0]

# for setting a timeout for requests
class TimeoutException(Exception): pass

@contextmanager
def _time_limit(seconds):
    def signal_handler(sigint, frame):
        raise TimeoutException
    signal.signal(signal.SIGALRM, signal_handler)
    signal.alarm(seconds)
    try:
        yield
    finally:
        signal.alarm(0)

def run():
    counter = 0
    for entry in db["chart_entries"]:
        # each book has an unique id on Amazon, present in the URL
        book_id = entry["link"].split("/")[2]
        base_url = "https://amazon.com"
        url = base_url + entry["link"]

        # printing for monitoring
        counter += 1
        print(counter, entry["book_name"], url)

        # check if book was already scraped
        exists = db["books"].find_one(book_id=book_id)
        if exists:
            # skip if scraped
            print("SKIPPING: ALREADY PROCESSED")
            continue

        # get the HTML in the page through ScrapingBee
        # throws an Exception if request does not complete in 60s
        with _time_limit(60):
            r = spb_client.get(url, params={"timeout": "10000", "return_page_source": "true"})

        # write HTML response to file, to inspect if required
        with open("last.html", "w") as f:
            f.write(str(r.content))
            f.close()

        # create dict to store scraped data
        book_data = dict(
            book_id=book_id,
            name=entry["book_name"],
            author=entry["author_name"],
            url=url,
        )

        soup = BeautifulSoup(r.content, features="html5lib")
        if "/dogsofamazon" in str(r.content):
            # this is 404 page
            # mark as missing and move to next book
            book_data["missing"] = True
            print("BOOK PAGE MISSING, SKIPPING")
            # add book as missing to DB
            db["books"].insert(book_data)
            continue

        # scrape required fields
        categories = list(map(
            lambda s: s.text.strip(),
            soup.select("#wayfinding-breadcrumbs_feature_div a.a-link-normal"),
        ))
        book_data["categories"] = "|".join(categories)

        details = {}

        # most metadata is available as key value pairs
        # the format is slightly different for regular and audible links
        detail_soups = soup.select("#detailBullets_feature_div ul:first-child li .a-list-item")
        if len(detail_soups)>0:
            for item in detail_soups:
                spans = item.select("span")
                key = spans[0].text.replace("\u200F", "").replace("\u200E", "").replace("\n","")
                key = html.unescape(key).split(":")[0].strip()
                key = key.lower().replace(" ", "_").replace("-","_")
                value = html.unescape(spans[1].text).strip()
                if "_date" in key:
                    value = _parse_date(value)
                book_data["details_" + key] = value
        else:
            detail_soups = soup.select("#audibleProductDetails tr, #prodDetails tr")
            assert len(detail_soups)>0
            for item in detail_soups:
                key = item.select_one("th").text.lower()
                key = key.strip().replace(" ","_").replace("-","_").replace(".","_")
                value = item.select_one("td").text.strip()
                if "_date" in key:
                    value = _parse_date(value)
                book_data["details_" + key] = value

        # add scraped data to DB
        db["books"].insert(book_data)

    # indicate a successful run over all entries without errors
    return "FINISHED"

for i in range(400):
    # allow for 400 restarts in case of known exceptions
    # if run completes without exceptions, break the loop, exit the program
    try:
        if run()=="FINISHED":
            break
    # most exceptions are below are network or Amazon related
    # they are fixed by simply retrying
    except AssertionError:
        print("ASSERTION ERROR, RESTARTING")
        time.sleep(20)
        continue
    except TimeoutException:
        print("TIMED OUT, RESTARTING")
        time.sleep(60)
        continue
    except requests.exceptions.ChunkedEncodingError:
        print("CHUNKED ENCONDING ERROR, RESTARTING")
        time.sleep(2)
        continue
    except Exception:
        # any Exception other than the above indicates a bug in the code
        # bug is to be fixed and program is to be restarted
        print(traceback.format_exc())
        break

from datetime import datetime, timedelta

import dataset
import matplotlib.animation as animation
import matplotlib.image as image
import matplotlib.pyplot as plt
from matplotlib.ticker import PercentFormatter
import numpy as np
import pandas as pd

# formatting settings for matplotlib
plt.rcParams["font.family"] = "Noto Sans"
plt.rcParams["font.size"] = 16

# common function for adding our logo to plots
logo = image.imread("logo.png")
def _add_logo(fig):
    logo_ax = fig.add_axes([0.02, 0.82, 0.13, 0.13], anchor="NE", zorder=1)
    logo_ax.imshow(logo)
    logo_ax.axis("off")

# connect to db
db = dataset.connect("sqlite:///data.db")

# get data for the 4 charts from DB
mostread_fiction = list(db["chart_entries"].find(chart_name="mostread/fiction"))
mostsold_fiction = list(db["chart_entries"].find(chart_name="mostsold/fiction"))
mostread_nonfiction = list(db["chart_entries"].find(chart_name="mostread/nonfiction"))
mostsold_nonfiction = list(db["chart_entries"].find(chart_name="mostsold/nonfiction"))

# read all necessary book metadata into a dict
book_metadata = {}
date_keys = ["details_publication_date", "details_audible_com_release_date", "details_release_date"]
for book in db["books"]:
    book_id = book["book_id"]
    book_metadata[book_id] = book
    for key in date_keys:
        if book[key]:
            book_metadata[book_id][key] = datetime(*list(map(int, book[key].split("-"))))

def table_longest_charting(entries, filename):
    books_dict = {}
    for entry in entries:
        book_name = entry["book_name"]
        if book_name not in books_dict:
            # start counter if it does not exist
            books_dict[book_name] = {"weeks": 0}
        # increment the counter
        books_dict[book_name]["weeks"] += 1

    index = [] # book names
    weeks = [] # number of weeks spent on charts
    for book_name in books_dict:
        index.append(book_name)
        weeks.append(books_dict[book_name]["weeks"])

    df = pd.DataFrame({"Weeks": weeks}, index=index)
    df = df.sort_values(by="Weeks", ascending=False)

    # identify the value for 20th position
    # filter out everything below it
    # this makes sure all books tied at 20th position are included
    cutoff_value = df["Weeks"].iloc[19]
    df = df[df["Weeks"] >= cutoff_value]
    df.to_markdown(filename, index=True)

# run above function on all 4 datasets
table_longest_charting(mostread_fiction, "longest-charting-most-read-fiction.md")
table_longest_charting(mostsold_fiction, "longest-charting-most-sold-fiction.md")
table_longest_charting(mostread_nonfiction, "longest-charting-most-read-nonfiction.md")
table_longest_charting(mostsold_nonfiction, "longest-charting-most-sold-nonfiction.md")

def stints(entries, chart_name):
    file_suffix = chart_name.lower().replace(" ", "-")

    # to keep track of books by week
    weekwise_books = {}
    for entry in entries:
        if entry["date"] not in weekwise_books:
            weekwise_books[entry["date"]] = set()
        book_name = entry["book_name"]
        weekwise_books[entry["date"]].add(book_name)

    # the dates to iterate over
    dates = sorted(weekwise_books.keys())

    # to keep track of books that entered and left
    stints = []

    # to keep track of books that are currently staying on the charts
    open_stints = {}

    last_date = "2024-05-05"
    for date in dates:
        week_book_names = weekwise_books[date]

        # for books on the chart in this week
        for book_name in week_book_names:
            # check if this book entered the charts this week
            # if yes, initialize stint
            if book_name not in open_stints:
                open_stints[book_name] = {
                    "book_name": book_name,
                    "start_date": date,
                    "weeks": 0,
                }

            # increment weeks in stint
            open_stints[book_name]["weeks"] += 1

        # for books that were staying on the charts
        open_stint_book_names = list(open_stints.keys())
        for book_name in open_stint_book_names:
            # check if a book exited this week
            if book_name not in week_book_names or date==last_date:
                # record the stint weeks
                stints.append(open_stints[book_name])
                # end the stint
                del open_stints[book_name]

    # make a dataframe with the stints
    data = {"Book Name": [], "Start Date": [], "Weeks": []}
    index = []
    for obj in stints:
        data["Book Name"].append(obj["book_name"])
        data["Start Date"].append(obj["start_date"])
        data["Weeks"].append(obj["weeks"])
        index.append(obj["book_name"])

    df = pd.DataFrame(data, index)
    df = df.sort_values(by="Weeks", ascending=False)

    # write the longest stints
    cutoff_value = df["Weeks"].iloc[19]
    top_df = df[df["Weeks"] >= cutoff_value]
    top_df.to_markdown("longest-stints-" + file_suffix + ".md", index=False)

    # make a dataframe with number of stints per book
    stint_count_dict = {x: index.count(x) for x in set(index)}
    counts_index = list(stint_count_dict.keys())
    counts_data = {"Book Name": [], "Stints": []}
    for book_name in counts_index:
        counts_data["Book Name"].append(book_name)
        counts_data["Stints"].append(stint_count_dict[book_name])

    stint_count_df = pd.DataFrame(counts_data, counts_index)
    stint_count_df = stint_count_df.sort_values(by="Stints", ascending=False)

    # write books with most number of stints
    cutoff_value = stint_count_df["Stints"].iloc[19]
    top_df = stint_count_df[stint_count_df["Stints"] >= cutoff_value]
    top_df.to_markdown("most-stints-"+file_suffix+".md", index=False)

# run above function on the 4 datasets
stints(mostread_fiction, "Most Read Fiction")
stints(mostread_nonfiction, "Most Read Non Fiction")
stints(mostsold_fiction, "Most Sold Fiction")
stints(mostsold_nonfiction, "Most Sold Non Fiction")

def debut_trends(entries):
    # common for both plots

    # to keep track over iterations
    debuted_book_names = set()
    debut_positions = []
    times_to_debut = []

    for entry in entries:
        book_name = entry["book_name"]
        book_id = entry["link"].split("/")[2]
        if book_name not in debuted_book_names:
            book = book_metadata[book_id]
            if book["details_publication_date"]:
                chart_date = datetime(*list(map(int, entry["date"].split("-"))))
                diff = chart_date - book["details_publication_date"]
                times_to_debut.append(int(diff.days/7))
            debuted_book_names.add(book_name)
            debut_positions.append(entry["chart_position"])
    return debut_positions, times_to_debut

def plot_debut_position(debut_positions, chart_name):
    file_suffix = chart_name.lower().replace(" ", "-")

    # bin the positions
    positions = list(range(1,21))
    positionwise_debut_percentages = list(map(
        lambda val: debut_positions.count(val)/len(debut_positions),
        positions,
    ))

    # plot bar chart
    fig = plt.figure(figsize=(16, 9))
    bar_plot = plt.bar(
        positions, positionwise_debut_percentages,
        color="#ffc91f",
    )
    plt.xlabel("Debut Position on Chart")
    plt.ylabel("No. of Books")
    plt.gca().yaxis.set_major_formatter(PercentFormatter(1))
    plt.xticks(positions)
    plt.bar_label(bar_plot, fmt=lambda x: f'{round(x*100, 1)}%')
    plt.title(
        "Amazon Book Charts\nDebut Position Distribution - " + chart_name,
    )
    _add_logo(fig)
    plt.savefig("debut-position-distribution-" + file_suffix + ".png")
    plt.clf()

def plot_times_to_debut(times_to_debut_a, times_to_debut_b, chart_name, legend):
    file_suffix = chart_name.lower().replace(" ", "-").replace(".", "")

    # Compute histogram values first
    histogram_a, _bins = np.histogram(times_to_debut_a, bins=range(0, max(times_to_debut_a), 1))
    cumulative_histogram_a = np.cumsum(histogram_a)
    normalized_cumulative_histogram_a = cumulative_histogram_a/cumulative_histogram_a[-1]

    histogram_b, _bins = np.histogram(times_to_debut_b, bins=range(0, max(times_to_debut_b), 1))
    cumulative_histogram_b = np.cumsum(histogram_b)
    normalized_cumulative_histogram_b = cumulative_histogram_b/cumulative_histogram_b[-1]

    # plot computed histogram as line chart
    # avoided plt.hist due to limited formatting options
    fig = plt.figure(figsize=(16, 9))
    plt.plot(normalized_cumulative_histogram_a[:500], color = "#ffc91f", linewidth=2)
    plt.plot(normalized_cumulative_histogram_b[:500], color = "#1fffcb", linewidth=2)
    plt.ylim(0,1)
    plt.xlabel("Weeks After Publication")
    plt.ylabel("% of Books Debuted")
    plt.gca().yaxis.set_major_formatter(PercentFormatter(1))
    plt.title("Amazon Book Charts\nCumulative Time To Debut - " + chart_name)
    plt.legend(legend)

    _add_logo(fig)
    plt.savefig(f"time-to-debut-{file_suffix}.png")
    plt.clf()


# Most Read Debut Trends
debut_positions_1, times_to_debut_1  = debut_trends(mostread_fiction)
debut_positions_2, times_to_debut_2 = debut_trends(mostread_nonfiction)
plot_debut_position(debut_positions_1, "Most Read Fiction")
plot_debut_position(debut_positions_2, "Most Read Non Fiction")
plot_times_to_debut(
    times_to_debut_1,
    times_to_debut_2,
    "Most Read Fiction vs. Non Fiction",
    ["Fiction", "Non Fiction"],
)

# Most Sold Debut Trends
debut_positions_3, times_to_debut_3 = debut_trends(mostsold_fiction)
debut_positions_4, times_to_debut_4 = debut_trends(mostsold_nonfiction)
plot_debut_position(debut_positions_3, "Most Sold Fiction")
plot_debut_position(debut_positions_4, "Most Sold Non Fiction")
plot_times_to_debut(
    times_to_debut_3,
    times_to_debut_4,
    "Most Sold Fiction vs. Non Fiction",
    ["Fiction", "Non Fiction"]
)

# used one of 6 colors for each book
# color of a book is fixed through the timelapse
# helps to keep track of a book as it moves

def animated_bar_chart(entries, chart_name):
    file_suffix = chart_name.lower().replace(" ", "-")

    # dates to iterate over
    dates = list(set([x["date"] for x in entries]))
    dates.sort()

    fig = plt.figure(figsize=(12, 9))
    colorset = ["#ffcb1f", "#53ff1f", "#1fffcb", "#1f53ff", "#cb1fff", "#ff1f53"]
    charted_books = {}

    # returns a plot for a given date
    def animate(i):
        print(i+1, end="\r"*4)

        # clear previous plot
        plt.clf()

        # get entries for given date
        date = dates[i]
        date_obj = datetime(*map(int, date.split("-")))
        date_entries = list(filter(lambda entry: entry["date"]==date, entries))
        date_entries.sort(key=lambda entry: entry["chart_position"])

        # get book names and number of weeks on chart
        for entry in date_entries:
            # if its a new entrant, start counter and allot a color
            if entry["book_name"] not in charted_books:
                color = colorset[len(charted_books)%len(colorset)]
                charted_books[entry["book_name"]] = {"color": color, "weeks": 0}
            # increment counter for new entrants & previously charted books
            charted_books[entry["book_name"]]["weeks"] += 1

        # construct a data frame to get top 20
        names = []
        weeks_on_chart = []
        colors = []
        for book_name in charted_books:
            names.append(book_name)
            weeks_on_chart.append(charted_books[book_name]["weeks"])
            colors.append(charted_books[book_name]["color"])

        df = pd.DataFrame({"Name": names, "Weeks": weeks_on_chart, "Color": colors})
        df = df.sort_values(by=["Weeks", "Name"], ascending=False).head(20)

        # draw the bar plot for this frame
        bar_plot = plt.barh(
            df["Name"][::-1],
            df["Weeks"][::-1],
            color=df["Color"][::-1],
            height=0.3,
        )

        # customize the plot
        ax = plt.gca()
        ax.get_xaxis().set_visible(False)
        ax.spines["top"].set_visible(False)
        ax.spines["right"].set_visible(False)
        ax.spines["bottom"].set_visible(False)
        ax.spines["left"].set_visible(False)

        plt.tick_params(left=False)
        plt.bar_label(bar_plot, padding=10)

        date_str = date_obj.strftime("%-d %b %Y")
        plt.title(
            "\n".join([
                f"Number of Weeks on Amazon's {chart_name} Chart",
                "Date: " + date_str
            ]),
            x=1.05, y=1.02, ha="right",
        )
        plt.subplots_adjust(left=0.6, top=0.87)
        _add_logo(fig)

    # generate plot for all dates and put it together
    ani = animation.FuncAnimation(fig, animate, frames=len(dates))
    # save as GIF
    ani.save(f"animated-{file_suffix}.gif", writer=animation.PillowWriter(fps=3))

animated_bar_chart(mostread_fiction, "Most Read Fiction")
animated_bar_chart(mostread_nonfiction, "Most Read Non Fiction")
animated_bar_chart(mostsold_fiction, "Most Sold Fiction")
animated_bar_chart(mostsold_nonfiction, "Most Sold Non Fiction")