"""Invoice operations for RoboKassa API."""
import json
from decimal import Decimal
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union, cast
if TYPE_CHECKING:
from aiorobokassa.api._protocols import ClientProtocol
from aiorobokassa.constants import DEFAULT_SIGNATURE_ALGORITHM, INVOICE_API_BASE_URL
from aiorobokassa.enums import InvoiceType, SignatureAlgorithm
from aiorobokassa.exceptions import APIError
from aiorobokassa.models.receipt import Receipt
from aiorobokassa.models.requests import InvoiceItem, InvoiceResponse
from aiorobokassa.utils.jwt import create_jwt_token
[docs]
class InvoiceMixin:
"""Mixin for invoice operations."""
[docs]
async def create_invoice(
self,
out_sum: Union[Decimal, float, int, str],
description: str,
invoice_type: Union[InvoiceType, str] = InvoiceType.ONE_TIME,
inv_id: Optional[int] = None,
culture: Optional[str] = None,
merchant_comments: Optional[str] = None,
invoice_items: Optional[List[InvoiceItem]] = None,
receipt: Optional[Union[Receipt, str, Dict[str, Any]]] = None,
user_fields: Optional[Dict[str, str]] = None,
success_url: Optional[str] = None,
success_url_method: str = "GET",
fail_url: Optional[str] = None,
fail_url_method: str = "GET",
signature_algorithm: Union[str, SignatureAlgorithm] = DEFAULT_SIGNATURE_ALGORITHM,
) -> InvoiceResponse:
"""
Create invoice via Invoice API (JWT-based).
Args:
out_sum: Payment amount
description: Payment description
invoice_type: Invoice type (OneTime or Reusable)
inv_id: Invoice ID (optional)
culture: Language code (ru, en) (optional)
merchant_comments: Internal comment for staff (optional)
invoice_items: List of invoice items for fiscalization (optional)
receipt: Receipt data for fiscalization - Receipt model, JSON string or dict (optional).
If provided and invoice_items is not, receipt items will be converted to invoice_items.
user_fields: Additional user parameters (optional)
success_url: Success redirect URL (optional)
success_url_method: HTTP method for success URL (GET or POST)
fail_url: Fail redirect URL (optional)
fail_url_method: HTTP method for fail URL (GET or POST)
signature_algorithm: Signature algorithm (optional, default: MD5)
Returns:
InvoiceResponse with invoice information (id, url, inv_id, encoded_id)
Raises:
APIError: If invoice creation fails
ValueError: If both invoice_items and receipt are provided
"""
if TYPE_CHECKING:
client = cast("ClientProtocol", self)
else:
client = self # type: ignore[assignment]
if isinstance(invoice_type, InvoiceType):
invoice_type_str = invoice_type.value
else:
invoice_type_str = str(invoice_type)
payload: Dict[str, Any] = {
"MerchantLogin": client.merchant_login,
"InvoiceType": invoice_type_str,
"OutSum": float(Decimal(str(out_sum))),
"Description": description,
}
if inv_id is not None:
payload["InvId"] = inv_id
if culture:
payload["Culture"] = culture
if merchant_comments:
payload["MerchantComments"] = merchant_comments
if user_fields:
payload["UserFields"] = user_fields
if invoice_items and receipt:
raise ValueError("Cannot provide both invoice_items and receipt. Use only one.")
if receipt and not invoice_items:
if isinstance(receipt, Receipt):
receipt_model = receipt
elif isinstance(receipt, dict):
receipt_model = Receipt.from_dict(receipt)
elif isinstance(receipt, str):
receipt_dict = json.loads(receipt)
receipt_model = Receipt.from_dict(receipt_dict)
else:
raise ValueError("receipt must be Receipt model, JSON string or dict")
invoice_items = []
for receipt_item in receipt_model.items:
cost_value: Union[float, Decimal] = 0.0
if receipt_item.cost is not None:
cost_value = float(receipt_item.cost)
elif receipt_item.sum is not None:
cost_value = float(receipt_item.sum / Decimal(str(receipt_item.quantity)))
invoice_item = InvoiceItem(
name=receipt_item.name,
quantity=receipt_item.quantity,
cost=cost_value,
tax=receipt_item.tax,
payment_method=receipt_item.payment_method,
payment_object=receipt_item.payment_object,
nomenclature_code=receipt_item.nomenclature_code,
)
invoice_items.append(invoice_item)
if receipt_model.sno:
payload["Sno"] = receipt_model.sno.value
if invoice_items:
payload["InvoiceItems"] = [item.to_api_dict() for item in invoice_items]
if success_url:
payload["SuccessUrl2Data"] = {"Url": success_url, "Method": success_url_method}
if fail_url:
payload["FailUrl2Data"] = {"Url": fail_url, "Method": fail_url_method}
secret_key = f"{client.merchant_login}:{client.password1}"
jwt_token = create_jwt_token(payload, secret_key, signature_algorithm)
response = await client._post(
f"{INVOICE_API_BASE_URL}/CreateInvoice",
json=jwt_token,
)
async with response:
result = await response.json()
if not result.get("isSuccess", False):
error_message = result.get("errorMessage", "Failed to create invoice")
raise APIError(f"Invoice creation failed: {error_message}")
return InvoiceResponse.from_api_response(result)
[docs]
async def deactivate_invoice(
self,
inv_id: Optional[int] = None,
invoice_id: Optional[str] = None,
encoded_id: Optional[str] = None,
signature_algorithm: Union[str, SignatureAlgorithm] = DEFAULT_SIGNATURE_ALGORITHM,
) -> None:
"""
Deactivate invoice.
Args:
inv_id: Invoice ID (number specified by merchant)
invoice_id: Invoice identifier (returned from create_invoice)
encoded_id: Encoded invoice ID (last part of invoice URL)
signature_algorithm: Signature algorithm (optional, default: MD5)
Raises:
APIError: If deactivation fails
ValueError: If no identifier is provided
"""
if TYPE_CHECKING:
client = cast("ClientProtocol", self)
else:
client = self # type: ignore[assignment]
if not any([inv_id, invoice_id, encoded_id]):
raise ValueError(
"At least one identifier (inv_id, invoice_id, or encoded_id) must be provided"
)
payload: Dict[str, Any] = {"MerchantLogin": client.merchant_login}
if inv_id is not None:
payload["InvId"] = inv_id
if invoice_id:
payload["Id"] = invoice_id
if encoded_id:
payload["EncodedId"] = encoded_id
secret_key = f"{client.merchant_login}:{client.password1}"
jwt_token = create_jwt_token(payload, secret_key, signature_algorithm)
response = await client._post(
f"{INVOICE_API_BASE_URL}/DeactivateInvoice",
json=jwt_token,
)
async with response:
result = await response.json()
if not result.get("isSuccess", False):
error_message = result.get("errorMessage", "Failed to deactivate invoice")
raise APIError(f"Invoice deactivation failed: {error_message}")