added my nalog management

2026-01-05 19:43:40 +03:00
parent c1593d7868
commit 475a30caf4
9 changed files with 174 additions and 5 deletions
@@ -0,0 +1,103 @@
+import asyncio
+import logging
+from datetime import datetime
+from decimal import Decimal
+from typing import Optional
+
+from nalogo import Client
+from nalogo.dto.income import (
+    AtomDateTime,
+    IncomeClient,
+    IncomeRequest,
+    IncomeServiceItem,
+    PaymentType,
+)
+
+
+class NalogoService:
+    def __init__(self, inn: Optional[str], password: Optional[str]) -> None:
+        self.inn = inn.strip() if inn else None
+        self.password = password
+        self.configured = bool(self.inn and self.password)
+        self._client = Client() if self.configured else None
+        self._auth_lock = asyncio.Lock()
+
+        if not self.configured:
+            logging.warning("Nalogo credentials are missing. Receipt sending disabled.")
+
+    async def _ensure_authenticated(self) -> bool:
+        if not self._client:
+            return False
+
+        async with self._auth_lock:
+            token_data = await self._client.auth_provider.get_token()
+            if token_data:
+                return True
+
+            try:
+                token_json = await self._client.create_new_access_token(
+                    self.inn,
+                    self.password,
+                )
+                await self._client.authenticate(token_json)
+                logging.info("Nalogo authentication succeeded.")
+                return True
+            except Exception:
+                logging.exception("Nalogo authentication failed.")
+                return False
+
+    async def create_income_receipt(
+        self,
+        *,
+        item_name: str,
+        amount: float,
+        quantity: float = 1.0,
+        client: Optional[IncomeClient] = None,
+        operation_time: Optional[datetime] = None,
+    ) -> Optional[str]:
+        if not self.configured:
+            return None
+        if not await self._ensure_authenticated():
+            return None
+
+        try:
+            service_item = IncomeServiceItem(
+                name=item_name,
+                amount=Decimal(str(amount)),
+                quantity=Decimal(str(quantity)),
+            )
+            total_amount = service_item.get_total_amount()
+            request = IncomeRequest(
+                operation_time=(
+                    AtomDateTime.from_datetime(operation_time)
+                    if operation_time
+                    else AtomDateTime.now()
+                ),
+                request_time=AtomDateTime.now(),
+                services=[service_item],
+                total_amount=str(total_amount),
+                client=client or IncomeClient(),
+                payment_type=PaymentType.ACCOUNT,
+                ignore_max_total_income_restriction=False,
+            )
+            response = await self._client.http_client.post(
+                "/income",
+                json_data=request.model_dump(),
+            )
+            payload = response.json()
+            receipt_uuid = (
+                payload.get("approvedReceiptUuid")
+                or payload.get("receiptUuid")
+                or payload.get("receipt_uuid")
+            )
+            if receipt_uuid:
+                logging.info("Nalogo receipt created: %s", receipt_uuid)
+            else:
+                logging.info("Nalogo receipt created without a UUID in response.")
+            return receipt_uuid
+        except Exception:
+            logging.exception("Failed to create Nalogo receipt.")
+            return None
+
+    async def close(self) -> None:
+        return None