diff --git a/app/lib/__init__.py b/app/lib/__init__.py new file mode 100644 index 00000000..a57d7d8b --- /dev/null +++ b/app/lib/__init__.py @@ -0,0 +1 @@ +# Local libraries diff --git a/app/lib/nalogo/__init__.py b/app/lib/nalogo/__init__.py new file mode 100644 index 00000000..cfb8a954 --- /dev/null +++ b/app/lib/nalogo/__init__.py @@ -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 +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", +] diff --git a/app/lib/nalogo/_http.py b/app/lib/nalogo/_http.py new file mode 100644 index 00000000..9f233365 --- /dev/null +++ b/app/lib/nalogo/_http.py @@ -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) diff --git a/app/lib/nalogo/auth.py b/app/lib/nalogo/auth.py new file mode 100644 index 00000000..c5150a92 --- /dev/null +++ b/app/lib/nalogo/auth.py @@ -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 diff --git a/app/lib/nalogo/client.py b/app/lib/nalogo/client.py new file mode 100644 index 00000000..dc9d8b15 --- /dev/null +++ b/app/lib/nalogo/client.py @@ -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) diff --git a/app/lib/nalogo/dto/__init__.py b/app/lib/nalogo/dto/__init__.py new file mode 100644 index 00000000..1cbed53e --- /dev/null +++ b/app/lib/nalogo/dto/__init__.py @@ -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", +] diff --git a/app/lib/nalogo/dto/device.py b/app/lib/nalogo/dto/device.py new file mode 100644 index 00000000..da462d5c --- /dev/null +++ b/app/lib/nalogo/dto/device.py @@ -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}, + } diff --git a/app/lib/nalogo/dto/income.py b/app/lib/nalogo/dto/income.py new file mode 100644 index 00000000..e0a29cfe --- /dev/null +++ b/app/lib/nalogo/dto/income.py @@ -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, + } diff --git a/app/lib/nalogo/dto/invoice.py b/app/lib/nalogo/dto/invoice.py new file mode 100644 index 00000000..16ce7989 --- /dev/null +++ b/app/lib/nalogo/dto/invoice.py @@ -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, + } diff --git a/app/lib/nalogo/dto/payment_type.py b/app/lib/nalogo/dto/payment_type.py new file mode 100644 index 00000000..a63cb2b2 --- /dev/null +++ b/app/lib/nalogo/dto/payment_type.py @@ -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] diff --git a/app/lib/nalogo/dto/tax.py b/app/lib/nalogo/dto/tax.py new file mode 100644 index 00000000..090c1c82 --- /dev/null +++ b/app/lib/nalogo/dto/tax.py @@ -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] diff --git a/app/lib/nalogo/dto/user.py b/app/lib/nalogo/dto/user.py new file mode 100644 index 00000000..00856100 --- /dev/null +++ b/app/lib/nalogo/dto/user.py @@ -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, + } diff --git a/app/lib/nalogo/exceptions.py b/app/lib/nalogo/exceptions.py new file mode 100644 index 00000000..498febe4 --- /dev/null +++ b/app/lib/nalogo/exceptions.py @@ -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) diff --git a/app/lib/nalogo/income.py b/app/lib/nalogo/income.py new file mode 100644 index 00000000..6b3727f7 --- /dev/null +++ b/app/lib/nalogo/income.py @@ -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] diff --git a/app/lib/nalogo/payment_type.py b/app/lib/nalogo/payment_type.py new file mode 100644 index 00000000..714efff9 --- /dev/null +++ b/app/lib/nalogo/payment_type.py @@ -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 diff --git a/app/lib/nalogo/receipt.py b/app/lib/nalogo/receipt.py new file mode 100644 index 00000000..f8664b91 --- /dev/null +++ b/app/lib/nalogo/receipt.py @@ -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] diff --git a/app/lib/nalogo/tax.py b/app/lib/nalogo/tax.py new file mode 100644 index 00000000..5c845a5d --- /dev/null +++ b/app/lib/nalogo/tax.py @@ -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] diff --git a/app/lib/nalogo/user.py b/app/lib/nalogo/user.py new file mode 100644 index 00000000..4cdf7cac --- /dev/null +++ b/app/lib/nalogo/user.py @@ -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] diff --git a/app/services/nalogo_queue_service.py b/app/services/nalogo_queue_service.py index 57fb3813..3134bdfc 100644 --- a/app/services/nalogo_queue_service.py +++ b/app/services/nalogo_queue_service.py @@ -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: diff --git a/app/services/nalogo_service.py b/app/services/nalogo_service.py index 52e0c0e2..e2c13b67 100644 --- a/app/services/nalogo_service.py +++ b/app/services/nalogo_service.py @@ -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 diff --git a/app/services/payment/yookassa.py b/app/services/payment/yookassa.py index 8b34b219..5a315570 100644 --- a/app/services/payment/yookassa.py +++ b/app/services/payment/yookassa.py @@ -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: diff --git a/requirements.txt b/requirements.txt index 055ee40e..54cbe774 100644 --- a/requirements.txt +++ b/requirements.txt @@ -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