fdsa

2026-08-29 10:49:36 +00:00
parent 103983846d
commit 73fba17711
11 changed files with 632 additions and 107 deletions
@@ -1,30 +1,85 @@
+import logging
 import sys

+import taskiq
+
 from aiohttp import ClientSession
 from config import settings as sett
-from printing import sent_to_printer
-from settings import settings
+from nats.js.api import ConsumerConfig, StreamConfig
+from taskiq import SmartRetryMiddleware
+from taskiq_nats import PullBasedJetStreamBroker
+
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="[%(asctime)s][%(name)s][%(levelname)s] %(message)s",
+)
+logger = logging.getLogger(__name__)

 sys.path.extend(["./"])

-
-async def startup(ctx):
-    ctx["session"] = ClientSession()
-    print(f"Worker listening for: {sett.QUEUE_NAME}")
-    # print(f"Worker printing on: {sett.WORKER_PRINTER_ADDRESS}")
+broker = PullBasedJetStreamBroker(
+    [sett.NATS_URL],
+    subject=f"{sett.SUBJECT}.{sett.DURABLE}",
+    stream_name=f"{sett.STREAM}",
+    durable=f"{sett.DURABLE}",
+    stream_config=StreamConfig(
+        name=f"{sett.STREAM}",
+        subjects=[f"{sett.SUBJECT}.>"],
+    ),
+    consumer_config=ConsumerConfig(
+        durable_name=f"{sett.DURABLE}",
+        filter_subject=f"{sett.SUBJECT}.{sett.DURABLE}",
+    ),
+).with_middlewares(
+    SmartRetryMiddleware(
+        default_retry_count=5,
+        default_delay=5,
+        use_delay_exponent=True,
+        default_retry_label=True,
+    )
+)


-async def shutdown(ctx):
-    await ctx["session"].close()
+@broker.on_event("startup")
+async def startup(state: taskiq.TaskiqState) -> None:
+    state.session = ClientSession()


-class WorkerSettings:
-    """
-    Settings for the ARQ worker.
-    """
+@broker.on_event("shutdown")
+async def shutdown(state: taskiq.TaskiqState) -> None:
+    await state.session.close()

-    queue_name = f"barker:print:{sett.QUEUE_NAME}"
-    redis_settings = settings
-    functions: list = [sent_to_printer]
-    on_startup = startup
-    on_shutdown = shutdown
+
+@broker.task(task_name="nat_print")
+async def nat_print(data: str, cut_code: str) -> None:
+    if sett.DEBUG:
+        logger.info(
+            "DEV MODE - Print Job Received:\n%s\n[Cut code: %r]", data, cut_code
+        )
+        return
+
+    try:
+        with open(sett.ADDRESS, "w") as printer:
+            printer.write(data)
+            printer.write("\n")
+            printer.write(cut_code)
+    except LookupError as e:
+        logger.error("Lookup error while writing to printer: %s", e)
+        raise
+    except FileNotFoundError as e:
+        logger.error("Printer address %s not found: %s", sett.ADDRESS, e)
+        raise
+    except Exception:
+        logger.exception("Unexpected error occurred while printing to %s", sett.ADDRESS)
+        raise
+
+    # GS = "\x1d"
+    # PAPER_CUT = GS + "\x56"
+    # # For the printer ESC/POS Reference
+    # # https://reference.epson-biz.com/modules/ref_escpos/index.php?content_id=87
+    # # The m codes are in decimal and can be converted using the following table
+    # # https://www.eso.org/~ndelmott/ascii.html
+    # # The \x03 in all the following is to feed 3 lines before cut it can be increased or reduced
+    # FUNCTION_B_FULL_CUT = "\x41\x03"
+    # FUNCTION_B_PARTIAL_CUT = "\x42\x03"