mirror of
https://github.com/BEDOLAGA-DEV/remnawave-bedolaga-telegram-bot.git
synced 2026-01-19 19:32:10 +00:00
refactor(nalogo): восстановить описание чеков из настроек и использовать локальную библиотеку
- Добавлено восстановление описания чека из настроек при обработке очереди - Передача telegram_user_id и amount_kopeks через всю цепочку создания чеков - Переход на локальную исправленную версию библ
This commit is contained in:
gy9vin
1
app/lib/__init__.py
Normal file
1
app/lib/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
# Local libraries
|
||||
37
app/lib/nalogo/__init__.py
Normal file
37
app/lib/nalogo/__init__.py
Normal file
@@ -0,0 +1,37 @@
|
||||
"""
|
||||
Asynchronous Python client for Russian self-employed tax service (Moy Nalog).
|
||||
|
||||
This is a Python port of the PHP library shoman4eg/moy-nalog,
|
||||
providing async HTTP client for interaction with lknpd.nalog.ru API.
|
||||
|
||||
Original PHP library: https://github.com/shoman4eg/moy-nalog
|
||||
Author: Artem Dubinin <artem@dubinin.me>
|
||||
License: MIT
|
||||
"""
|
||||
|
||||
from .client import Client
|
||||
from .exceptions import (
|
||||
ClientException,
|
||||
DomainException,
|
||||
ForbiddenException,
|
||||
NotFoundException,
|
||||
PhoneException,
|
||||
ServerException,
|
||||
UnauthorizedException,
|
||||
UnknownErrorException,
|
||||
ValidationException,
|
||||
)
|
||||
|
||||
__version__ = "1.0.0"
|
||||
__all__ = [
|
||||
"Client",
|
||||
"ClientException",
|
||||
"DomainException",
|
||||
"ForbiddenException",
|
||||
"NotFoundException",
|
||||
"PhoneException",
|
||||
"ServerException",
|
||||
"UnauthorizedException",
|
||||
"UnknownErrorException",
|
||||
"ValidationException",
|
||||
]
|
||||
188
app/lib/nalogo/_http.py
Normal file
188
app/lib/nalogo/_http.py
Normal file
@@ -0,0 +1,188 @@
|
||||
"""
|
||||
Internal HTTP client and authentication middleware.
|
||||
Based on PHP library's AuthenticationPlugin and HTTP architecture.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
from abc import ABC, abstractmethod
|
||||
from http import HTTPStatus
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
|
||||
from .exceptions import raise_for_status
|
||||
|
||||
|
||||
class AuthProvider(ABC):
|
||||
"""Abstract interface for authentication provider."""
|
||||
|
||||
@abstractmethod
|
||||
async def get_token(self) -> dict[str, Any] | None:
|
||||
"""Get current access token data."""
|
||||
|
||||
@abstractmethod
|
||||
async def refresh(self, refresh_token: str) -> dict[str, Any] | None:
|
||||
"""Refresh access token using refresh token."""
|
||||
|
||||
|
||||
class AsyncHTTPClient:
|
||||
"""
|
||||
Async HTTP client with automatic token refresh on 401 responses.
|
||||
|
||||
Based on PHP's AuthenticationPlugin behavior:
|
||||
- Adds Bearer authorization header
|
||||
- On 401 response, attempts token refresh once
|
||||
- Retries request with new token (max 2 attempts)
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
base_url: str,
|
||||
auth_provider: AuthProvider,
|
||||
default_headers: dict[str, str] | None = None,
|
||||
timeout: float = 10.0,
|
||||
):
|
||||
self.base_url = base_url
|
||||
self.auth_provider = auth_provider
|
||||
self.default_headers = default_headers or {}
|
||||
self.timeout = timeout
|
||||
self._refresh_lock = asyncio.Lock()
|
||||
self.max_retries = 2 # Same as PHP AuthenticationPlugin::RETRY_LIMIT
|
||||
|
||||
async def _get_auth_headers(self) -> dict[str, str]:
|
||||
"""Get authorization headers from current token."""
|
||||
token_data = await self.auth_provider.get_token()
|
||||
if not token_data or "token" not in token_data:
|
||||
return {}
|
||||
|
||||
return {"Authorization": f"Bearer {token_data['token']}"}
|
||||
|
||||
async def _handle_401_response(
|
||||
self, client: httpx.AsyncClient, request: httpx.Request
|
||||
) -> httpx.Response | None:
|
||||
"""
|
||||
Handle 401 response by refreshing token and retrying request.
|
||||
|
||||
Uses asyncio.Lock to prevent concurrent refresh attempts,
|
||||
similar to PHP's retry storage mechanism.
|
||||
"""
|
||||
async with self._refresh_lock:
|
||||
token_data = await self.auth_provider.get_token()
|
||||
if not token_data or "refreshToken" not in token_data:
|
||||
return None
|
||||
|
||||
# Attempt token refresh
|
||||
new_token_data = await self.auth_provider.refresh(
|
||||
token_data["refreshToken"]
|
||||
)
|
||||
if not new_token_data or "token" not in new_token_data:
|
||||
return None
|
||||
|
||||
# Update request with new authorization header
|
||||
new_auth_headers = {"Authorization": f"Bearer {new_token_data['token']}"}
|
||||
request.headers.update(new_auth_headers)
|
||||
|
||||
# Retry request with new token
|
||||
return await client.send(request)
|
||||
|
||||
async def request(
|
||||
self,
|
||||
method: str,
|
||||
path: str,
|
||||
headers: dict[str, str] | None = None,
|
||||
json_data: dict[str, Any] | None = None,
|
||||
**kwargs: Any,
|
||||
) -> httpx.Response:
|
||||
"""
|
||||
Make HTTP request with automatic auth and 401 retry logic.
|
||||
|
||||
Args:
|
||||
method: HTTP method (GET, POST, etc.)
|
||||
path: API path (e.g., "/income")
|
||||
headers: Additional headers
|
||||
json_data: JSON request body
|
||||
**kwargs: Additional httpx.AsyncClient.request arguments
|
||||
|
||||
Returns:
|
||||
httpx.Response object
|
||||
|
||||
Raises:
|
||||
Domain exceptions via raise_for_status()
|
||||
"""
|
||||
# Prepare headers
|
||||
request_headers = self.default_headers.copy()
|
||||
auth_headers = await self._get_auth_headers()
|
||||
request_headers.update(auth_headers)
|
||||
if headers:
|
||||
request_headers.update(headers)
|
||||
|
||||
# Prepare request parameters
|
||||
request_kwargs = {
|
||||
"method": method,
|
||||
"url": self.base_url + path,
|
||||
"headers": request_headers,
|
||||
"timeout": self.timeout,
|
||||
**kwargs,
|
||||
}
|
||||
|
||||
if json_data is not None:
|
||||
request_kwargs["json"] = json_data
|
||||
|
||||
async with httpx.AsyncClient() as client:
|
||||
# Initial request
|
||||
response = await client.request(**request_kwargs)
|
||||
|
||||
# Handle 401 with token refresh (max 1 retry)
|
||||
if response.status_code == HTTPStatus.UNAUTHORIZED:
|
||||
# Build request object for retry
|
||||
request = client.build_request(**request_kwargs)
|
||||
retry_response = await self._handle_401_response(client, request)
|
||||
if retry_response is not None:
|
||||
response = retry_response
|
||||
|
||||
# Check for domain exceptions
|
||||
raise_for_status(response)
|
||||
|
||||
return response
|
||||
|
||||
async def get(
|
||||
self,
|
||||
path: str,
|
||||
headers: dict[str, str] | None = None,
|
||||
**kwargs: Any,
|
||||
) -> httpx.Response:
|
||||
"""GET request."""
|
||||
return await self.request("GET", path, headers=headers, **kwargs)
|
||||
|
||||
async def post(
|
||||
self,
|
||||
path: str,
|
||||
json_data: dict[str, Any] | None = None,
|
||||
headers: dict[str, str] | None = None,
|
||||
**kwargs: Any,
|
||||
) -> httpx.Response:
|
||||
"""POST request with JSON data."""
|
||||
return await self.request(
|
||||
"POST", path, headers=headers, json_data=json_data, **kwargs
|
||||
)
|
||||
|
||||
async def put(
|
||||
self,
|
||||
path: str,
|
||||
json_data: dict[str, Any] | None = None,
|
||||
headers: dict[str, str] | None = None,
|
||||
**kwargs: Any,
|
||||
) -> httpx.Response:
|
||||
"""PUT request with JSON data."""
|
||||
return await self.request(
|
||||
"PUT", path, headers=headers, json_data=json_data, **kwargs
|
||||
)
|
||||
|
||||
async def delete(
|
||||
self,
|
||||
path: str,
|
||||
headers: dict[str, str] | None = None,
|
||||
**kwargs: Any,
|
||||
) -> httpx.Response:
|
||||
"""DELETE request."""
|
||||
return await self.request("DELETE", path, headers=headers, **kwargs)
|
||||
257
app/lib/nalogo/auth.py
Normal file
257
app/lib/nalogo/auth.py
Normal file
@@ -0,0 +1,257 @@
|
||||
"""
|
||||
Authentication provider implementation.
|
||||
Based on PHP library's Authenticator class.
|
||||
"""
|
||||
|
||||
import json
|
||||
import uuid
|
||||
from http import HTTPStatus
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
|
||||
from ._http import AuthProvider
|
||||
from .dto.device import DeviceInfo
|
||||
from .exceptions import raise_for_status
|
||||
|
||||
|
||||
def generate_device_id() -> str:
|
||||
"""Generate device ID similar to PHP's DeviceIdGenerator."""
|
||||
return str(uuid.uuid4()).replace("-", "")[:21].lower()
|
||||
|
||||
|
||||
# DeviceInfo is now imported from dto.device
|
||||
|
||||
|
||||
class AuthProviderImpl(AuthProvider):
|
||||
"""
|
||||
Authentication provider implementation.
|
||||
|
||||
Provides methods for:
|
||||
- Username/password authentication (INN + password)
|
||||
- Phone-based authentication (2-step: challenge + verify)
|
||||
- Token refresh
|
||||
- Token storage (in-memory or file-based)
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
base_url: str = "https://lknpd.nalog.ru/api",
|
||||
storage_path: str | None = None,
|
||||
device_id: str | None = None,
|
||||
):
|
||||
self.base_url_v1 = f"{base_url}/v1"
|
||||
self.base_url_v2 = f"{base_url}/v2"
|
||||
self.storage_path = storage_path
|
||||
self.device_id = device_id or generate_device_id()
|
||||
self.device_info = DeviceInfo(sourceDeviceId=self.device_id)
|
||||
self._token_data: dict[str, Any] | None = None
|
||||
|
||||
# Default headers similar to PHP Authenticator
|
||||
self.default_headers = {
|
||||
"Referrer": "https://lknpd.nalog.ru/auth/login",
|
||||
"Content-Type": "application/json",
|
||||
"Accept": "application/json, text/plain, */*",
|
||||
"Accept-Language": "ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7",
|
||||
}
|
||||
|
||||
# Load token from storage if available
|
||||
if self.storage_path:
|
||||
self._load_token_from_storage()
|
||||
|
||||
def _load_token_from_storage(self) -> None:
|
||||
"""Load token from file storage."""
|
||||
if not self.storage_path:
|
||||
return
|
||||
storage_path = Path(self.storage_path)
|
||||
if not storage_path.exists():
|
||||
return
|
||||
|
||||
try:
|
||||
with storage_path.open(encoding="utf-8") as f:
|
||||
self._token_data = json.load(f)
|
||||
except (json.JSONDecodeError, OSError):
|
||||
# Ignore errors, token will be None
|
||||
pass
|
||||
|
||||
def _save_token_to_storage(self) -> None:
|
||||
"""Save token to file storage."""
|
||||
if not self.storage_path or not self._token_data:
|
||||
return
|
||||
|
||||
storage_path = Path(self.storage_path)
|
||||
try:
|
||||
# Ensure directory exists
|
||||
storage_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
with storage_path.open("w", encoding="utf-8") as f:
|
||||
json.dump(self._token_data, f, ensure_ascii=False, indent=2)
|
||||
except OSError:
|
||||
# Ignore storage errors
|
||||
pass
|
||||
|
||||
async def get_token(self) -> dict[str, Any] | None:
|
||||
"""Get current access token data."""
|
||||
return self._token_data
|
||||
|
||||
async def set_token(self, token_json: str) -> None:
|
||||
"""
|
||||
Set access token from JSON string.
|
||||
|
||||
Args:
|
||||
token_json: JSON string containing token data
|
||||
"""
|
||||
try:
|
||||
self._token_data = json.loads(token_json)
|
||||
self._save_token_to_storage()
|
||||
except json.JSONDecodeError as e:
|
||||
raise ValueError(f"Invalid token JSON: {e}") from e
|
||||
|
||||
async def create_new_access_token(self, username: str, password: str) -> str:
|
||||
"""
|
||||
Create new access token using INN and password.
|
||||
|
||||
Mirrors PHP Authenticator::createAccessToken().
|
||||
|
||||
Args:
|
||||
username: INN (tax identification number)
|
||||
password: Password
|
||||
|
||||
Returns:
|
||||
JSON string with token data
|
||||
|
||||
Raises:
|
||||
Domain exceptions for authentication errors
|
||||
"""
|
||||
request_data = {
|
||||
"username": username,
|
||||
"password": password,
|
||||
"deviceInfo": self.device_info.model_dump(),
|
||||
}
|
||||
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.post(
|
||||
f"{self.base_url_v1}/auth/lkfl",
|
||||
json=request_data,
|
||||
headers=self.default_headers,
|
||||
timeout=10.0,
|
||||
)
|
||||
|
||||
raise_for_status(response)
|
||||
|
||||
# Store and return token
|
||||
token_json = response.text
|
||||
await self.set_token(token_json)
|
||||
return token_json
|
||||
|
||||
async def create_phone_challenge(self, phone: str) -> dict[str, Any]:
|
||||
"""
|
||||
Start phone-based authentication challenge.
|
||||
|
||||
Mirrors PHP ApiClient::createPhoneChallenge() - uses v2 API.
|
||||
|
||||
Args:
|
||||
phone: Phone number (e.g., "79000000000")
|
||||
|
||||
Returns:
|
||||
Dictionary with challengeToken, expireDate, expireIn
|
||||
|
||||
Raises:
|
||||
Domain exceptions for API errors
|
||||
"""
|
||||
request_data = {
|
||||
"phone": phone,
|
||||
"requireTpToBeActive": True,
|
||||
}
|
||||
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.post(
|
||||
f"{self.base_url_v2}/auth/challenge/sms/start",
|
||||
json=request_data,
|
||||
headers=self.default_headers,
|
||||
timeout=10.0,
|
||||
)
|
||||
|
||||
raise_for_status(response)
|
||||
return response.json() # type: ignore[no-any-return]
|
||||
|
||||
async def create_new_access_token_by_phone(
|
||||
self, phone: str, challenge_token: str, verification_code: str
|
||||
) -> str:
|
||||
"""
|
||||
Complete phone-based authentication with SMS code.
|
||||
|
||||
Mirrors PHP Authenticator::createAccessTokenByPhone().
|
||||
|
||||
Args:
|
||||
phone: Phone number
|
||||
challenge_token: Token from create_phone_challenge()
|
||||
verification_code: SMS verification code
|
||||
|
||||
Returns:
|
||||
JSON string with token data
|
||||
|
||||
Raises:
|
||||
Domain exceptions for authentication errors
|
||||
"""
|
||||
request_data = {
|
||||
"phone": phone,
|
||||
"code": verification_code,
|
||||
"challengeToken": challenge_token,
|
||||
"deviceInfo": self.device_info.model_dump(),
|
||||
}
|
||||
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.post(
|
||||
f"{self.base_url_v1}/auth/challenge/sms/verify",
|
||||
json=request_data,
|
||||
headers=self.default_headers,
|
||||
timeout=10.0,
|
||||
)
|
||||
|
||||
raise_for_status(response)
|
||||
|
||||
# Store and return token
|
||||
token_json = response.text
|
||||
await self.set_token(token_json)
|
||||
return token_json
|
||||
|
||||
async def refresh(self, refresh_token: str) -> dict[str, Any] | None:
|
||||
"""
|
||||
Refresh access token using refresh token.
|
||||
|
||||
Mirrors PHP Authenticator::refreshAccessToken().
|
||||
|
||||
Args:
|
||||
refresh_token: Refresh token string
|
||||
|
||||
Returns:
|
||||
New token data dictionary or None if refresh failed
|
||||
"""
|
||||
request_data = {
|
||||
"deviceInfo": self.device_info.model_dump(),
|
||||
"refreshToken": refresh_token,
|
||||
}
|
||||
|
||||
try:
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.post(
|
||||
f"{self.base_url_v1}/auth/token",
|
||||
json=request_data,
|
||||
headers=self.default_headers,
|
||||
timeout=10.0,
|
||||
)
|
||||
|
||||
# PHP version only checks for 200 status
|
||||
if response.status_code != HTTPStatus.OK:
|
||||
return None
|
||||
|
||||
# Store and return new token data
|
||||
token_json = response.text
|
||||
await self.set_token(token_json)
|
||||
return self._token_data
|
||||
|
||||
except Exception:
|
||||
# Silently fail refresh attempts like PHP version
|
||||
return None
|
||||
235
app/lib/nalogo/client.py
Normal file
235
app/lib/nalogo/client.py
Normal file
@@ -0,0 +1,235 @@
|
||||
"""
|
||||
Main client facade for Moy Nalog API.
|
||||
Based on PHP library's ApiClient class.
|
||||
"""
|
||||
|
||||
import json
|
||||
from typing import Any
|
||||
|
||||
from ._http import AsyncHTTPClient
|
||||
from .auth import AuthProviderImpl
|
||||
from .income import IncomeAPI
|
||||
from .payment_type import PaymentTypeAPI
|
||||
from .receipt import ReceiptAPI
|
||||
from .tax import TaxAPI
|
||||
from .user import UserAPI
|
||||
|
||||
|
||||
class Client:
|
||||
"""
|
||||
Main async client for Moy Nalog API.
|
||||
|
||||
Provides factory methods for API modules and authentication methods.
|
||||
Maps to PHP ApiClient functionality with async support.
|
||||
|
||||
Example:
|
||||
>>> client = Client()
|
||||
>>> token = await client.create_new_access_token("inn", "password")
|
||||
>>> await client.authenticate(token)
|
||||
>>> income_api = client.income()
|
||||
>>> result = await income_api.create("Service", 100, 1)
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
base_url: str = "https://lknpd.nalog.ru/api",
|
||||
storage_path: str | None = None,
|
||||
device_id: str | None = None,
|
||||
timeout: float = 10.0,
|
||||
):
|
||||
"""
|
||||
Initialize Moy Nalog API client.
|
||||
|
||||
Args:
|
||||
base_url: API base URL (default: https://lknpd.nalog.ru/api)
|
||||
storage_path: Optional file path for token storage
|
||||
device_id: Optional device ID (auto-generated if not provided)
|
||||
timeout: HTTP request timeout in seconds
|
||||
"""
|
||||
self.base_url = base_url
|
||||
self.timeout = timeout
|
||||
|
||||
# Initialize auth provider
|
||||
self.auth_provider = AuthProviderImpl(
|
||||
base_url=base_url,
|
||||
storage_path=storage_path,
|
||||
device_id=device_id,
|
||||
)
|
||||
|
||||
# Initialize HTTP client with auth middleware
|
||||
self.http_client = AsyncHTTPClient(
|
||||
base_url=f"{base_url}/v1",
|
||||
auth_provider=self.auth_provider,
|
||||
default_headers={
|
||||
"Content-Type": "application/json",
|
||||
"Accept": "application/json, text/plain, */*",
|
||||
"Accept-Language": "ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7",
|
||||
"Referrer": "https://lknpd.nalog.ru/auth/login",
|
||||
},
|
||||
timeout=timeout,
|
||||
)
|
||||
|
||||
# User profile data (for receipt operations)
|
||||
self._user_profile: dict[str, Any] | None = None
|
||||
|
||||
async def create_new_access_token(self, username: str, password: str) -> str:
|
||||
"""
|
||||
Create new access token using INN and password.
|
||||
|
||||
Maps to PHP ApiClient::createNewAccessToken().
|
||||
|
||||
Args:
|
||||
username: INN (tax identification number)
|
||||
password: Password
|
||||
|
||||
Returns:
|
||||
JSON string with access token data
|
||||
|
||||
Raises:
|
||||
UnauthorizedException: For invalid credentials
|
||||
DomainException: For other API errors
|
||||
"""
|
||||
return await self.auth_provider.create_new_access_token(username, password)
|
||||
|
||||
async def create_phone_challenge(self, phone: str) -> dict[str, Any]:
|
||||
"""
|
||||
Start phone-based authentication challenge.
|
||||
|
||||
Maps to PHP ApiClient::createPhoneChallenge().
|
||||
|
||||
Args:
|
||||
phone: Phone number (e.g., "79000000000")
|
||||
|
||||
Returns:
|
||||
Dictionary with challengeToken, expireDate, expireIn
|
||||
|
||||
Raises:
|
||||
DomainException: For API errors
|
||||
"""
|
||||
return await self.auth_provider.create_phone_challenge(phone)
|
||||
|
||||
async def create_new_access_token_by_phone(
|
||||
self, phone: str, challenge_token: str, verification_code: str
|
||||
) -> str:
|
||||
"""
|
||||
Complete phone-based authentication with SMS code.
|
||||
|
||||
Maps to PHP ApiClient::createNewAccessTokenByPhone().
|
||||
|
||||
Args:
|
||||
phone: Phone number
|
||||
challenge_token: Token from create_phone_challenge()
|
||||
verification_code: SMS verification code
|
||||
|
||||
Returns:
|
||||
JSON string with access token data
|
||||
|
||||
Raises:
|
||||
UnauthorizedException: For invalid verification
|
||||
DomainException: For other API errors
|
||||
"""
|
||||
return await self.auth_provider.create_new_access_token_by_phone(
|
||||
phone, challenge_token, verification_code
|
||||
)
|
||||
|
||||
async def authenticate(self, access_token: str) -> None:
|
||||
"""
|
||||
Authenticate client with access token.
|
||||
|
||||
Maps to PHP ApiClient::authenticate().
|
||||
|
||||
Args:
|
||||
access_token: JSON string with token data
|
||||
|
||||
Raises:
|
||||
ValueError: For invalid token JSON
|
||||
"""
|
||||
await self.auth_provider.set_token(access_token)
|
||||
|
||||
# Parse token to extract user profile (like PHP version)
|
||||
try:
|
||||
token_data = json.loads(access_token)
|
||||
if "profile" in token_data:
|
||||
self._user_profile = token_data["profile"]
|
||||
except json.JSONDecodeError:
|
||||
# If token parsing fails, profile will remain None
|
||||
pass
|
||||
|
||||
async def get_access_token(self) -> str | None:
|
||||
"""
|
||||
Get current access token (may be refreshed).
|
||||
|
||||
Maps to PHP ApiClient::getAccessToken().
|
||||
|
||||
Returns:
|
||||
Current access token JSON string or None
|
||||
"""
|
||||
token_data = await self.auth_provider.get_token()
|
||||
if token_data:
|
||||
return json.dumps(token_data)
|
||||
return None
|
||||
|
||||
def income(self) -> IncomeAPI:
|
||||
"""
|
||||
Get Income API instance.
|
||||
|
||||
Maps to PHP ApiClient::income().
|
||||
|
||||
Returns:
|
||||
IncomeAPI instance for creating/cancelling receipts
|
||||
"""
|
||||
return IncomeAPI(self.http_client)
|
||||
|
||||
def receipt(self) -> ReceiptAPI:
|
||||
"""
|
||||
Get Receipt API instance.
|
||||
|
||||
Maps to PHP ApiClient::receipt().
|
||||
|
||||
Returns:
|
||||
ReceiptAPI instance for accessing receipt data
|
||||
|
||||
Raises:
|
||||
ValueError: If user is not authenticated (no profile data)
|
||||
"""
|
||||
if not self._user_profile or "inn" not in self._user_profile:
|
||||
raise ValueError("User profile not available. Please authenticate first.")
|
||||
|
||||
return ReceiptAPI(
|
||||
http_client=self.http_client,
|
||||
base_endpoint=self.base_url,
|
||||
user_inn=self._user_profile["inn"],
|
||||
)
|
||||
|
||||
def payment_type(self) -> PaymentTypeAPI:
|
||||
"""
|
||||
Get PaymentType API instance.
|
||||
|
||||
Maps to PHP ApiClient::paymentType().
|
||||
|
||||
Returns:
|
||||
PaymentTypeAPI instance for managing payment methods
|
||||
"""
|
||||
return PaymentTypeAPI(self.http_client)
|
||||
|
||||
def tax(self) -> TaxAPI:
|
||||
"""
|
||||
Get Tax API instance.
|
||||
|
||||
Maps to PHP ApiClient::tax().
|
||||
|
||||
Returns:
|
||||
TaxAPI instance for tax information and history
|
||||
"""
|
||||
return TaxAPI(self.http_client)
|
||||
|
||||
def user(self) -> UserAPI:
|
||||
"""
|
||||
Get User API instance.
|
||||
|
||||
Maps to PHP ApiClient::user().
|
||||
|
||||
Returns:
|
||||
UserAPI instance for user information
|
||||
"""
|
||||
return UserAPI(self.http_client)
|
||||
46
app/lib/nalogo/dto/__init__.py
Normal file
46
app/lib/nalogo/dto/__init__.py
Normal file
@@ -0,0 +1,46 @@
|
||||
"""Data Transfer Objects for Moy Nalog API."""
|
||||
|
||||
from .device import DeviceInfo
|
||||
from .income import (
|
||||
AtomDateTime,
|
||||
CancelCommentType,
|
||||
CancelRequest,
|
||||
IncomeClient,
|
||||
IncomeRequest,
|
||||
IncomeServiceItem,
|
||||
IncomeType,
|
||||
PaymentType,
|
||||
)
|
||||
from .invoice import InvoiceClient, InvoiceServiceItem
|
||||
from .payment_type import PaymentType as PaymentTypeModel
|
||||
from .payment_type import PaymentTypeCollection
|
||||
from .tax import History, HistoryRecords, Payment, PaymentRecords, Tax
|
||||
from .user import UserType
|
||||
|
||||
__all__ = [
|
||||
"AtomDateTime",
|
||||
"CancelCommentType",
|
||||
"CancelRequest",
|
||||
# Device DTOs
|
||||
"DeviceInfo",
|
||||
"History",
|
||||
"HistoryRecords",
|
||||
"IncomeClient",
|
||||
"IncomeRequest",
|
||||
"IncomeServiceItem",
|
||||
# Income DTOs
|
||||
"IncomeType",
|
||||
"InvoiceClient",
|
||||
# Invoice DTOs
|
||||
"InvoiceServiceItem",
|
||||
"Payment",
|
||||
"PaymentRecords",
|
||||
"PaymentType",
|
||||
"PaymentTypeCollection",
|
||||
# Payment Type DTOs
|
||||
"PaymentTypeModel",
|
||||
# Tax DTOs
|
||||
"Tax",
|
||||
# User DTOs
|
||||
"UserType",
|
||||
]
|
||||
44
app/lib/nalogo/dto/device.py
Normal file
44
app/lib/nalogo/dto/device.py
Normal file
@@ -0,0 +1,44 @@
|
||||
"""
|
||||
Device-related DTO models.
|
||||
Based on PHP library's DTO\\DeviceInfo class.
|
||||
"""
|
||||
|
||||
from typing import Any, ClassVar
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
class DeviceInfo(BaseModel):
|
||||
"""
|
||||
Device information model for API requests.
|
||||
Maps to PHP DTO\\DeviceInfo.
|
||||
"""
|
||||
|
||||
# Constants from PHP class
|
||||
SOURCE_TYPE_WEB: ClassVar[str] = "WEB"
|
||||
APP_VERSION: ClassVar[str] = "1.0.0"
|
||||
USER_AGENT: ClassVar[str] = (
|
||||
"Mozilla/5.0 (Macintosh; Intel Mac OS X 11_2_2) AppleWebKit/537.36 "
|
||||
"(KHTML, like Gecko) Chrome/88.0.4324.192 Safari/537.36"
|
||||
)
|
||||
|
||||
source_type: str = Field(
|
||||
default=SOURCE_TYPE_WEB,
|
||||
alias="sourceType",
|
||||
description="Source type (usually WEB)",
|
||||
)
|
||||
source_device_id: str = Field(..., alias="sourceDeviceId", description="Device ID")
|
||||
app_version: str = Field(
|
||||
default=APP_VERSION, alias="appVersion", description="Application version"
|
||||
)
|
||||
user_agent: str = Field(default=USER_AGENT, description="User agent string")
|
||||
|
||||
def model_dump(self, **kwargs: Any) -> dict[str, Any]:
|
||||
"""Custom serialization to match PHP jsonSerialize format."""
|
||||
_ = kwargs
|
||||
return {
|
||||
"sourceType": self.source_type,
|
||||
"sourceDeviceId": self.source_device_id,
|
||||
"appVersion": self.app_version,
|
||||
"metaDetails": {"userAgent": self.user_agent},
|
||||
}
|
||||
255
app/lib/nalogo/dto/income.py
Normal file
255
app/lib/nalogo/dto/income.py
Normal file
@@ -0,0 +1,255 @@
|
||||
"""
|
||||
Income-related DTO models.
|
||||
Based on PHP library's DTO and Enum classes.
|
||||
|
||||
MODIFIED: Исправлена работа с временем для nalog.ru
|
||||
"""
|
||||
|
||||
from datetime import UTC, datetime, timezone, timedelta
|
||||
from decimal import Decimal
|
||||
from enum import Enum
|
||||
from typing import Any
|
||||
|
||||
from pydantic import BaseModel, Field, field_serializer, field_validator
|
||||
|
||||
|
||||
class IncomeType(str, Enum):
|
||||
"""Income type enumeration. Maps to PHP Enum\\IncomeType."""
|
||||
|
||||
FROM_INDIVIDUAL = "FROM_INDIVIDUAL"
|
||||
FROM_LEGAL_ENTITY = "FROM_LEGAL_ENTITY"
|
||||
FROM_FOREIGN_AGENCY = "FROM_FOREIGN_AGENCY"
|
||||
|
||||
|
||||
class PaymentType(str, Enum):
|
||||
"""Payment type enumeration. Maps to PHP Enum\\PaymentType."""
|
||||
|
||||
CASH = "CASH"
|
||||
ACCOUNT = "ACCOUNT"
|
||||
|
||||
|
||||
class CancelCommentType(str, Enum):
|
||||
"""Cancel comment type enumeration. Maps to PHP Enum\\CancelCommentType."""
|
||||
|
||||
CANCEL = "Чек сформирован ошибочно"
|
||||
REFUND = "Возврат средств"
|
||||
|
||||
|
||||
# Московская таймзона UTC+3
|
||||
MOSCOW_TZ = timezone(timedelta(hours=3))
|
||||
|
||||
|
||||
class AtomDateTime(BaseModel):
|
||||
"""
|
||||
DateTime wrapper for ISO/ATOM serialization.
|
||||
Maps to PHP DTO\\DateTime behavior.
|
||||
|
||||
Используем московское время с явным offset +03:00
|
||||
"""
|
||||
|
||||
value: datetime = Field(default_factory=lambda: datetime.now(MOSCOW_TZ))
|
||||
|
||||
@field_serializer("value")
|
||||
def serialize_datetime(self, dt: datetime) -> str:
|
||||
"""Serialize datetime with timezone offset (NOT Z suffix)."""
|
||||
# Убеждаемся что есть московская таймзона
|
||||
if dt.tzinfo is None:
|
||||
dt = dt.replace(tzinfo=MOSCOW_TZ)
|
||||
elif dt.tzinfo == UTC:
|
||||
dt = dt.astimezone(MOSCOW_TZ)
|
||||
|
||||
# Возвращаем с offset +03:00, НЕ конвертируем в UTC
|
||||
return dt.isoformat()
|
||||
|
||||
@classmethod
|
||||
def now(cls) -> "AtomDateTime":
|
||||
"""Create AtomDateTime with current Moscow time."""
|
||||
return cls(value=datetime.now(MOSCOW_TZ))
|
||||
|
||||
@classmethod
|
||||
def from_datetime(cls, dt: datetime) -> "AtomDateTime":
|
||||
"""Create AtomDateTime from datetime object."""
|
||||
return cls(value=dt)
|
||||
|
||||
|
||||
class IncomeServiceItem(BaseModel):
|
||||
"""
|
||||
Service item for income creation.
|
||||
Maps to PHP DTO\\IncomeServiceItem.
|
||||
"""
|
||||
|
||||
name: str = Field(..., description="Service name/description")
|
||||
amount: Decimal = Field(..., description="Service amount", gt=0)
|
||||
quantity: Decimal = Field(..., description="Service quantity", gt=0)
|
||||
|
||||
@field_validator("name")
|
||||
@classmethod
|
||||
def validate_name(cls, v: str) -> str:
|
||||
"""Validate name is not empty."""
|
||||
if not v.strip():
|
||||
raise ValueError("Name cannot be empty")
|
||||
return v.strip()
|
||||
|
||||
@field_validator("amount", "quantity")
|
||||
@classmethod
|
||||
def validate_positive_decimal(cls, v: Decimal) -> Decimal:
|
||||
"""Validate decimal values are positive."""
|
||||
if v <= 0:
|
||||
raise ValueError("Amount and quantity must be greater than 0")
|
||||
return v
|
||||
|
||||
@field_serializer("amount", "quantity")
|
||||
def serialize_decimal(self, value: Decimal) -> str:
|
||||
"""Serialize Decimal as string (like PHP BigDecimal)."""
|
||||
return str(value)
|
||||
|
||||
def get_total_amount(self) -> Decimal:
|
||||
"""Calculate total amount (amount * quantity)."""
|
||||
return self.amount * self.quantity
|
||||
|
||||
def model_dump(self, **kwargs: Any) -> dict[str, Any]:
|
||||
"""Custom serialization to match PHP jsonSerialize format."""
|
||||
_ = kwargs
|
||||
return {
|
||||
"name": self.name,
|
||||
"amount": str(self.amount),
|
||||
"quantity": str(self.quantity),
|
||||
}
|
||||
|
||||
|
||||
class IncomeClient(BaseModel):
|
||||
"""
|
||||
Client information for income creation.
|
||||
Maps to PHP DTO\\IncomeClient.
|
||||
"""
|
||||
|
||||
contact_phone: str | None = Field(default=None, description="Client contact phone")
|
||||
display_name: str | None = Field(default=None, description="Client display name")
|
||||
income_type: IncomeType = Field(
|
||||
default=IncomeType.FROM_INDIVIDUAL, description="Income type"
|
||||
)
|
||||
inn: str | None = Field(default=None, description="Client INN (tax ID)")
|
||||
|
||||
@field_validator("inn")
|
||||
@classmethod
|
||||
def validate_inn(cls, v: str | None, info: Any) -> str | None:
|
||||
"""Validate INN format for legal entities."""
|
||||
_ = info
|
||||
if v is None:
|
||||
return v
|
||||
|
||||
# Remove any whitespace
|
||||
v = v.strip()
|
||||
if not v:
|
||||
return None
|
||||
|
||||
# Check if it's numeric
|
||||
if not v.isdigit():
|
||||
raise ValueError("INN must contain only numbers")
|
||||
|
||||
# Check length (10 for legal entities, 12 for individuals)
|
||||
if len(v) not in [10, 12]:
|
||||
raise ValueError("INN length must be 10 or 12 digits")
|
||||
|
||||
return v
|
||||
|
||||
@field_validator("display_name")
|
||||
@classmethod
|
||||
def validate_display_name_for_legal_entity(
|
||||
cls, v: str | None, info: Any
|
||||
) -> str | None:
|
||||
"""Validate display name is provided for legal entities."""
|
||||
_ = info
|
||||
# Note: This validation is applied in the API layer in PHP,
|
||||
# but we can do basic validation here
|
||||
if v is not None:
|
||||
v = v.strip()
|
||||
if not v:
|
||||
return None
|
||||
return v
|
||||
|
||||
def model_dump(self, **kwargs: Any) -> dict[str, Any]:
|
||||
"""Custom serialization to match PHP jsonSerialize format."""
|
||||
_ = kwargs
|
||||
return {
|
||||
"contactPhone": self.contact_phone,
|
||||
"displayName": self.display_name,
|
||||
"incomeType": self.income_type.value,
|
||||
"inn": self.inn,
|
||||
}
|
||||
|
||||
|
||||
class IncomeRequest(BaseModel):
|
||||
"""
|
||||
Complete income creation request.
|
||||
Maps to PHP request structure in Income::createMultipleItems().
|
||||
"""
|
||||
|
||||
operation_time: AtomDateTime = Field(default_factory=AtomDateTime.now)
|
||||
request_time: AtomDateTime = Field(default_factory=AtomDateTime.now)
|
||||
services: list[IncomeServiceItem] = Field(..., min_length=1)
|
||||
total_amount: str = Field(..., description="Total amount as string")
|
||||
client: IncomeClient = Field(default_factory=IncomeClient)
|
||||
payment_type: PaymentType = Field(default=PaymentType.CASH)
|
||||
ignore_max_total_income_restriction: bool = Field(default=False)
|
||||
|
||||
@field_validator("services")
|
||||
@classmethod
|
||||
def validate_services(cls, v: list[IncomeServiceItem]) -> list[IncomeServiceItem]:
|
||||
"""Validate services list is not empty."""
|
||||
if not v:
|
||||
raise ValueError("Services cannot be empty")
|
||||
return v
|
||||
|
||||
def model_dump(self, **kwargs: Any) -> dict[str, Any]:
|
||||
"""Custom serialization to match PHP request format."""
|
||||
_ = kwargs
|
||||
return {
|
||||
"operationTime": self.operation_time.serialize_datetime(
|
||||
self.operation_time.value
|
||||
),
|
||||
"requestTime": self.request_time.serialize_datetime(
|
||||
self.request_time.value
|
||||
),
|
||||
"services": [service.model_dump() for service in self.services],
|
||||
"totalAmount": self.total_amount,
|
||||
"client": self.client.model_dump(),
|
||||
"paymentType": self.payment_type.value,
|
||||
"ignoreMaxTotalIncomeRestriction": self.ignore_max_total_income_restriction,
|
||||
}
|
||||
|
||||
|
||||
class CancelRequest(BaseModel):
|
||||
"""
|
||||
Income cancellation request.
|
||||
Maps to PHP request structure in Income::cancel().
|
||||
"""
|
||||
|
||||
operation_time: AtomDateTime = Field(default_factory=AtomDateTime.now)
|
||||
request_time: AtomDateTime = Field(default_factory=AtomDateTime.now)
|
||||
comment: CancelCommentType = Field(..., description="Cancellation reason")
|
||||
receipt_uuid: str = Field(..., description="Receipt UUID to cancel")
|
||||
partner_code: str | None = Field(default=None, description="Partner code")
|
||||
|
||||
@field_validator("receipt_uuid")
|
||||
@classmethod
|
||||
def validate_receipt_uuid(cls, v: str) -> str:
|
||||
"""Validate receipt UUID is not empty."""
|
||||
if not v.strip():
|
||||
raise ValueError("Receipt UUID cannot be empty")
|
||||
return v.strip()
|
||||
|
||||
def model_dump(self, **kwargs: Any) -> dict[str, Any]:
|
||||
"""Custom serialization to match PHP request format."""
|
||||
_ = kwargs
|
||||
return {
|
||||
"operationTime": self.operation_time.serialize_datetime(
|
||||
self.operation_time.value
|
||||
),
|
||||
"requestTime": self.request_time.serialize_datetime(
|
||||
self.request_time.value
|
||||
),
|
||||
"comment": self.comment.value,
|
||||
"receiptUuid": self.receipt_uuid,
|
||||
"partnerCode": self.partner_code,
|
||||
}
|
||||
69
app/lib/nalogo/dto/invoice.py
Normal file
69
app/lib/nalogo/dto/invoice.py
Normal file
@@ -0,0 +1,69 @@
|
||||
"""
|
||||
Invoice-related DTO models.
|
||||
Based on PHP library's DTO classes for invoices.
|
||||
|
||||
Note: Invoice functionality is marked as "Not implemented" in the PHP library,
|
||||
but DTO structures are provided for future implementation.
|
||||
"""
|
||||
|
||||
from decimal import Decimal
|
||||
from typing import Any
|
||||
|
||||
from pydantic import BaseModel, Field, field_serializer
|
||||
|
||||
|
||||
class InvoiceServiceItem(BaseModel):
|
||||
"""
|
||||
Invoice service item model.
|
||||
Maps to PHP DTO\\InvoiceServiceItem.
|
||||
|
||||
Similar to IncomeServiceItem but for invoices.
|
||||
"""
|
||||
|
||||
name: str = Field(..., description="Service name/description")
|
||||
amount: Decimal = Field(..., description="Service amount", gt=0)
|
||||
quantity: Decimal = Field(..., description="Service quantity", gt=0)
|
||||
|
||||
@field_serializer("amount", "quantity")
|
||||
def serialize_decimal(self, value: Decimal) -> str:
|
||||
"""Serialize Decimal as string."""
|
||||
return str(value)
|
||||
|
||||
def get_total_amount(self) -> Decimal:
|
||||
"""Calculate total amount (amount * quantity)."""
|
||||
return self.amount * self.quantity
|
||||
|
||||
def model_dump(self, **kwargs: Any) -> dict[str, Any]:
|
||||
"""Custom serialization to match API format."""
|
||||
_ = kwargs
|
||||
return {
|
||||
"name": self.name,
|
||||
"amount": str(self.amount),
|
||||
"quantity": str(self.quantity),
|
||||
}
|
||||
|
||||
|
||||
class InvoiceClient(BaseModel):
|
||||
"""
|
||||
Invoice client information model.
|
||||
Maps to PHP DTO\\InvoiceClient.
|
||||
|
||||
Similar to IncomeClient but for invoices.
|
||||
"""
|
||||
|
||||
contact_phone: str | None = Field(
|
||||
None, alias="contactPhone", description="Client contact phone"
|
||||
)
|
||||
display_name: str | None = Field(
|
||||
None, alias="displayName", description="Client display name"
|
||||
)
|
||||
inn: str | None = Field(None, description="Client INN")
|
||||
|
||||
def model_dump(self, **kwargs: Any) -> dict[str, Any]:
|
||||
"""Custom serialization to match API format."""
|
||||
_ = kwargs
|
||||
return {
|
||||
"contactPhone": self.contact_phone,
|
||||
"displayName": self.display_name,
|
||||
"inn": self.inn,
|
||||
}
|
||||
73
app/lib/nalogo/dto/payment_type.py
Normal file
73
app/lib/nalogo/dto/payment_type.py
Normal file
@@ -0,0 +1,73 @@
|
||||
"""
|
||||
PaymentType-related DTO models.
|
||||
Based on PHP library's Model/PaymentType classes.
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
class PaymentType(BaseModel):
|
||||
"""
|
||||
Payment type model.
|
||||
Maps to PHP Model\\PaymentType\\PaymentType.
|
||||
"""
|
||||
|
||||
id: int = Field(..., description="Payment type ID")
|
||||
type: str = Field(..., description="Payment type")
|
||||
bank_name: str = Field(..., alias="bankName", description="Bank name")
|
||||
bank_bik: str = Field(..., alias="bankBik", description="Bank BIK")
|
||||
corr_account: str = Field(
|
||||
..., alias="corrAccount", description="Correspondent account"
|
||||
)
|
||||
favorite: bool = Field(..., description="Is favorite payment type")
|
||||
phone: str | None = Field(None, description="Phone number")
|
||||
bank_id: str | None = Field(None, alias="bankId", description="Bank ID")
|
||||
current_account: str = Field(
|
||||
..., alias="currentAccount", description="Current account"
|
||||
)
|
||||
available_for_pa: bool = Field(
|
||||
..., alias="availableForPa", description="Available for PA"
|
||||
)
|
||||
|
||||
def is_favorite(self) -> bool:
|
||||
"""Check if this payment type is marked as favorite."""
|
||||
return self.favorite
|
||||
|
||||
def model_dump(self, **kwargs: Any) -> dict[str, Any]:
|
||||
"""Custom serialization to match API format."""
|
||||
_ = kwargs
|
||||
return {
|
||||
"id": self.id,
|
||||
"type": self.type,
|
||||
"bankName": self.bank_name,
|
||||
"bankBik": self.bank_bik,
|
||||
"corrAccount": self.corr_account,
|
||||
"favorite": self.favorite,
|
||||
"phone": self.phone,
|
||||
"bankId": self.bank_id,
|
||||
"currentAccount": self.current_account,
|
||||
"availableForPa": self.available_for_pa,
|
||||
}
|
||||
|
||||
|
||||
class PaymentTypeCollection(BaseModel):
|
||||
"""
|
||||
Collection of payment types.
|
||||
Maps to PHP Model\\PaymentType\\PaymentTypeCollection.
|
||||
"""
|
||||
|
||||
payment_types: list[PaymentType] = Field(default_factory=list)
|
||||
|
||||
def __iter__(self) -> Any:
|
||||
"""Make collection iterable."""
|
||||
return iter(self.payment_types)
|
||||
|
||||
def __len__(self) -> int:
|
||||
"""Get collection length."""
|
||||
return len(self.payment_types)
|
||||
|
||||
def __getitem__(self, index: int) -> PaymentType:
|
||||
"""Get item by index."""
|
||||
return self.payment_types[index]
|
||||
96
app/lib/nalogo/dto/tax.py
Normal file
96
app/lib/nalogo/dto/tax.py
Normal file
@@ -0,0 +1,96 @@
|
||||
"""
|
||||
Tax-related DTO models.
|
||||
Based on PHP library's Model/Tax classes.
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
class Tax(BaseModel):
|
||||
"""
|
||||
Tax information model.
|
||||
Maps to PHP Model\\Tax\\Tax.
|
||||
"""
|
||||
|
||||
# Tax model fields would be defined based on API response structure
|
||||
# Since we don't have the exact PHP model structure, we'll use flexible Dict
|
||||
data: dict[str, Any] = Field(default_factory=dict, description="Tax data")
|
||||
|
||||
def model_dump(self, **kwargs: Any) -> dict[str, Any]:
|
||||
"""Return the raw data dictionary."""
|
||||
_ = kwargs
|
||||
return self.data
|
||||
|
||||
|
||||
class History(BaseModel):
|
||||
"""
|
||||
Tax history entry model.
|
||||
Maps to PHP Model\\Tax\\History.
|
||||
"""
|
||||
|
||||
# History fields would be defined based on API response
|
||||
data: dict[str, Any] = Field(default_factory=dict, description="History entry data")
|
||||
|
||||
def model_dump(self, **kwargs: Any) -> dict[str, Any]:
|
||||
"""Return the raw data dictionary."""
|
||||
_ = kwargs
|
||||
return self.data
|
||||
|
||||
|
||||
class HistoryRecords(BaseModel):
|
||||
"""
|
||||
Collection of tax history records.
|
||||
Maps to PHP Model\\Tax\\HistoryRecords.
|
||||
"""
|
||||
|
||||
records: list[History] = Field(default_factory=list, description="History records")
|
||||
|
||||
def __iter__(self) -> Any:
|
||||
"""Make collection iterable."""
|
||||
return iter(self.records)
|
||||
|
||||
def __len__(self) -> int:
|
||||
"""Get collection length."""
|
||||
return len(self.records)
|
||||
|
||||
def __getitem__(self, index: int) -> History:
|
||||
"""Get item by index."""
|
||||
return self.records[index]
|
||||
|
||||
|
||||
class Payment(BaseModel):
|
||||
"""
|
||||
Tax payment model.
|
||||
Maps to PHP Model\\Tax\\Payment.
|
||||
"""
|
||||
|
||||
# Payment fields would be defined based on API response
|
||||
data: dict[str, Any] = Field(default_factory=dict, description="Payment data")
|
||||
|
||||
def model_dump(self, **kwargs: Any) -> dict[str, Any]:
|
||||
"""Return the raw data dictionary."""
|
||||
_ = kwargs
|
||||
return self.data
|
||||
|
||||
|
||||
class PaymentRecords(BaseModel):
|
||||
"""
|
||||
Collection of tax payment records.
|
||||
Maps to PHP Model\\Tax\\PaymentRecords.
|
||||
"""
|
||||
|
||||
records: list[Payment] = Field(default_factory=list, description="Payment records")
|
||||
|
||||
def __iter__(self) -> Any:
|
||||
"""Make collection iterable."""
|
||||
return iter(self.records)
|
||||
|
||||
def __len__(self) -> int:
|
||||
"""Get collection length."""
|
||||
return len(self.records)
|
||||
|
||||
def __getitem__(self, index: int) -> Payment:
|
||||
"""Get item by index."""
|
||||
return self.records[index]
|
||||
122
app/lib/nalogo/dto/user.py
Normal file
122
app/lib/nalogo/dto/user.py
Normal file
@@ -0,0 +1,122 @@
|
||||
"""
|
||||
User-related DTO models.
|
||||
Based on PHP library's Model/User classes.
|
||||
"""
|
||||
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
|
||||
from pydantic import BaseModel, Field, field_validator
|
||||
|
||||
|
||||
class UserType(BaseModel):
|
||||
"""
|
||||
User profile model.
|
||||
Maps to PHP Model\\User\\UserType.
|
||||
"""
|
||||
|
||||
id: int = Field(..., description="User ID")
|
||||
last_name: str | None = Field(None, alias="lastName", description="Last name")
|
||||
display_name: str = Field(..., alias="displayName", description="Display name")
|
||||
middle_name: str | None = Field(None, alias="middleName", description="Middle name")
|
||||
email: str | None = Field(None, description="Email address")
|
||||
phone: str = Field(..., description="Phone number")
|
||||
inn: str = Field(..., description="INN (tax identification number)")
|
||||
snils: str | None = Field(None, description="SNILS")
|
||||
avatar_exists: bool = Field(..., alias="avatarExists", description="Avatar exists")
|
||||
initial_registration_date: datetime | None = Field(
|
||||
None, alias="initialRegistrationDate", description="Initial registration date"
|
||||
)
|
||||
registration_date: datetime | None = Field(
|
||||
None, alias="registrationDate", description="Registration date"
|
||||
)
|
||||
first_receipt_register_time: datetime | None = Field(
|
||||
None,
|
||||
alias="firstReceiptRegisterTime",
|
||||
description="First receipt registration time",
|
||||
)
|
||||
first_receipt_cancel_time: datetime | None = Field(
|
||||
None,
|
||||
alias="firstReceiptCancelTime",
|
||||
description="First receipt cancellation time",
|
||||
)
|
||||
hide_cancelled_receipt: bool = Field(
|
||||
..., alias="hideCancelledReceipt", description="Hide cancelled receipts"
|
||||
)
|
||||
register_available: bool | str | None = Field(
|
||||
None, alias="registerAvailable", description="Register available (mixed type)"
|
||||
)
|
||||
status: str | None = Field(None, description="User status")
|
||||
restricted_mode: bool = Field(
|
||||
..., alias="restrictedMode", description="Restricted mode"
|
||||
)
|
||||
pfr_url: str | None = Field(None, alias="pfrUrl", description="PFR URL")
|
||||
login: str | None = Field(None, description="Login")
|
||||
|
||||
@field_validator(
|
||||
"initial_registration_date",
|
||||
"registration_date",
|
||||
"first_receipt_register_time",
|
||||
"first_receipt_cancel_time",
|
||||
mode="before",
|
||||
)
|
||||
@classmethod
|
||||
def parse_datetime(cls, v: Any) -> datetime | None:
|
||||
"""Parse datetime strings from API."""
|
||||
if v is None or v == "":
|
||||
return None
|
||||
if isinstance(v, str):
|
||||
try:
|
||||
return datetime.fromisoformat(v.replace("Z", "+00:00"))
|
||||
except ValueError:
|
||||
return None
|
||||
return v # type: ignore[no-any-return]
|
||||
|
||||
def is_avatar_exists(self) -> bool:
|
||||
"""Check if user has avatar."""
|
||||
return self.avatar_exists
|
||||
|
||||
def is_hide_cancelled_receipt(self) -> bool:
|
||||
"""Check if cancelled receipts are hidden."""
|
||||
return self.hide_cancelled_receipt
|
||||
|
||||
def is_restricted_mode(self) -> bool:
|
||||
"""Check if user is in restricted mode."""
|
||||
return self.restricted_mode
|
||||
|
||||
def model_dump(self, **kwargs: Any) -> dict[str, Any]:
|
||||
"""Custom serialization to match API format."""
|
||||
_ = kwargs
|
||||
|
||||
def serialize_datetime(dt: datetime | None) -> str | None:
|
||||
if dt is None:
|
||||
return None
|
||||
return dt.isoformat()
|
||||
|
||||
return {
|
||||
"id": self.id,
|
||||
"lastName": self.last_name,
|
||||
"displayName": self.display_name,
|
||||
"middleName": self.middle_name,
|
||||
"email": self.email,
|
||||
"phone": self.phone,
|
||||
"inn": self.inn,
|
||||
"snils": self.snils,
|
||||
"avatarExists": self.avatar_exists,
|
||||
"initialRegistrationDate": serialize_datetime(
|
||||
self.initial_registration_date
|
||||
),
|
||||
"registrationDate": serialize_datetime(self.registration_date),
|
||||
"firstReceiptRegisterTime": serialize_datetime(
|
||||
self.first_receipt_register_time
|
||||
),
|
||||
"firstReceiptCancelTime": serialize_datetime(
|
||||
self.first_receipt_cancel_time
|
||||
),
|
||||
"hideCancelledReceipt": self.hide_cancelled_receipt,
|
||||
"registerAvailable": self.register_available,
|
||||
"status": self.status,
|
||||
"restrictedMode": self.restricted_mode,
|
||||
"pfrUrl": self.pfr_url,
|
||||
"login": self.login,
|
||||
}
|
||||
158
app/lib/nalogo/exceptions.py
Normal file
158
app/lib/nalogo/exceptions.py
Normal file
@@ -0,0 +1,158 @@
|
||||
"""
|
||||
Domain exceptions for Moy Nalog API.
|
||||
Mirrors PHP library's exception hierarchy and error handling.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import re
|
||||
from http import HTTPStatus
|
||||
|
||||
import httpx
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class DomainException(Exception): # noqa: N818 для совместимости публичного API
|
||||
"""Base domain exception for all Moy Nalog API errors."""
|
||||
|
||||
def __init__(self, message: str, response: httpx.Response | None = None):
|
||||
super().__init__(message)
|
||||
self.response = response
|
||||
|
||||
# Log the error with response details (without sensitive data)
|
||||
if response:
|
||||
self._log_error_details(message, response)
|
||||
|
||||
def _log_error_details(self, message: str, response: httpx.Response) -> None:
|
||||
"""Log error details while avoiding sensitive information."""
|
||||
# Mask potential sensitive data in URLs and headers
|
||||
safe_url = self._mask_sensitive_url(str(response.url))
|
||||
safe_headers = self._mask_sensitive_headers(dict(response.headers))
|
||||
|
||||
logger.error(
|
||||
"API Error: %s | Status: %d | URL: %s | Headers: %s | Body: %s",
|
||||
message,
|
||||
response.status_code,
|
||||
safe_url,
|
||||
safe_headers,
|
||||
self._get_safe_response_body(response),
|
||||
)
|
||||
|
||||
def _mask_sensitive_url(self, url: str) -> str:
|
||||
"""Mask potential sensitive data in URL."""
|
||||
# Replace tokens/keys with asterisks
|
||||
patterns = [
|
||||
(r"(token=)[^&]*", r"\1***"),
|
||||
(r"(key=)[^&]*", r"\1***"),
|
||||
(r"(secret=)[^&]*", r"\1***"),
|
||||
]
|
||||
|
||||
for pattern, replacement in patterns:
|
||||
url = re.sub(pattern, replacement, url)
|
||||
return url
|
||||
|
||||
def _mask_sensitive_headers(self, headers: dict[str, str]) -> dict[str, str]:
|
||||
"""Mask sensitive headers."""
|
||||
safe_headers = headers.copy()
|
||||
sensitive_keys = ["authorization", "x-api-key", "cookie", "set-cookie"]
|
||||
|
||||
for key in sensitive_keys:
|
||||
if key.lower() in [h.lower() for h in safe_headers]:
|
||||
# Find the actual key (case-insensitive)
|
||||
actual_key = next(k for k in safe_headers if k.lower() == key.lower())
|
||||
safe_headers[actual_key] = "***"
|
||||
|
||||
return safe_headers
|
||||
|
||||
def _get_safe_response_body(self, response: httpx.Response) -> str:
|
||||
"""Get response body with potential sensitive data masked."""
|
||||
try:
|
||||
body = response.text[:1000] # Limit body size for logging
|
||||
# Mask potential tokens in JSON responses
|
||||
patterns = [
|
||||
(r'("token":\s*")[^"]*(")', r"\1***\2"),
|
||||
(r'("refreshToken":\s*")[^"]*(")', r"\1***\2"),
|
||||
(r'("password":\s*")[^"]*(")', r"\1***\2"),
|
||||
(r'("secret":\s*")[^"]*(")', r"\1***\2"),
|
||||
]
|
||||
|
||||
for pattern, replacement in patterns:
|
||||
body = re.sub(pattern, replacement, body)
|
||||
|
||||
return body
|
||||
except Exception:
|
||||
return "[Failed to read response body]"
|
||||
|
||||
|
||||
class ValidationException(DomainException):
|
||||
"""HTTP 400 - Validation error."""
|
||||
|
||||
|
||||
class UnauthorizedException(DomainException):
|
||||
"""HTTP 401 - Authentication required or invalid credentials."""
|
||||
|
||||
|
||||
class ForbiddenException(DomainException):
|
||||
"""HTTP 403 - Access forbidden."""
|
||||
|
||||
|
||||
class NotFoundException(DomainException):
|
||||
"""HTTP 404 - Resource not found."""
|
||||
|
||||
|
||||
class ClientException(DomainException):
|
||||
"""HTTP 406 - Client error (e.g., wrong Accept headers)."""
|
||||
|
||||
|
||||
class PhoneException(DomainException):
|
||||
"""HTTP 422 - Phone-related error (SMS, verification, etc.)."""
|
||||
|
||||
|
||||
class ServerException(DomainException):
|
||||
"""HTTP 500 - Internal server error."""
|
||||
|
||||
|
||||
class UnknownErrorException(DomainException):
|
||||
"""Unknown HTTP error code."""
|
||||
|
||||
|
||||
def raise_for_status(response: httpx.Response) -> None:
|
||||
"""
|
||||
Raise appropriate domain exception based on HTTP status code.
|
||||
|
||||
Maps status codes to exceptions exactly like PHP ErrorHandler:
|
||||
- 400: ValidationException
|
||||
- 401: UnauthorizedException
|
||||
- 403: ForbiddenException
|
||||
- 404: NotFoundException
|
||||
- 406: ClientException
|
||||
- 422: PhoneException
|
||||
- 500: ServerException
|
||||
- default: UnknownErrorException
|
||||
|
||||
Args:
|
||||
response: httpx.Response object
|
||||
|
||||
Raises:
|
||||
DomainException: Appropriate exception for status code
|
||||
"""
|
||||
if response.status_code < HTTPStatus.BAD_REQUEST:
|
||||
return
|
||||
|
||||
body = response.text
|
||||
|
||||
if response.status_code == HTTPStatus.BAD_REQUEST:
|
||||
raise ValidationException(body, response)
|
||||
if response.status_code == HTTPStatus.UNAUTHORIZED:
|
||||
raise UnauthorizedException(body, response)
|
||||
if response.status_code == HTTPStatus.FORBIDDEN:
|
||||
raise ForbiddenException(body, response)
|
||||
if response.status_code == HTTPStatus.NOT_FOUND:
|
||||
raise NotFoundException(body, response)
|
||||
if response.status_code == HTTPStatus.NOT_ACCEPTABLE:
|
||||
raise ClientException("Wrong Accept headers", response)
|
||||
if response.status_code == HTTPStatus.UNPROCESSABLE_ENTITY:
|
||||
raise PhoneException(body, response)
|
||||
if response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR:
|
||||
raise ServerException(body, response)
|
||||
raise UnknownErrorException(body, response)
|
||||
195
app/lib/nalogo/income.py
Normal file
195
app/lib/nalogo/income.py
Normal file
@@ -0,0 +1,195 @@
|
||||
"""
|
||||
Income API implementation.
|
||||
Based on PHP library's Api\\Income class.
|
||||
"""
|
||||
|
||||
from datetime import datetime
|
||||
from decimal import Decimal
|
||||
from typing import Any
|
||||
|
||||
from ._http import AsyncHTTPClient
|
||||
from .dto.income import (
|
||||
AtomDateTime,
|
||||
CancelCommentType,
|
||||
CancelRequest,
|
||||
IncomeClient,
|
||||
IncomeRequest,
|
||||
IncomeServiceItem,
|
||||
IncomeType,
|
||||
PaymentType,
|
||||
)
|
||||
|
||||
|
||||
class IncomeAPI:
|
||||
"""
|
||||
Income API for creating and managing receipts.
|
||||
|
||||
Provides async methods for:
|
||||
- Creating income receipts (single or multiple items)
|
||||
- Cancelling income receipts
|
||||
|
||||
Maps to PHP Api\\Income functionality.
|
||||
"""
|
||||
|
||||
def __init__(self, http_client: AsyncHTTPClient):
|
||||
self.http = http_client
|
||||
|
||||
async def create(
|
||||
self,
|
||||
name: str,
|
||||
amount: Decimal | float | int | str,
|
||||
quantity: Decimal | float | int | str = 1,
|
||||
operation_time: datetime | None = None,
|
||||
client: IncomeClient | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
Create income receipt with single service item.
|
||||
|
||||
Maps to PHP Income::create() method.
|
||||
|
||||
Args:
|
||||
name: Service name/description
|
||||
amount: Service amount (converted to Decimal)
|
||||
quantity: Service quantity (converted to Decimal, default: 1)
|
||||
operation_time: Operation datetime (default: now)
|
||||
client: Client information (default: individual client)
|
||||
|
||||
Returns:
|
||||
Dictionary with response data including approvedReceiptUuid
|
||||
|
||||
Raises:
|
||||
ValidationException: For validation errors
|
||||
DomainException: For other API errors
|
||||
"""
|
||||
# Convert to IncomeServiceItem
|
||||
service_item = IncomeServiceItem(
|
||||
name=name,
|
||||
amount=Decimal(str(amount)),
|
||||
quantity=Decimal(str(quantity)),
|
||||
)
|
||||
|
||||
return await self.create_multiple_items([service_item], operation_time, client)
|
||||
|
||||
async def create_multiple_items(
|
||||
self,
|
||||
services: list[IncomeServiceItem],
|
||||
operation_time: datetime | None = None,
|
||||
client: IncomeClient | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
Create income receipt with multiple service items.
|
||||
|
||||
Maps to PHP Income::createMultipleItems() method.
|
||||
|
||||
Args:
|
||||
services: List of service items
|
||||
operation_time: Operation datetime (default: now)
|
||||
client: Client information (default: individual client)
|
||||
|
||||
Returns:
|
||||
Dictionary with response data including approvedReceiptUuid
|
||||
|
||||
Raises:
|
||||
ValidationException: For validation errors (empty items, invalid amounts, etc.)
|
||||
DomainException: For other API errors
|
||||
"""
|
||||
if not services:
|
||||
raise ValueError("Services cannot be empty")
|
||||
|
||||
# Validate client for legal entity (mirrors PHP validation)
|
||||
if client and client.income_type == IncomeType.FROM_LEGAL_ENTITY:
|
||||
if not client.inn:
|
||||
raise ValueError("Client INN cannot be empty for legal entity")
|
||||
if not client.display_name:
|
||||
raise ValueError("Client DisplayName cannot be empty for legal entity")
|
||||
|
||||
# Calculate total amount (mirrors PHP BigDecimal logic)
|
||||
total_amount = sum(item.get_total_amount() for item in services)
|
||||
|
||||
# Create request object
|
||||
request = IncomeRequest(
|
||||
operation_time=(
|
||||
AtomDateTime.from_datetime(operation_time)
|
||||
if operation_time
|
||||
else AtomDateTime.now()
|
||||
),
|
||||
request_time=AtomDateTime.now(),
|
||||
services=services,
|
||||
total_amount=str(total_amount),
|
||||
client=client or IncomeClient(),
|
||||
payment_type=PaymentType.CASH,
|
||||
ignore_max_total_income_restriction=False,
|
||||
)
|
||||
|
||||
# Make API request
|
||||
response = await self.http.post("/income", json_data=request.model_dump())
|
||||
return response.json() # type: ignore[no-any-return]
|
||||
|
||||
async def cancel(
|
||||
self,
|
||||
receipt_uuid: str,
|
||||
comment: CancelCommentType | str,
|
||||
operation_time: datetime | None = None,
|
||||
request_time: datetime | None = None,
|
||||
partner_code: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
Cancel income receipt.
|
||||
|
||||
Maps to PHP Income::cancel() method.
|
||||
|
||||
Args:
|
||||
receipt_uuid: Receipt UUID to cancel
|
||||
comment: Cancellation reason (enum or string)
|
||||
operation_time: Operation datetime (default: now)
|
||||
request_time: Request datetime (default: now)
|
||||
partner_code: Partner code (optional)
|
||||
|
||||
Returns:
|
||||
Dictionary with cancellation response data
|
||||
|
||||
Raises:
|
||||
ValidationException: For validation errors (empty UUID, invalid comment)
|
||||
DomainException: For other API errors
|
||||
"""
|
||||
# Validate receipt UUID
|
||||
if not receipt_uuid.strip():
|
||||
raise ValueError("Receipt UUID cannot be empty")
|
||||
|
||||
# Convert comment to enum if string
|
||||
if isinstance(comment, str):
|
||||
# Try to find matching enum value
|
||||
comment_enum = None
|
||||
for enum_val in CancelCommentType:
|
||||
if enum_val.value == comment:
|
||||
comment_enum = enum_val
|
||||
break
|
||||
|
||||
if comment_enum is None:
|
||||
valid_comments = [e.value for e in CancelCommentType]
|
||||
raise ValueError(
|
||||
f"Comment is invalid. Must be one of: {valid_comments}"
|
||||
)
|
||||
|
||||
comment = comment_enum
|
||||
|
||||
# Create request object
|
||||
request = CancelRequest(
|
||||
operation_time=(
|
||||
AtomDateTime.from_datetime(operation_time)
|
||||
if operation_time
|
||||
else AtomDateTime.now()
|
||||
),
|
||||
request_time=(
|
||||
AtomDateTime.from_datetime(request_time)
|
||||
if request_time
|
||||
else AtomDateTime.now()
|
||||
),
|
||||
comment=comment,
|
||||
receipt_uuid=receipt_uuid.strip(),
|
||||
partner_code=partner_code,
|
||||
)
|
||||
|
||||
# Make API request
|
||||
response = await self.http.post("/cancel", json_data=request.model_dump())
|
||||
return response.json() # type: ignore[no-any-return]
|
||||
60
app/lib/nalogo/payment_type.py
Normal file
60
app/lib/nalogo/payment_type.py
Normal file
@@ -0,0 +1,60 @@
|
||||
"""
|
||||
PaymentType API implementation.
|
||||
Based on PHP library's Api\\PaymentType class.
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
|
||||
from ._http import AsyncHTTPClient
|
||||
|
||||
|
||||
class PaymentTypeAPI:
|
||||
"""
|
||||
PaymentType API for managing payment methods.
|
||||
|
||||
Provides async methods for:
|
||||
- Getting payment types table
|
||||
- Finding favorite payment type
|
||||
|
||||
Maps to PHP Api\\PaymentType functionality.
|
||||
"""
|
||||
|
||||
def __init__(self, http_client: AsyncHTTPClient):
|
||||
self.http = http_client
|
||||
|
||||
async def table(self) -> list[dict[str, Any]]:
|
||||
"""
|
||||
Get all available payment types.
|
||||
|
||||
Maps to PHP PaymentType::table().
|
||||
|
||||
Returns:
|
||||
List of payment type dictionaries with bank information
|
||||
|
||||
Raises:
|
||||
DomainException: For API errors
|
||||
"""
|
||||
response = await self.http.get("/payment-type/table")
|
||||
return response.json() # type: ignore[no-any-return]
|
||||
|
||||
async def favorite(self) -> dict[str, Any] | None:
|
||||
"""
|
||||
Get favorite payment type.
|
||||
|
||||
Maps to PHP PaymentType::favorite().
|
||||
Finds the first payment type marked as favorite from the table.
|
||||
|
||||
Returns:
|
||||
Payment type dictionary if favorite found, None otherwise
|
||||
|
||||
Raises:
|
||||
DomainException: For API errors
|
||||
"""
|
||||
payment_types = await self.table()
|
||||
|
||||
# Find first payment type with favorite=True
|
||||
for payment_type in payment_types:
|
||||
if payment_type.get("favorite", False):
|
||||
return payment_type
|
||||
|
||||
return None
|
||||
73
app/lib/nalogo/receipt.py
Normal file
73
app/lib/nalogo/receipt.py
Normal file
@@ -0,0 +1,73 @@
|
||||
"""
|
||||
Receipt API implementation.
|
||||
Based on PHP library's Api\\Receipt class.
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
|
||||
from ._http import AsyncHTTPClient
|
||||
|
||||
|
||||
class ReceiptAPI:
|
||||
"""
|
||||
Receipt API for accessing receipt information.
|
||||
|
||||
Provides async methods for:
|
||||
- Getting receipt print URL
|
||||
- Getting receipt JSON data
|
||||
|
||||
Maps to PHP Api\\Receipt functionality.
|
||||
"""
|
||||
|
||||
def __init__(self, http_client: AsyncHTTPClient, base_endpoint: str, user_inn: str):
|
||||
self.http = http_client
|
||||
self.base_endpoint = base_endpoint
|
||||
self.user_inn = user_inn
|
||||
|
||||
def print_url(self, receipt_uuid: str) -> str:
|
||||
"""
|
||||
Compose receipt print URL.
|
||||
|
||||
Maps to PHP Receipt::printUrl() method.
|
||||
This method composes the URL without making HTTP request.
|
||||
|
||||
Args:
|
||||
receipt_uuid: Receipt UUID
|
||||
|
||||
Returns:
|
||||
Complete URL for receipt printing
|
||||
|
||||
Raises:
|
||||
ValueError: If receipt_uuid is empty
|
||||
"""
|
||||
if not receipt_uuid.strip():
|
||||
raise ValueError("Receipt UUID cannot be empty")
|
||||
|
||||
# Compose URL like PHP: sprintf('/receipt/%s/%s/print', $this->profile->getInn(), $receiptUuid)
|
||||
path = f"/receipt/{self.user_inn}/{receipt_uuid.strip()}/print"
|
||||
return f"{self.base_endpoint}{path}"
|
||||
|
||||
async def json(self, receipt_uuid: str) -> dict[str, Any]:
|
||||
"""
|
||||
Get receipt data in JSON format.
|
||||
|
||||
Maps to PHP Receipt::json() method.
|
||||
|
||||
Args:
|
||||
receipt_uuid: Receipt UUID
|
||||
|
||||
Returns:
|
||||
Dictionary with receipt JSON data
|
||||
|
||||
Raises:
|
||||
ValueError: If receipt_uuid is empty
|
||||
DomainException: For API errors
|
||||
"""
|
||||
if not receipt_uuid.strip():
|
||||
raise ValueError("Receipt UUID cannot be empty")
|
||||
|
||||
# Make GET request like PHP: sprintf('/receipt/%s/%s/json', $this->profile->getInn(), $receiptUuid)
|
||||
path = f"/receipt/{self.user_inn}/{receipt_uuid.strip()}/json"
|
||||
response = await self.http.get(path)
|
||||
|
||||
return response.json() # type: ignore[no-any-return]
|
||||
83
app/lib/nalogo/tax.py
Normal file
83
app/lib/nalogo/tax.py
Normal file
@@ -0,0 +1,83 @@
|
||||
"""
|
||||
Tax API implementation.
|
||||
Based on PHP library's Api\\Tax class.
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
|
||||
from ._http import AsyncHTTPClient
|
||||
|
||||
|
||||
class TaxAPI:
|
||||
"""
|
||||
Tax API for tax information and history.
|
||||
|
||||
Provides async methods for:
|
||||
- Getting current tax information
|
||||
- Getting tax history by OKTMO
|
||||
- Getting payment records
|
||||
|
||||
Maps to PHP Api\\Tax functionality.
|
||||
"""
|
||||
|
||||
def __init__(self, http_client: AsyncHTTPClient):
|
||||
self.http = http_client
|
||||
|
||||
async def get(self) -> dict[str, Any]:
|
||||
"""
|
||||
Get current tax information.
|
||||
|
||||
Maps to PHP Tax::get().
|
||||
|
||||
Returns:
|
||||
Dictionary with current tax data
|
||||
|
||||
Raises:
|
||||
DomainException: For API errors
|
||||
"""
|
||||
response = await self.http.get("/taxes")
|
||||
return response.json() # type: ignore[no-any-return]
|
||||
|
||||
async def history(self, oktmo: str | None = None) -> dict[str, Any]:
|
||||
"""
|
||||
Get tax history.
|
||||
|
||||
Maps to PHP Tax::history().
|
||||
|
||||
Args:
|
||||
oktmo: Optional OKTMO code for filtering
|
||||
|
||||
Returns:
|
||||
Dictionary with tax history records
|
||||
|
||||
Raises:
|
||||
DomainException: For API errors
|
||||
"""
|
||||
request_data = {"oktmo": oktmo}
|
||||
response = await self.http.post("/taxes/history", json_data=request_data)
|
||||
return response.json() # type: ignore[no-any-return]
|
||||
|
||||
async def payments(
|
||||
self, oktmo: str | None = None, only_paid: bool = False
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
Get tax payment records.
|
||||
|
||||
Maps to PHP Tax::payments().
|
||||
|
||||
Args:
|
||||
oktmo: Optional OKTMO code for filtering
|
||||
only_paid: If True, return only paid records
|
||||
|
||||
Returns:
|
||||
Dictionary with payment records
|
||||
|
||||
Raises:
|
||||
DomainException: For API errors
|
||||
"""
|
||||
request_data = {
|
||||
"oktmo": oktmo,
|
||||
"onlyPaid": only_paid,
|
||||
}
|
||||
response = await self.http.post("/taxes/payments", json_data=request_data)
|
||||
return response.json() # type: ignore[no-any-return]
|
||||
40
app/lib/nalogo/user.py
Normal file
40
app/lib/nalogo/user.py
Normal file
@@ -0,0 +1,40 @@
|
||||
"""
|
||||
User API implementation.
|
||||
Based on PHP library's Api\\User class.
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
|
||||
from ._http import AsyncHTTPClient
|
||||
|
||||
|
||||
class UserAPI:
|
||||
"""
|
||||
User API for user information.
|
||||
|
||||
Provides async methods for:
|
||||
- Getting current user information
|
||||
|
||||
Maps to PHP Api\\User functionality.
|
||||
"""
|
||||
|
||||
def __init__(self, http_client: AsyncHTTPClient):
|
||||
self.http = http_client
|
||||
|
||||
async def get(self) -> dict[str, Any]:
|
||||
"""
|
||||
Get current user information.
|
||||
|
||||
Maps to PHP User::get().
|
||||
|
||||
Returns:
|
||||
Dictionary with user profile data including:
|
||||
- id, inn, displayName, email, phone
|
||||
- registration dates, status, restrictions
|
||||
- avatar, receipt settings, etc.
|
||||
|
||||
Raises:
|
||||
DomainException: For API errors
|
||||
"""
|
||||
response = await self.http.get("/user")
|
||||
return response.json() # type: ignore[no-any-return]
|
||||
@@ -164,13 +164,31 @@ class NalogoQueueService:
|
||||
|
||||
# Пытаемся отправить чек
|
||||
try:
|
||||
# Восстанавливаем описание из сохранённых данных
|
||||
telegram_user_id = receipt_data.get("telegram_user_id")
|
||||
amount_kopeks = receipt_data.get("amount_kopeks")
|
||||
|
||||
# Формируем описание заново из настроек (если есть данные)
|
||||
if amount_kopeks is not None:
|
||||
receipt_name = settings.get_balance_payment_description(
|
||||
amount_kopeks, telegram_user_id
|
||||
)
|
||||
else:
|
||||
# Fallback на сохранённое имя
|
||||
receipt_name = receipt_data.get(
|
||||
"name",
|
||||
settings.get_balance_payment_description(int(amount * 100), telegram_user_id)
|
||||
)
|
||||
|
||||
receipt_uuid = await self._nalogo_service.create_receipt(
|
||||
name=receipt_data.get("name", "Интернет-сервис - Пополнение баланса"),
|
||||
name=receipt_name,
|
||||
amount=amount,
|
||||
quantity=receipt_data.get("quantity", 1),
|
||||
client_info=receipt_data.get("client_info"),
|
||||
payment_id=payment_id,
|
||||
queue_on_failure=False, # Не добавлять в очередь повторно автоматически
|
||||
telegram_user_id=telegram_user_id,
|
||||
amount_kopeks=amount_kopeks,
|
||||
)
|
||||
|
||||
if receipt_uuid:
|
||||
|
||||
@@ -1,11 +1,11 @@
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from zoneinfo import ZoneInfo
|
||||
from datetime import datetime, timezone, timedelta
|
||||
from typing import Optional, Dict, Any
|
||||
from decimal import Decimal
|
||||
|
||||
from nalogo import Client
|
||||
from nalogo.dto.income import IncomeClient, IncomeType
|
||||
# Используем локальную исправленную версию библиотеки
|
||||
from app.lib.nalogo import Client
|
||||
from app.lib.nalogo.dto.income import IncomeClient, IncomeType, MOSCOW_TZ
|
||||
|
||||
from app.config import settings
|
||||
from app.utils.cache import cache
|
||||
@@ -81,6 +81,8 @@ class NaloGoService:
|
||||
quantity: int,
|
||||
client_info: Optional[Dict[str, Any]],
|
||||
payment_id: Optional[str] = None,
|
||||
telegram_user_id: Optional[int] = None,
|
||||
amount_kopeks: Optional[int] = None,
|
||||
) -> bool:
|
||||
"""Добавить чек в очередь для отложенной отправки."""
|
||||
receipt_data = {
|
||||
@@ -89,6 +91,8 @@ class NaloGoService:
|
||||
"quantity": quantity,
|
||||
"client_info": client_info,
|
||||
"payment_id": payment_id,
|
||||
"telegram_user_id": telegram_user_id,
|
||||
"amount_kopeks": amount_kopeks,
|
||||
"created_at": datetime.now().isoformat(),
|
||||
"attempts": 0,
|
||||
}
|
||||
@@ -129,6 +133,8 @@ class NaloGoService:
|
||||
client_info: Optional[Dict[str, Any]] = None,
|
||||
payment_id: Optional[str] = None,
|
||||
queue_on_failure: bool = True,
|
||||
telegram_user_id: Optional[int] = None,
|
||||
amount_kopeks: Optional[int] = None,
|
||||
) -> Optional[str]:
|
||||
"""Создание чека о доходе.
|
||||
|
||||
@@ -139,6 +145,8 @@ class NaloGoService:
|
||||
client_info: Информация о клиенте (опционально)
|
||||
payment_id: ID платежа для логирования
|
||||
queue_on_failure: Добавить в очередь при временной недоступности
|
||||
telegram_user_id: Telegram ID пользователя для формирования описания
|
||||
amount_kopeks: Сумма в копейках для формирования описания
|
||||
|
||||
Returns:
|
||||
UUID чека или None при ошибке
|
||||
@@ -154,7 +162,10 @@ class NaloGoService:
|
||||
if not auth_success:
|
||||
# Если сервис недоступен — добавляем в очередь
|
||||
if queue_on_failure:
|
||||
await self._queue_receipt(name, amount, quantity, client_info, payment_id)
|
||||
await self._queue_receipt(
|
||||
name, amount, quantity, client_info, payment_id,
|
||||
telegram_user_id, amount_kopeks
|
||||
)
|
||||
return None
|
||||
|
||||
income_api = self.client.income()
|
||||
@@ -169,16 +180,12 @@ class NaloGoService:
|
||||
inn=client_info.get("inn")
|
||||
)
|
||||
|
||||
# Используем время из настроек (TZ env)
|
||||
local_tz = ZoneInfo(settings.TIMEZONE)
|
||||
local_time = datetime.now(local_tz)
|
||||
|
||||
# Библиотека использует UTC время автоматически
|
||||
result = await income_api.create(
|
||||
name=name,
|
||||
amount=Decimal(str(amount)),
|
||||
quantity=quantity,
|
||||
client=income_client,
|
||||
operation_time=local_time
|
||||
)
|
||||
|
||||
receipt_uuid = result.get("approvedReceiptUuid")
|
||||
@@ -196,7 +203,10 @@ class NaloGoService:
|
||||
f"(payment_id={payment_id}, сумма={amount}₽)"
|
||||
)
|
||||
if queue_on_failure:
|
||||
await self._queue_receipt(name, amount, quantity, client_info, payment_id)
|
||||
await self._queue_receipt(
|
||||
name, amount, quantity, client_info, payment_id,
|
||||
telegram_user_id, amount_kopeks
|
||||
)
|
||||
else:
|
||||
logger.error("Ошибка создания чека в NaloGO: %s", error, exc_info=True)
|
||||
return None
|
||||
|
||||
@@ -969,7 +969,10 @@ class YooKassaPaymentMixin:
|
||||
|
||||
# Создаем чек через NaloGO (если NALOGO_ENABLED=true)
|
||||
if hasattr(self, "nalogo_service") and self.nalogo_service:
|
||||
await self._create_nalogo_receipt(payment)
|
||||
await self._create_nalogo_receipt(
|
||||
payment,
|
||||
telegram_user_id=user.telegram_id if user else None,
|
||||
)
|
||||
|
||||
return True
|
||||
|
||||
@@ -1024,6 +1027,7 @@ class YooKassaPaymentMixin:
|
||||
async def _create_nalogo_receipt(
|
||||
self,
|
||||
payment: "YooKassaPayment",
|
||||
telegram_user_id: Optional[int] = None,
|
||||
) -> None:
|
||||
"""Создание чека через NaloGO для успешного платежа."""
|
||||
if not hasattr(self, "nalogo_service") or not self.nalogo_service:
|
||||
@@ -1032,13 +1036,18 @@ class YooKassaPaymentMixin:
|
||||
|
||||
try:
|
||||
amount_rubles = payment.amount_kopeks / 100
|
||||
receipt_name = "Интернет-сервис - Пополнение баланса"
|
||||
# Формируем описание из настроек (включает сумму и ID пользователя)
|
||||
receipt_name = settings.get_balance_payment_description(
|
||||
payment.amount_kopeks, telegram_user_id
|
||||
)
|
||||
|
||||
receipt_uuid = await self.nalogo_service.create_receipt(
|
||||
name=receipt_name,
|
||||
amount=amount_rubles,
|
||||
quantity=1,
|
||||
payment_id=payment.yookassa_payment_id,
|
||||
telegram_user_id=telegram_user_id,
|
||||
amount_kopeks=payment.amount_kopeks,
|
||||
)
|
||||
|
||||
if receipt_uuid:
|
||||
|
||||
@@ -20,7 +20,8 @@ python-multipart==0.0.9
|
||||
yookassa==3.9.0
|
||||
|
||||
# NaloGO для чеков в налоговую
|
||||
nalogo
|
||||
# nalogo - используем локальную исправленную версию в app/lib/nalogo/
|
||||
httpx # зависимость для nalogo
|
||||
|
||||
# Логирование и мониторинг
|
||||
structlog==23.2.0
|
||||
|
||||
Reference in New Issue
Block a user