Вчасно-каса
Інтеграція K2 ERP з Вчасно.Каса через Python та Device Manager
Device Manager від Вчасно.Каса — локальний інтеграційний застосунок, який встановлюється на компʼютер, сервер або Android-пристрій і надає локальне REST API для роботи з ПРРО Вчасно.Каса, принтерами, банківськими терміналами та іншими POS-пристроями.
У цій статті описана рекомендована інтеграція:
K2 ERP ↓ Python integration service ↓ HTTP REST API ↓ Device Manager Вчасно.Каса ↓ ПРРО Вчасно.Каса / принтер / термінал
Мета інтеграції
Мета інтеграції — дати K2 ERP можливість:
- фіскалізувати чеки продажу;
- робити чеки повернення;
- відкривати та закривати зміну;
- формувати X-звіти та Z-звіти;
- працювати з оплатами;
- отримувати результат уже проведеної операції;
- безпечно повторювати запити при timeout або втраті звʼязку;
- працювати, коли ПРРО тимчасово переходить в офлайн-режим;
- логувати всі фіскальні операції у K2 ERP.
Короткий висновок
| Статус | Рішення | Коментар |
|---|---|---|
| Рекомендовано | Інтегрувати K2 ERP через локальний Device Manager API | Це офіційний сценарій для локальної інтеграції з Вчасно.Каса. |
| Рекомендовано | Писати окремий Python-сервіс-посередник | Не варто напряму вбудовувати HTTP-виклики Вчасно.Каса у бізнес-логіку K2 ERP. |
| Обовʼязково | Передавати унікальний `tag` для кожної фіскальної операції | Це основа транзакційності та захисту від дублювання чеків. |
| Важливо | Timeout для ПРРО — не менше 20 секунд | Потрібно врахувати можливий автоматичний перехід каси в офлайн. |
| Не робити | Не відправляти паралельні запити на одну й ту саму касу | Device Manager поверне помилку `1105`: пристрій зайнятий. |
| Не робити | Не створювати чек повторно з новим `tag`, якщо попередній запит завершився timeout | Треба повторити запит із тим самим `tag`, інакше можливий дубль чека. |
Компоненти інтеграції
| Компонент | Роль |
|---|---|
| K2 ERP | Джерело документів продажу, повернення, оплат, зміни та облікових даних. |
| Python integration service | Посередник між K2 ERP та Device Manager. Формує JSON, надсилає запити, обробляє відповіді, веде лог. |
| Device Manager | Локальний застосунок Вчасно.Каса, який приймає REST API запити. |
| ПРРО Вчасно.Каса | Фіскалізація чеків, відкриття/закриття зміни, звітність. |
| Принтер | Друк нефіскальних або службових документів, якщо використовується. |
| Банківський термінал | Оплата карткою, повернення, звірка, якщо підключено. |
Локальне API Device Manager
Після встановлення та запуску Device Manager піднімає локальний вебсервер.
За замовчуванням:
http://localhost:3939
Якщо K2 ERP або Python-сервіс знаходиться на іншому пристрої в локальній мережі:
http://{{dm_ip}}:3939
де `Шаблон:Dm ip` — IP-адреса пристрою, на якому встановлено Device Manager.
Вебінтерфейси Device Manager
| URL | Призначення |
|---|---|
http://localhost:3939/dm/vchasno-kasa/
|
Налаштування ПРРО Вчасно.Каса. |
http://localhost:3939/dm/
|
Налаштування POS-пристроїв: принтери, термінали, інші пристрої. |
Основні API endpoint-и
| Endpoint | Призначення |
|---|---|
/dm/execute
|
Основний endpoint для виконання фіскальних операцій по ПРРО. |
/dm/execute-prn
|
Виконання операцій друку. |
/dm/execute-pkg
|
Пакетний режим: ПРРО + термінал + принтер. |
Типи завдань
У запитах Device Manager використовуються три ключові поля:
| Поле | Опис |
|---|---|
device
|
Назва пристрою у Device Manager. Наприклад: K2_MAIN_KASA.
|
type
|
Тип завдання. |
task
|
Номер конкретного завдання. |
Значення поля type
| Значення | Тип | Призначення |
|---|---|---|
1
|
fiscal | Робота з ПРРО: чек, повернення, зміна, Z-звіт. |
2
|
doc | Друк документів або робота з принтером. |
3
|
pay | Робота з банківським терміналом. |
Основні fiscal task
| Task | Операція | Використання в K2 ERP |
|---|---|---|
0
|
Відкриття зміни | Початок роботи касира або торгової точки. |
1
|
Чек продажу | Продаж товарів або послуг. |
2
|
Чек повернення | Повернення товару або скасування продажу за процедурою повернення. |
3
|
Службове внесення | Внесення готівки в касу. |
4
|
Службова видача | Вилучення готівки з каси. |
11
|
Z-звіт | Закриття зміни. |
Рекомендована архітектура для K2 ERP
Загальна схема
K2 ERP ↓ Fiscal Integration Module ↓ Python service ↓ Device Manager REST API ↓ Вчасно.Каса
Чому потрібен окремий Python-сервіс
| Причина | Пояснення |
|---|---|
| Ізоляція інтеграції | K2 ERP не має напряму залежати від формату Device Manager API. |
| Транзакційність | Python-сервіс контролює `tag`, повтори, статуси та логування. |
| Черга запитів | Для однієї каси запити треба виконувати послідовно. |
| Обробка timeout | Python-сервіс може повторно перевірити чек за тим самим `tag`. |
| Офлайн-режим | Python-сервіс може коректно обробити затримки та статуси офлайн-роботи. |
| Масштабування | Можна підключити кілька кас, кожну з власною чергою. |
Встановлення Device Manager
Файли встановлення доступні на сторінці пакетів Device Manager.
Підтримуються:
- Windows x32 та x64;
- Debian-based Linux x86_64;
- Red Hat-based Linux x86_64;
- Arch Linux;
- Android.
| ОС | Підтримка | Коментар |
|---|---|---|
| Windows | x32 / x64 | Версії нижче Windows 7 SP1 або Server 2008 не підтримуються. |
| Debian Linux | x86_64 | Ubuntu, Debian, Mint. Нижче Ubuntu 18.04 не підтримується. |
| Red Hat Linux | x86_64 | CentOS, Fedora. Рекомендовано CentOS 7+. |
| Arch Linux | x86_64 | Потрібні додатково `curl` та `jq`. |
| Android | Android 6+ | Рекомендовано Android 10+. |
| ARM Linux | Не підтримується | AArch32 та AArch64 не підтримуються для Linux-пакетів. |
Початкове налаштування
Крок 1. Встановити Device Manager
- Завантажити пакет для потрібної ОС.
- Встановити застосунок.
- Запустити Device Manager.
- Перевірити, що вебсервер доступний на порту `3939`.
Перевірка з компʼютера, де встановлено Device Manager:
curl http://localhost:3939
Якщо K2 ERP або Python-сервіс працює на іншому сервері:
curl http://192.168.1.50:3939
Крок 2. Перевірити доступ до вебінтерфейсу
Відкрити у браузері:
http://localhost:3939/dm/vchasno-kasa/
або з іншого пристрою:
http://192.168.1.50:3939/dm/vchasno-kasa/
Крок 3. Додати тестову касу
У вебінтерфейсі Device Manager потрібно:
- додати тестову ПРРО-касу;
- вказати токен або необхідні параметри Вчасно.Каса;
- налаштувати ключ підпису;
- налаштувати податкові групи;
- налаштувати типи оплат;
- присвоїти касі зрозумілу назву.
Рекомендована назва тестової каси:
K2_TEST_KASA
Рекомендована назва робочої каси:
K2_MAIN_KASA
| Статус | Правило |
|---|---|
| Добре | Використовувати стабільні імена пристроїв: `K2_TEST_KASA`, `K2_MAIN_KASA`. |
| Погано | Називати пристрої випадковими назвами або змінювати назву після інтеграції. |
Транзакційність через tag
Для кожної фіскальної операції K2 ERP має передавати унікальний `tag`.
`tag` — це ідентифікатор операції, який дозволяє:
- повторно отримати результат уже проведеного чека;
- не створити дубль при timeout;
- звірити операції між K2 ERP та Device Manager;
- відновити стан після збою живлення або мережі.
Правило формування tag
Рекомендований формат:
K2:{document_type}:{document_id}:{operation_type}:{attempt_group}
Приклад:
K2:SALE:INV-000123:FISCAL:20260506
Або UUID:
9f1d9f9d-32fb-4d3d-bd71-6a1b2a7c5f7a
| Статус | Правило |
|---|---|
| Обовʼязково | Один документ K2 ERP = один стабільний `tag` для фіскалізації. |
| Обовʼязково | При повторі після timeout використовувати той самий `tag`. |
| Заборонено | Генерувати новий `tag` при повторній відправці того самого чека. |
Офлайн-режим Вчасно.Каса
Device Manager може певний час працювати з ПРРО в офлайн-режимі, якщо виникають проблеми з онлайн-фіскалізацією: недоступність ДПС, проблеми з АЦСК, проблеми з ключем або мережею.
Для K2 ERP це означає:
- не треба одразу вважати довший запит помилкою;
- timeout для ПРРО має бути не менше 20 секунд;
- відповідь Device Manager може затриматись через спробу автоматичного переходу в офлайн;
- потрібно зберігати `tag` і повторювати перевірку результату;
- потрібно логувати, чи чек був фіскалізований онлайн або офлайн, якщо така інформація є у відповіді.
Типові помилки офлайн-режиму
| Код | Значення | Дія в K2 ERP |
|---|---|---|
1052
|
ПРРО заборонено переходити в офлайн | Показати помилку касиру, звернутись до адміністратора. |
1053
|
Неможливо перейти в офлайн, бо ПРРО вже офлайн | Перевірити поточний стан ПРРО. |
1054
|
Неможливо перейти в онлайн, бо ПРРО вже онлайн | Не критична помилка стану, перевірити логіку переходу. |
1056
|
Відсутні офлайн-номери | Перевідкрити зміну онлайн або звернутись до підтримки. |
1062
|
ПРРО знаходиться в офлайн-режимі понад дозволений час | Дозволити тільки закриття зміни, звернутись до підтримки. |
1122
|
Закінчились офлайн-номери | Дочекатися реєстрації офлайн-чеків та переходу онлайн. |
Обмеження паралельної роботи
Device Manager може обробляти різні запити паралельно, але для одного конкретного пристрою операції виконуються синхронно.
Це означає:
- два одночасні запити на одну касу виконуватись не будуть;
- перший запит буде виконуватись;
- наступний запит може отримати помилку `1105`;
- для однієї каси у Python-сервісі потрібно зробити чергу.
Помилка 1105
res: 1105 errortxt: Пристрій зайнятий
| Статус | Рекомендація |
|---|---|
| Правильно | Робити окрему FIFO-чергу на кожну касу. |
| Можливо | При `1105` повторити запит через коротку паузу. |
| Неправильно | Відправляти всі чеки паралельно на одну касу. |
Таймаути
| Операція | Рекомендований timeout | Коментар |
|---|---|---|
| ПРРО: чек, зміна, звіти | 20 секунд або більше | Через можливий перехід в офлайн. |
| Принтер | 6 секунд або більше | Друк зазвичай швидкий. |
| Термінал | 310 секунд | Очікування картки та відповідь банку можуть тривати довше. |
| Пакетний режим | Сума timeout-ів | ПРРО + термінал + принтер. |
Python-клієнт для Device Manager
Встановлення залежностей
pip install requests pydantic
Базовий HTTP-клієнт
from __future__ import annotations
import logging
import time
import uuid
from dataclasses import dataclass
from typing import Any, Dict, Optional
import requests
logger = logging.getLogger(__name__)
class VchasnoKasaError(Exception):
pass
class VchasnoKasaTimeoutError(VchasnoKasaError):
pass
class VchasnoKasaDeviceBusyError(VchasnoKasaError):
pass
class VchasnoKasaBusinessError(VchasnoKasaError):
def __init__(self, response: Dict[str, Any]):
self.response = response
message = response.get("errortxt") or "Vchasno.Kasa business error"
super().__init__(message)
@dataclass
class DeviceManagerConfig:
base_url: str = "http://localhost:3939"
device: str = "K2_TEST_KASA"
fiscal_timeout: int = 25
terminal_timeout: int = 310
username: Optional[str] = None
password: Optional[str] = None
class VchasnoDeviceManagerClient:
def __init__(self, config: DeviceManagerConfig):
self.config = config
self.session = requests.Session()
if config.username and config.password:
self.session.auth = (config.username, config.password)
def _url(self, path: str) -> str:
return f"{self.config.base_url.rstrip('/')}{path}"
def execute(
self,
payload: Dict[str, Any],
timeout: Optional[int] = None,
) -> Dict[str, Any]:
url = self._url("/dm/execute")
request_timeout = timeout or self.config.fiscal_timeout
try:
response = self.session.post(
url,
json=payload,
timeout=request_timeout,
)
response.raise_for_status()
data = response.json()
except requests.Timeout as exc:
raise VchasnoKasaTimeoutError(
f"Device Manager timeout after {request_timeout}s"
) from exc
except requests.RequestException as exc:
raise VchasnoKasaError(f"Device Manager HTTP error: {exc}") from exc
self._validate_response(data)
return data
def _validate_response(self, data: Dict[str, Any]) -> None:
res = data.get("res")
task_status = data.get("task_status")
if res == 1105:
raise VchasnoKasaDeviceBusyError(data.get("errortxt", "Пристрій зайнятий"))
if task_status == 3 or (isinstance(res, int) and res > 0):
raise VchasnoKasaBusinessError(data)
def find_by_tag(self, tag: str) -> Dict[str, Any]:
payload = {
"device": self.config.device,
"tag": tag,
}
return self.execute(payload)
def fiscal_request(
self,
task: int,
tag: str,
fiscal_data: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
payload: Dict[str, Any] = {
"device": self.config.device,
"type": 1,
"task": task,
"tag": tag,
}
if fiscal_data:
payload.update(fiscal_data)
return self.execute(payload)
Відкриття зміни
Python-метод
def open_shift(client: VchasnoDeviceManagerClient, k2_shift_id: str) -> Dict[str, Any]:
tag = f"K2:SHIFT:{k2_shift_id}:OPEN"
return client.fiscal_request(
task=0,
tag=tag,
fiscal_data={
# Тут можуть бути додаткові параметри,
# якщо вони потрібні згідно з актуальною документацією Вчасно.Каса.
},
)
Приклад використання
config = DeviceManagerConfig(
base_url="http://localhost:3939",
device="K2_TEST_KASA",
)
client = VchasnoDeviceManagerClient(config)
result = open_shift(client, k2_shift_id="SHIFT-2026-05-06-001")
print(result)
Фіскалізація чека продажу
Мапінг документа K2 ERP у чек Вчасно.Каса
| K2 ERP | Device Manager / Вчасно.Каса |
|---|---|
| Номер документа | `tag`, службовий номер або коментар |
| Дата документа | Дата операції |
| Каса / торгова точка | `device` |
| Товар | Рядок чека |
| Кількість | Кількість у рядку чека |
| Ціна | Ціна одиниці |
| Знижка | Знижка рядка або чека |
| Податкова група | Податкова група ПРРО |
| Оплата | Тип оплати: готівка, картка, інше |
Шаблон payload для продажу
Увага: точні назви полів для товарів, оплат, податкових груп і знижок потрібно звірити з актуальною сторінкою `API для роботи з ПРРО`. Нижче наведено інтеграційний шаблон, який показує структуру на рівні K2 ERP → Python.
def build_sale_payload(k2_document: dict) -> dict:
"""
Перетворює документ продажу K2 ERP у payload для Device Manager.
Важливо:
- назви вкладених полів fiscal/check/goods/payments потрібно звірити
з актуальною документацією Вчасно.Каса;
- стабільний tag має зберігатися в K2 ERP.
"""
tag = f"K2:SALE:{k2_document['id']}"
goods = []
for line in k2_document["lines"]:
goods.append(
{
"name": line["name"],
"code": line.get("sku") or line["id"],
"cnt": float(line["quantity"]),
"price": round(float(line["price"]), 2),
"sum": round(float(line["quantity"]) * float(line["price"]), 2),
"tax": line.get("tax_group", 1),
}
)
payments = []
for payment in k2_document["payments"]:
payments.append(
{
"type": payment["type"], # cash/card/etc — згідно з налаштуваннями ПРРО
"sum": round(float(payment["amount"]), 2),
}
)
return {
"device": k2_document["device"],
"type": 1,
"task": 1,
"tag": tag,
# Назву цього блоку потрібно звірити з актуальним fiscal API.
"fiscal": {
"goods": goods,
"payments": payments,
"comment": f"K2 ERP document {k2_document['number']}",
},
}
Відправка чека продажу
def fiscalize_sale(
client: VchasnoDeviceManagerClient,
k2_document: dict,
) -> dict:
payload = build_sale_payload(k2_document)
try:
return client.execute(payload, timeout=client.config.fiscal_timeout)
except VchasnoKasaTimeoutError:
# Якщо не отримали відповідь, не створюємо новий чек.
# Перевіряємо результат по тому самому tag.
return client.find_by_tag(payload["tag"])
except VchasnoKasaDeviceBusyError:
# Пристрій зайнятий. У production краще ставити документ у чергу.
time.sleep(2)
return client.execute(payload, timeout=client.config.fiscal_timeout)
Приклад документа K2 ERP
k2_document = {
"id": "INV-000123",
"number": "INV-000123",
"device": "K2_TEST_KASA",
"lines": [
{
"id": "ITEM-001",
"sku": "482000000001",
"name": "Кава",
"quantity": 2,
"price": 50.00,
"tax_group": 1,
},
{
"id": "ITEM-002",
"sku": "482000000002",
"name": "Круасан",
"quantity": 1,
"price": 75.00,
"tax_group": 1,
},
],
"payments": [
{
"type": "cash",
"amount": 175.00,
}
],
}
Перевірка чека за tag
Якщо K2 ERP не отримала відповідь від Device Manager, потрібно не створювати новий чек, а перевірити попередній результат за `tag`.
def check_receipt_result(
client: VchasnoDeviceManagerClient,
device: str,
tag: str,
) -> dict:
payload = {
"device": device,
"tag": tag,
}
return client.execute(payload)
Логіка K2 ERP при timeout
1. K2 ERP створила документ продажу. 2. Python-сервіс сформував tag. 3. Python-сервіс відправив чек у Device Manager. 4. Виник timeout. 5. Python-сервіс НЕ створює новий tag. 6. Python-сервіс виконує пошук за тим самим tag. 7. Якщо чек знайдено — записує фіскальний результат у K2 ERP. 8. Якщо чек не знайдено — вирішує, чи повторювати відправку з тим самим tag.
Закриття зміни / Z-звіт
def close_shift(client: VchasnoDeviceManagerClient, k2_shift_id: str) -> dict:
tag = f"K2:SHIFT:{k2_shift_id}:ZREPORT"
return client.fiscal_request(
task=11,
tag=tag,
)
| Статус | Правило |
|---|---|
| Обовʼязково | Z-звіт має бути повʼязаний із конкретною зміною K2 ERP. |
| Обовʼязково | Результат Z-звіту треба зберігати в K2 ERP. |
| Не робити | Не закривати зміну повторно з новим tag після timeout. |
Черга операцій для однієї каси
Оскільки Device Manager не виконує паралельні операції на одному ПРРО, у Python-сервісі потрібно зробити чергу.
Спрощена in-memory черга
import queue
import threading
from typing import Callable
class FiscalQueue:
def __init__(self):
self._queue: queue.Queue[Callable[[], None]] = queue.Queue()
self._worker = threading.Thread(target=self._run, daemon=True)
self._worker.start()
def put(self, job: Callable[[], None]) -> None:
self._queue.put(job)
def _run(self) -> None:
while True:
job = self._queue.get()
try:
job()
except Exception:
logger.exception("Fiscal job failed")
finally:
self._queue.task_done()
Production-рекомендація
Для production краще використовувати не in-memory чергу, а одну з таких схем:
- таблиця черги в базі K2 ERP;
- окрема таблиця `fiscal_queue`;
- Redis Queue;
- Celery;
- RabbitMQ;
- Kafka, якщо вже використовується в інфраструктурі.
Таблиця fiscal_operations у K2 ERP
Рекомендовано створити окрему таблицю для контролю фіскальних операцій.
| Поле | Тип | Опис |
|---|---|---|
| id | UUID / bigint | Внутрішній ID операції. |
| k2_document_id | string | ID документа K2 ERP. |
| operation_type | string | sale, return, open_shift, z_report. |
| device | string | Назва каси у Device Manager. |
| tag | string | Унікальний tag операції. |
| status | string | new, sent, success, failed, unknown, retry. |
| request_json | json | Запит до Device Manager. |
| response_json | json | Відповідь Device Manager. |
| res | int | Код результату Device Manager. |
| res_action | int | Рекомендована дія за відповіддю. |
| errortxt | text | Текст помилки. |
| fiscal_number | string | Фіскальний номер чека, якщо є. |
| created_at | datetime | Дата створення. |
| updated_at | datetime | Дата оновлення. |
Обробка task_status
Device Manager повертає `task_status`.
| task_status | Значення | Дія в K2 ERP |
|---|---|---|
| 1 | Завдання виконано | Позначити операцію як успішну. |
| 2 | Знайдено раніше виконану операцію за tag | Позначити як успішну, не створювати новий чек. |
| 3 | Помилка | Аналізувати `res` і `res_action`. |
Обробка res_action
| res_action | Значення | Дія |
|---|---|---|
| 0 | Завдання пройшло | Зберегти результат. |
| 1 | Помилка, можна повторити | Повторити з тим самим `tag`. |
| 2 | Колізія | Повторити останнє завдання з тим самим `tag`. |
| 3 | Помилка даних або стану | Не повторювати автоматично. Потрібне виправлення. |
Типові помилки та реакція K2 ERP
| Код | Помилка | Реакція |
|---|---|---|
1091
|
Пристрій не знайдено | Перевірити назву `device` у K2 ERP та Device Manager. |
1105
|
Пристрій зайнятий | Повторити через чергу або зачекати завершення попередньої операції. |
1082
|
Неможливо зберегти дані до БД | Перевірити права запуску Device Manager або стан бази. |
1058
|
Чек з tag не знайдено | Якщо це перевірка після timeout — можна повторити запит із тим самим tag. |
1092
|
Зміна закрита | Відкрити зміну перед продажем. |
1121
|
Дата на пристрої не співпадає з сервером | Виправити час на пристрої та перезапустити. |
1125
|
Застаріла версія Device Manager | Оновити Device Manager. |
2000
|
Невірний JSON | Виправити payload у Python-сервісі. |
2001
|
Невірний токен Вчасно.Каса | Оновити токен у налаштуваннях ПРРО. |
Логування в Python
import json
import logging
from datetime import datetime
logging.basicConfig(
filename="k2_vchasno_kasa.log",
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
)
def log_operation(
operation_type: str,
tag: str,
request_payload: dict,
response_payload: dict | None = None,
error: Exception | None = None,
) -> None:
log_data = {
"operation_type": operation_type,
"tag": tag,
"request": request_payload,
"response": response_payload,
"error": str(error) if error else None,
"logged_at": datetime.utcnow().isoformat(),
}
if error:
logging.error(json.dumps(log_data, ensure_ascii=False))
else:
logging.info(json.dumps(log_data, ensure_ascii=False))
Тестування інтеграції
Етап 1. Перевірка доступності Device Manager
import requests
response = requests.get("http://localhost:3939", timeout=5)
print(response.status_code)
print(response.text[:500])
Етап 2. Перевірка доступу до API
client = VchasnoDeviceManagerClient(
DeviceManagerConfig(
base_url="http://localhost:3939",
device="K2_TEST_KASA",
)
)
try:
result = client.find_by_tag("TEST-NOT-EXISTING-TAG")
print(result)
except VchasnoKasaBusinessError as exc:
print("Expected business response:", exc.response)
Етап 3. Тест відкриття зміни
result = open_shift(client, "TEST-SHIFT-001")
print(result)
Етап 4. Тест продажу
result = fiscalize_sale(client, k2_document)
print(result)
Етап 5. Тест повтору з тим самим tag
- Відправити чек.
- Зберегти `tag`.
- Повторно виконати запит з тим самим `tag`.
- Переконатися, що Device Manager не створює дубль, а повертає результат попереднього чека.
Етап 6. Тест timeout
У Python-сервісі штучно поставити малий timeout:
try:
client.execute(payload, timeout=1)
except VchasnoKasaTimeoutError:
result = client.find_by_tag(payload["tag"])
print(result)
Етап 7. Тест зайнятого пристрою
- Відправити два запити одночасно на одну касу.
- Переконатися, що один із них може отримати `1105`.
- Перевірити, що черга K2 ERP повторює операцію пізніше.
Тестування офлайн-режиму
Офлайн-режим потрібно тестувати обережно та тільки на тестовій касі.
Що перевірити
| Тест | Очікуваний результат |
|---|---|
| Відсутність інтернету | Device Manager або ПРРО переходить у дозволений офлайн-режим, якщо це налаштовано. |
| Фіскалізація в офлайн | Чек створюється з офлайн-ознаками, якщо офлайн дозволений і є офлайн-номери. |
| Відсутні офлайн-номери | Повертається помилка, операція не вважається успішною. |
| Повернення онлайн | Офлайн-чеки мають бути синхронізовані. |
| Timeout під час переходу в офлайн | K2 ERP повторно перевіряє результат за тим самим `tag`. |
| Статус | Правило |
|---|---|
| Правильно | Зберігати всі офлайн-операції в K2 ERP зі статусом синхронізації. |
| Контроль | Перевіряти помилки `1052`, `1056`, `1062`, `1122`. |
| Небезпечно | Вважати timeout фіскалізації невдалим без перевірки за `tag`. |
Рекомендована структура Python-проєкту
k2_vchasno_integration/
├── app.py
├── config.py
├── device_manager.py
├── fiscal_service.py
├── mapper.py
├── queue.py
├── models.py
├── errors.py
├── logging_config.py
└── tests/
├── test_mapper.py
├── test_tag.py
├── test_timeout.py
├── test_response_handling.py
└── test_fiscal_flow.py
Призначення модулів
| Файл | Призначення |
|---|---|
| `device_manager.py` | HTTP-клієнт для Device Manager. |
| `fiscal_service.py` | Бізнес-логіка фіскалізації, повторів, перевірки за tag. |
| `mapper.py` | Перетворення документів K2 ERP у JSON для Вчасно.Каса. |
| `queue.py` | Черга операцій по кожній касі. |
| `models.py` | DTO / Pydantic-моделі. |
| `errors.py` | Класи помилок інтеграції. |
| `tests/` | Автоматизовані тести. |
Production checklist
| Статус | Перевірка |
|---|---|
| □ | Device Manager встановлено. |
| □ | Device Manager доступний на `http://localhost:3939` або по IP. |
| □ | Тестову касу додано. |
| □ | Робочу касу додано. |
| □ | Назва `device` у K2 ERP збігається з назвою в Device Manager. |
| □ | Для кожного документа K2 ERP формується стабільний `tag`. |
| □ | Timeout для ПРРО не менше 20 секунд. |
| □ | Реалізовано повторну перевірку за `tag` після timeout. |
| □ | Реалізовано чергу на кожну касу. |
| □ | Обробляється помилка `1105`. |
| □ | Обробляються `task_status`, `res`, `res_action`. |
| □ | Протестовано офлайн-режим. |
| □ | Протестовано втрату інтернету. |
| □ | Протестовано timeout. |
| □ | Протестовано повторну відправку з тим самим `tag`. |
| □ | Немає повторної фіскалізації з новим `tag` після timeout. |
| □ | Немає паралельних запитів на одну касу. |
Рекомендований алгоритм фіскалізації в K2 ERP
1. Користувач проводить документ продажу в K2 ERP. 2. K2 ERP створює запис у fiscal_operations. 3. Python-сервіс формує стабільний tag. 4. Python-сервіс формує JSON для Device Manager. 5. Запит потрапляє в чергу конкретної каси. 6. Python-сервіс відправляє POST /dm/execute. 7. Якщо відповідь успішна: 7.1. Зберегти фіскальні реквізити. 7.2. Позначити документ як фіскалізований. 8. Якщо timeout: 8.1. Виконати пошук за тим самим tag. 8.2. Якщо чек знайдено — зберегти результат. 8.3. Якщо чек не знайдено — повторити з тим самим tag або поставити в retry. 9. Якщо res_action = 1: 9.1. Повторити з тим самим tag. 10. Якщо res_action = 2: 10.1. Повторити останню операцію з тим самим tag. 11. Якщо res_action = 3: 11.1. Не повторювати автоматично. 11.2. Показати помилку користувачу або адміністратору.
Висновок
Для інтеграції K2 ERP з Вчасно.Каса рекомендовано використовувати Device Manager як локальний API-шлюз, а між K2 ERP та Device Manager створити окремий Python-сервіс.
Ключові вимоги до інтеграції:
- усі фіскальні операції мають мати стабільний `tag`;
- timeout не можна трактувати як однозначну помилку;
- після timeout потрібно перевіряти результат за тим самим `tag`;
- на одну касу має бути одна послідовна черга операцій;
- потрібно обробляти `task_status`, `res`, `res_action`;
- офлайн-режим потрібно підтримувати на рівні логіки повторів, статусів і звірки;
- усі запити та відповіді потрібно логувати в K2 ERP.