0
\$\begingroup\$

I've done a script to sort fanfics by the bayesian average of views and rating. I've used views because it took much longer to scrape the dlp website for reviews.

I leave all the code so that it works. But I'm most interested in the following functions:

get_percentile(conn, column, nth)
get_top_100(conn)
get_bayesian_ranking(conn)
print_html(rows)
print_markdown(rows)
print_text(rows)
print_database()

Here is the code:

# Formula for the bayesian rating:
# bayesianAverage = (v ÷ (v+m)) × R + (m ÷ (v+m)) × C
# R = average for the fanfic (mean) = rating
# v = number of views for the fanfic = views
# C = minimum votes required to be listed in the Top 100 (10,000)
# m = the mean vote across the whole report

import re
import sqlite3
import requests
import logging
import traceback
from bs4 import BeautifulSoup

dlp_url = 'https://forums.darklordpotter.net/'

def parse_library(cursor):
    dlp_forums_url = dlp_url + 'forums/'
    parse_forum(cursor, dlp_forums_url + 'almost-recommended.41/')
    parse_forum(cursor, dlp_forums_url + 'general-fics.28/')
    parse_forum(cursor, dlp_forums_url + 'humor.27/')
    parse_forum(cursor, dlp_forums_url + 'the-alternates.16/')
    parse_forum(cursor, dlp_forums_url + 'dark-arts.14/')
    parse_forum(cursor, dlp_forums_url + 'romance.17/')
    parse_forum(cursor, dlp_forums_url + 'restricted-section.35/')

def parse_forum(cursor, web_url: str):
    r = requests.get(web_url)
    web = BeautifulSoup(r.text, 'lxml')
    number_pages = web.find('div', class_='PageNav')['data-last']

    parse_page(cursor, web)
    for page_number in range(2, int(number_pages) + 1):
        page_url = web_url + 'page-' + str(page_number)
        r = requests.get(page_url)
        page = BeautifulSoup(r.text, 'lxml')
        parse_page(cursor, page)

def parse_page(cursor, page):
    discussion_list = page.find_all('li', class_='discussionListItem')
    for discussion in discussion_list:
        parse_discussion(cursor, discussion)

def parse_discussion(cursor, discussion):
    if is_sticky(discussion):
        return
    fic = {}
    discussion_main = discussion \
        .find('div', class_='listBlock main') \
        .find('h3', class_='title') \
        .find('a', class_='PreviewTooltip')
    href = discussion_main['href']
    fic['dlp_url'] = dlp_url + href
    discussion_text = discussion_main.text
    logging.info(f"Scraping '{discussion_text}'")
    parse_status(discussion, fic)
    parse_rating(discussion, fic)
    parse_views(discussion, fic)
    parse_title(discussion_main, fic)
    insert_in_database(cursor, fic)

def is_sticky(discussion) -> bool:
    if discussion.find('span', class_='fa fa-thumb-tack') is not None:
        return True
    else:
        return False

def parse_status(discussion, fic):
    status = ""
    try:
        status = discussion.find('a', class_='prefixLink').find('span', class_='prefix').text
    except AttributeError as e:
        logging.warning("No status")
    fic['status'] = status

def parse_title(discussion_main, fic):
    title = ""
    author = ""
    age_rating = ""
    try:
        regex = r"^(?P<title>.*)( by )(?P<author>.*)( - )(?P<age_rating>.*)"
        matches = re.search(regex, discussion_main.text)
        title = matches.group('title')
        author = matches.group('author')
        age_rating = matches.group('age_rating')
    except Exception as e:
        logging.warning(f"Couldn't parse {discussion_main.text}, {fic['status']}, {fic['rating']}, {fic['views']}, {fic['dlp_url']}")
    fic['title'] = title
    fic['author'] = author
    fic['age_rating'] = age_rating

def parse_rating(discussion, fic):
    rating = 0.0
    try:
        rating = float(discussion.find('span', class_='ratings')['title'])
    except AttributeError as e:
        logging.warning("No rating")
    fic['rating'] = rating

def parse_views(discussion, fic):
    views = 0
    try:
        views = int(discussion.find('div', class_='listBlock stats pairsJustified').find('dl', class_='minor').find('dd').text.replace(',', ''))
    except AttributeError as e:
        logging.warning("No views")
    fic['views'] = views

def create_database_schema(cursor):
    cursor.execute("DROP TABLE IF EXISTS fic")
    sql = '''CREATE TABLE fic(
                id INTEGER PRIMARY KEY,
                title TEXT NOT NULL,
                author TEXT,
                age_rating INT,
                status TEXT,
                rating FLOAT,
                views INT,
                dlp_url TEXT
            )'''
    cursor.execute(sql)

def insert_in_database(cursor, fic):
    if not fic['title']:
        return
    logging.info(f'Insert {{{fic["title"]}, {fic["author"]}, {fic["age_rating"]}, {fic["status"]}, {fic["rating"]}, {fic["views"]}, {fic["dlp_url"]}}}')
    try:
        sql = '''INSERT INTO fic(title, author, age_rating, status, rating, views, dlp_url)
                 VALUES (?,?,?,?,?,?,?)'''
        cursor.execute(sql, (fic['title'], fic['author'], fic['age_rating'], fic['status'], fic['rating'], fic['views'], fic['dlp_url']))
    except:
        logging.error(traceback.print_exception(type(e), e, e.__traceback__))
        logging.error(f'Failed to insert {{{fic["title"]}, {fic["author"]}, {fic["age_rating"]}, {fic["status"]}, {fic["rating"]}, {fic["views"]}, {fic["dlp_url"]}}}')

def get_percentile(conn, column, nth):
    nth /= 100
    # https://stackoverflow.com/a/1123642/13477305
    sql = f'''
        SELECT {column}
        FROM fic
        ORDER BY {column} ASC
        LIMIT 1
        OFFSET ROUND((
            SELECT COUNT(*)
            FROM fic) * {nth} - 1);
    '''
    return conn.execute(sql).fetchone()[0]

def get_top_100(conn):
    sql = '''
        SELECT ROUND((m * c + views * rating) /
               (c + views), 2) as bayesian,
               title, views, status, age_rating, rating, dlp_url
        FROM (
            SELECT AVG(rating) OVER () as m,
                   AVG(views) OVER () as c,
                   *
             FROM fic
            WHERE views > 10000
        )
        ORDER BY bayesian DESC;
    '''
    return conn.cursor().execute(sql).fetchall()

def get_bayesian_ranking(conn):
    _25th_percentile_views = get_percentile(conn, "views", 25)
    sql = f'''
        SELECT ROUND((m * {_25th_percentile_views} + views * rating) /
               ({_25th_percentile_views} + views), 2) as bayesian,
               rating, views, status, age_rating, title, dlp_url
        FROM (
            SELECT AVG(rating) OVER () as m,
                   AVG(views) OVER () as c,
                   *
            FROM fic
        )
        ORDER BY bayesian DESC;
    '''
    return conn.cursor().execute(sql).fetchall()

def print_html(rows):
    print('<table><tr><td>Bayesian</td><td>Rating</td><td>Views</td>'
          '<td>Status</td><td>Title</td></tr>')
    for row in rows:
        print(f'''
<tr>
  <td>{row["bayesian"]}</td>
  <td>{row["rating"]}</td>
  <td>{row["views"]}</td>
  <td>{row["status"]}</td>
  <td>[url={row["dlp_url"]}]{row["title"]}[/url]</td>
</tr>
''')
    print('</table>')

def print_markdown(rows):
    print('| Bayesian | Rating | Views | Status | Title |')
    print('| --- | --- | --- | --- | --- |')
    for row in rows:
        print(f'| {row["bayesian"]} | {row["rating"]} | {row["views"]} '
              f'| {row["status"]} | [{row["title"]}]({row["dlp_url"]}) |')

def print_text(rows):
    print('Bayesian Rating Views Status Title')
    for row in rows:
        print(f'{row["bayesian"]} {row["rating"]} {row["views"]} '
              f'{row["status"]} {row["title"]}')

def print_database():
    conn = sqlite3.connect('dlp.sqlite3')
    conn.row_factory = sqlite3.Row
    #rows = get_top_100(conn)
    rows = get_bayesian_ranking(conn)
    #print_html(rows)
    print_markdown(rows)
    #print_text(rows)

def main():
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(message)s",
        handlers=[
            logging.FileHandler("debug.log"),
            logging.StreamHandler()
        ]
    )
    conn = sqlite3.connect('dlp.sqlite3')
    cursor = conn.cursor()
    create_database_schema(cursor)
    parse_library(cursor)
    insert_missing(cursor)
    conn.commit()
    conn.close()

if __name__ == "__main__":
    #main()
    print_database()
\$\endgroup\$

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy