Пет-проєкт, який став SaaS: як я написав фінансовий Telegram-бот і подружив його з Google Sheets API

💡 Усі статті, обговорення, новини про DevOps — в одному місці. Приєднуйтесь до DevOps спільноти!

Хочу поділитися досвідом розробки свого пет-проєкту — Telegram-бота Fibby. Ідея була простою: я ненавидів відкривати окремі додатки для запису витрат, тому вирішив зробити так, щоб транзакції можна було просто писати в месенджер, а зберігалися б вони у звичних Google Таблицях з красивими дашбордами.

Здавалося б, що складного — прикрутити бота до таблиці? Але коли справа дійшла до продакшену, лімітів Google API та асинхронної обробки платежів, почалися цікаві інженерні виклики. Про них і розповім.

Архітектура: чому такий стек?

Я хотів зробити систему максимально стабільною та масштабованою. Тому зупинився на такому стеку:

• Core: Python з бібліотекою telebot (pyTelegramBotAPI) для швидкої обробки повідомлень.
• База даних: Supabase (PostgreSQL). Вона потрібна не для самих фінансових даних (вони в Google Sheets), а для зберігання профілів користувачів, статусів підписок та — найголовніше — черги транзакцій.
• Sync Engine: Окремий скрипт sheet_worker, який відповідає виключно за спілкування з Google API.

Виклик 1: Авторизація та доступ до чужих таблиць

Як безпечно писати дані в таблицю користувача? Я відмовився від ідеї змушувати людей створювати власні Google Cloud проєкти і генерувати ключі — це вбило б конверсію на етапі онбордингу.

Рішення: використання Service Account Architecture. Користувач просто робить копію нашого еталонного шаблону таблиці (з уже налаштованими дашбордами та формулами QUERY) і дає права «Редактора» нашому сервісному email-у ([email protected]).

Після цього юзер відправляє боту команду /set_sheet <посилання>. Бот парсить URL, витягує spreadsheet і робить тестовий запис. Якщо успішно — прив’язує URL до профілю в Supabase:

@bot.message_handler(commands=["set_sheet"])
def cmd_set_sheet(message):
    chat_id = message.chat.id
    parts = message.text.split(maxsplit=1)
    if len(parts) < 2:
        bot.send_message(chat_id, "❌ Формат: /set_sheet <https://docs.google.com/spreadsheets/>...")
        return
    url = parts[1].strip()
    if "docs.google.com/spreadsheets" not in url:
        bot.send_message(chat_id, "❌ Це посилання не схоже на Google Таблицю.")
        return
    try:
        # Тестове відкриття — якщо прав немає, тут полетить Exception
        test_ss = gc.open_by_url(url)
        worksheets = [ws.title for ws in test_ss.worksheets()]
        # Перевіряємо, чи є обов'язкові аркуші (Expenses, Income)
        if "Expenses" not in worksheets or "Income" not in worksheets:
            bot.send_message(chat_id, "❌ Відсутні обов'язкові аркуші Expenses / Income.")
            return
        # Все ок — зберігаємо в Supabase
        supabase.table("users").upsert({
            "telegram_id": chat_id,
            "spreadsheet_url": url
        }).execute()
        bot.send_message(chat_id, "✅ Таблицю підключено!")
    except Exception as e:
        bot.send_message(chat_id, f"❌ Помилка. Перевірте, чи надано доступ Редактора.")

Тут gc — це клієнт gspread, авторизований через Service Account. Ніякого OAuth-флоу для користувача, ніяких токенів в інтерфейсі. Максимально просто.

Виклик 2: Біль від Google Sheets API Rate Limits (і як нас врятував sheet_worker)

Ті, хто працював з Google Sheets API, знають їхні жорсткі ліміти (Rate Limiting). Близько 60 запитів на хвилину на проєкт. Якщо кілька юзерів одночасно почнуть вводити витрати, або хтось вирішить поклацати кнопки, API швидко поверне 429 Too Many Requests, і бот впаде.

Щоб цього уникнути, я реалізував асинхронну чергу.

Коли користувач надсилає витрату, бот не йде в Google Sheets. Він миттєво записує транзакцію в таблицю transactions у Supabase зі статусом pending. Для юзера це виглядає як миттєва відповідь бота «✅ Витрату записано!»:

# bot.py — обробник кнопки категорії витрати
result = supabase.table("transactions").insert({
    "telegram_id": chat_id,
    "amount": amount,
    "category": category,
    "type": "exp",
    "created_at": datetime.now(kyiv_tz).isoformat(),
    "sync_status": "pending"   # <-- ключове поле
}).execute()
bot.edit_message_text(
    f"✅ Витрату записано!\\n📅 {now}\\n📁 {category}\\n💸 {amount:.2f}",
    chat_id, call.message.message_id
)

Далі в гру вступає sheet_worker.py. Це окремий процес-демон, який:

1. Постійно опитує Supabase на наявність записів зі статусом pending.
2. Атомарно захоплює транзакцію (pending → processing), щоб два воркери не взяли одну й ту ж.
3. Виконує запис у Google Sheets через gspread.
4. Змінює статус у БД на synced.

Ось спрощений, але реальний код воркера:

# sheet_worker.py — спрощена версія продакшен-коду
import time
import gspread
from google.oauth2.service_account import Credentials
from supabase import create_client
gc = gspread.authorize(
    Credentials.from_service_account_file("credentials.json", scopes=[...])
)
supabase = create_client(SUPABASE_URL, SUPABASE_KEY)
while True:
    # 1. Беремо pending-транзакції (по 5, щоб не перевищити ліміти)
    response = (supabase.table("transactions")
        .select("*")
        .eq("sync_status", "pending")
        .order("created_at")
        .limit(5)
        .execute())
    transactions = response.data
    if not transactions:
        time.sleep(5)
        continue
    for t in transactions:
        trans_id = t["id"]
        # 2. Атомарне захоплення: pending → processing
        #    Якщо інший worker вже взяв — claim.data буде пустим
        claim = (supabase.table("transactions")
            .update({"sync_status": "processing"})
            .eq("id", trans_id)
            .eq("sync_status", "pending")
            .execute())
        if not claim.data:
            continue  # вже захоплено іншим воркером
        # 3. Знаходимо таблицю юзера
        user = (supabase.table("users")
            .select("spreadsheet_url")
            .eq("telegram_id", t["telegram_id"])
            .execute())
        sheet_url = user.data[0]["spreadsheet_url"]
        try:
            
# 4. Пишемо в Google Sheets
            ss = gc.open_by_url(sheet_url)
            sheet_name = "Expenses" if t["type"] == "exp" else "Income"
            ws = ss.worksheet(sheet_name)
            ws.insert_row(
                [formatted_time, t["category"], float(t["amount"])],
                index=3,
                value_input_option="USER_ENTERED"
            )
            # 5. Позначаємо як synced
            supabase.table("transactions").update(
                {"sync_status": "synced"}
            ).eq("id", trans_id).execute()
            # Спимо між операціями — 1.5с ≈ 40 запитів/хв, безпечно
            time.sleep(1.5)
        except gspread.exceptions.APIError as e:
            if "429" in str(e):
                # Rate limit! Повертаємо в pending і засинаємо на 30с
                supabase.table("transactions").update(
                    {"sync_status": "pending"}
                ).eq("id", trans_id).execute()
                time.sleep(30)
                break  # виходимо з циклу, почнемо заново
            else:
                # Інша помилка API — позначаємо як error
                supabase.table("transactions").update(
                    {"sync_status": "error"}
                ).eq("id", trans_id).execute()

Основна ідея: навіть якщо сервери Google «ляжуть» або ми зловимо ліміт, транзакція просто полежить у Supabase і запишеться пізніше. Жодних втрачених даних. А pending → processing з подвійним .eq() працює як примітивний optimistic lock — гарантує, що два воркери не запишуть одну транзакцію двічі.

Виклик 3: Вебхуки, підписки та WayForPay

Оскільки бот платний (SaaS модель), потрібно було автоматизувати видачу доступу. Я інтегрував WayForPay.

Флоу працює повністю на вебхуках:

1. Користувач обирає тариф (/tariff).
2. Бот генерує платіжне посилання, вшиваючи telegram_id юзера в поле orderReference.
3. Юзер оплачує.
4. WayForPay шле POST-запит на наш ендпоінт /api/payment_callback.
5. Скрипт ловить запит, валідує HMAC-підпис, дістає telegram_id і додає дні підписки.

Ось як це виглядає в коді — обробка вебхуку з валідацією підпису:

# webhook_wayforpay.py
@app.route('/api/payment_callback', methods=['POST'])
def payment_callback():
    data = request.get_json(force=True, silent=True)
    # Витягуємо параметри
    merchant_account   = data.get('merchantAccount', '')
    order_reference    = data.get('orderReference', '')
    amount             = str(data.get('amount', ''))
    currency           = data.get('currency', '')
    auth_code          = str(data.get('authCode', ''))
    card_pan           = data.get('cardPan', '')
    transaction_status = data.get('transactionStatus', '')
    reason_code        = str(data.get('reasonCode', ''))
    received_signature = data.get('merchantSignature', '')
    # Валідація HMAC-MD5 підпису (формат WayForPay)
    sign_string = ";".join([
        merchant_account, order_reference, amount, currency,
        auth_code, card_pan, transaction_status, reason_code
    ])
    expected_signature = hmac.new(
        MERCHANT_SECRET_KEY,
        sign_string.encode('utf-8'),
        hashlib.md5
    ).hexdigest()
    if expected_signature != received_signature:
        return jsonify({"error": "Invalid signature"}), 400
    if transaction_status == 'Approved':
        # orderReference = "123456789_1712345678_3m"
        parts = order_reference.split('_')
        telegram_id = int(parts[0])
        months = int(parts[-1][:-1])  # "3m" → 3
        add_subscription_days(telegram_id, months * 30)
    # WayForPay вимагає підписану відповідь — інакше повторюватиме запит
    resp_time = int(time.time())
    resp_signature = hmac.new(
        MERCHANT_SECRET_KEY,
        f"{order_reference};accept;{resp_time}".encode('utf-8'),
        hashlib.md5
    ).hexdigest()
    return jsonify({
        "orderReference": order_reference,
        "status": "accept",
        "time": resp_time,
        "signature": resp_signature
    }), 200

А add_subscription_days — це проста, але важлива функція, яка розумно обробляє подовження:

def add_subscription_days(telegram_id: int, days_to_add: int = 30):
    user = supabase.table("users").select("subscription_ends_at") \\
        .eq("telegram_id", telegram_id).execute()
    sub_end_str = user.data[0].get("subscription_ends_at")
    now = datetime.now(pytz.timezone('Europe/Kyiv'))
    if sub_end_str:
        sub_dt = datetime.fromisoformat(sub_end_str.replace('Z', '+00:00'))
        # Якщо підписка ще діє — додаємо до кінця
        # Якщо вже спливла — додаємо від сьогодні
        new_end = (sub_dt if sub_dt > now else now) + timedelta(days=days_to_add)
    else:
        new_end = now + timedelta(days=days_to_add)
    supabase.table("users").update({
        "subscription_ends_at": new_end.isoformat(),
        "is_first_month_used": True
    }).eq("telegram_id", telegram_id).execute()
    # Відправляємо радісне повідомлення в Telegram
    bot.send_message(telegram_id, "Ура! 🥳 Оплата пройшла успішно!")

Окремо реалізована cron-джоба для утримання клієнтів (Retention). Вона щодня чекає базу і, якщо бачить, що у юзера підписка закінчується завтра або щойно скінчилася — тригерить відправку нагадування:

# notify_workers.py — запускається cron щодня о 11:00
for user in users:
    sub_dt = datetime.fromisoformat(sub_end_str).astimezone(kyiv_tz)
    tomorrow_start = (now + timedelta(days=1)).replace(hour=0, minute=0, second=0)
    tomorrow_end   = tomorrow_start + timedelta(days=1)
    if tomorrow_start <= sub_dt <= tomorrow_end:
        bot.send_message(telegram_id, "⚠️ Ваша підписка спливає завтра!")
    yesterday_start = (now - timedelta(days=1)).replace(hour=0, minute=0, second=0)
    yesterday_end   = yesterday_start + timedelta(days=1)
    if yesterday_start <= sub_dt <= yesterday_end:
        bot.send_message(telegram_id, "⏳ Ваша підписка спливла!")

Плани на майбутнє

Зараз Fibby повністю покриває мої потреби та працює для перших користувачів, але технічний беклог ще великий:

1. AI Financial Advisor: Хочу прикрутити LLM (наприклад, OpenAI API), щоб бот не просто збирав дані, а аналізував патерни і міг написати: «Ти витратив на каву на 20% більше цього місяця».
2. Інтеграція з Monobank API: Для тих, хто хоче повної автоматизації без ручного вводу.
3. Batch writes: Зараз sheet_worker пише кожну транзакцію окремим insert_row. Наступний крок — групувати записи одного юзера і використовувати batch_update для мінімізації кількості запитів до Google API.

Робити продукти, які вирішують власні щоденні проблеми — це крутий досвід. Буду радий почути ваші думки щодо архітектури: як би ви організували чергу для Google API? Чи варто переходити на RabbitMQ/Celery замість поллінгу Supabase? Чи є досвід із batch-записами через Google Sheets API?

Якщо цікаво подивитись, як це виглядає наживо — ось посилання на бота: fibby.pro і @botfibby

Дякую за увагу!