Как автоматически получать выписку из Монобанки для благотворительных организаций

Сегодня «МоноБанка» — один из самых известных инструментов для благотворителей. Про МоноБанки знают все, и доверия к этому сервису больше, чем к остальным.

Рядовому пользователю завести монобанку и вести её несложно, а вот юрлицу приходится изрядно повозиться. Да и работа с банковским сервисом на стороне Универсалбанка временами превращается в настоящий ад. 

Когда мы с единомышленниками создали благотворительный фонд, я взялся продумывать автоматизацию процессов, чтобы всё было удобно. Трудностей я ждал откуда угодно, но только не от Монобанка. А когда начал в это погружаться, понял: вся конструкция — как декорации в кино.  На экране всё красиво, а за краем кадра — крашеный картон. Косяки старается сглаживать служба поддержки Моно, она много общается с пользователями. Старается честно, но не всесильна. Ведь там, где заканчивается Моно, начинается Универсалбанк, и вот тут уже тянет нафталином, а от большинства сервисов — едким трупным запахом Legacy…

Как выписка по МоноБанке работает у обычного пользователя? Открыли приложение, создали банку — и в реальном времени видите, кто донатит. А знаете, как это работает у юр. лиц через приложение? … а никак не работает :-)). Нет такого в приложении! 🙂

Выписки по монобанке приходят каждый день на…. электронную почту, ту самую, которую вы указали при регистрации. Ну и на том спасибо, что не в почтовый ящик Укрпочты 🙂 Никакого API нет, хотя для физлиц и ФОПов у Монобанка API есть, и вполне нормальный. Но через МоноAPI до монобанки фонда не достучаться, это уже территория Универсалбанка!…  

А ЧТО ТАКОЕ УНИВЕРСАЛБАНК ПО ЧАСТИ IT-СЕРВИСОВ? ВОТ ЗДЕСЬ ВКЛЮЧАЙТЕ ВООБРАЖЕНИЕ:

Представьте постапокалиптическую пустыню: свалка техники под вековой пылью, где-то воет ветер, катится перекати-поле. Слева стоит полуржавый школьный автобус, такой же, как в американских фильмах, из него торчат старые провода витой пары. Наружу выглядывает очень уставший бородатый мужик в рваной одежде, в руках у него старая клавиатура. В окнах автобуса видны обжитые рабочие места со старыми кинескопными мониторами из 90-х.

В этом автобусе живут программисты Legacy-кода, а в кадре — их техлид, которого кредитами и ипотеками для сотрудников держат здесь с 90-х. За автобусом раскинулось большое кладбище, оно уходит вдаль, и к каждой могиле от автобуса тянется витая пара. Это кладбище Legacy-кода, который эти программисты пытаются поддерживать. А на заднем плане виден яркий современный парк развлечений под названием Mono Park 🙂 

Как-то так я себе и представляю всю эту компанию, когда сталкиваюсь с их ИТ-сервисами. А вот так это нарисовал искусственный интеллект Dall-E. По-моему, метко:)

Извините, отвлёкся. Так вот, выписки по монобанке приходят вам на электронную почту, а чтобы вы совсем уж не чувствовали себя человеком, приходят они в ZIP-архиве, внутри которого лежит один-единственный файл в формате CSV. «Удобно», правда?   

По традиции я подёргал поддержку Монобанка: вдруг есть способ выгрузить выписку поудобнее и посовременнее. Способа не нашлось, так что я сел писать код, который быстро закроет задачу: времени на это особо нет.

Что мне было нужно:

1. Каждый раз, когда приходит письмо с выпиской, вытащить из него данные и добавить их в базу.
2. Собрать отчёт по конкретной банке сбора за нужный период в форматах PDF и JSON.
3. Полученный JSON отправить POST-запросом на сайт, чтобы на странице сбора показать список донатов и комментарии от неравнодушных людей, за что им огромное спасибо!

Ниже приведу сильно урезанный код, который всё это делает, только вместо записи в базу сохраняет данные в файл.

Что делает скрипт:

1. Подключается к почтовому серверу по IMAP и забирает письма с вложенными ZIP-файлами.
2. Распаковывает ZIP-архивы и достаёт из них CSV-файлы.
3. Фильтрует данные по нужному критерию (например, по полю «Референс»).
4. Отправляет данные на сервер POST-запросом в формате JSON.
5. Формирует отчёты в форматах CSV и PDF.

Нужные библиотеки Python:

pip install imaplib email zipfile pandas fpdf requests numpy

Полный скрипт выглядит так:

import imaplib
import email
import zipfile
import pandas as pd
from io import BytesIO
from datetime import datetime, timedelta
import json
import requests
from fpdf import FPDF
import numpy as np

# Email account details (replace with your actual data)
IMAP_SERVER = 'mail.yourdomain.com'
IMAP_PORT = 993
EMAIL_ACCOUNT = 'your_email@yourdomain.com'
PASSWORD = 'your_password'

# URL for the POST request to send data to the server
POST_URL = 'https://yourdomain.com/your_script.php'

# Authorization token (replace with your actual token)
AUTH_TOKEN = 'your_secure_token'

# Reference filter for data selection (if empty, all data will be selected)
REFERENCE_FILTER = 'F_966'  # You can replace this or leave it empty

# Connect to the email server using IMAP
mail = imaplib.IMAP4_SSL(IMAP_SERVER, IMAP_PORT)
mail.login(EMAIL_ACCOUNT, PASSWORD)
mail.select('inbox')

# Search for emails from a specific sender over the last month
today = datetime.today()
since_date = (today - timedelta(days=30)).strftime("%d-%b-%Y")
status, email_ids = mail.search(None, f'(FROM "SENDER_EMAIL@monobank.com" SINCE {since_date})')

# Check if any emails were found
if email_ids[0]:
    print(f"Emails found: {len(email_ids[0].split())}")
else:
    print("No emails found.")

# List to store combined data
all_data = []

# Process each email
for e_id in email_ids[0].split():
    status, email_data = mail.fetch(e_id, '(RFC822)')
    for response_part in email_data:
        if isinstance(response_part, tuple):
            msg = email.message_from_bytes(response_part[1])
            print(f"Processing email: {msg['Subject']}")

            # Process each part of the email
            for part in msg.walk():
                if part.get_content_maintype() == 'multipart':
                    continue
                if part.get('Content-Disposition') is None:
                    continue

                # Look for zip attachments
                filename = part.get_filename()
                if filename and filename.endswith('.zip'):
                    print(f"Found archive: {filename}")

                    # Process each zip file found
                    zip_data = BytesIO(part.get_payload(decode=True))
                    try:
                        with zipfile.ZipFile(zip_data, 'r') as zip_ref:
                            for file in zip_ref.namelist():
                                if file.endswith('.csv'):
                                    print(f"Extracted file: {file}")
                                    # Read each CSV file from the archive using UTF-8 encoding
                                    with zip_ref.open(file) as csv_file:
                                        data = pd.read_csv(csv_file, encoding='utf-8')

                                        # Filter data by reference if REFERENCE_FILTER is set
                                        if REFERENCE_FILTER:
                                            filtered_data = data[data['Референс благодійної банки'] == REFERENCE_FILTER]
                                        else:
                                            filtered_data = data

                                        # Convert the data to a list of dictionaries and add to combined list
                                        all_data.extend(filtered_data.to_dict(orient='records'))
                    except zipfile.BadZipFile:
                        print(f"Error opening archive: {filename}")

# Check if there is any data to save
if all_data:
    # Export to CSV
    combined_df = pd.DataFrame(all_data)  # Create DataFrame from combined data
    combined_df.to_csv('combined_data.csv', index=False, encoding='utf-8')  # Save as CSV
    print("Data successfully saved to 'combined_data.csv'.")

    # Function to clean data before sending
    def clean_record(record):
        """Replace invalid NaN or infinite values with 0."""
        for key, value in record.items():
            if isinstance(value, float):
                if np.isnan(value) or np.isinf(value):
                    record[key] = 0  # Replace NaN or infinity with 0
        return record

    # Clean all data records
    all_data = [clean_record(record) for record in all_data]

    # Send the cleaned data to the server via POST request
    headers = {
        'Authorization': f'Bearer {AUTH_TOKEN}',
        'Content-Type': 'application/json'
    }

    # Convert the data to JSON
    json_data = json.dumps(all_data, ensure_ascii=False, indent=4)

    # Send data to the server
    response = requests.post(POST_URL, data=json_data.encode('utf-8'), headers=headers)

    if response.status_code == 200:
        print("Data successfully sent to the server.")
    else:
        print(f"Error sending data: {response.status_code} - {response.text}")

    # Export data to PDF
    pdf = FPDF(orientation='L', unit='mm', format='A4')  # Landscape orientation
    pdf.add_page()

    # Add fonts (replace the path with your actual font files)
    pdf.add_font('Roboto', '', './Roboto/Roboto-Regular.ttf', uni=True)
    pdf.add_font('Roboto', 'B', './Roboto/Roboto-Bold.ttf', uni=True)

    # Set font for title
    pdf.set_font('Roboto', 'B', 14)
    pdf.cell(280, 10, txt="Виписка", ln=True, align='C')

    # Include "Reference" and "Bank Name" only once before the table
    pdf.set_font('Roboto', '', 12)
    reference_value = all_data[0].get("Референс благодійної банки", "")
    bank_name = all_data[0].get("Назва благодійної банки", "")
    pdf.ln(10)
    pdf.cell(280, 10, f"Референс благодійної банки: {reference_value}", ln=True)
    pdf.cell(280, 10, f"Назва благодійної банки: {bank_name}", ln=True)

    # Define columns to include and their widths
    columns_to_include = ['Дата платежу', 'Час платежу', 'Сума платежу', 'Метод поповнення банки', 'Коментар до платежу']
    col_widths = [40, 30, 40, 100, 70]  # Set column widths

    # Print table headers
    for i, column in enumerate(columns_to_include):
        pdf.cell(col_widths[i], 10, txt=column, border=1, align='C')
    pdf.ln()

    # Print rows with a combination of cell() and multi_cell()
    for record in all_data:
        row_height = 10  # Default row height

        # Calculate max line count for each cell
        line_counts = [
            pdf.get_string_width(str(record['Дата платежу'])) // col_widths[0] + 1,
            pdf.get_string_width(str(record['Час платежу'])) // col_widths[1] + 1,
            pdf.get_string_width(str(record['Сума платежу'])) // col_widths[2] + 1,
            pdf.get_string_width(str(record['Метод поповнення банки'])) // col_widths[3] + 1,
            pdf.get_string_width(str(record['Коментар до платежу'])) // col_widths[4] + 1
        ]
        max_lines = max(line_counts)  # Calculate the maximum line count in the record

        # Adjust row height based on the maximum line count
        row_height = 10 * max_lines

        # Print regular columns using cell()
        pdf.cell(col_widths[0], row_height, txt=str(record['Дата платежу']), border=1)
        pdf.cell(col_widths[1], row_height, txt=str(record['Час платежу']), border=1)
        pdf.cell(col_widths[2], row_height, txt=str(record['Сума платежу']), border=1)

        # Use multi_cell for long text in "Метод поповнення банки" and "Коментар до платежу"
        x, y = pdf.get_x(), pdf.get_y()  # Save current coordinates
        pdf.multi_cell(col_widths[3], row_height / max_lines, txt=str(record['Метод поповнення банки']), border=1)
        # Move to the next column
        pdf.set_xy(x + col_widths[3], y)

        # Update coordinates for "Коментар до платежу"
        x, y = pdf.get_x(), pdf.get_y()
        pdf.multi_cell(col_widths[4], row_height / max_lines, txt=str(record['Коментар до платежу']), border=1)
        pdf.set_xy(x + col_widths[4], y)

        # Move to the next row
        pdf.ln()

    # Save the PDF
    pdf.output("combined_data.pdf")
    print("Data successfully saved to 'combined_data.pdf'.")
else:
    print("No data found.")

# Close the email session
mail.logout()
Code language: PHP (php)

Теперь по порядку.
Данные для подключения к почтовому серверу: сюда подставьте свой почтовый сервер, учётную запись и пароль.

IMAP_SERVER = 'mail.yourdomain.com'
EMAIL_ACCOUNT = 'your_email@yourdomain.com'
PASSWORD = 'your_password'
Code language: JavaScript (javascript)

URL для POST-запроса: здесь указываю адрес сервера, куда скрипт отправит json с данными.

POST_URL = 'https://yourdomain.com/your_script.php'
Code language: JavaScript (javascript)

Токен авторизации: чтобы скрипт на сайте не задалбывали всякие боты, я передаю в запросе токен.  

AUTH_TOKEN = 'your_secure_token'
Code language: JavaScript (javascript)

А здесь уникальный ID банки. Если сборов много, их нужно как-то разделять. У меня сделано так: указан идентификатор — фильтруем, поле пустое — показываем всё.

# Reference filter for data selection (if empty, all data will be selected)
REFERENCE_FILTER = 'F_966'  # You can replace this or leave it empty
Code language: PHP (php)

Здесь задаю адрес, с которого Моно присылает выписку, и период, за который разбирать письма. Для примера беру 1 месяц. 

# Search for emails from a specific sender over the last month
today = datetime.today()
since_date = (today - timedelta(days=30)).strftime("%d-%b-%Y")
status, email_ids = mail.search(None, f'(FROM "SENDER_EMAIL@monobank.com" SINCE {since_date})')
Code language: PHP (php)

Вроде бы и всё. Результат вывожу в PDF, но с форматированием документа особо не заморачивался. Для моих задач хватило.

Скрипт висит у меня в CRON и запускается каждый день в 11:00

0 11 * * * /usr/bin/python3 /path_to_your_script/your_script.py
Code language: JavaScript (javascript)

У меня всё!
Спасибо за внимание, надеюсь, было полезно!

Michael Dvornichenko Архитектор решений · SK.AI