Skip to content

aa.gusti / opensipkd-base

Go to a project
Commit 3d130165 by aa.gustiana@gmail.com
Options

add text_printers table and update related models and views 

Co-authored-by: Copilot <copilot@github.com>
1 parent d6a6af2d
Inline Side-by-side
Showing 3 changed files with 119 additions and 8 deletions
opensipkd/base/alembic/versions/d2dc53a168c8_initial_table_creation.py → opensipkd/base/alembic/versions/663de8bbd563_text_printer_init.py
"""Initial table creation """text_printer init
Revision ID: d2dc53a168c8 Revision ID: 663de8bbd563
Revises: 1be80caf7f0a Revises: 1be80caf7f0a
Create Date: 2026-04-30 01:39:46.558393 Create Date: 2026-04-30 04:55:03.400917
""" """
# revision identifiers, used by Alembic. # revision identifiers, used by Alembic.
revision = 'd2dc53a168c8' revision = '663de8bbd563'
down_revision = '1be80caf7f0a' down_revision = '1be80caf7f0a'
branch_labels = None branch_labels = None
depends_on = None depends_on = None
...@@ -19,6 +19,10 @@ import sqlalchemy as sa ...@@ -19,6 +19,10 @@ import sqlalchemy as sa
def upgrade(): def upgrade():
# ### commands auto generated by Alembic - please adjust! ### # ### commands auto generated by Alembic - please adjust! ###
op.create_table('text_printers', op.create_table('text_printers',
sa.Column('is_epson', sa.Integer(), nullable=True),
sa.Column('queue', sa.String(length=16), nullable=True),
sa.Column('port', sa.Integer(), nullable=True),
sa.Column('timeout', sa.Integer(), nullable=True),
sa.Column('nama', sa.String(length=128), nullable=False), sa.Column('nama', sa.String(length=128), nullable=False),
sa.Column('kode', sa.String(length=32), nullable=False), sa.Column('kode', sa.String(length=32), nullable=False),
sa.Column('status', sa.SmallInteger(), nullable=False), sa.Column('status', sa.SmallInteger(), nullable=False),
......
opensipkd/base/models/utils.py
from sqlalchemy import Column from sqlalchemy import Column, Integer, String
from . import Base, NamaModel, TABLE_ARGS from . import Base, NamaModel, TABLE_ARGS
class TextPrinters(Base, NamaModel): class TextPrinters(Base, NamaModel):
__tablename__ = 'text_printers' __tablename__ = 'text_printers'
__table_args__ = TABLE_ARGS __table_args__ = TABLE_ARGS
is_epson = Column(Integer(), default=0)
queue = Column(String(16), default='lp')
port = Column(Integer(), default=515)
timeout = Column(Integer(), default=10)
opensipkd/base/views/printers.py
import logging import logging
import re import re
from turtle import title
from wsgiref.validate import validator
import colander import colander
from deform import (widget,) from deform import (widget,)
from pyramid.i18n import TranslationStringFactory from pyramid.i18n import TranslationStringFactory
...@@ -15,7 +17,109 @@ SESS_ADD_FAILED = 'Tambah partner gagal' ...@@ -15,7 +17,109 @@ SESS_ADD_FAILED = 'Tambah partner gagal'
SESS_EDIT_FAILED = 'Edit partner gagal' SESS_EDIT_FAILED = 'Edit partner gagal'
def print_text(user, text):
from opensipkd.base.models import TextPrinters
printer = TextPrinters.query().filter_by(create_uid=user.id, status=1).first()
if not printer:
log.error(
f"User {user.id} does not have an active printer configured.")
raise Exception("No active printer configured for the user.")
# Replace with your printer's IP
hostname = printer.kode if printer else "127.0.0.1"
queue = printer.queue if printer else "lp" # Common default queue name
port = printer.port if printer else 515 # Default LPR port
timeout = printer.timeout if printer else 10 # Timeout in seconds
# Set to True if the printer is an Epson model that requires specific control codes
is_epson = printer.is_epson if printer else False
from pylpr import LprClient
with LprClient(
hostname=hostname,
port=port,
timeout=timeout,
queue=queue,
# recv_buffer=args.recv_buffer,
# username=args.username,
# job_name=args.job_name,
# file_name=getattr(args.print_file, 'name', None) if hasattr(
# args, 'print_file') and args.print_file else None,
# label=args.label,
# use_reserved_port=args.use_reserved_port
) as lpr:
# if args.print_queue:
# # Send the 'Print any waiting jobs' command (0x01) to the specified queue
# lpr._send_lpr_command(1, queue.encode('ascii'))
# print(
# f"Sent 'Print any waiting jobs' command to queue '{args.queue}'")
# return
# if args.status:
# lpr._send_lpr_command(
# 3, (args.queue + " " + args.status).encode('ascii'), noreceive=True)
# print(
# f"Sent 'Send queue state (short)' command with attributes '{args.status}' to queue '{args.queue}'")
# # Receive and print the response from the printer
# response = lpr.receive()
# print("Received queue state (short) response:")
# print(response.decode('utf-8', errors='replace'))
# return
# if args.longstatus:
# lpr._send_lpr_command(
# 4, (args.queue + " " + args.longstatus).encode('ascii'), noreceive=True)
# print(
# f"Sent 'Send queue state (long)' command with attributes '{args.longstatus}' to queue '{args.queue}'")
# # Receive and print the response from the printer
# response = lpr.receive()
# print("Received queue state (long) response:")
# print(response.decode('utf-8', errors='replace'))
# return
# if args.remove:
# # Send the 'Remove jobs' command (0x05) to the specified queue
# lpr._send_lpr_command(
# 5, (args.queue + " " + args.remove).encode('ascii'))
# print(
# f"Sent 'Remove jobs' command with attributes '{args.remove}' to queue '{args.queue}'")
# return
if is_epson:
lpr.send(
lpr.EXIT_PACKET_MODE
+ lpr.INITIALIZE_PRINTER
+ text.encode('utf-8')
+ lpr.FF
)
else:
lpr.send(text.encode('utf-8'))
# lpr.send(text.encode('utf-8'))
class AddSchema(NamaSchema): class AddSchema(NamaSchema):
nama = colander.SchemaNode(
colander.String(),
title="Nama Printer",
validator=colander.Length(max=64)
)
kode = colander.SchemaNode(
colander.String(),
title="IP Address",
widget=widget.TextInputWidget(mask="999.999.999.999",))
port = colander.SchemaNode(
colander.Integer(),
title="Port",
default=515)
is_epson = colander.SchemaNode(
colander.Integer(),
widget=widget.CheckboxWidget(true_val="1", false_val="0"),
title="Epson Printer",
default=0)
queue = colander.SchemaNode(
colander.String(),
validator=colander.Length(max=16),
title="Queue Name",
default='lp')
timeout = colander.SchemaNode(
colander.Integer(),
title="Timeout",
default=10)
status = colander.SchemaNode( status = colander.SchemaNode(
colander.Integer(), colander.Integer(),
widget=widget.CheckboxWidget(true_val="1", false_val="0"), widget=widget.CheckboxWidget(true_val="1", false_val="0"),
...@@ -63,7 +167,6 @@ class Views(BaseView): ...@@ -63,7 +167,6 @@ class Views(BaseView):
if not self.req.has_permission('admin'): if not self.req.has_permission('admin'):
query = query.filter_by(create_uid=self.req.user.id) query = query.filter_by(create_uid=self.req.user.id)
return query return query
def form_validator(self, form, value): def form_validator(self, form, value):
exc = colander.Invalid(form, None) exc = colander.Invalid(form, None)
...@@ -84,11 +187,11 @@ class Views(BaseView): ...@@ -84,11 +187,11 @@ class Views(BaseView):
if err: if err:
exc['kode'] = '\n'.join(err) exc['kode'] = '\n'.join(err)
raise exc raise exc
def after_save(self, values, row): def after_save(self, values, row):
self.db_session.flush() self.db_session.flush()
if values['status'] == 1: if values['status'] == 1:
printers = TextPrinters.query().filter_by(create_uid=row.create_uid).all() printers = TextPrinters.query().filter_by(create_uid=row.create_uid).all()
for printer in printers: for printer in printers:
if printer.id == row.id: if printer.id == row.id:
continue continue
......
Markdown is supported
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!