Skip to content

aa.gusti / opensipkd-base

Go to a project
Commit 51d26cb1 by Ari Agung Prasetiyo
Options

Merge branch '5.0.3' of https://git.opensipkd.com/aa.gusti/opensipkd-base into 5.0.3

2 parents 266eb772 5be4c97a
Inline Side-by-side
Showing 18 changed files with 406 additions and 180 deletions
CHANGES.txt
5.0.4 2026-06-01
1. Penambahan parameter
- ws_print_url
- is_pylpr
2. Penambahan Table
- tabel text_printer
3. Penambahan URL
- printer
- printer-add
- printer-edit
- printer-edit
- printer-delete
5.0.3 2026-04-28
- Penambahan field session_id pada table users digunakan untuk membatasi jumlah
login hanya bisa ada pada satu device
......
opensipkd/base/__init__.py
......@@ -520,14 +520,18 @@ class BaseApp():
self.reg_nip = 0
self.single_device = "false"
self.is_pylpr = False
self.wsp_url = ""
self.wsp_client_id = ""
self.wsp_client_key = ""
def get_route_file(self, filename="routes.csv"):
fullpath = os.path.join(self.base_dir, 'scripts', 'data', filename)
return open(fullpath)
return open(fullpath, encoding="utf-8")
def static_view(self, config, settings=None):
if not settings:
settings = get_settings()
settings = settings or config.registry.settings
# if not settings:
# settings = get_settings()
self.temp_files = settings.get("temp_files")
if not os.path.exists(self.temp_files):
os.makedirs(self.temp_files)
......@@ -564,6 +568,9 @@ class BaseApp():
self.login_tpl = get_params("login_tpl", "", settings=settings)
self.login_captcha = int(get_params(
"login_captcha", 0, settings=settings))
self.wsp_url = settings.get("wsp_url", "")
self.wsp_client_id = settings.get("wsp_client_id", "")
self.wsp_client_key = settings.get("wsp_client_key", "")
def add_menu(self, config, route_menus, parent=None, paket="opensipkd.base.views",
template_path="views/templates/"):
......@@ -784,18 +791,33 @@ def set_routes(config, app_id=None):
from pyramid.httpexceptions import HTTPBadRequest, HTTPFound
from pyramid.security import forget
from pyramid.view import exception_view_config
from pyramid.response import Response
import json
@exception_view_config(HTTPBadRequest)
def bad_request_view(exc, request):
if request.matched_route.name=='base-login' :
# Bersihkan sesi autentikasi (logout)
headers = forget(request)
# Arahkan ulang ke halaman login (misalnya route 'login')
request.session.flash("Permintaan tidak valid. Silakan ulangi kembali. atau origin tidak diizinkan.", "error")
_logging.debug(f"Bad Request: {exc} from {request.url}")
referrer = request.route_url('base-home')
referrer = request.route_url('base-login')
response = HTTPFound(location=referrer)
response.headers.extend(headers)
return response
elif request.is_xhr:
payload = {
"status": "error",
"message": exc.detail or str(exc)
}
return Response(
json_body=payload,
status=400,
content_type='application/json'
)
return exc
from .depreciated_base import *
opensipkd/base/views/base_views.py
......@@ -15,7 +15,6 @@ from pyramid.request import Response
from sqlalchemy import Table
# from opensipkd.base.views.upload import tmpstore
from opensipkd.tools.captcha import img_captcha
from opensipkd.tools import dmy, get_settings, get_ext, \
date_from_str, get_random_string, Upload, InvalidExtension, mem_tmp_store
from opensipkd.tools.buttons import (
......@@ -77,8 +76,6 @@ class CSRFSchema(colander.Schema):
import re
from pyramid.interfaces import IRoutesMapper
from pyramid.threadlocal import get_current_registry
import io
import xlsxwriter
class BaseView(object):
def __init__(self, request):
......@@ -107,7 +104,9 @@ class BaseView(object):
self.list_route = 'home'
# self.list_col_defs = ""
# self.list_cols = ""
self.list_report = (btn_csv, btn_pdf)
self.list_report = (btn_csv, btn_xls)
self.list_csvpdf = (btn_csv, btn_pdf)
self.list_pdf = (btn_pdf,)
self.list_buttons = (btn_add,)
self.list_upload = (btn_upload,)
self.list_view_field = None
......@@ -556,7 +555,8 @@ class BaseView(object):
elif url_dict['act'] == 'csv':
return self.csv_response(**kwargs)
elif url_dict['act'] == 'csv1':
return self.csv1_response(**kwargs)
elif url_dict['act'] == 'pdf':
return self.pdf_response(**kwargs)
......@@ -671,7 +671,7 @@ class BaseView(object):
filename = jasper_export(self.report_file)
return file_response(self.req, filename=filename[0])
def csv_response(self, **kwargs):
def csv1_response(self, **kwargs):
query = self.table.query_register()
row = query.first()
header = row._mapping.keys()
......@@ -725,6 +725,12 @@ class BaseView(object):
for i, value in enumerate(row):
if isinstance(value, str):
row[i] = self.remove_tags(value)
elif isinstance(value, datetime):
is_aware_utc = value.tzinfo is not None \
and value.tzinfo.utcoffset(value) is not None
if is_aware_utc:
value = value.astimezone().replace(tzinfo=None)
row[i] = value
value = {
'header': header,
......@@ -732,11 +738,14 @@ class BaseView(object):
}
return value
def xls_response(self, **kwargs):
value = self.xls_data()
return xls_response(self.req, value)
def csv_response(self, **kwargs):
value = self.xls_data()
return csv_response(self.req, value)
def get_bindings(self, row=None):
return {"row": row}
......
opensipkd/base/views/dati2.py
......@@ -30,13 +30,16 @@ class AddSchema(colander.Schema):
validator=colander.Length(max=32),
oid="kode")
kode = colander.SchemaNode(colander.String(),
validator=colander.Length(max=32), oid="kode")
validator=colander.Length(max=32), oid="kode",
global_search=True,)
kategori = colander.SchemaNode(colander.String(),
widget=widget.SelectWidget(
values=kategori_dati2),
validator=colander.Length(max=32),
oid="kode")
nama = colander.SchemaNode(colander.String(), oid="nama")
oid="kode",
global_search=True,)
nama = colander.SchemaNode(colander.String(), oid="nama",
global_search=True,)
class EditSchema(AddSchema):
......
opensipkd/base/views/departemen.py
......@@ -86,8 +86,10 @@ class EditSchema(AddSchema):
class ListSchema(colander.Schema):
id = colander.SchemaNode(colander.String(), title="Action", visible=False)
kode = colander.SchemaNode(colander.String(), title="Kode")
nama = colander.SchemaNode(colander.String(), title="Nama")
kode = colander.SchemaNode(colander.String(), title="Kode",
global_search=True,)
nama = colander.SchemaNode(colander.String(), title="Nama",
global_search=True,)
status = colander.SchemaNode(colander.Boolean(), title="Status", width='50pt',
widget=widget.CheckboxWidget())
level_id = colander.SchemaNode(
......
opensipkd/base/views/desa.py
......@@ -43,12 +43,15 @@ class AddSchema(colander.Schema):
oid="kecamatan_id",
title="Kecamatan")
kode = colander.SchemaNode(colander.String(),
validator=colander.Length(max=32), oid="kode")
validator=colander.Length(max=32), oid="kode",
global_search=True,)
kategori = colander.SchemaNode(colander.String(),
widget=widget.SelectWidget(
values=kategori_desa),
validator=colander.Length(max=32), oid="kode")
nama = colander.SchemaNode(colander.String(), oid="nama")
validator=colander.Length(max=32), oid="kode",
global_search=True,)
nama = colander.SchemaNode(colander.String(), oid="nama",
global_search=True,)
class EditSchema(AddSchema):
......
opensipkd/base/views/groups.py
......@@ -42,8 +42,10 @@ class EditSchema(AddSchema):
class ListSchema(colander.Schema):
id = colander.SchemaNode(colander.Integer(), visible=False, title="Action")
group_name = colander.SchemaNode(
colander.String(), )
description = colander.SchemaNode(colander.String())
colander.String(),
global_search=True, )
description = colander.SchemaNode(colander.String(),
global_search=True,)
class Views(BaseView):
......
opensipkd/base/views/jabatan.py
......@@ -32,17 +32,21 @@ class AddSchema(colander.Schema):
kode = colander.SchemaNode(
colander.String(),
validator=colander.Length(max=32),
oid="kode")
oid="kode",
global_search=True,)
nama = colander.SchemaNode(
colander.String(),
oid="nama")
oid="nama",
global_search=True,)
nama_pendek = colander.SchemaNode(
colander.String(),
oid="nama_pendek",
global_search=True,
missing=colander.drop)
nama_lain = colander.SchemaNode(
colander.String(),
oid="nama_lain",
global_search=True,
missing=colander.drop)
jenis = colander.SchemaNode(
colander.Integer(),
......
opensipkd/base/views/js/printer_form.js 0 → 100644
$(function () {
$('#deformcheck').click(function (e) {
var url = '%s';
e.preventDefault();
console.log('Check button clicked: ' + url);
$.get(url, function (data) {
console.log("Data loaded:", data);
$('#errors').hide();
$('#success').text(data.message).show();
}).fail(function (jqXHR, textStatus, errorThrown,) {
var err = ("Error printing: " + errorThrown + " " + jqXHR.responseJSON.message);
$('#success').hide();
$('#errors').text(err).show();
});
});
});
\ No newline at end of file
opensipkd/base/views/kecamatan.py
......@@ -41,8 +41,10 @@ class AddSchema(colander.Schema):
oid="dati2_id")
kode = colander.SchemaNode(
colander.String(),
validator=colander.Length(max=32), oid="kode")
nama = colander.SchemaNode(colander.String(), oid="nama")
validator=colander.Length(max=32), oid="kode",
global_search=True,)
nama = colander.SchemaNode(colander.String(), oid="nama",
global_search=True,)
class EditSchema(AddSchema):
......
opensipkd/base/views/partner.py
......@@ -70,15 +70,18 @@ class ListSchema(colander.Schema):
validator=colander.Length(max=32),
oid="kode",
title="Kode",
width="100pt")
width="100pt",
global_search=True)
nama = colander.SchemaNode(
colander.String(),
validator=colander.Length(max=64),
oid="nama")
oid="nama",
global_search=True)
email = colander.SchemaNode(
colander.String(),
oid="email")
oid="email",
global_search=True)
idcard = colander.SchemaNode(
colander.String(),
oid="idcard",
......
opensipkd/base/views/permission.py
......@@ -18,12 +18,14 @@ class ListSchema(colander.Schema):
perm_name = colander.SchemaNode(
colander.String(),
oid="perm_name",
title="Nama")
title="Nama",
global_search=True)
description = colander.SchemaNode(
colander.String(),
missing=colander.drop,
oid="description",
title="Diskripsi")
title="Diskripsi",
global_search=True)
class EditSchema(ListSchema):
pass
......
opensipkd/base/views/printers.py
from pyramid.httpexceptions import HTTPBadRequest
import logging
import re
import colander
from deform import (widget,)
from pyramid.i18n import TranslationStringFactory
from ..models import TextPrinters
from . import BaseView, NamaSchema
import opensipkd
import platform
import websockets
import json
import os
import asyncio
from opensipkd.base import BASE_CLASS
from opensipkd.tools.buttons import btn_check
from opensipkd.tools import get_random_string
from ..models import TextPrinters, User
from . import BaseView
log = logging.getLogger(__name__)
......@@ -14,146 +24,59 @@ _ = TranslationStringFactory("opensipkd")
SESS_ADD_FAILED = 'Tambah partner gagal'
SESS_EDIT_FAILED = 'Edit partner gagal'
TEST_PRINT_DATA = \
"""*** TEST PRINT FROM WEB SERVER ***
This is a test print message sent from the server to the client.
def print_text(user_id, text=None, filename=None):
from opensipkd.base.models import TextPrinters
printer = TextPrinters.query().filter_by(user_id=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
ip = printer.kode if printer else "127.0.0.1"
ips = ip.split(".") # Support multiple IPs separated by commas
ips = [int(ip.strip()) for ip in ips if ip.strip()] # Clean up whitespace and empty entries
hostname = ".".join(map(str, ips)) if ips 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 = bool(printer.is_epson) if printer else False
from opensipkd.base import BASE_CLASS
from opensipkd.tools import get_random_string
import os
if not BASE_CLASS.is_pylpr:
import platform
current_os = platform.system().lower()
if current_os == "windows":
cmd = 'lpr'
else: # Darwin is macOS
cmd = 'rlpr'
if text and not filename:
filename = os.path.join(BASE_CLASS.temp_files,
get_random_string(16) + ".txt")
with open(filename, "w", encoding="utf-8") as f:
f.write(text)
try:
import subprocess
subprocess.run(
[cmd, f"--printer={queue}@{hostname}:{port}", filename], check=True)
except subprocess.CalledProcessError as e:
log.error(f"Failed to print using {cmd}: {e}")
raise Exception(f"Failed to print the document. {e}, {hostname}, {port}, {queue}") from e
else:
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:
try:
lpr.send(text.encode('utf-8'))
except Exception as e:
log.error(f"Failed to send print job: {e}")
raise Exception(
f"Failed to print the document. {e}, {hostname}, {port}, {queue}") from e
# lpr.send(text.encode('utf-8'))
If you see this on your printer, the connection and printing functionality are working
correctly."""
class AddSchema(NamaSchema):
class AddSchema(colander.Schema):
user_id = colander.SchemaNode(
colander.Integer(),
widget=widget.SelectWidget(),
title="User ID",
global_search=False,
searchable=True,
search_method="numeric",
default=10)
nama = colander.SchemaNode(
colander.String(),
title="Nama Printer",
validator=colander.Length(max=64)
validator=colander.Length(max=64),
global_search=True,
)
is_epson = colander.SchemaNode(
colander.Integer(),
widget=widget.SelectWidget(),
title="Printer/Network Type",
default=0)
kode = colander.SchemaNode(
colander.String(),
title="IP Address",
widget=widget.TextInputWidget(mask="999.999.999.999",))
widget=widget.TextInputWidget(mask="999.999.999.999",),
missing="",
global_search=True,
)
port = colander.SchemaNode(
colander.Integer(),
title="Port",
missing=0,
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",
global_search=True,
missing="",
default='lp')
timeout = colander.SchemaNode(
colander.Integer(),
title="Timeout",
default=10)
user_id = colander.SchemaNode(
colander.Integer(),
widget=widget.SelectWidget(),
title="User ID",
missing=0,
default=10)
status = colander.SchemaNode(
......@@ -167,12 +90,16 @@ class AddSchema(NamaSchema):
mask="999.999.999.999", # Use specialized mask format
mask_mapping={'Z': {'pattern': '[0-9]', 'optional': True}},
)
from opensipkd.base.models import User
schema["user_id"].widget.values = User.get_list() # Use a simple text input for user_id
request = kw.get('request')
if request and not request.has_permission('admin'):
schema["user_id"].widget = widget.HiddenWidget()
schema["user_id"].default = request.user.id
# Use a simple text input for user_id
schema["user_id"].widget.values = User.get_list()
schema["user_id"].aligned = "text-left"
schema["is_epson"].aligned = "text-left"
epson_values = [
("0", "Other"),
("1", "Epson Type"),]
if BASE_CLASS.wsp_url:
epson_values.append(("2", "Web Socket"))
schema["is_epson"].widget.values = epson_values
class EditSchema(AddSchema):
......@@ -183,7 +110,12 @@ class EditSchema(AddSchema):
class ListSchema(EditSchema):
pass
def after_bind(self, schema, kw):
super().after_bind(schema, kw)
request = kw.get('request')
if request and not request.has_permission('admin'):
schema["user_id"].widget = widget.HiddenWidget()
schema["user_id"].default = request.user.id
class Views(BaseView):
......@@ -199,6 +131,18 @@ class Views(BaseView):
self.save_state = True
self.allow_check = True
self.list_view_field = 'nama'
self.filter_columns = True
self.allow_delete = True
route_name = request.matched_route.name
id_ = request.matchdict.get('id')
if id_ and route_name.endswith('view'):
import os
folder = os.path.abspath(opensipkd.base.views.__file__)
filename = os.path.join(os.path.dirname(
folder), "js", "printer_form.js")
url = self.req.route_url(self.list_route)+f'/check/act?id={id_}'
with open(filename, "r", encoding="utf-8") as f:
self.form_scripts = f.read() % url
def list_filter(self, query, **kwargs):
if not self.req.has_permission('admin'):
......@@ -206,29 +150,51 @@ class Views(BaseView):
return query
def form_validator(self, form, value):
exc = colander.Invalid(form, None)
err = {}
err_format = 'Format IP harus 123.123.123.123'
found = TextPrinters.query().filter(
TextPrinters.nama == value['nama'],
TextPrinters.user_id == value['user_id']
).first()
if found:
if str(found.id) != self.req.matchdict.get('id'):
err["nama"] = 'Nama printer sudah digunakan untuk user ini'
is_epson = value['is_epson']
if is_epson < 2:
if not value.get('port'):
err["port"] = 'Port harus diisi untuk tipe Web Socket'
if not value.get('queue'):
err["queue"] = 'Queue harus diisi untuk tipe Web Socket'
if (not value.get('kode')
or not re.match(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}', value['kode'])):
err["kode"] = 'Kode harus diisi untuk tipe Web Socket'
else:
ip = value['kode'].split('.')
err = []
err_format = 'Format IP salah'
if not re.match(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}', value['kode']):
err.append(err_format+' 123.123.123.123')
if len(ip) != 4:
err.append(err_format+' harus 4 oktet')
err["kode"] = err_format+' terdiri dari 4 oktet'
else:
for i in ip:
if not i.isdigit():
err.append(err_format)
err["kode"] = err_format + \
' setiap oktet harus angka'
break
if int(i) < 0 or int(i) > 255:
err.append(err_format+' setiap oktet harus antara 0-255')
if ip[0] == '0':
err.append(err_format+' tidak boleh diawali 0')
err["kode"] = err_format + \
' setiap oktet harus antara 0-255'
break
if ip[0] == '000':
err["kode"] = err_format+' tidak boleh diawali 000'
break
if err:
exc['kode'] = '\n'.join(err)
exc = colander.Invalid(form, None)
for key, val in err.items():
exc[key] = val
raise exc
def after_save(self, values, row):
self.db_session.flush()
if values['status'] == 1:
printers = TextPrinters.query().filter_by(create_uid=row.create_uid).all()
printers = TextPrinters.query().filter_by(user_id=row.user_id).all()
for printer in printers:
if printer.id == row.id:
continue
......@@ -236,3 +202,164 @@ class Views(BaseView):
self.db_session.add(printer)
self.db_session.flush()
return row
def view_buttons(self, row):
buttons = super().view_buttons(row)
buttons += (btn_check,)
return buttons
def next_act(self, **kwargs):
id_ = self.req.GET.get('id')
try:
return asyncio.run(print_text(print_id=id_, text="Test Print"))
# return {'success': True,
# 'message': "Print job sent successfully."}
except Exception as e:
raise HTTPBadRequest(explanation=str(e)) from e
async def print_text(user_id=None, print_id=None, text=None, filename=None,
ws_url=BASE_CLASS.wsp_url):
log.debug(f"Starting print_text with user_id={user_id}, print_id={print_id}, filename={filename}")
if not user_id and not print_id:
log.error("User ID or Print ID must be provided.")
raise Exception("User ID or Print ID must be provided.")
if user_id:
printer = TextPrinters.query().filter_by(user_id=user_id, status=1).first()
else:
printer = TextPrinters.query().filter_by(id=print_id, status=1).first()
if not printer:
log.error(
f"User {user_id} or Print ID {print_id} does not have an active printer configured.")
raise Exception("No active printer configured for the user.")
is_ws = printer.is_epson == 2 if printer else False
timeout = printer.timeout if printer else 10 # Timeout in seconds
if is_ws:
printer_name = printer.nama if printer else "unknown"
printer_name = BASE_CLASS.wsp_client_id+'_'+ printer_name
ws_url = ws_url or BASE_CLASS.wsp_url
if not ws_url:
log.error("WebSocket URL is not configured.")
raise Exception("WebSocket URL is not configured.")
try:
async with websockets.connect(ws_url) as websocket:
# Send user on connect
log.info(f"Connected to server at {ws_url} as {printer_name}")
# data = {"action": "print",
# "printer": self.printer,
# "message": message
# }
data = {
"action": "login",
"client_id": BASE_CLASS.wsp_client_id,
"api_key": BASE_CLASS.wsp_client_key,
"printer": BASE_CLASS.wsp_client_id
}
await websocket.send(json.dumps(data))
text = await websocket.recv()
log.info(f"Received from server: {text}")
try:
resp = json.loads(text)
except json.JSONDecodeError as e:
log.error(f"Failed to decode server response: {e}")
raise Exception(f"Invalid response from server: {text}") from e
if resp.get("status") != True:
log.error(
f"Login failed: {resp.get('message', 'No message provided')}")
raise Exception(f"Login failed: {resp.get('message', 'No message provided')}")
# Wait for server to process login
if filename and os.path.isfile(filename):
ext = os.path.splitext(filename)[1].lower()
if ext in {".prn", ".txt"}:
with open(filename, "r", encoding="utf-8") as f:
message = f.read()
else:
with open(filename, "rb") as file:
message = file.read()
else:
message = TEST_PRINT_DATA
data = {"action": "print",
"printer": printer_name,
"filename": os.path.basename(filename) if filename else "test_print.txt",
"message": message
}
await websocket.send(json.dumps(data))
resp = await websocket.recv()
data = json.loads(resp)
if not data.get("status"):
log.error(
f"Server returned an error: {data.get('message')}")
raise Exception(f"Server error: {data.get('message')}")
# Wait for any response from server
log.info(f"Received from server: {resp}")
return data
except (websockets.ConnectionClosed, ConnectionRefusedError, OSError) as e:
msg = f"Failed to connect to WebSocket server at {ws_url}: {e}"
log.error(msg)
raise Exception(msg) from e
# Replace with your printer's IP
ip = printer.kode if printer else "127.0.0.1"
ips = ip.split(".") # Support multiple IPs separated by commas
# Clean up whitespace and empty entries
ips = [int(ip.strip()) for ip in ips if ip.strip()]
hostname = ".".join(map(str, ips)) if ips 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
# Set to True if the printer is an Epson model that requires specific control codes
is_epson = printer.is_epson == 1 if printer else False
if not BASE_CLASS.is_pylpr:
current_os = platform.system().lower()
if current_os == "windows":
cmd = 'lpr'
else: # Darwin is macOS
cmd = 'rlpr'
if text and not filename:
filename = os.path.join(BASE_CLASS.temp_files,
get_random_string(16) + ".txt")
with open(filename, "w", encoding="utf-8") as f:
f.write(text)
try:
import subprocess
subprocess.run(
[cmd, f"--printer={queue}@{hostname}:{port}", filename], check=True)
except subprocess.CalledProcessError as e:
log.error(f"Failed to print using {cmd}: {e}")
raise Exception(
f"Failed to print the document. {e}, {hostname}, {port}, {queue}") from e
finally:
if filename and os.path.isfile(filename):
try:
os.remove(filename)
except Exception as e:
log.warning(f"Failed to delete temp file {filename}: {e}")
else:
from pylpr import LprClient
with LprClient(
hostname=hostname,
port=port,
timeout=timeout,
queue=queue,
) as lpr:
if is_epson:
lpr.send(
lpr.EXIT_PACKET_MODE
+ lpr.INITIALIZE_PRINTER
+ text.encode('utf-8')
+ lpr.FF
)
else:
try:
lpr.send(text.encode('utf-8'))
except Exception as e:
log.error(f"Failed to send print job: {e}")
raise Exception(
f"Failed to print the document. {e}, {hostname}, {port}, {queue}") from e
lpr.send(text.encode('utf-8'))
opensipkd/base/views/provinsi.py
......@@ -27,14 +27,17 @@ def provinsi_widget(node, kw):
class AddSchema(colander.Schema):
kode = colander.SchemaNode(colander.String(),
oid="kode",
validator=colander.Length(max=32), )
validator=colander.Length(max=32),
global_search=True, )
kategori = colander.SchemaNode(colander.String(),
widget=widget.SelectWidget(
values=kategori_provinsi),
validator=colander.Length(max=32), oid="kode")
nama = colander.SchemaNode(colander.String(), oid="nama")
nama = colander.SchemaNode(colander.String(), oid="nama",
global_search=True,)
ibu_kota = colander.SchemaNode(
colander.String(), oid="nama", missing=colander.drop)
colander.String(), oid="nama", missing=colander.drop,
global_search=True,)
class EditSchema(AddSchema):
......
opensipkd/base/views/user.py
import os
import re
from shapely import node
import colander
from deform import (widget, )
from opensipkd.base import BASE_CLASS
from pyramid.i18n import TranslationStringFactory
from sqlalchemy import (func, true, )
from sqlalchemy import (func )
from ziggurat_foundations.models.services.user import UserService
from opensipkd.tools import create_now
from opensipkd.tools.buttons import btn_delete
from . import BaseView
# from .company import company_widget
from .user_login import (
regenerate_security_code, send_email_security_code, generate_api_key, )
from ..models.users import (DBSession, User, Group, UserGroup,)
......@@ -25,9 +23,11 @@ class ListSchema(colander.Schema):
id = colander.SchemaNode(colander.String(),
title="Action",
searchable=False)
email = colander.SchemaNode(colander.String())
email = colander.SchemaNode(colander.String(),
global_search=True)
user_name = colander.SchemaNode(colander.String(),
title=_("user-name", default="User Name"))
title=_("user-name", default="User Name"),
global_search=True)
status = colander.SchemaNode(colander.Integer(),
widget=widget.CheckboxWidget(),
width=50, searchable=False)
......
opensipkd/base/views/user_area.py
......@@ -14,14 +14,16 @@ class ListSchema(colander.Schema):
user_name = colander.SchemaNode(
colander.String(),
field=User.user_name,
title="User"
title="User",
global_search=True,
)
desa_kd = colander.SchemaNode(
colander.String(),
field=ResDesa.kode)
desa_name = colander.SchemaNode(
colander.String(),
field=ResDesa.nama)
field=ResDesa.nama,
global_search=True,)
class AddSchema(colander.Schema):
......
opensipkd/detable/detable.py
......@@ -190,6 +190,14 @@ class DeTable(field.Field):
window.location = o%sUri+'/csv/act%s&'+paramString;}
""" % (
tableid, tableid, params),
"csv2": """{
params = o%s.ajax.params();
paramString = $.param(params);
console.log(params);
window.location = o%sUri+'/csv2/act%s&'+paramString;}
""" % (
tableid, tableid, params),
"xls": """{
params = o%s.ajax.params();
paramString = $.param(params);
......
pyproject.toml
......@@ -38,6 +38,8 @@ dependencies = [
'wheezy.captcha',
'ziggurat-foundations',
'zope.sqlalchemy',
'websockets',
"asyncio",
]
requires-python = '>= 3.8'
authors = [
......
Markdown is supported
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!