Coupon API — пример на Python

Пример для Python получает исходы Sport Line API, создаёт ординар, экспресс или несколько ординаров и запрашивает актуальные состояния купонов.

Требования и настройки

Используйте Python 3.11 или новее. Установите библиотеку requests:

python3 -m pip install requests

Сохраните код в coupon_example.py и задайте переменные окружения:

export COUPON_API_BASE_URL='https://YOUR_COUPON_API_DOMAIN'
export COUPON_LOGIN='YOUR_LOGIN'
export COUPON_PASSWORD='YOUR_PASSWORD'
export SPORT_LINE_BASE_URL='https://YOUR_SPORT_LINE_DOMAIN'
export SPORT_LINE_PACKAGE='YOUR_SPORT_LINE_API_KEY'

Дополнительные настройки:

Полный исполняемый пример

from __future__ import annotations

import json
import os
import sys
from typing import Any
from urllib.parse import quote

import requests


def required_env(name: str) -> str:
    value = os.environ.get(name, '').strip()
    if not value:
        raise RuntimeError(f'{name} is not configured')
    return value


COUPON_BASE_URL = required_env('COUPON_API_BASE_URL').rstrip('/')
COUPON_LOGIN = required_env('COUPON_LOGIN')
COUPON_PASSWORD = required_env('COUPON_PASSWORD')
SPORT_LINE_BASE_URL = required_env('SPORT_LINE_BASE_URL').rstrip('/')
SPORT_LINE_PACKAGE = required_env('SPORT_LINE_PACKAGE')

EXAMPLE_MODE = os.environ.get('COUPON_EXAMPLE_MODE', 'single')
AMOUNT = float(os.environ.get('COUPON_AMOUNT', '10'))
CURRENCY = os.environ.get('COUPON_CURRENCY', 'USD')
LINE_TYPE = os.environ.get('SPORT_LINE_TYPE', 'live')
LANGUAGE = os.environ.get('DOCUMENT_LANGUAGE', 'en')

if EXAMPLE_MODE not in {'single', 'express', 'multi'}:
    raise RuntimeError('COUPON_EXAMPLE_MODE must be single, express, or multi')
if LINE_TYPE not in {'live', 'line'}:
    raise RuntimeError('SPORT_LINE_TYPE must be live or line')
if AMOUNT <= 0:
    raise RuntimeError('COUPON_AMOUNT must be a positive number')

SESSION = requests.Session()
SESSION.headers.update({'Accept': 'application/json'})


def request_json(
    url: str,
    *,
    method: str = 'GET',
    headers: dict[str, str] | None = None,
    body: dict[str, Any] | None = None,
) -> dict[str, Any]:
    response = SESSION.request(
        method,
        url,
        headers=headers,
        json=body,
        timeout=(5, 15),
    )
    try:
        payload = response.json()
    except requests.exceptions.JSONDecodeError as error:
        raise RuntimeError(f'HTTP {response.status_code}: invalid JSON') from error

    if not response.ok:
        message = payload.get('error_message', 'request failed')
        raise RuntimeError(f'HTTP {response.status_code}: {message}')
    return payload


def require_coupon_success(payload: dict[str, Any], operation: str) -> Any:
    if payload.get('code') != 1:
        code = payload.get('error_code', 'unknown')
        message = payload.get('error_message', 'error')
        raise RuntimeError(f'{operation}: {code} {message}')
    return payload['body']


def login() -> str:
    payload = request_json(
        f'{COUPON_BASE_URL}/api/partner/login',
        method='POST',
        body={'username': COUPON_LOGIN, 'password': COUPON_PASSWORD},
    )
    return require_coupon_success(payload, 'Login')['token']


def sport_line_get(path: str) -> dict[str, Any]:
    return request_json(
        f'{SPORT_LINE_BASE_URL}{path}',
        headers={'Package': SPORT_LINE_PACKAGE},
    )


def find_games(limit: int) -> list[dict[str, Any]]:
    menu = sport_line_get(f'/v1/menu/{LINE_TYPE}/{LANGUAGE}')
    games: dict[str, dict[str, Any]] = {}

    for sport in menu.get('body', []):
        for country in sport.get('sub', []):
            for tournament in country.get('sub', []):
                path = (
                    f"/v1/events/{sport['id']}/{tournament['id']}"
                    f'/sub/50/{LINE_TYPE}/{LANGUAGE}'
                )
                result = sport_line_get(path)
                for group in result.get('body', []):
                    for game in group.get('events_list', []):
                        games[str(game['game_id'])] = game
                        if len(games) >= limit:
                            return list(games.values())

    raise RuntimeError(f'Only {len(games)} suitable events found; {limit} required')


def find_outcome(value: Any) -> dict[str, Any] | None:
    if isinstance(value, list):
        for item in value:
            outcome = find_outcome(item)
            if outcome is not None:
                return outcome
        return None

    if not isinstance(value, dict):
        return None

    rate = value.get('oc_rate')
    block = value.get('oc_block', False)
    if (
        isinstance(value.get('oc_pointer'), str)
        and isinstance(rate, (int, float))
        and not isinstance(rate, bool)
        and rate > 1
        and block in (False, 0, None)
        and value.get('op_id') is None
    ):
        return value

    for child in value.values():
        outcome = find_outcome(child)
        if outcome is not None:
            return outcome
    return None


def get_selection(game: dict[str, Any]) -> dict[str, Any]:
    event = sport_line_get(
        f"/v1/event/{game['game_id']}/group/{LINE_TYPE}/{LANGUAGE}",
    )
    outcome = find_outcome(event.get('body'))
    if outcome is None:
        raise RuntimeError(f"No available outcome for game {game['game_id']}")

    technical_pointer = outcome['oc_pointer'].replace('|', '#')
    player_part = f"#{outcome['op_id']}" if outcome.get('op_id') is not None else ''
    return {
        'pointer': (
            f"{LINE_TYPE}#{technical_pointer}#{outcome['oc_rate']}{player_part}"
        ),
        'game_id': game['game_id'],
        'event': f"{game['opp_1_name']} — {game['opp_2_name']}",
        'market': outcome.get('oc_group_name'),
        'outcome': outcome.get('oc_name'),
        'coefficient': outcome['oc_rate'],
    }


def coupon_request(
    token: str,
    path: str,
    *,
    method: str = 'GET',
    body: dict[str, Any] | None = None,
) -> Any:
    payload = request_json(
        f'{COUPON_BASE_URL}{path}',
        method=method,
        headers={'Authorization': f'Bearer {token}'},
        body=body,
    )
    return require_coupon_success(payload, path)


COUPON_STATUSES = {
    0: 'NEW', 2: 'WIN', 4: 'LOSE', 8: 'RETURN', 15: 'UPDATE',
}
BET_STATUSES = {
    0: 'NET', 1: 'WIN', 2: 'LOSE', 3: 'RETURN', 4: 'RECALCULATE',
    21: 'HALF_WIN', 22: 'HALF_LOSE', 23: 'PUSH',
}


def summarize_coupon(coupon: dict[str, Any]) -> dict[str, Any]:
    return {
        'coupon_code': coupon['coupon_code'],
        'status': coupon['status'],
        'status_name': COUPON_STATUSES.get(coupon['status'], 'UNKNOWN'),
        'amount': coupon['amount'],
        'coefficient': coupon['coef'],
        'real_win': coupon.get('real_win'),
        'events': [
            {
                'game_id': event['game_id'],
                'bet': event['bet_name'],
                'status': event['status'],
                'status_name': BET_STATUSES.get(event['status'], 'UNKNOWN'),
            }
            for event in coupon.get('events_data', [])
        ],
    }


def print_json(label: str, value: Any) -> None:
    print(label)
    print(json.dumps(value, ensure_ascii=False, indent=2))


def main() -> None:
    required_games = 1 if EXAMPLE_MODE == 'single' else 2
    games = find_games(required_games)
    selections = [get_selection(game) for game in games]

    print_json(
        'Selected outcomes:',
        [
            {key: value for key, value in selection.items() if key != 'pointer'}
            for selection in selections
        ],
    )

    token = login()
    placement = coupon_request(
        token,
        '/api/partner/coupons/place',
        method='POST',
        body={
            'list_bets': [selection['pointer'] for selection in selections],
            'amount': AMOUNT,
            'currency': CURRENCY,
            'callback_url': None,
            'lang': LANGUAGE,
            'mode': 'reject',
            'mode_type': None,
            'multi': EXAMPLE_MODE == 'multi',
        },
    )

    created = placement['coupons']
    print_json('Created coupons:', [summarize_coupon(item) for item in created])
    coupon_codes = [item['coupon_code'] for item in created]

    current = [
        coupon_request(
            token,
            f'/api/partner/coupons/get?coupon_code={quote(code)}',
        )
        for code in coupon_codes
    ]
    print_json('Current state:', [summarize_coupon(item) for item in current])

    active = coupon_request(token, '/api/partner/coupons/active')
    print_json('Active coupons:', [summarize_coupon(item) for item in active])

    calculated = coupon_request(
        token,
        '/api/partner/coupons/calculated?time=120',
    )
    print_json('Recently settled:', [summarize_coupon(item) for item in calculated])

    batch = coupon_request(
        token,
        '/api/partner/coupons/results',
        method='POST',
        body={'coupon_ids': coupon_codes},
    )
    print_json(
        'Batch result:',
        [summarize_coupon(item) for item in batch['coupons']],
    )


if __name__ == '__main__':
    try:
        main()
    except (RuntimeError, requests.RequestException) as error:
        print(str(error), file=sys.stderr)
        raise SystemExit(1) from error

Запуск

COUPON_EXAMPLE_MODE=single python3 coupon_example.py
COUPON_EXAMPLE_MODE=express python3 coupon_example.py
COUPON_EXAMPLE_MODE=multi python3 coupon_example.py

Каждый запуск создаёт реальные купоны. При multi значение COUPON_AMOUNT применяется к каждому отдельному ординару. После сетевого timeout сначала проверьте наличие купона, а затем решайте, можно ли повторять запрос.

См. также PHP-пример, JavaScript-пример и «Статусы и расчёт выплаты».