import re import sqlite3 from pypos.core.utils.db_helper import connect_sqlite from pypos.core.utils.path_utils import get_db_path from pypos.core.utils.sql_query_builder import render_sql_template class ReprintDetailItem: def __init__(self, kode, nama, harga, jumlah, diskon, satuan): self.produk_kode = kode self.produk_nama = nama self.produk_ord_hrg = harga self.produk_ord_jml = jumlah self.produk_ord_diskon = diskon self.produk_satuan = satuan self.produk_jenis = "" class HistoryReprintService: def __init__(self, db_path=None, print_controller=None, print_controller_factory=None): self._print_controller = print_controller self._print_controller_factory = print_controller_factory self._db_path = db_path or get_db_path() def _build_default_print_controller(self): settings_module = __import__( "pypos.modules.printer.controllers.printer_settings_controller", fromlist=["PrinterSettingsController"], ) print_module = __import__( "pypos.modules.printer.controllers.print_controller", fromlist=["PrintController"], ) settings_controller = settings_module.PrinterSettingsController() return print_module.PrintController(settings_controller) def _get_print_controller(self): controller = getattr(self, "_print_controller", None) if controller is not None: return controller factory = getattr(self, "_print_controller_factory", None) if callable(factory): controller = factory() else: controller = self._build_default_print_controller() self._print_controller = controller return controller # edited by glg def _to_non_negative_int(self, value, default=0): try: parsed = int(float(value)) except (TypeError, ValueError, OverflowError): parsed = int(default or 0) return parsed if parsed > 0 else 0 # edited by glg def _extract_point_from_diskon_log(self, diskon_log): text = str(diskon_log or "").strip() if not text: return 0 match = re.search(r"(?:^|;)point=([^;]*)", text) if not match: return 0 return self._to_non_negative_int(match.group(1), 0) # edited by glg def _extract_ppn_mode_from_diskon_log(self, diskon_log): text = str(diskon_log or "").strip().lower() if not text: return "" match = re.search(r"(?:^|;)ppn_mode=([^;]*)", text) if not match: return "" mode = str(match.group(1) or "").strip().lower() return mode if mode in {"include", "exclude"} else "" # edited by glg def _fetch_point_context(self, transaksi_id): conn = connect_sqlite(self._db_path) try: cursor = conn.cursor() cursor.execute("PRAGMA table_info(transaksi)") cols = {str(row[1]) for row in cursor.fetchall() if row and len(row) > 1} select_cols = [] if "point_transaksi" in cols: select_cols.append("point_transaksi") if "diskon_log" in cols: select_cols.append("diskon_log") if not select_cols: return {"point_transaksi": 0, "diskon_log": ""} query = render_sql_template( "SELECT {select_cols} FROM transaksi WHERE id = ? LIMIT 1", select_cols=", ".join(select_cols), ) cursor.execute( query, (transaksi_id,), ) row = cursor.fetchone() if not row: return {"point_transaksi": 0, "diskon_log": ""} row_map = {} for idx, col_name in enumerate(select_cols): row_map[col_name] = row[idx] point_transaksi = self._to_non_negative_int( row_map.get("point_transaksi"), self._extract_point_from_diskon_log(row_map.get("diskon_log")), ) return { "point_transaksi": point_transaksi, "diskon_log": str(row_map.get("diskon_log") or ""), } except (sqlite3.Error, TypeError, ValueError, AttributeError, IndexError): return {"point_transaksi": 0, "diskon_log": ""} finally: conn.close() def _build_transaksi_data(self, transaksi_id, invoice_number, header_data): customer_id = header_data[0] if len(header_data) > 0 else 0 customer_nama = header_data[1] if len(header_data) > 1 else "" diskon_persen = header_data[2] if len(header_data) > 2 else 0 ppn_persen = header_data[3] if len(header_data) > 3 else 0 transaksi_nilai = header_data[4] if len(header_data) > 4 else 0 dtime = header_data[5] if len(header_data) > 5 else "" kasir_id = header_data[6] if len(header_data) > 6 else 0 kasir_nama = header_data[7] if len(header_data) > 7 else "" metode_pembayaran = header_data[8] if len(header_data) > 8 else "" bank_id = header_data[9] if len(header_data) > 9 else 0 bank_nama = header_data[10] if len(header_data) > 10 else "" settlement_id = header_data[11] if len(header_data) > 11 else 1 jumlah_bayar = header_data[12] if len(header_data) > 12 else 0 kembalian = header_data[13] if len(header_data) > 13 else 0 approval_code = header_data[14] if len(header_data) > 14 else "" no_kartu = header_data[15] if len(header_data) > 15 else "" header_ppn_mode = str(header_data[16] if len(header_data) > 16 else "").strip().lower() if header_ppn_mode not in {"include", "exclude"}: header_ppn_mode = "" point_ctx = self._fetch_point_context(transaksi_id) ppn_mode = header_ppn_mode or self._extract_ppn_mode_from_diskon_log(point_ctx.get("diskon_log", "")) transaksi_data = ( transaksi_id, invoice_number, customer_id, customer_nama, diskon_persen, ppn_persen, transaksi_nilai, dtime, kasir_id, kasir_nama, "simpan_transaksi", None, metode_pembayaran, settlement_id, ) transaksi_data_dict = { "jumlah_bayar": jumlah_bayar, "kembalian": kembalian, "metode_pembayaran": metode_pembayaran, "bank_nama": bank_nama if bank_nama else "", "approval_code": approval_code if approval_code else "", "no_kartu": no_kartu if no_kartu else "", "transaksi_dibayar": jumlah_bayar, "transaksi_dibayar_return": kembalian, "nomer": invoice_number, "dtime": dtime, "kasir_nama": kasir_nama, "customer_nama": customer_nama, "skip_logo": True, "point_transaksi": point_ctx.get("point_transaksi", 0), "ppn_mode": ppn_mode, "diskon_log": point_ctx.get("diskon_log", ""), } return transaksi_data, transaksi_data_dict def _build_detail_data(self, detail_rows): detail_data = [] for item in detail_rows: diskon = item[4] if len(item) > 4 else 0 satuan = item[5] if len(item) > 5 else "" detail_data.append( ReprintDetailItem( kode=item[0], nama=item[1], harga=item[2], jumlah=item[3], diskon=diskon, satuan=satuan, ) ) return detail_data def validate_payload(self, transaksi_id, invoice_number, header_data, detail_rows): if not transaksi_id: return False, "ID transaksi tidak valid." if not invoice_number: return False, "Nomor invoice tidak valid." if not header_data or len(header_data) < 16: return False, "Data header transaksi tidak lengkap." if not detail_rows: return False, "Detail transaksi kosong." return True, "" def reprint(self, transaksi_id, invoice_number, header_data, detail_rows): transaksi_data, transaksi_data_dict = self._build_transaksi_data( transaksi_id, invoice_number, header_data, ) detail_data = self._build_detail_data(detail_rows) success = self._get_print_controller().print_struk( transaksi_data=transaksi_data, detail_data=detail_data, transaksi_data_dict=transaksi_data_dict, index_printer=None, copy_no=1, ) if success: return True, "" return False, "Print gagal. Pastikan printer terhubung."