В последнем материале мы подготовили базовый макет дашборда при помощи библиотеки dash-bootstrap-components с двумя графиками: scatter plot и российской картой, которые подробно разбирали ранее. Сегодня продолжим наполнять дашборд информацией: встроим в него таблицы и фильтр данных по пивоварням.
Получение таблиц
Сами таблицы будем описывать в макете в файле application.py, но информацию, которую они отображают лаконичнее будет получить в отдельном модуле. Создадим файл get_tables.py: в нём будет функция, передающая готовую таблицу класса Table библиотеки dbc в application.py. В этом материале мы опишем только таблицу лучших пивоварен России, но на GithHub будут представлены все три.
В таблицах по заведениям и пивоварням мы реализуем фильтр по городам, но изначально города в собранных с Untappd данных записаны на латинице. Для запросов мы будем переводить русскоязычные наименования городов на английский при помощи библиотеки Google Translate. Кроме того, одни и те же города могут называться по-разному — например, «Москва» на латинице где-то записана как «Moskva», а где-то как «Moscow». Для этого дополнительно настроим маппинг наименований города и заранее создадим словарь с корректными наименованиями основных городов. Он пригодится в самом конце.
import pandas as pd
import dash_bootstrap_components as dbc
from clickhouse_driver import Client
import numpy as np
from googletrans import Translator
translator = Translator()
client = Client(host='12.34.56.78', user='default', password='', port='9000', database='')
city_names = {
'Moskva': 'Москва',
'Moscow': 'Москва',
'СПБ': 'Санкт-Петербург',
'Saint Petersburg': 'Санкт-Петербург',
'St Petersburg': 'Санкт-Петербург',
'Nizhnij Novgorod': 'Нижний Новгород',
'Tula': 'Тула',
'Nizhniy Novgorod': 'Нижний Новгород',
}
Таблица лучших пивоварен
Таблица, о которой идёт речь в материале, будет показывать топ-10 лучших российских пивоварен с изменением рейтинга. То есть мы сравниваем данные за два периода: [30 дней назад; сегодня] и [60 дней назад; 30 дней назад] и смотрим, как менялось место пивоварни в рейтинге. Соответственно, мы опишем следующие колонки: место в рейтинге, название пивоварни, ассортимент сортов пива, рейтинг пивоварни на untappd, изменение места и количество чекинов у этой пивоварни.
Опишем функцию get_top_russian_breweries, которая отправляет запрос к Clickhouse, получает общий топ пивоварен России, формирует данные и возвращает готовый для вывода DataFrame. Отправим два запроса — топ пивоварен за последние 30 дней и топ пивоварен за предыдущие 30 дней. Следующий запрос будет отбирать лучшие пивоварни, основываясь на количестве отзывов о пиве данной пивоварни.
Забираем данные из базы
def get_top_russian_breweries(checkins_n=250):
top_n_brewery_today = client.execute(f'''
SELECT rt.brewery_id,
rt.brewery_name,
beer_pure_average_mult_count/count_for_that_brewery as avg_rating,
count_for_that_brewery as checkins FROM (
SELECT
brewery_id,
dictGet('breweries', 'brewery_name', toUInt64(brewery_id)) as brewery_name,
sum(rating_score) AS beer_pure_average_mult_count,
count(rating_score) AS count_for_that_brewery
FROM beer_reviews t1
ANY LEFT JOIN venues AS t2 ON t1.venue_id = t2.venue_id
WHERE isNotNull(venue_id) AND (created_at >= (today() - 30)) AND (venue_country = 'Россия')
GROUP BY
brewery_id,
brewery_name) rt
WHERE (checkins>={checkins_n})
ORDER BY avg_rating DESC
LIMIT 10
'''
)
top_n_brewery_n_days = client.execute(f'''
SELECT rt.brewery_id,
rt.brewery_name,
beer_pure_average_mult_count/count_for_that_brewery as avg_rating,
count_for_that_brewery as checkins FROM (
SELECT
brewery_id,
dictGet('breweries', 'brewery_name', toUInt64(brewery_id)) as brewery_name,
sum(rating_score) AS beer_pure_average_mult_count,
count(rating_score) AS count_for_that_brewery
FROM beer_reviews t1
ANY LEFT JOIN venues AS t2 ON t1.venue_id = t2.venue_id
WHERE isNotNull(venue_id) AND (created_at >= (today() - 60) AND created_at <= (today() - 30)) AND (venue_country = 'Россия')
GROUP BY
brewery_id,
brewery_name) rt
WHERE (checkins>={checkins_n})
ORDER BY avg_rating DESC
LIMIT 10
'''
)
Формируем из полученных строк два DataFrame:
top_n = len(top_n_brewery_today)
column_names = ['brewery_id', 'brewery_name', 'avg_rating', 'checkins']
top_n_brewery_today_df = pd.DataFrame(top_n_brewery_today, columns=column_names).replace(np.nan, 0)
top_n_brewery_today_df['brewery_pure_average'] = round(top_n_brewery_today_df.avg_rating, 2)
top_n_brewery_today_df['brewery_rank'] = list(range(1, top_n + 1))
top_n_brewery_n_days = pd.DataFrame(top_n_brewery_n_days, columns=column_names).replace(np.nan, 0)
top_n_brewery_n_days['brewery_pure_average'] = round(top_n_brewery_n_days.avg_rating, 2)
top_n_brewery_n_days['brewery_rank'] = list(range(1, len(top_n_brewery_n_days) + 1))
А затем в итераторе считаем, как изменилось место за последнее время у пивоварни. Обработаем исключение на случай, если 60 дней назад этой пивоварни в нашей базе ещё не было.
rank_was_list = []
for brewery_id in top_n_brewery_today_df.brewery_id:
try:
rank_was_list.append(
top_n_brewery_n_days[top_n_brewery_n_days.brewery_id == brewery_id].brewery_rank.item())
except ValueError:
rank_was_list.append('–')
top_n_brewery_today_df['rank_was'] = rank_was_list
Теперь пройдёмся по полученным колонкам с текущими местами и изменениями. Если они не пустые, то при положительном изменении добавим к записи стрелочку вверх. При отрицательном — стрелочку вниз.
diff_rank_list = []
for rank_was, rank_now in zip(top_n_brewery_today_df['rank_was'], top_n_brewery_today_df['brewery_rank']):
if rank_was != '–':
difference = rank_was - rank_now
if difference > 0:
diff_rank_list.append(f'↑ +{difference}')
elif difference < 0:
diff_rank_list.append(f'↓ {difference}')
else:
diff_rank_list.append('–')
else:
diff_rank_list.append(rank_was)
Наконец, разметим итоговый DataFrame и вставим в него колонку с текущим местом. При этом у топ-3 будет отображаться эмодзи с золотым кубком.
df = top_n_brewery_today_df[['brewery_name', 'avg_rating', 'checkins']].round(2)
df.insert(2, 'Изменение', diff_rank_list)
df.columns = ['НАЗВАНИЕ', 'РЕЙТИНГ', 'ИЗМЕНЕНИЕ', 'ЧЕКИНОВ']
df.insert(0, 'МЕСТО',
list('🏆 ' + str(i) if i in [1, 2, 3] else str(i) for i in range(1, len(df) + 1)))
return df
#Выбор пивоварен с фильтром по городам
Одна из функций нашего дашборда — просмотр топа пивоварен по конкретному городу. Для корректной работы напишем скрипт, который для каждого из списка российских городов получает топ пивоварен по числу чекинов и записывает данные по каждому городу в отдельные csv-файлы. В сущности, он мало чем отличается от предыдущего — рассмотрим главные отличия.
Прежде всего, функция принимает конкретный город. Мы уже отметили, что города в базе данных записаны на латинице — поэтому сначала переводим наименование города. В случае с Санкт-Петербургом, Нижним Новгородом и Пермью придётся перевести вручную: например, Санкт-Петербург переводится в Google Translate как St. Petersburg вместо ожидаемого Saint Petersburg.
ru_city = venue_city
if ru_city == 'Санкт-Петербург':
en_city = 'Saint Petersburg'
elif ru_city == 'Нижний Новгород':
en_city = 'Nizhnij Novgorod'
elif ru_city == 'Пермь':
en_city = 'Perm'
else:
en_city = translator.translate(ru_city, dest='en').text
Следующее отличие — запрос к базе. Нам нужно добавить в него условие совпадения по городу, чтобы получать чекины только в запрошенном городе:
WHERE (rt.venue_city='{ru_city}' OR rt.venue_city='{en_city}')
Наконец, сформированный DataFrame мы не возвращаем, а сохраняем в директорию data/cities.
df = top_n_brewery_today_df[['brewery_name', 'venue_city', 'avg_rating', 'checkins']].round(2)
df.insert(3, 'Изменение', diff_rank_list)
df.columns = ['НАЗВАНИЕ', 'ГОРОД', 'РЕЙТИНГ', 'ИЗМЕНЕНИЕ', 'ЧЕКИНОВ']
df.to_csv(f'data/cities/{en_city}.csv', index=False) # saving DF
print(f'{en_city}.csv updated!')
Обновление таблиц по расписанию
Наш дашборд будет использовать библиотеку apscheduler для вызова последней функции по расписанию и обновления таблиц по городам. Следующие строки добавим в файл application.py: scheduler будет обновлять данные для каждого города из списка all_cities ежедневно в 13:30 по МСК.
from apscheduler.schedulers.background import BackgroundScheduler
from get_tables import update_best_breweries
all_cities = sorted(['Москва', 'Сергиев Посад', 'Санкт-Петербург', 'Владимир',
'Красная Пахра', 'Воронеж', 'Екатеринбург', 'Ярославль', 'Казань',
'Ростов-на-Дону', 'Краснодар', 'Тула', 'Курск', 'Пермь', 'Нижний Новгород'])
scheduler = BackgroundScheduler()
@scheduler.scheduled_job('cron', hour=10, misfire_grace_time=30)
def update_data():
for city in all_cities:
update_best_breweries(city)
scheduler.start()
Формирование таблицы
Наконец, опишем заключительную функцию get_top_russian_breweries_table(venue_city, checkins_n=250) — она будет принимать город, количество чекинов и будет возвращать сформированную таблицу dbc. Второй параметр — checkins_n будет отсеивать пивоварни, у которых чекинов меньше значения этой переменной. Если город не указан, сразу вызываем ранее описанную get_top_russian_breweries(checkins_n) — она вернёт общую статистику за последнее время. В противном случае снова переводим города на латиницу.
if venue_city == None:
selected_df = get_top_russian_breweries(checkins_n)
else:
ru_city = venue_city
if ru_city == 'Санкт-Петербург':
en_city = 'Saint Petersburg'
elif ru_city == 'Нижний Новгород':
en_city = 'Nizhnij Novgorod'
elif ru_city == 'Пермь':
en_city = 'Perm'
else:
en_city = translator.translate(ru_city, dest='en').text
Читаем все строки из таблицы с нужным городом и проверяем количество чекинов каждой пивоварни. В самом начале материала мы завели словарь city_names. При помощи функции map() мы пишем лямбда-выражение, которое возвращает значение ключа словаря city_names только если входной аргумент из колонки df[‘ГОРОД’] совпадает с каким-либо из ключей в city_names. В случае, если совпадения не будет возвращает просто x во избежание np.Nan.
Например, для наименования «СПБ» в колонке df[‘ГОРОД’] вернётся значение «Санкт-Петербург», так как такой ключ есть в city_names. Для «Воронеж» название таким и останется, так как совпадающий ключ не найден. В конце удаляем возможные дубликаты из DataFrame, добавляем колонку с номером места пивоварни и забираем себе первые 10 строк — это и будет топ-10 пивоварен по нужному городу.
df = pd.read_csv(f'data/cities/{en_city}.csv')
df = df.loc[df['ЧЕКИНОВ'] >= checkins_n]
df['ГОРОД'] = df['ГОРОД'].map(lambda x: city_names[x] if (x in city_names) else x)
df.drop_duplicates(subset=['НАЗВАНИЕ', 'ГОРОД'], keep='first', inplace=True)
df.insert(0, 'МЕСТО', list('🏆 ' + str(i) if i in [1, 2, 3] else str(i) for i in range(1, len(df) + 1)))
selected_df = df.head(10)
Вне зависимости от того, получали мы DataFrame общей функцией get_top_russian_breweries() или по конкретному городу, собираем таблицу, задаём стили и возвращаем готовый dbc-объект.
Вёрстка в Dash Bootstrap Components
table = dbc.Table.from_dataframe(selected_df, striped=False,
bordered=False, hover=True,
size='sm',
style={'background-color': '#ffffff',
'font-family': 'Proxima Nova Regular',
'text-align':'center',
'fontSize': '12px'},
className='table borderless'
)
return table
Структура вёрстки
Опишем в application.py слайдер, таблицу и Dropdown-фильтр с выбором города.
О вёрстке дашборда при помощи Dash Bootstrap Components мы говорили в предыдущем материале цикла
checkins_slider_tab_1 = dbc.CardBody(
dbc.FormGroup(
[
html.H6('Количество чекинов', style={'text-align': 'center'})),
dcc.Slider(
id='checkin_n_tab_1',
min=0,
max=250,
step=25,
value=250,
loading_state={'is_loading': True},
marks={i: i for i in list(range(0, 251, 25))}
),
],
),
style={'max-height': '80px',
'padding-top': '25px'
}
)
top_breweries = dbc.Card(
[
dbc.CardBody(
[
dbc.FormGroup(
[
html.H6('Фильтр городов', style={'text-align': 'center'}),
dcc.Dropdown(
id='city_menu',
options=[{'label': i, 'value': i} for i in all_cities],
multi=False,
placeholder='Выберите город',
style={'font-family': 'Proxima Nova Regular'}
),
],
),
html.P(id="tab-1-content", className="card-text"),
],
),
],
)
И для обновления таблицы по фильтру и слайдеру с минимальным количеством чекинов опишем callback с вызовом get_top_russian_breweries_table(city, checkin_n):
@app.callback(
Output("tab-1-content", "children"), [Input("city_menu", "value"),
Input("checkin_n_tab_1", "value")]
)
def table_content(city, checkin_n):
return get_top_russian_breweries_table(city, checkin_n)
Готово! Напомню, в материале описан пример создания только одной таблицы. На данный момент дашборд помимо лучших пивоварен выдаёт лучшие и худшие сорта пива, а также средний рейтинг пива по регионам и отношение количества чекинов каждой пивоварни к её средней оценке.
Полный код проекта доступен на GitHub
