Интеграция API ProverkaCheka в Python — пошаговое руководство

API ProverkaCheka позволяет автоматизировать проверку чеков Kaspi прямо из вашего Python-приложения. Отправьте PDF-файл чека, получите структурированный JSON-ответ с результатом проверки — подлинность, сумма, получатель, наличие дубликатов. В этом руководстве мы разберём интеграцию от первого запроса до полноценного production-решения.

Предварительные требования

• Python 3.8 или выше
• Библиотека requests (установка: pip install requests)
• API-ключ ProverkaCheka (получите через @ProverkaChekakzbot)
• Тестовый PDF-чек Kaspi

Шаг 1: Установка зависимостей

Создайте виртуальное окружение и установите библиотеку requests:

python -m venv venv
source venv/bin/activate  # Linux/macOS
# или venv\Scripts\activate  # Windows

pip install requests

Шаг 2: Первый запрос — базовая проверка чека

Минимальный пример отправки PDF-чека на проверку:

import requests

API_URL = "https://api.proverkacheka.kz/upload"
API_KEY = "your_api_key_here"

def check_receipt(file_path):
    """Проверить чек Kaspi по PDF-файлу."""
    with open(file_path, "rb") as f:
        files = {"file": (file_path, f, "application/pdf")}
        data = {"api_key": API_KEY}
        response = requests.post(API_URL, files=files, data=data)

    return response.json()

# Использование
result = check_receipt("receipt.pdf")
print(result)

Этот код открывает PDF-файл, отправляет его на API как multipart/form-data и получает JSON-ответ. Разберём, что содержит ответ.

Шаг 3: Структура ответа API

API возвращает JSON-объект со следующими полями:

Шаг 4: Проверка с ИИН/БИН

Для дополнительной верификации передайте параметр iin — ИИН или БИН получателя платежа. API проверит, что деньги были отправлены именно этому получателю.

def check_receipt_with_iin(file_path, iin):
    """Проверить чек с валидацией ИИН/БИН получателя."""
    with open(file_path, "rb") as f:
        files = {"file": (file_path, f, "application/pdf")}
        data = {
            "api_key": API_KEY,
            "iin": iin
        }
        response = requests.post(API_URL, files=files, data=data)

    return response.json()

# Пример: проверяем, что оплата пришла на ИИН 123456789012
result = check_receipt_with_iin("receipt.pdf", "123456789012")

if result.get("status") == "valid":
    print(f"Чек подлинный. Сумма: {result['amount']} тг")
else:
    print(f"Проверка не пройдена: {result.get('message')}")

Шаг 5: Обработка ошибок

В production-коде важно обработать все возможные сценарии ошибок. Вот расширенная функция:

import requests
from requests.exceptions import (
    ConnectionError, Timeout, RequestException
)

API_URL = "https://api.proverkacheka.kz/upload"
API_KEY = "your_api_key_here"
TIMEOUT = 30  # секунд

class ReceiptCheckError(Exception):
    """Исключение при ошибке проверки чека."""
    pass

def check_receipt_safe(file_path, iin=None):
    """
    Проверить чек Kaspi с обработкой ошибок.

    Returns:
        dict: Результат проверки
    Raises:
        ReceiptCheckError: При ошибке API или сети
        FileNotFoundError: Если файл не найден
    """
    # Проверяем, что файл существует и это PDF
    if not file_path.lower().endswith(".pdf"):
        raise ReceiptCheckError("Файл должен быть в формате PDF")

    try:
        with open(file_path, "rb") as f:
            files = {"file": (file_path, f, "application/pdf")}
            data = {"api_key": API_KEY}
            if iin:
                data["iin"] = iin

            response = requests.post(
                API_URL,
                files=files,
                data=data,
                timeout=TIMEOUT
            )

    except FileNotFoundError:
        raise FileNotFoundError(f"Файл не найден: {file_path}")
    except ConnectionError:
        raise ReceiptCheckError(
            "Нет соединения с API. Проверьте интернет."
        )
    except Timeout:
        raise ReceiptCheckError(
            "Превышено время ожидания ответа от API."
        )
    except RequestException as e:
        raise ReceiptCheckError(f"Ошибка запроса: {e}")

    # Проверяем HTTP-статус
    if response.status_code != 200:
        raise ReceiptCheckError(
            f"API вернул статус {response.status_code}"
        )

    result = response.json()

    # Проверяем, нужен ли повторный запрос
    if result.get("retry_after"):
        raise ReceiptCheckError(
            f"Kaspi временно недоступен. "
            f"Повторите через {result['retry_after']} сек."
        )

    return result

Шаг 6: Парсинг и интерпретация результатов

Создадим удобную обёртку, которая интерпретирует ответ API и возвращает понятный результат:

from dataclasses import dataclass
from typing import Optional

@dataclass
class ReceiptResult:
    """Результат проверки чека."""
    is_valid: bool
    is_duplicate: bool
    check_number: str
    amount: float
    seller_name: str
    message: str
    raw_response: dict

def parse_receipt_result(api_response: dict) -> ReceiptResult:
    """Парсит ответ API в структурированный объект."""
    status = api_response.get("status", "error")

    return ReceiptResult(
        is_valid=(status == "valid"),
        is_duplicate=(status == "duplicate"),
        check_number=api_response.get("check_number", ""),
        amount=api_response.get("amount", 0.0),
        seller_name=api_response.get("ip_name", ""),
        message=api_response.get("message", "Неизвестная ошибка"),
        raw_response=api_response,
    )

# Пример использования
result = check_receipt_safe("receipt.pdf", iin="123456789012")
parsed = parse_receipt_result(result)

if parsed.is_valid:
    print(f"Чек #{parsed.check_number} подлинный")
    print(f"Сумма: {parsed.amount} тг")
    print(f"Продавец: {parsed.seller_name}")
elif parsed.is_duplicate:
    print(f"Дубликат! Чек #{parsed.check_number} уже использован")
else:
    print(f"Ошибка: {parsed.message}")

Шаг 7: Пакетная обработка чеков

Если нужно проверить несколько чеков за раз (например, при массовом наборе на курс), используйте последовательную обработку с задержкой:

import os
import time
import csv

def batch_check_receipts(folder_path, iin=None, output_csv="results.csv"):
    """
    Проверить все PDF-чеки в папке и сохранить результаты в CSV.

    Args:
        folder_path: Путь к папке с PDF-файлами
        iin: ИИН/БИН для проверки (опционально)
        output_csv: Путь к файлу результатов
    """
    pdf_files = [
        f for f in os.listdir(folder_path)
        if f.lower().endswith(".pdf")
    ]
    print(f"Найдено {len(pdf_files)} PDF-файлов")

    results = []
    for i, filename in enumerate(pdf_files, 1):
        file_path = os.path.join(folder_path, filename)
        print(f"[{i}/{len(pdf_files)}] Проверяю {filename}...")

        try:
            result = check_receipt_safe(file_path, iin)
            parsed = parse_receipt_result(result)
            results.append({
                "file": filename,
                "status": "valid" if parsed.is_valid else "invalid",
                "check_number": parsed.check_number,
                "amount": parsed.amount,
                "seller": parsed.seller_name,
                "message": parsed.message,
            })
        except ReceiptCheckError as e:
            results.append({
                "file": filename,
                "status": "error",
                "check_number": "",
                "amount": 0,
                "seller": "",
                "message": str(e),
            })

        # Пауза между запросами
        if i < len(pdf_files):
            time.sleep(1)

    # Сохраняем результаты в CSV
    with open(output_csv, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=results[0].keys())
        writer.writeheader()
        writer.writerows(results)

    # Статистика
    valid = sum(1 for r in results if r["status"] == "valid")
    invalid = sum(1 for r in results if r["status"] == "invalid")
    errors = sum(1 for r in results if r["status"] == "error")
    print(f"\nГотово! Валидных: {valid}, Невалидных: {invalid}, Ошибок: {errors}")
    print(f"Результаты сохранены в {output_csv}")

# Использование
batch_check_receipts("./receipts/", iin="123456789012")

Шаг 8: Интеграция с Flask

Пример endpoint для веб-приложения на Flask, который принимает PDF от пользователя и проверяет через API:

from flask import Flask, request, jsonify
import tempfile
import os

app = Flask(__name__)

ALLOWED_EXTENSIONS = {"pdf"}
MAX_FILE_SIZE = 5 * 1024 * 1024  # 5 МБ

def allowed_file(filename):
    return (
        "." in filename
        and filename.rsplit(".", 1)[1].lower() in ALLOWED_EXTENSIONS
    )

@app.route("/verify-payment", methods=["POST"])
def verify_payment():
    """Endpoint для проверки чека от пользователя."""
    if "file" not in request.files:
        return jsonify({"error": "Файл не загружен"}), 400

    file = request.files["file"]
    if not file.filename or not allowed_file(file.filename):
        return jsonify({"error": "Нужен PDF-файл"}), 400

    # Проверяем размер файла
    file.seek(0, os.SEEK_END)
    size = file.tell()
    file.seek(0)
    if size > MAX_FILE_SIZE:
        return jsonify({"error": "Файл слишком большой (макс. 5 МБ)"}), 400

    # Сохраняем временно и проверяем
    with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tmp:
        file.save(tmp.name)
        try:
            result = check_receipt_safe(tmp.name, iin="123456789012")
            parsed = parse_receipt_result(result)
            return jsonify({
                "valid": parsed.is_valid,
                "duplicate": parsed.is_duplicate,
                "amount": parsed.amount,
                "check_number": parsed.check_number,
                "message": parsed.message,
            })
        except ReceiptCheckError as e:
            return jsonify({"error": str(e)}), 502
        finally:
            os.unlink(tmp.name)  # Удаляем временный файл

Шаг 9: Повторные попытки при недоступности Kaspi

Иногда Kaspi API может быть временно недоступен. Ответ API будет содержать поле retry_after с рекомендуемым интервалом в секундах. Реализуем автоматический повтор:

import time

def check_receipt_with_retry(file_path, iin=None, max_retries=3):
    """Проверить чек с автоматическим повтором при ошибках."""
    for attempt in range(1, max_retries + 1):
        try:
            with open(file_path, "rb") as f:
                files = {"file": (file_path, f, "application/pdf")}
                data = {"api_key": API_KEY}
                if iin:
                    data["iin"] = iin
                response = requests.post(
                    API_URL, files=files, data=data, timeout=TIMEOUT
                )

            result = response.json()

            # Если Kaspi недоступен — ждём и повторяем
            retry_after = result.get("retry_after")
            if retry_after and attempt < max_retries:
                print(
                    f"Kaspi недоступен, повтор через "
                    f"{retry_after} сек (попытка {attempt}/{max_retries})"
                )
                time.sleep(retry_after)
                continue

            return result

        except (ConnectionError, Timeout):
            if attempt < max_retries:
                wait = 2 ** attempt  # Экспоненциальная задержка
                print(f"Ошибка сети, повтор через {wait} сек...")
                time.sleep(wait)
            else:
                raise ReceiptCheckError(
                    f"Не удалось подключиться после {max_retries} попыток"
                )

Шаг 10: Полный рабочий скрипт

Вот готовый скрипт, который можно использовать из командной строки:

#!/usr/bin/env python3
"""
Скрипт проверки чеков Kaspi через API ProverkaCheka.
Использование: python verify.py receipt.pdf [--iin 123456789012]
"""
import sys
import argparse
import requests

API_URL = "https://api.proverkacheka.kz/upload"
# API-ключ из переменной окружения
import os
API_KEY = os.environ.get("PROVERKACHEKA_API_KEY", "")

def main():
    parser = argparse.ArgumentParser(
        description="Проверка чеков Kaspi через ProverkaCheka API"
    )
    parser.add_argument("file", help="Путь к PDF-чеку")
    parser.add_argument("--iin", help="ИИН/БИН получателя для проверки")
    args = parser.parse_args()

    if not API_KEY:
        print("Ошибка: установите PROVERKACHEKA_API_KEY")
        print("export PROVERKACHEKA_API_KEY=your_key_here")
        sys.exit(1)

    if not os.path.exists(args.file):
        print(f"Файл не найден: {args.file}")
        sys.exit(1)

    print(f"Проверяю {args.file}...")

    with open(args.file, "rb") as f:
        files = {"file": (args.file, f, "application/pdf")}
        data = {"api_key": API_KEY}
        if args.iin:
            data["iin"] = args.iin

        try:
            resp = requests.post(API_URL, files=files, data=data, timeout=30)
        except requests.RequestException as e:
            print(f"Ошибка сети: {e}")
            sys.exit(1)

    result = resp.json()
    status = result.get("status", "error")

    if status == "valid":
        print(f"  Статус:  ПОДЛИННЫЙ")
        print(f"  Чек:     {result.get('check_number', 'N/A')}")
        print(f"  Сумма:   {result.get('amount', 'N/A')} тг")
        print(f"  Продавец:{result.get('ip_name', 'N/A')}")
    elif status == "duplicate":
        print(f"  Статус:  ДУБЛИКАТ")
        print(f"  Чек:     {result.get('check_number', 'N/A')}")
        print(f"  Сообщение: {result.get('message', '')}")
    else:
        print(f"  Статус:  ОШИБКА")
        print(f"  Сообщение: {result.get('message', 'Неизвестная ошибка')}")

if __name__ == "__main__":
    main()

Рекомендации по безопасности

• Валидируйте входные файлы. Проверяйте расширение (.pdf), размер (не более 5 МБ) и MIME-тип перед отправкой на API.
• Удаляйте временные файлы. Если сохраняете загруженный PDF во временную директорию, обязательно удалите его после проверки.
• Логируйте результаты. Сохраняйте результат каждой проверки в базу данных для аудита и разрешения споров.
• Ограничивайте доступ. Endpoint проверки должен требовать аутентификации пользователя — не позволяйте анонимам проверять чеки за ваш счёт.

Типичные ошибки при интеграции

Что дальше

После базовой интеграции вы можете расширить функциональность:

• Webhook-уведомления. Настройте получение результатов проверки через webhook вместо polling.
• Асинхронная обработка. Используйте aiohttp вместо requests для неблокирующих запросов в async-приложениях.
• Интеграция с Django/FastAPI. Те же принципы работают с любым Python-фреймворком — меняется только обёртка.
• Мониторинг. Отслеживайте количество проверок, процент отклонённых чеков и время ответа API.

Полная документация API доступна по адресу proverkacheka.kz/docs.