2036 lines
63 KiB
Python
2036 lines
63 KiB
Python
# -----------------------------------------------------
|
||
# API Calls
|
||
# -----------------------------------------------------
|
||
import logging
|
||
import data
|
||
import requests
|
||
import time
|
||
import threading
|
||
import requests
|
||
|
||
from konstanty import *
|
||
from pydantic import BaseModel, SecretStr, TypeAdapter
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_api_session: requests.Session | None = None
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_hb_session: requests.Session | None = None
|
||
_hb_thread: threading.Thread | None = None
|
||
_hb_stop_event = threading.Event()
|
||
|
||
|
||
def get_hb_session() -> requests.Session:
|
||
global _hb_session
|
||
if _hb_session is None:
|
||
_hb_session = requests.Session()
|
||
return _hb_session
|
||
|
||
def get_api_session() -> requests.Session:
|
||
global _api_session
|
||
if _api_session is None:
|
||
_api_session = requests.Session()
|
||
return _api_session
|
||
|
||
class ApiContext(BaseModel):
|
||
user: str # uživatel pokladny
|
||
base_url: str
|
||
refresh_url: str
|
||
client_id: str # číslo terminálu
|
||
id_kas: str # číslo pokladny
|
||
username: str # jméno zakázky
|
||
password: SecretStr # heslo zakázky
|
||
token: str = ""
|
||
refresh_token: str = ""
|
||
#debug: bool = True
|
||
|
||
# --- heartbeat
|
||
def start_heartbeat(ctx: ApiContext, interval: int = HEART_BEAT):
|
||
global _hb_thread
|
||
if _hb_thread and _hb_thread.is_alive():
|
||
logger.info("Heartbeat already running")
|
||
return
|
||
_hb_stop_event.clear()
|
||
hb_session = get_hb_session()
|
||
def loop():
|
||
logger.info("Heartbeat thread started")
|
||
while not _hb_stop_event.is_set():
|
||
if not ctx.token:
|
||
time.sleep(interval)
|
||
continue
|
||
try:
|
||
ok, _, new_token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/heartbeat/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
session=hb_session,
|
||
params={"id_kas": ctx.id_kas},
|
||
)
|
||
if new_token:
|
||
ctx.token = new_token
|
||
except Exception as e:
|
||
logger.warning(f"Heartbeat failed: {e}")
|
||
time.sleep(interval)
|
||
logger.info("Heartbeat thread stopped")
|
||
_hb_thread = threading.Thread(
|
||
target=loop,
|
||
name="heartbeat",
|
||
daemon=True,
|
||
)
|
||
_hb_thread.start()
|
||
|
||
def stop_heartbeat(ctx: ApiContext | None = None):
|
||
global _hb_thread, _hb_session
|
||
logger.info("Stopping heartbeat")
|
||
_hb_stop_event.set()
|
||
if _hb_thread:
|
||
_hb_thread.join(timeout=2)
|
||
_hb_thread = None
|
||
if _hb_session:
|
||
try:
|
||
_hb_session.close()
|
||
except Exception:
|
||
pass
|
||
_hb_session = None
|
||
|
||
|
||
# --- nacteni setupu
|
||
def load_setup_API(ctx: ApiContext) -> data.PosSetup:
|
||
ok, response, token = call_api(
|
||
method="GET", base_url=ctx.base_url, endpoint="/setup/",
|
||
client_id=ctx.client_id, token=ctx.token, refresh_token=ctx.refresh_token,
|
||
params={"id_kas": ctx.id_kas}, )
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Setup pro pokladnu {ctx.id_kas} nebyl nacten")
|
||
logger.debug(f"\nSetup OK\n{response}")
|
||
setup = data.PosSetup.model_validate(response)
|
||
return setup
|
||
|
||
def load_platby_API(ctx: ApiContext) -> list[data.PaymentType]:
|
||
ok, response, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/platby/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params={"id_kas": ctx.id_kas},
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Platby pokladne {ctx.id_kas} neboli nacitane\n{response}")
|
||
return [
|
||
data.PaymentType.model_validate(item)
|
||
for item in response
|
||
]
|
||
|
||
def save_platby_API(ctx: ApiContext, platby: list[data.PaymentType]) -> dict:
|
||
ok, response, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/platby/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params={"id_kas": ctx.id_kas},
|
||
json=[
|
||
data.PaymentType.model_validate(p).model_dump()
|
||
for p in platby
|
||
],
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Platby pokladne {ctx.id_kas} neboli ulozene\n{response}")
|
||
return response
|
||
|
||
def load_fooddat_API(ctx: ApiContext) -> list[data.FoodDat]:
|
||
ok, response, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/fooddat/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Fooddat nebol nacitany\n{response}")
|
||
return [data.FoodDat.model_validate(item) for item in response]
|
||
|
||
def save_fooddat_API(ctx: ApiContext, items: list[data.FoodDat]) -> dict:
|
||
ok, response, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/fooddat/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
json=[
|
||
data.FoodDat.model_validate(item).model_dump()
|
||
for item in items
|
||
],
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Fooddat nebol ulozeny\n{response}")
|
||
return response
|
||
|
||
def load_bankterms_API(ctx: ApiContext) -> list[data.BankTerm]:
|
||
ok, response, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/bankterm/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Bankove terminaly neboli nacitane\n{response}")
|
||
return [
|
||
data.BankTerm.model_validate(item)
|
||
for item in response
|
||
]
|
||
|
||
def create_print_job_API(ctx: ApiContext, job: data.PrintJobCreate) -> data.PrintJob:
|
||
payload = data.PrintJobCreate.model_validate(job).model_dump(mode="json")
|
||
ok, response, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/print/jobs/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
json=payload,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Tlacovy job nebol vytvoreny\n{response}")
|
||
return data.PrintJob.model_validate(response)
|
||
|
||
def load_print_jobs_API(
|
||
ctx: ApiContext,
|
||
status: str | None = None,
|
||
printer_no: str | None = None,
|
||
limit: int = 100,
|
||
) -> list[data.PrintJob]:
|
||
params = {
|
||
"id_kas": ctx.id_kas,
|
||
"limit": limit,
|
||
}
|
||
if status:
|
||
params["status"] = status
|
||
if printer_no:
|
||
params["printer_no"] = printer_no
|
||
ok, response, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/print/jobs/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params=params,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Tlacove joby neboli nacitane\n{response}")
|
||
return [data.PrintJob.model_validate(item) for item in response]
|
||
|
||
def create_kitchen_print_jobs_API(
|
||
ctx: ApiContext,
|
||
ucet: data.Ucet,
|
||
kind: str = "bon",
|
||
room_name: str = "",
|
||
pos_name: str = "",
|
||
required: bool = True,
|
||
priority: int = 50,
|
||
) -> list[data.PrintJob]:
|
||
payload = data.KitchenPrintRequest(
|
||
id_kas=ctx.id_kas,
|
||
kind=kind,
|
||
ucet=ucet,
|
||
room_name=room_name,
|
||
pos_name=pos_name,
|
||
required=required,
|
||
priority=priority,
|
||
).model_dump(mode="json")
|
||
ok, response, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/print/kitchen/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
json=payload,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Kuchynsky tlacovy job nebol vytvoreny\n{response}")
|
||
return [data.PrintJob.model_validate(item) for item in response]
|
||
|
||
def create_receipt_print_jobs_API(
|
||
ctx: ApiContext,
|
||
ucet: data.Ucet,
|
||
kind: str = "receipt",
|
||
printer_no: str = "",
|
||
title: str = "",
|
||
pos_name: str = "",
|
||
headers: list[str] | None = None,
|
||
footers: list[str] | None = None,
|
||
required: bool = False,
|
||
priority: int = 40,
|
||
copies: int = 1,
|
||
) -> list[data.PrintJob]:
|
||
payload = data.ReceiptPrintRequest(
|
||
id_kas=ctx.id_kas,
|
||
kind=kind,
|
||
ucet=ucet,
|
||
printer_no=printer_no,
|
||
title=title,
|
||
pos_name=pos_name,
|
||
headers=headers or [],
|
||
footers=footers or [],
|
||
required=required,
|
||
priority=priority,
|
||
copies=copies,
|
||
).model_dump(mode="json")
|
||
ok, response, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/print/receipt/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
json=payload,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Tlacovy job uctu nebol vytvoreny\n{response}")
|
||
return [data.PrintJob.model_validate(item) for item in response]
|
||
|
||
def create_closure_print_jobs_API(
|
||
ctx: ApiContext,
|
||
text: str,
|
||
printer_no: str,
|
||
clsrep_no: str | None = None,
|
||
kind: str = "closure",
|
||
title: str = "Uzavierka",
|
||
required: bool = False,
|
||
priority: int = 35,
|
||
copies: int = 1,
|
||
) -> list[data.PrintJob]:
|
||
payload = data.ClosurePrintRequest(
|
||
id_kas=ctx.id_kas,
|
||
kind=kind,
|
||
printer_no=printer_no,
|
||
clsrep_no=clsrep_no,
|
||
title=title,
|
||
text=text,
|
||
required=required,
|
||
priority=priority,
|
||
copies=copies,
|
||
).model_dump(mode="json")
|
||
ok, response, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/print/closure/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
json=payload,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Tlacovy job uzavierky nebol vytvoreny\n{response}")
|
||
return [data.PrintJob.model_validate(item) for item in response]
|
||
|
||
def load_usage_report_API(
|
||
ctx: ApiContext,
|
||
mode: str = "current",
|
||
date_from: str = "",
|
||
date_to: str = "",
|
||
days_back: int = 0,
|
||
) -> data.UsageReportOut:
|
||
params = {
|
||
"id_kas": ctx.id_kas,
|
||
"mode": mode,
|
||
}
|
||
if date_from:
|
||
params["date_from"] = date_from
|
||
if date_to:
|
||
params["date_to"] = date_to
|
||
if days_back:
|
||
params["days_back"] = days_back
|
||
ok, response, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/usage/report/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params=params,
|
||
timeout=60,
|
||
retries=0,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Nacitanie prezerania spotreby zlyhalo\n{response}")
|
||
return data.UsageReportOut.model_validate(response)
|
||
|
||
def render_receipt_preview_API(
|
||
ctx: ApiContext,
|
||
ucet: data.Ucet,
|
||
kind: str = "receipt",
|
||
printer_no: str = "",
|
||
title: str = "",
|
||
pos_name: str = "",
|
||
headers: list[str] | None = None,
|
||
footers: list[str] | None = None,
|
||
) -> data.ReceiptPrintPreviewOut:
|
||
payload = data.ReceiptPrintRequest(
|
||
id_kas=ctx.id_kas,
|
||
kind=kind,
|
||
ucet=ucet,
|
||
printer_no=printer_no,
|
||
title=title,
|
||
pos_name=pos_name,
|
||
headers=headers or [],
|
||
footers=footers or [],
|
||
copies=1,
|
||
).model_dump(mode="json")
|
||
ok, response, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/print/receipt/preview/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
json=payload,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Nahlad uctu nebol vyrenderovany\n{response}")
|
||
return data.ReceiptPrintPreviewOut.model_validate(response)
|
||
|
||
def print_fiscal_receipt_API(
|
||
ctx: ApiContext,
|
||
ucet: data.Ucet,
|
||
printer_no: str = "",
|
||
title: str = "",
|
||
pos_name: str = "",
|
||
headers: list[str] | None = None,
|
||
footers: list[str] | None = None,
|
||
) -> data.FiscalReceiptPrintOut:
|
||
payload = data.FiscalReceiptPrintRequest(
|
||
id_kas=ctx.id_kas,
|
||
ucet=ucet,
|
||
printer_no=printer_no,
|
||
title=title,
|
||
pos_name=pos_name,
|
||
headers=headers or [],
|
||
footers=footers or [],
|
||
).model_dump(mode="json")
|
||
ok, response, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/print/fiscal/receipt/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
json=payload,
|
||
timeout=420,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Fiskalny doklad nebol vytlaceny\n{response}")
|
||
return data.FiscalReceiptPrintOut.model_validate(response)
|
||
|
||
def print_fiscal_receipt_copy_API(
|
||
ctx: ApiContext,
|
||
ucet: data.Ucet,
|
||
printer_no: str = "",
|
||
bill_id: str = "",
|
||
) -> data.FiscalReceiptPrintOut:
|
||
payload = data.FiscalReceiptCopyRequest(
|
||
id_kas=ctx.id_kas,
|
||
ucet=ucet,
|
||
printer_no=printer_no,
|
||
bill_id=bill_id,
|
||
).model_dump(mode="json")
|
||
ok, response, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/print/fiscal/receipt/copy/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
json=payload,
|
||
timeout=420,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Fiskalna kopia dokladu nebola vytlacena\n{response}")
|
||
return data.FiscalReceiptPrintOut.model_validate(response)
|
||
|
||
def print_fiscal_cash_operation_API(
|
||
ctx: ApiContext,
|
||
operation: str,
|
||
amount: float,
|
||
payment: data.PaymentType,
|
||
printer_no: str = "",
|
||
author: str = "",
|
||
pos_name: str = "",
|
||
) -> data.FiscalCashOperationOut:
|
||
payload = data.FiscalCashOperationRequest(
|
||
id_kas=ctx.id_kas,
|
||
operation=operation,
|
||
amount=amount,
|
||
payment=payment,
|
||
printer_no=printer_no,
|
||
author=author,
|
||
pos_name=pos_name,
|
||
).model_dump(mode="json")
|
||
ok, response, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/print/fiscal/cash-operation/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
json=payload,
|
||
timeout=180,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Fiskalny vklad/vyber nebol vykonany\n{response}")
|
||
return data.FiscalCashOperationOut.model_validate(response)
|
||
|
||
def claim_print_jobs_API(
|
||
ctx: ApiContext,
|
||
agent_id: str,
|
||
printers: list[str] | None = None,
|
||
limit: int = 10,
|
||
) -> list[data.PrintJob]:
|
||
payload = data.PrintJobClaimRequest(
|
||
id_kas=ctx.id_kas,
|
||
agent_id=agent_id,
|
||
printers=printers or [],
|
||
limit=limit,
|
||
).model_dump(mode="json")
|
||
ok, response, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/print/jobs/claim/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
json=payload,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Tlacove joby neboli pridelene agentovi\n{response}")
|
||
return [data.PrintJob.model_validate(item) for item in response]
|
||
|
||
def update_print_job_status_API(
|
||
ctx: ApiContext,
|
||
job_id: int,
|
||
status: str,
|
||
result: dict | None = None,
|
||
error: str = "",
|
||
) -> data.PrintJob:
|
||
payload = data.PrintJobStatusUpdate(
|
||
status=status,
|
||
result=result or {},
|
||
error=error or "",
|
||
).model_dump(mode="json")
|
||
ok, response, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint=f"/print/jobs/{job_id}/status",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
json=payload,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Stav tlacoveho jobu nebol ulozeny\n{response}")
|
||
return data.PrintJob.model_validate(response)
|
||
|
||
def retry_print_job_API(ctx: ApiContext, job_id: int) -> data.PrintJob:
|
||
ok, response, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint=f"/print/jobs/{int(job_id)}/retry",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Tlacovy job nebol restartovany\n{response}")
|
||
return data.PrintJob.model_validate(response)
|
||
|
||
def load_fiscal_printer_status_API(
|
||
ctx: ApiContext,
|
||
printer_no: str,
|
||
timeout: float = 10.0,
|
||
) -> data.PrinterStatusOut:
|
||
ok, response, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/print/fiscal/status/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params={
|
||
"id_kas": ctx.id_kas,
|
||
"printer_no": printer_no,
|
||
"timeout": timeout,
|
||
},
|
||
timeout=max(float(timeout or 10.0) + 5.0, 15.0),
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Stav fiskalnej tlaciarne nebol nacitany\n{response}")
|
||
return data.PrinterStatusOut.model_validate(response)
|
||
|
||
def process_local_print_jobs_API(
|
||
ctx: ApiContext,
|
||
agent_id: str,
|
||
printers: list[str] | None = None,
|
||
limit: int = 10,
|
||
timeout: float = 10.0,
|
||
) -> list[data.PrintJob]:
|
||
payload = data.PrintJobClaimRequest(
|
||
id_kas=ctx.id_kas,
|
||
agent_id=agent_id,
|
||
printers=printers or [],
|
||
limit=limit,
|
||
).model_dump(mode="json")
|
||
ok, response, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/print/jobs/process-local/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params={"timeout": timeout},
|
||
json=payload,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Tlacove joby neboli spracovane\n{response}")
|
||
return [data.PrintJob.model_validate(item) for item in response]
|
||
|
||
def load_printer_status_API(ctx: ApiContext) -> list[data.PrinterStatusOut]:
|
||
ok, response, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/print/printers/status/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params={"id_kas": ctx.id_kas},
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Stav tlaciarni nebol nacitany\n{response}")
|
||
return [data.PrinterStatusOut.model_validate(item) for item in response]
|
||
|
||
def load_print_worker_diagnostics_API(
|
||
ctx: ApiContext,
|
||
id_kas: str | None = None,
|
||
limit: int = 100,
|
||
) -> dict:
|
||
params = {"limit": limit}
|
||
if id_kas:
|
||
params["id_kas"] = id_kas
|
||
ok, response, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/print/worker/diagnostics/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params=params,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Diagnostika tlace nebola nacitana\n{response}")
|
||
return response if isinstance(response, dict) else {}
|
||
|
||
def save_printer_status_API(
|
||
ctx: ApiContext,
|
||
status: data.PrinterStatusIn,
|
||
) -> data.PrinterStatusOut:
|
||
payload = data.PrinterStatusIn.model_validate(status).model_dump(mode="json")
|
||
ok, response, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/print/printers/status/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
json=payload,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Stav tlaciarne nebol ulozeny\n{response}")
|
||
return data.PrinterStatusOut.model_validate(response)
|
||
|
||
def load_uvery_API(
|
||
ctx: ApiContext,
|
||
q: str = "",
|
||
limit: int = 2000,
|
||
) -> list[data.UverFirma]:
|
||
ok, response, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/uvery/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params={
|
||
"q": q,
|
||
"limit": limit,
|
||
},
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Uvery neboli nacitane\n{response}")
|
||
return [
|
||
data.UverFirma.model_validate(item)
|
||
for item in response
|
||
]
|
||
|
||
def save_uver_API(ctx: ApiContext, firma: data.UverFirma) -> data.UverFirma:
|
||
payload = data.UverFirma.model_validate(firma).model_dump()
|
||
ok, response, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/uvery/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
json=payload,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Uverova firma nebola ulozena\n{response}")
|
||
return data.UverFirma.model_validate(response)
|
||
|
||
def load_hotel_receptions_API(ctx: ApiContext) -> list[data.HotelReception]:
|
||
ok, response, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/hotel/receptions/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Recepcie neboli nacitane\n{response}")
|
||
return [
|
||
data.HotelReception.model_validate(item)
|
||
for item in response
|
||
]
|
||
|
||
def load_hotel_rooms_API(
|
||
ctx: ApiContext,
|
||
reception_id: int,
|
||
) -> data.HotelRoomsResponse:
|
||
ok, response, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/hotel/rooms/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params={
|
||
"reception_id": reception_id,
|
||
"id_kas": ctx.id_kas,
|
||
},
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Izby recepcie neboli nacitane\n{response}")
|
||
return data.HotelRoomsResponse.model_validate(response)
|
||
|
||
def load_hotel_guests_API(
|
||
ctx: ApiContext,
|
||
reception_id: int,
|
||
room_id: str = "",
|
||
room_code: str = "",
|
||
account_id: str = "",
|
||
) -> list[data.HotelGuest]:
|
||
ok, response, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/hotel/guests/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params={
|
||
"reception_id": reception_id,
|
||
"id_kas": ctx.id_kas,
|
||
"room_id": room_id,
|
||
"room_code": room_code,
|
||
"account_id": account_id,
|
||
},
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Hostia izby neboli nacitani\n{response}")
|
||
return [
|
||
data.HotelGuest.model_validate(item)
|
||
for item in response
|
||
]
|
||
|
||
def check_hotel_card_API(
|
||
ctx: ApiContext,
|
||
reception_id: int,
|
||
card_code: str,
|
||
) -> data.HotelCardResult:
|
||
ok, response, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/hotel/card/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
json={
|
||
"reception_id": reception_id,
|
||
"id_kas": ctx.id_kas,
|
||
"card_code": card_code,
|
||
},
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Hotelova karta nebola overena\n{response}")
|
||
return data.HotelCardResult.model_validate(response)
|
||
|
||
def prepare_hotel_charge_API(
|
||
ctx: ApiContext,
|
||
ucet: data.Ucet,
|
||
) -> data.HotelChargePreparation:
|
||
ok, response, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/hotel/charge/prepare/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
json=ucet.model_dump(mode="json"),
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Hotelovy ucet sa nepodarilo pripravit\n{response}")
|
||
return data.HotelChargePreparation.model_validate(response)
|
||
|
||
def send_hotel_charge_API(
|
||
ctx: ApiContext,
|
||
ucet: data.Ucet,
|
||
) -> data.HotelChargeSendResult:
|
||
ok, response, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/hotel/charge/send/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
json=ucet.model_dump(mode="json"),
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Hotelovy ucet sa nepodarilo odoslat\n{response}")
|
||
return data.HotelChargeSendResult.model_validate(response)
|
||
|
||
def load_import_parameters_API(ctx: ApiContext) -> list[dict]:
|
||
ok, response, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/import_parameters/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Zoznam parametrov nebol nacitany\n{response}")
|
||
return response
|
||
|
||
def load_setup_parameters_API(
|
||
ctx: ApiContext,
|
||
include_defaults: bool = True,
|
||
) -> list[data.SetupParameterValue]:
|
||
ok, response, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/setup/parameters/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params={
|
||
"id_kas": ctx.id_kas,
|
||
"include_defaults": "true" if include_defaults else "false",
|
||
},
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Parametre pokladne {ctx.id_kas} neboli nacitane\n{response}")
|
||
return [
|
||
data.SetupParameterValue.model_validate(item)
|
||
for item in response
|
||
]
|
||
|
||
def save_setup_parameters_API(
|
||
ctx: ApiContext,
|
||
parameters: list[data.SetupParameterValue],
|
||
) -> dict:
|
||
payload = [
|
||
data.SetupParameterValue.model_validate(item).model_dump()
|
||
for item in parameters
|
||
]
|
||
ok, response, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/setup/parameters/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params={"id_kas": ctx.id_kas},
|
||
json=payload,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Parametre pokladne {ctx.id_kas} neboli ulozene\n{response}")
|
||
return response
|
||
|
||
def load_postgres_connection_API(ctx: ApiContext) -> data.PostgresConnectionOut:
|
||
ok, response, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/postgres/connection/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"PostgreSQL pripojenie nebolo nacitane\n{response}")
|
||
return data.PostgresConnectionOut.model_validate(response)
|
||
|
||
def save_postgres_connection_API(
|
||
ctx: ApiContext,
|
||
connection: data.PostgresConnection,
|
||
) -> data.PostgresConnectionOut:
|
||
payload = data.PostgresConnection.model_validate(connection).model_dump(by_alias=True)
|
||
ok, response, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/postgres/connection/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
json=payload,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"PostgreSQL pripojenie nebolo ulozene\n{response}")
|
||
return data.PostgresConnectionOut.model_validate(response)
|
||
|
||
def load_postgres_status_API(
|
||
ctx: ApiContext,
|
||
test_connection: bool = True,
|
||
) -> data.PostgresStatus:
|
||
ok, response, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/postgres/status/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params={
|
||
"id_kas": ctx.id_kas,
|
||
"test_connection": "true" if test_connection else "false",
|
||
},
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"PostgreSQL status nebol nacitany\n{response}")
|
||
return data.PostgresStatus.model_validate(response)
|
||
|
||
def load_limity_API(ctx: ApiContext) -> list[data.LimitTable]:
|
||
ok, response, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/limity/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params={"id_kas": ctx.id_kas},
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Limity neboli nacitane\n{response}")
|
||
return [data.LimitTable.model_validate(item) for item in response]
|
||
|
||
def load_limit_ucet_API(ctx: ApiContext, id_limit: int, id_den: int) -> data.Ucet:
|
||
ok, response, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/limity/ucet/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params={
|
||
"id_kas": ctx.id_kas,
|
||
"id_limit": str(id_limit),
|
||
"id_den": str(id_den),
|
||
},
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Limitovy ucet nebol nacitany\n{response}")
|
||
return data.Ucet.model_validate(response)
|
||
|
||
def release_limit_API(ctx: ApiContext, id_limit: int) -> data.LimitLockResult:
|
||
ok, response, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/limity/release/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params={
|
||
"id_kas": ctx.id_kas,
|
||
"id_limit": str(id_limit),
|
||
},
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Limitovy semafor nebol uvolneny\n{response}")
|
||
return data.LimitLockResult.model_validate(response)
|
||
|
||
def save_limit_ucet_API(ctx: ApiContext, ucet: data.Ucet) -> data.Ucet:
|
||
ok, response, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/limity/ucet/save/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
json=ucet.model_dump(mode="json"),
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Limitovy ucet nebol ulozeny\n{response}")
|
||
return data.Ucet.model_validate(response)
|
||
|
||
def finish_limit_ucet_API(ctx: ApiContext, ucet: data.Ucet) -> data.Ucet:
|
||
ok, response, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/limity/ucet/finish/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
json=ucet.model_dump(mode="json"),
|
||
timeout=120,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Limitovy ucet nebol uzavrety\n{response}")
|
||
return data.Ucet.model_validate(response)
|
||
|
||
def clear_limit_ucet_API(ctx: ApiContext, ucet: data.Ucet) -> data.LimitLockResult:
|
||
ok, response, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/limity/ucet/clear/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
json=ucet.model_dump(mode="json"),
|
||
timeout=60,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Limitovy ucet nebol odznaceny\n{response}")
|
||
return data.LimitLockResult.model_validate(response)
|
||
|
||
# --- login_API
|
||
def login_API(ctx: ApiContext) -> str:
|
||
ok, response, _ = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/login/",
|
||
client_id=ctx.client_id,
|
||
json={
|
||
"username": ctx.username,
|
||
"password": ctx.password.get_secret_value(),
|
||
"id_kas": ctx.id_kas,
|
||
},
|
||
)
|
||
if not ok:
|
||
raise RuntimeError(
|
||
"Login failed – bad credentials or server unavailable"
|
||
)
|
||
ctx.token = response["access_token"]
|
||
ctx.refresh_token = response["refresh_token"]
|
||
logger.info(
|
||
f"\nPřihlášení úspěšné"
|
||
f"\ntoken={ctx.token}"
|
||
f"\nrefresh_token={ctx.refresh_token}\n"
|
||
)
|
||
return response["version_API"], response["database_name"]
|
||
|
||
def logout_API(ctx: ApiContext):
|
||
ok, resp, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/logout/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params={
|
||
"id_kas": ctx.id_kas,
|
||
},
|
||
)
|
||
# pokud server vrátí nový token (nemusí), aktualizuj
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(
|
||
f"Logout selhal\n{resp}"
|
||
)
|
||
return resp
|
||
|
||
|
||
# replace cenik na jeden request
|
||
def debugrepl2_cenik_API(ctx: ApiContext, items: list[data.CenPolCreate]) -> None:
|
||
#import tst_data
|
||
# --- Pripadne smazani ceniku
|
||
ok, resp, token = call_api(
|
||
method="DELETE",
|
||
base_url=ctx.base_url,
|
||
endpoint=f"/cenik/pokl/{ctx.id_kas}",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if ok:
|
||
print(f"\nSmazan cenik pokladny {ctx.id_kas}")
|
||
else:
|
||
print(f"Mazani ceniku pokladny {ctx.id_kas} selhalo nebo neexistuje")
|
||
# --- BATCH payload
|
||
payload = [
|
||
i.model_dump(exclude={"id"})
|
||
for i in items
|
||
]
|
||
# --- Ulozeni celeho ceniku jednim requestem
|
||
ok, resp, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint=f"/cenik/items/{ctx.id_kas}",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
json=payload
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(
|
||
"Uložení ceníku selhalo\n"
|
||
f"SERVER: {resp}"
|
||
)
|
||
print("Ceník uložen celý úspěšně")
|
||
|
||
|
||
# --- replace cenik, jen pro ladeni !!!!
|
||
def debugrepl_cenik_API(ctx: ApiContext) -> None:
|
||
import tst_data
|
||
cenik = tst_data.create_tst_cenik(ctx.id_kas)
|
||
# --- Pripadne smazani ceniku pro pokladnu 01
|
||
ok, resp, token = call_api(
|
||
method="DELETE", base_url=ctx.base_url, endpoint=f"/cenik/pokl/{ctx.id_kas}",
|
||
client_id=ctx.client_id, token=ctx.token, refresh_token=ctx.refresh_token )
|
||
update_ctx_token(ctx, token)
|
||
if ok:
|
||
print(f'\nSmazan cenik pokladny {ctx.id_kas} ')
|
||
else:
|
||
print(f'Mazani ceniku pokladny {ctx.id_kas} selhalo nebo neexistuje')
|
||
# --- Ulozeni ceniku na server
|
||
for i in cenik.cenpol:
|
||
payload = i.model_dump(exclude={"id"})
|
||
ok, resp, _ = call_api(
|
||
method="POST", base_url=ctx.base_url, endpoint="/save_cenpol/",
|
||
client_id=ctx.client_id, token=ctx.token, refresh_token=ctx.refresh_token,
|
||
json=payload)
|
||
if not ok:
|
||
raise RuntimeError(
|
||
f"Uložení položky {i.id} ({i.ch_name}) selhalo\n"
|
||
f"SERVER: {resp}\n"
|
||
f"PAYLOAD: {payload}")
|
||
print("Ceník uložen celý úspěšně")
|
||
|
||
def load_users_API(ctx: ApiContext) -> list[data.UserOut]:
|
||
ok, resp, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/users/",
|
||
client_id=ctx.client_id,
|
||
params={"id_kas": ctx.id_kas},
|
||
token=ctx.token,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Load users selhal\n{resp}")
|
||
# resp je list dict → převedeme na modely
|
||
return [data.UserOut(**u) for u in resp]
|
||
|
||
def reset_users_API(ctx: ApiContext, users: list[data.UserIn]):
|
||
ok, resp, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/users/reset/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
json=[u.model_dump() for u in users],
|
||
params={"id_kas": ctx.id_kas},
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Reset users selhal\n{resp}")
|
||
return True
|
||
|
||
#Milan 15.04.26 - doplnene id_kas
|
||
def login_user_API(ctx: ApiContext, heslo: str) -> data.UserLoginOut:
|
||
ok, resp, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/users/login/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
json={"heslo": heslo,"kas":ctx.id_kas},
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
logger.debug(f'User login {resp}')
|
||
if not ok:
|
||
raise RuntimeError("Neplatné přihlášení")
|
||
return data.UserLoginOut(**resp)
|
||
|
||
# --- mapa stolu
|
||
def save_mapa_stolu_API(ctx, mapa):
|
||
ok, resp, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/mapa_stolu/",
|
||
json=mapa.model_dump(),
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Save mapa_stolu failed\n{resp}")
|
||
return ok, resp
|
||
|
||
def load_mapa_stolu_API(ctx):
|
||
ok, resp, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/mapa_stolu/",
|
||
params={"id_kas": ctx.id_kas},
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Load mapa_stolu failed\n{resp}")
|
||
return data.MapaStolu.model_validate(resp)
|
||
|
||
# --- nacte stoly/ucty ze serveru, jen pro moji pokladnu a closed == False
|
||
def load_stoly_API(
|
||
ctx: ApiContext,
|
||
closed: bool = False,
|
||
onlynonclsrep: bool = True,
|
||
limit: int | None = None,
|
||
) -> list[data.UcetSelect]:
|
||
params = {
|
||
"id_kas": ctx.id_kas,
|
||
"closed": "true" if closed else "false",
|
||
"onlynonclsrep": "true" if onlynonclsrep else "false",
|
||
}
|
||
if limit is not None:
|
||
params["limit"] = str(limit)
|
||
ok, resp, token = call_api(
|
||
method="GET", base_url=ctx.base_url, endpoint="/ucty/",
|
||
client_id=ctx.client_id, token=ctx.token, refresh_token=ctx.refresh_token,
|
||
params=params)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f'List of check/tables fail\n {resp}')
|
||
ucty_row = [data.UcetSelect(**u) for u in resp["ucty"]]
|
||
logger.debug(f'\n{ucty_row}')
|
||
return ucty_row
|
||
|
||
# --- pripoji ucet k jiz existujicimu resp. vytvori
|
||
def merge_ucet_API(ctx: ApiContext, source_ucet: data.Ucet, target_stul: str,) -> dict:
|
||
payload = {
|
||
"ucet": source_ucet.model_dump(),
|
||
"target_stul": target_stul, }
|
||
ok, resp, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/ucet/merge/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
json=payload, )
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(
|
||
f"Merge účtu na stůl {target_stul} selhalo\n{resp}" )
|
||
return resp
|
||
|
||
# --- nacte ucet ze serveru dle ucis (uzavreny)
|
||
def load_ucet_by_ucislo_API(ctx: ApiContext, ucislo: str) -> data.Ucet:
|
||
ok, resp, token = call_api(method="GET", base_url=ctx.base_url, endpoint="/ucet/",
|
||
client_id=ctx.client_id, token=ctx.token, refresh_token=ctx.refresh_token,
|
||
params={"ucislo": ucislo, "id_kas": ctx.id_kas },)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Načtení účtu číslo {ucislo} selhalo\n{resp}" )
|
||
logger.debug(f"\nNačtený uzavřený účet z DB:\n{resp}\n")
|
||
return data.Ucet(**resp)
|
||
|
||
# --- načte účet ze serveru
|
||
def load_ucet_API(ctx: ApiContext, stul: str, block: bool = True) -> data.Ucet:
|
||
ok, resp, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/ucet/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params={
|
||
"stul": stul,
|
||
"id_kas": ctx.id_kas,
|
||
"block": "true" if block else "false",
|
||
},
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
# resp je text chyby ze serveru (např. "… je blokován: 02|01|…")
|
||
logger.warning(f"load_ucet_API FAILED stul={stul}: {resp}")
|
||
raise RuntimeError(resp)
|
||
logger.debug(f"Načtený účet z DB: {resp}")
|
||
return data.Ucet(**resp)
|
||
|
||
|
||
# --- ulozeni uctu
|
||
def save_ucet_API(ctx: ApiContext, ucet: data.Ucet) -> data.Ucet:
|
||
payload = ucet.model_dump()
|
||
#print(payload)
|
||
ok, resp, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/ucet/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
json=payload, )
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
#print("STATUS:", status)
|
||
#print("RESPONSE:", resp)
|
||
raise RuntimeError(f"Uložení účtu selhalo\n{resp}")
|
||
# server vrátil ucislo
|
||
if resp.get("ucislo"):
|
||
ucet.ucislo = resp["ucislo"]
|
||
return ucet
|
||
|
||
def open_block_ucet_API(ctx: ApiContext, stul: str):
|
||
"""
|
||
Atomicky:
|
||
- otevře účet (pokud neexistuje → vytvoří dummy)
|
||
- zablokuje ho pro aktuální zařízení
|
||
"""
|
||
ok, resp, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/ucet/open/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params={
|
||
"stul": stul,
|
||
"id_kas": ctx.id_kas,
|
||
},
|
||
)
|
||
|
||
# aktualizace tokenu (pokud server poslal nový)
|
||
update_ctx_token(ctx, token)
|
||
|
||
if not ok:
|
||
# resp je už text / dict s chybou – zatím řešíme přes Exception
|
||
raise Exception(resp)
|
||
|
||
return resp
|
||
|
||
# --- nacteni ceniku
|
||
def load_cenik_API(ctx: ApiContext) -> data.Cenik:
|
||
ok, resp, token = call_api(
|
||
method="GET", base_url=ctx.base_url, endpoint=f"/cenik/pokl/{ctx.id_kas}",
|
||
client_id=ctx.client_id, token=ctx.token, refresh_token=ctx.refresh_token)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError("f'Nacteni ceniku pokladny {ctx.id_kas} selhalo'")
|
||
cenik = data.Cenik.model_validate({"cenpol": resp})
|
||
logger.debug(f'Load cenik pro {ctx.id_kas}, nacteno polozek: {len(cenik.cenpol)}')
|
||
#cenik.prn()
|
||
return cenik
|
||
|
||
|
||
def load_cenik_texty_API(ctx: ApiContext, lang: str) -> list[data.CenikText]:
|
||
ok, resp, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/cenik/texty",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params={"lang": lang or "sk"},
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Nacteni jazykovych nazvov cenika {ctx.id_kas}/{lang} selhalo\n{resp}")
|
||
return [data.CenikText.model_validate(item) for item in (resp or [])]
|
||
|
||
|
||
def load_locales_API(ctx: ApiContext) -> list[str]:
|
||
ok, resp, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/locales/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Nacteni zoznamu jazykov selhalo\n{resp}")
|
||
return list((resp or {}).get("locales", []))
|
||
|
||
|
||
def load_locale_API(ctx: ApiContext, lang: str) -> dict:
|
||
ok, resp, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint=f"/locales/{lang or 'sk'}",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Nacteni lokalizacie {lang} selhalo\n{resp}")
|
||
return dict(resp or {})
|
||
|
||
|
||
def replace_cenik_texty_API(ctx: ApiContext, texty: list[data.CenikText], lang: str | None = None) -> dict:
|
||
params = {}
|
||
if lang:
|
||
params["lang"] = lang
|
||
ok, resp, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/cenik/texty",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params=params,
|
||
json=[item.model_dump() for item in texty],
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Ulozenie jazykovych nazvov cenika {ctx.id_kas} selhalo\n{resp}")
|
||
return resp
|
||
|
||
|
||
def load_zlavy_API(
|
||
ctx: ApiContext,
|
||
) -> data.Zlavy:
|
||
ok, resp, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint=f"/zlavy/{ctx.id_kas}",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params={},
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Nacteni zliav pokladny {ctx.id_kas} selhalo\n{resp}")
|
||
return data.Zlavy.model_validate({"id_kas": ctx.id_kas, "zlavy": resp})
|
||
|
||
def load_kasutxt_API(
|
||
ctx: ApiContext,
|
||
) -> data.KasUtxtRiadky:
|
||
ok, resp, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint=f"/kasutxt/{ctx.id_kas}",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params={},
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Nacteni hlaviciek uctov pokladny {ctx.id_kas} selhalo\n{resp}")
|
||
return resp
|
||
|
||
def replace_zlavy_API(ctx: ApiContext, zlavy: list[data.Zlava]) -> dict:
|
||
ok, resp, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint=f"/zlavy/{ctx.id_kas}",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
json=[z.model_dump(mode="json") for z in zlavy],
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(
|
||
f"Ulozenie zliav pokladny {ctx.id_kas} selhalo\n{resp}"
|
||
)
|
||
return resp
|
||
|
||
|
||
def _payload_dict(item) -> dict:
|
||
if isinstance(item, BaseModel):
|
||
return item.model_dump()
|
||
return dict(item)
|
||
|
||
|
||
def build_zlavy_from_tables(
|
||
hlav,
|
||
druhy=None,
|
||
kalky=None,
|
||
) -> list[data.Zlava]:
|
||
zlav_map: dict[int, data.Zlava] = {}
|
||
for row in hlav or []:
|
||
zlava = data.Zlava.model_validate(_payload_dict(row))
|
||
if druhy is not None:
|
||
zlava.druhy = []
|
||
if kalky is not None:
|
||
zlava.kalky = []
|
||
zlav_map[int(zlava.idriadok)] = zlava
|
||
|
||
for row in druhy or []:
|
||
druh = data.ZlavaDruh.model_validate(_payload_dict(row))
|
||
parent = zlav_map.get(int(druh.id_zlavy_hlav))
|
||
if parent:
|
||
parent.druhy.append(druh)
|
||
|
||
for row in kalky or []:
|
||
kalka = data.ZlavaKalka.model_validate(_payload_dict(row))
|
||
parent = zlav_map.get(int(kalka.id_zlavy_hlav))
|
||
if parent:
|
||
parent.kalky.append(kalka)
|
||
|
||
return list(zlav_map.values())
|
||
|
||
|
||
def replace_zlavy_tables_API(
|
||
ctx: ApiContext,
|
||
hlav,
|
||
druhy=None,
|
||
kalky=None,
|
||
) -> dict:
|
||
return replace_zlavy_API(
|
||
ctx,
|
||
build_zlavy_from_tables(hlav, druhy, kalky),
|
||
)
|
||
|
||
|
||
def delete_zlavy_API(ctx: ApiContext) -> bool:
|
||
ok, resp, token = call_api(
|
||
method="DELETE",
|
||
base_url=ctx.base_url,
|
||
endpoint=f"/zlavy/{ctx.id_kas}",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Mazani zliav pokladny {ctx.id_kas} selhalo\n{resp}")
|
||
return True
|
||
|
||
|
||
# --- nacteni FST menu
|
||
def load_fstmenu_API(ctx: ApiContext) -> data.FstMenuKasa:
|
||
ok, resp, token = call_api(
|
||
method="GET", base_url=ctx.base_url, endpoint=f"/fstmenu/pokl/{ctx.id_kas}",
|
||
client_id=ctx.client_id, token=ctx.token, refresh_token=ctx.refresh_token)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError("f'Nacteni fstmenu {ctx.id_kas} selhalo'")
|
||
fstmenu = [data.FstMenuKasa.model_validate(x) for x in resp]
|
||
logger.debug(f'Load fstmenu pro {ctx.id_kas}, nacteno polozek: {len(fstmenu)}')
|
||
print(fstmenu)
|
||
return fstmenu
|
||
|
||
# --- test je-li ucet blokovan
|
||
def is_ucet_blocked_API(ctx: ApiContext, stul: str) -> dict:
|
||
ok, resp, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/ucet/is_blocked/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params={"stul": stul, "id_kas": ctx.id_kas},)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Kontrola blokace selhala\n{resp}")
|
||
return resp
|
||
|
||
# --- unblock ucet stolu
|
||
def unblock_ucet_API(ctx: ApiContext, stul: str):
|
||
try: #nevraci chybu, neni-li ucet blokovan nebo neexistuje
|
||
unblock_ucet_hardAPI(ctx, stul)
|
||
except RuntimeError as e:
|
||
if "404" in str(e):
|
||
logger.warning(f"Unblock ignorován účet {stul} neexistuje nebo není blokován")
|
||
return
|
||
raise
|
||
|
||
def unblock_ucet_hardAPI(ctx: ApiContext, stul: str):
|
||
ok, resp, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/ucet/unblock/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params={"stul": stul, "id_kas": ctx.id_kas}, )
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Unblock účtu pro stůl {stul} selhalo\n{resp}")
|
||
return resp
|
||
|
||
def unblock_ucet_by_ucislo_API(ctx: ApiContext, ucislo: str):
|
||
ok, resp, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/ucet/unblock/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params={
|
||
"ucislo": ucislo,
|
||
"id_kas": ctx.id_kas,
|
||
},
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(
|
||
f"Unblock účtu č. {ucislo} selhalo\n{resp}"
|
||
)
|
||
return resp
|
||
|
||
|
||
# --- block ucet dle stolu
|
||
def block_ucet_API(ctx: ApiContext, stul: str) -> dict:
|
||
ok, resp, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/ucet/block/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params={"stul": stul, "id_kas": ctx.id_kas}, )
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(
|
||
f"Block účtu pro stůl {stul} selhal\n{resp}")
|
||
return resp
|
||
|
||
# --- delete uctu
|
||
def delete_ucet_API(ctx: ApiContext,*,ucislo: str | None = None,
|
||
stul: str | None = None,) -> dict:
|
||
# Smaže účet podle čísla účtu NEBO podle stolu.
|
||
# Přesně jeden z parametrů musí být zadán.
|
||
if (ucislo is None) == (stul is None):
|
||
raise ValueError(
|
||
"delete_ucet_API: zadej právě jeden parametr: ucislo NEBO stul" )
|
||
params = {}
|
||
params["id_kas"] = ctx.id_kas
|
||
if ucislo is not None:
|
||
params["ucislo"] = ucislo
|
||
else:
|
||
params["stul"] = stul
|
||
ok, resp, token = call_api(method="DELETE", base_url=ctx.base_url, endpoint="/ucet/",
|
||
client_id=ctx.client_id, token=ctx.token, refresh_token=ctx.refresh_token,
|
||
params=params, )
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(
|
||
f"Block účtu pro stůl {stul} selhal\n{resp}")
|
||
return resp
|
||
|
||
|
||
def update_ctx_token(ctx: ApiContext, new_token: str | None) -> None:
|
||
if new_token and new_token != ctx.token:
|
||
ctx.token = new_token
|
||
|
||
# --------------------------------------
|
||
# --- metoda pro komunikaci se serverem
|
||
# --------------------------------------
|
||
|
||
logger = logging.getLogger(__name__)
|
||
def call_api(
|
||
*,
|
||
method: str,
|
||
base_url: str,
|
||
endpoint: str,
|
||
client_id: str,
|
||
token: str | None = None,
|
||
refresh_token: str | None = None,
|
||
retries: int = API_RETRIES,
|
||
delay: float = API_DELAY,
|
||
session: requests.Session | None = None,
|
||
**kwargs,
|
||
):
|
||
#VRACÍ VŽDY: (ok: bool, resp: Any | str, new_token: str | None)
|
||
# HTTP chyby (4xx / 5xx) → ok=False (NEVYHAZUJE výjimku)
|
||
# Síťové chyby → retry → ok=False
|
||
url = f"{base_url}{endpoint}"
|
||
# --- session ---
|
||
s = session or requests
|
||
# --- headers ---
|
||
headers = kwargs.pop("headers", {}).copy()
|
||
headers["X-Client-ID"] = client_id
|
||
if token:
|
||
headers["Authorization"] = f"Bearer {token}"
|
||
kwargs["headers"] = headers
|
||
kwargs.setdefault("timeout", API_TIMEOUT)
|
||
current_token = token
|
||
for attempt in range(retries + 1):
|
||
try:
|
||
resp = s.request(method, url, **kwargs)
|
||
# 401 → refresh token (1×)
|
||
if resp.status_code == 401 and refresh_token:
|
||
logger.info("401 → refreshing access token")
|
||
new_token = refresh_access_token(
|
||
base_url=base_url,
|
||
refresh_token=refresh_token,
|
||
client_id=client_id,
|
||
)
|
||
if not new_token:
|
||
return False, "Token expired, refresh failed", None
|
||
current_token = new_token
|
||
headers["Authorization"] = f"Bearer {new_token}"
|
||
kwargs["headers"] = headers
|
||
resp = s.request(method, url, **kwargs)
|
||
# HTTP ERROR (4xx / 5xx)
|
||
if resp.status_code >= 400:
|
||
try:
|
||
return False, resp.json(), current_token
|
||
except ValueError:
|
||
return False, resp.text, current_token
|
||
# OK RESPONSE
|
||
try:
|
||
return True, resp.json(), current_token
|
||
except ValueError:
|
||
return True, resp.text, current_token
|
||
except requests.RequestException as e:
|
||
logger.warning(f"API request failed (attempt {attempt + 1}): {e}")
|
||
if attempt >= retries:
|
||
return False, str(e), None
|
||
time.sleep(delay)
|
||
return False, "Unknown error", None
|
||
|
||
# --- ziskani noveho tokenu
|
||
def refresh_access_token(base_url: str, refresh_token: str, client_id: str) -> str | None:
|
||
try:
|
||
r = requests.post(
|
||
f"{base_url}/refresh/",
|
||
headers={"X-Client-ID": client_id},
|
||
json={"refresh_token": f"Bearer {refresh_token}"},
|
||
timeout=5
|
||
)
|
||
r.raise_for_status()
|
||
return r.json().get("access_token")
|
||
except Exception as e:
|
||
#print("Refresh failed:", e)
|
||
return None
|
||
|
||
def clsrep_API(ctx: ApiContext,
|
||
ucislo_od: str | None,
|
||
ucislo_do: str | None,
|
||
save: bool = False,
|
||
cash_carry: list[dict] | None = None) -> data.ClosureReportOut:
|
||
params = {
|
||
"id_kas": ctx.id_kas,
|
||
}
|
||
if ucislo_od:
|
||
params["ucislo_od"] = ucislo_od
|
||
if ucislo_do:
|
||
params["ucislo_do"] = ucislo_do
|
||
endpoint = "/closure/save/" if save else "/closure/"
|
||
method = "POST" if save else "GET"
|
||
kwargs = {}
|
||
if save and cash_carry is not None:
|
||
kwargs["json"] = {"cash_carry": cash_carry}
|
||
ok, resp, token = call_api(
|
||
method=method,
|
||
base_url=ctx.base_url,
|
||
endpoint=endpoint,
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params=params,
|
||
timeout=120 if save else 60,
|
||
retries=0,
|
||
**kwargs,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
return None, f"Uzávěrka selhala\n{resp}"
|
||
try:
|
||
report = data.ClosureReportOut.model_validate(resp)
|
||
return report, None
|
||
except Exception as e:
|
||
return None, f"Chyba validace uzávěrky:\n{e}"
|
||
|
||
def load_closures_API( ctx: ApiContext):
|
||
params = {
|
||
"id_kas": ctx.id_kas,
|
||
}
|
||
ok, resp, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/closure/list/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params=params,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
return None, f"Načtení uzávěrek selhalo\n{resp}"
|
||
try:
|
||
closures = [
|
||
data.ClosureIntervalOut.model_validate(r)
|
||
for r in resp
|
||
]
|
||
return closures, None
|
||
except Exception as e:
|
||
return None, f"Chyba parsování uzávěrek\n{e}"
|
||
|
||
def load_closure_cash_state_API(
|
||
ctx: ApiContext,
|
||
clsrep_id: int | None = None,
|
||
status: str | None = None,
|
||
):
|
||
params = {"id_kas": ctx.id_kas}
|
||
if clsrep_id is not None:
|
||
params["clsrep_id"] = clsrep_id
|
||
if status:
|
||
params["status"] = status
|
||
ok, resp, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/closure/cash-state/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params=params,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
return None, f"Nacteni stavu uzavierkovych odvodu selhalo\n{resp}"
|
||
return resp, None
|
||
|
||
def load_closure_transfers_API(
|
||
ctx: ApiContext,
|
||
clsrep_id: int | None = None,
|
||
status: str | None = None,
|
||
):
|
||
params = {"id_kas": ctx.id_kas}
|
||
if clsrep_id is not None:
|
||
params["clsrep_id"] = clsrep_id
|
||
if status:
|
||
params["status"] = status
|
||
ok, resp, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/closure/transfers/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params=params,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
return None, f"Nacteni prenosu uzavierky selhalo\n{resp}"
|
||
return resp, None
|
||
|
||
def retry_closure_transfer_API(ctx: ApiContext, transfer_id: int):
|
||
ok, resp, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint=f"/closure/transfers/{int(transfer_id)}/retry",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
timeout=120,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
return None, f"Odoslanie prenosu uzavierky zlyhalo\n{resp}"
|
||
return resp, None
|
||
|
||
def closure_detail_API(ctx: ApiContext, clsrep_no: str):
|
||
params = {
|
||
"id_kas": ctx.id_kas,
|
||
"clsrep_no": clsrep_no
|
||
}
|
||
ok, resp, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/closure/detail/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params=params,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
return None, f"Načtení uzávěrky selhalo\n{resp}"
|
||
return resp, None
|
||
|
||
def load_ucty_notinclsrep_API(ctx):
|
||
ok, resp, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/ucty/notinclsrep/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
params={"id_kas": ctx.id_kas},
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
return None, resp
|
||
return [data.Ucet.model_validate(x) for x in resp], None
|
||
|
||
def load_printers_for_kasa_API(ctx: ApiContext) -> list[data.PrnDefShort]:
|
||
ok, resp, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/prndefkasa/",
|
||
client_id=ctx.client_id,
|
||
params={"id_kas": ctx.id_kas},
|
||
token=ctx.token,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Load prndef selhal\n{resp}")
|
||
# resp je list dict → převedeme na modely
|
||
return [data.PrnDefShort(**u) for u in resp]
|
||
|
||
def load_all_printers_API(ctx: ApiContext) -> list[data.PrnDefShort]:
|
||
ok, resp, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/prndef/",
|
||
client_id=ctx.client_id,
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Load prndef selhal\n{resp}")
|
||
return [data.PrnDefShort(**u) for u in resp]
|
||
|
||
def load_print_templates_API(ctx: ApiContext, kind: str = "bon") -> list[data.PrintTemplateOut]:
|
||
ok, resp, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/print/templates/",
|
||
client_id=ctx.client_id,
|
||
params={"kind": kind},
|
||
token=ctx.token,
|
||
refresh_token=ctx.refresh_token,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Load print templates selhal\n{resp}")
|
||
return [data.PrintTemplateOut.model_validate(item) for item in resp]
|
||
|
||
def load_pricelevels_API(ctx: ApiContext) -> list[data.HladinyRiadky]:
|
||
ok, resp, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/pricelevels/",
|
||
client_id=ctx.client_id,
|
||
params={"id_kas": ctx.id_kas},
|
||
token=ctx.token,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Load users selhal\n{resp}")
|
||
# resp je list dict → převedeme na modely
|
||
return [data.HladinyRiadky(**u) for u in resp]
|
||
|
||
def load_clientsettings_API(ctx: ApiContext) -> data.ClientSettings:
|
||
ok, resp, token = call_api(
|
||
method="GET",
|
||
base_url=ctx.base_url,
|
||
endpoint="/clientsettings/",
|
||
client_id=ctx.client_id,
|
||
params={"id_kas": ctx.id_kas},
|
||
token=ctx.token,
|
||
)
|
||
if not ok:
|
||
raise RuntimeError(f"Load clientsettings selhal\n{resp}")
|
||
update_ctx_token(ctx, token)
|
||
# resp je list dict → převedeme na modely
|
||
return resp
|
||
|
||
def save_clientsettings_API(ctx: ApiContext, prn_no:str, room_name:str) -> data.ClientSettings:
|
||
if not prn_no:
|
||
prn_no = ""
|
||
if not room_name:
|
||
room_name = ""
|
||
ok, resp, token = call_api(
|
||
method="POST",
|
||
base_url=ctx.base_url,
|
||
endpoint="/clientsettings/",
|
||
client_id=ctx.client_id,
|
||
params={"id_kas": ctx.id_kas, "prn_no": prn_no, "room_name": room_name},
|
||
token=ctx.token,
|
||
)
|
||
update_ctx_token(ctx, token)
|
||
if not ok:
|
||
raise RuntimeError(f"Zápis clientsettings selhal\n{resp}")
|
||
# resp je list dict → převedeme na modely
|
||
return resp
|