Добавляем графики в Notion

Многим не хватает графиков в Notion'e. Поэтому я решил напилить автоматическую штуку для их генерации.

Вот как это выглядит с моей стороны:


Всех интересующихся, как это реализовано, прошу под кат.

Часть 1. Постановка проблемы

Проблема, собственно, в том, что графиков в Notion'е нет, и информацию из таблички просто так визуализировать нельзя (а хочется). Соответственно, надо создать такую штуку, которая:

1. Будет брать список страниц, где потенциально могут быть графики
2. Будет из этих страниц собирать описание графика. Описания должны быть на страницах из-за понятности для читателя страницы + чтобы слишком часто не залезать в код исправлять.
3. Будет добавлять график сразу после его описания, удаляя предыдущую версию графика.
4. Будет делать это автоматически (раз в час), и, желательно, бесплатно.

Часть 2. Раскладываем по полочкам

Писать будем сервер на Django. Именно Django, потому что неофициальная либа для API Notion'a написана на питоне. Потом это все дело выгружаем на Heroku. Из IFTTT будем дергать нашу штуку на Heroku с определенной периодичностью.

Часть 3. Разбираемся, что нам нужно написать

1. Метод для того, чтобы отвечать на запрос от IFTTT
2. Метод для прочесывания страниц Notion'a и поиска описаний графиков
3. Метод для доставания данных для графика из таблицы
4. Метод для рисования графика и его добавления на страницу

Часть 4. Пишем код

Идем в Notion, жмякаем Ctrl + Shift + J, идем в Application -> Cookies, копируем token_v2 и называем его TOKEN. Это нужно для того, чтобы библиотека могла как-то взаимодействовать с API Notion'a.

Нам нужно как-то хранить список страниц, где потенциально может встретиться описание для графиков. Хранить мы это дело будет просто:

PAGES = [
    "https://www.notion.so/mixedself/Dashboard-40a3156030fd4d9cb1935993e1f2c7eb"
]

Для того, чтобы как-то распарсить само описание, нам нужны ключевые слова для:

1. Поля, данные которого будут у нас на оси X
2. Поля, данные которого будут у нас на оси Y
3. Урл на табличку

Выглядеть это будет вот так:

BASE_KEY = "Base:"
X_AXIS_KEY = "X axis:"
Y_AXIS_KEY = "Y axis:"

То есть пустое описание графика у нас будет выглядеть как-то вот так:

def get_empty_object():
    return {
        "database": "",
        "x": "",
        "y": ""
    }

Еще нам нужно как-то проверять, пустое ли описание. Для этого напилим специальную функцию. Если у нас все поля не пустые, тогда объект полный и мы можем приступать к отрисовке графика.

def is_not_empty(thing):
    return thing != ""


def check_for_completeness(object):
    return is_not_empty(object["database"]) and is_not_empty(object["x"]) and is_not_empty(object["y"])

Данные (это просто текст по факту) для того, чтобы сгенерировать описание, надо как-то очищать. Напишем для этого парочку функций. Небольшое пояснение: Notion хранит жирный шрифт (как на картинке над катом) __воттаквот__.

def br_text(text):
    return "__" + text + "__"


def clear_text(text):
    return text.replace(br_text(BASE_KEY), "").replace(BASE_KEY, "") \
        .replace(br_text(X_AXIS_KEY), "").replace(X_AXIS_KEY, "") \
        .replace(br_text(Y_AXIS_KEY), "").replace(Y_AXIS_KEY, "").strip()

Теперь напишем, пожалуй, главную функцию для нашей штучки. Под кодом пояснение, что тут происходит:

def plot():
    client = NotionClient(token_v2=TOKEN)

    for page in PAGES:
        blocks = client.get_block(page)
        thing = get_empty_object()

        for i in range(len(blocks.children)):
            block = blocks.children[i]
            print(block.type)

            if block.type != "image":
                title = block.title

                if BASE_KEY in title:
                    thing["database"] = clear_text(title).split("](")[0].replace("[", "")

                elif X_AXIS_KEY in title:
                    thing["x"] = clear_text(title)

                elif Y_AXIS_KEY in title:
                    thing["y"] = clear_text(title)

                    if check_for_completeness(thing):
                        # not last block
                        if i != len(blocks.children) - 1:
                            next_block = blocks.children[i + 1]

                            # if next block is picture, then it is previous
                            # version of the plot, then we should remove it
                            if blocks.children[i + 1].type == "image":
                                next_block.remove()

                        draw_plot(client, thing, block, blocks)
                        thing = get_empty_object()

Мы подключаем нашу библиотеку к Notion'у. Потом проходимся по массиву страниц, где у нас потенциально могут быть нужны графики. Каждую строчку страницы мы проверяем: есть ли там один из наших ключей или нет. Если вдруг есть — чистим оттуда текст и кладем в объект. Как только объект заполнится, мы проверяем, был ли уже тут сгенерированный график (если да, то удаляем) и идем рисовать новый график.

Далее напишем функцию для собирания данных из таблички.

def get_lines_array(thing, client):
    database = client.get_collection_view(thing["database"])
    rows = database.default_query().execute()
    lines_array = []

    for i in range(1, len(rows)):
        previous_row = rows[i - 1]
        current_row = rows[i]
        line = [(get_point_from_row(thing, previous_row)), (get_point_from_row(thing, current_row))]
        lines_array.append(line)

    return lines_array

Тут мы добываем базу, получаем все ее строки и проходимся по всем строкам, формируя набор линий от точки до точки.

А что такое get_point_from_row? Дело в том, что если у нас объект — это дата (часто ведь надо на оси X отображать именно дату), то ее просто так не отобразить, и надо ее дополнительно обработать:

def get_point_from_row(thing, row):
    x_property = row.get_property(thing["x"])
    y_property = row.get_property(thing["y"])

    if thing["x"] == "date":
        x_property = x_property.start

    if thing["y"] == "date":
        y_property = y_property.start

    return x_property, y_property

Теперь мы тащемта готовы к тому, чтобы отрисовать наш график.

def draw_plot(client, thing, block, page):
    photo = page.children.add_new(ImageBlock)
    photo.move_to(block, "after")

    array = get_lines_array(thing, client)
    print(array)

    for i in range(1, len(array)):
        points = reparse_points(array[i - 1:i][0])
        plt.plot(points[0], points[1], color="red")

    if not path.exists("images"):
        os.mkdir("images")

    if thing["x"] == "date":
        x_axis_dates()

    filename = "images/" + random_string(15) + ".png"
    plt.savefig(filename)

    print("Uploading " + filename)
    photo.upload_file(filename)

Здесь мы добавляем новый блок (с фото), перемещаем его под описание графика. Потом репарсим точки (об этом — ниже), отрисовываем линии с помощью matplotlib, сохраняем полученное изображение с рандомным имененм файла и загружаем его в блок картинки.

Рандомное имя файла мы можем получить вот так:

def random_string(string_length=10):
    letters = string.ascii_lowercase
    return ''.join(random.choice(letters) for i in range(string_length))

А точки нам надо репарсить из-за того, что matplotlib на вход принимает представление данных отличное от того, как это у нас сейчас реализовано.

def reparse_points(points):
    return [
        [points[0][0], points[1][0]],
        [points[0][1], points[1][1]],
    ]

Если присмотреться, в методе еще есть проверка, являются ли датой данные, которые у нас по оси Х. Если являются, тогда мы просто должны их корректно отобразить:

def x_axis_dates(ax=None, fig=None):
    if ax is None:
        ax = plt.gca()

    if fig is None:
        fig = plt.gcf()

    loc = mdates.AutoDateLocator()
    fmt = mdates.AutoDateFormatter(loc)

    ax.xaxis.set_major_locator(loc)
    ax.xaxis.set_major_formatter(fmt)

    fig.autofmt_xdate()

Теперь напишем функцию, которая запустит новый тред, когда у нас придет POST-запрос.
Почему именно POST? На всякий случай, чтобы если вдруг к вам на огонек заглянет кто-то, скрипт не запустился.

Почему именно новый тред? IFTTT, который мы будем использовать как триггер для работы этой штуки, не любит, когда ответа от сервера надо ждать очень долго (а в нашем случае ждать может быть долго), и через какое-то время он может прекратить триггерить штукенцию.

@csrf_exempt
def index(request):
    if request.method == "POST":
        thread = Thread(target=plot)
        thread.start()

        return HttpResponse("Hello, world.")

    else:
        return HttpResponse("Hello, world.")

Часть 5. IFTTT

Переходим на вкладку создания апплетов. Выбираем триггер (в нашем случае — это Date&time), ставим „каждый час“. Выбираем триггерируемым (то есть „that“) Webhook, указываем наш (пока что) локальный адрес, дабы потестить. Ну и все. Тестим.

Загружаем на Heroku

Вы думали, для чего мы возились с вот с этим триггерением со стороны IFTTT — это для того, чтобы не платить. Heroku предлагает бесплатный тариф для хостинга нашей штучки. Главное — чтобы сервис спал минимум 6 часов. А он точно будет спать, потому что мы его зовем работать каждый час, а не каждую минуту.

Далее делаем следующее. Идем в heroku создавать новый проект. Далее устанавливаем на свою операционную систему их клиент. А потом делаем все согласно инструкциям, появившимся после создания приложения.

Загрузив все на heroku, переходим в наш апплет и редактируем урл на новый.

Часть 5. IFTTT

Спасибо всем, кто дочитал до этого места. Надеюсь, эта статья вам чем-то помогла.

Можете почитать две другие мои статьи про Notion:

Автоматически экспортируем Google Forms в Notion с помощью IFTTT и Django

Делаем домашнюю библиотеку с Notion и Python