import json import sqlite3 from datetime import datetime from decimal import Decimal, InvalidOperation, ROUND_HALF_UP from pypos.core.utils.db_helper import connect_sqlite from pypos.core.utils.device_utils import get_active_device_info, get_device_id from pypos.core.utils.path_utils import get_db_path from pypos.core.utils.sql_query_builder import render_sql_template from pypos.modules.penjualan.config.penjualan_config import get_toko_id_from_config from pypos.modules.penjualan.models.detail_transaksi_model import DetailTransaksi class TransaksiPayloadService: # edited by glg def _to_decimal(self, value, default="0"): try: return Decimal(str(value if value is not None else default)) except (InvalidOperation, ValueError, TypeError): return Decimal(str(default)) # edited by glg def _to_rupiah_int(self, value): decimal_val = self._to_decimal(value, "0") return int(decimal_val.quantize(Decimal("1"), rounding=ROUND_HALF_UP)) def _normalisasi_harga(self, harga_text): if harga_text is None: return 0.0 text = str(harga_text).strip() if not text: return 0.0 if "," not in text and "." not in text: try: return float(text) except (TypeError, ValueError): return 0.0 if "," in text: text = text.replace(".", "") text = text.replace(",", ".") try: return float(text) except (TypeError, ValueError): return 0.0 def _parse_int(self, value): try: return int(str(value).strip()) except (TypeError, ValueError): return 0 def _parse_float(self, value): try: return float(str(value).strip().replace(",", ".")) except (TypeError, ValueError): return 0.0 # edited by glg def _as_positive_int(self, value, default=0): try: parsed = int(str(value).strip()) except (TypeError, ValueError): parsed = int(default or 0) return parsed if parsed > 0 else 0 # edited by glg def _safe_close_conn(self, conn): if conn is None: return try: conn.close() except sqlite3.Error: return # edited by glg def _lookup_cabang_nama(self, cabang_id): cabang_int = self._as_positive_int(cabang_id, 0) if cabang_int <= 0: return "" conn = None try: conn = connect_sqlite(get_db_path()) cur = conn.cursor() cur.execute( """ SELECT nama FROM per_cabang WHERE id = ? AND COALESCE(status, 1) = 1 AND COALESCE(trash, 0) = 0 LIMIT 1 """, (cabang_int,), ) row = cur.fetchone() if row and row[0]: return str(row[0]).strip() return "" except sqlite3.Error: return "" finally: self._safe_close_conn(conn) # edited by glg def _lookup_toko_nama(self, toko_id): toko_int = self._as_positive_int(toko_id, 0) if toko_int <= 0: return "" conn = None try: conn = connect_sqlite(get_db_path()) cur = conn.cursor() cur.execute( """ SELECT nama FROM company_profile WHERE toko_id = ? AND COALESCE(status, 1) = 1 AND COALESCE(trash, 0) = 0 ORDER BY id ASC LIMIT 1 """, (toko_int,), ) row = cur.fetchone() if row and row[0]: return str(row[0]).strip() return "" except sqlite3.Error: return "" finally: self._safe_close_conn(conn) # edited by glg # Source of truth gudang POS: default gudang pada per_cabang. # Semua POS dalam cabang yang sama wajib memakai gudang yang sama. def _lookup_cabang_default_gudang(self, cabang_id): cabang_int = self._as_positive_int(cabang_id, 0) if cabang_int <= 0: return {"gudang_id": 0, "gudang_nama": ""} cache = getattr(self, "_cabang_default_gudang_cache", None) if cache is None: cache = {} setattr(self, "_cabang_default_gudang_cache", cache) if cabang_int in cache: cached = cache.get(cabang_int) or {} return { "gudang_id": self._parse_int(cached.get("gudang_id")), "gudang_nama": str(cached.get("gudang_nama") or "").strip(), } conn = None result = {"gudang_id": 0, "gudang_nama": ""} try: conn = connect_sqlite(get_db_path()) cur = conn.cursor() cur.execute("PRAGMA table_info(per_cabang)") cols = {str(row[1]) for row in cur.fetchall() if row and len(row) > 1} if not cols: cache[cabang_int] = dict(result) return dict(result) gudang_id_candidates = [ "gudang_id", "gudangID", "default_gudang_id", "id_gudang", ] gudang_nama_candidates = [ "gudang_nama", "gudangName", "default_gudang_nama", "nama_gudang", ] gudang_id_col = next((c for c in gudang_id_candidates if c in cols), "") gudang_nama_col = next((c for c in gudang_nama_candidates if c in cols), "") if not gudang_id_col and not gudang_nama_col: cache[cabang_int] = dict(result) return dict(result) selected_cols = ["id"] if gudang_id_col: selected_cols.append(gudang_id_col) if gudang_nama_col: selected_cols.append(gudang_nama_col) if "status" in cols: where_status = " AND COALESCE(status, 1) = 1" else: where_status = "" if "trash" in cols: where_trash = " AND COALESCE(trash, 0) = 0" else: where_trash = "" query = render_sql_template( "SELECT {selected_cols} FROM per_cabang WHERE id = ?{where_status}{where_trash} LIMIT 1", selected_cols=", ".join(selected_cols), where_status=where_status, where_trash=where_trash, ) cur.execute(query, (cabang_int,)) row = cur.fetchone() if not row: cache[cabang_int] = dict(result) return dict(result) description = getattr(cur, "description", None) or [] row_map = {} for idx, desc in enumerate(description): if idx < len(row): row_map[str(desc[0])] = row[idx] parsed_gudang_id = self._parse_int( row_map.get(gudang_id_col) if gudang_id_col else 0 ) parsed_gudang_nama = str( row_map.get(gudang_nama_col) if gudang_nama_col else "" ).strip() result = { "gudang_id": parsed_gudang_id, "gudang_nama": parsed_gudang_nama, } cache[cabang_int] = dict(result) return dict(result) except (sqlite3.Error, TypeError, ValueError): cache[cabang_int] = dict(result) return dict(result) finally: self._safe_close_conn(conn) # edited by glg def _resolve_transaksi_context(self, user_info): user = user_info if isinstance(user_info, dict) else {} machine_id = str( user.get("machine_id") or user.get("device_id") or get_device_id() or "" ).strip() active_info = get_active_device_info(machine_id) or {} cabang_id = self._as_positive_int( user.get("cabang_id"), self._as_positive_int(active_info.get("cabang_id"), 0), ) toko_id = self._as_positive_int( user.get("toko_id"), self._as_positive_int(active_info.get("toko_id"), 0), ) if toko_id <= 0: toko_id = self._as_positive_int(get_toko_id_from_config(default=0), 0) gudang_raw = user.get("gudang_id") if gudang_raw in (None, ""): gudang_raw = user.get("gudangID") gudang_id_user = self._parse_int(gudang_raw) gudang_nama_user = str( user.get("gudang_nama") or user.get("nama_gudang") or "" ).strip() cabang_default_gudang = self._lookup_cabang_default_gudang(cabang_id) gudang_id_cabang = self._parse_int(cabang_default_gudang.get("gudang_id")) gudang_nama_cabang = str(cabang_default_gudang.get("gudang_nama") or "").strip() # edited by glg # Prioritas default cabang untuk memastikan semua POS dalam cabang pakai gudang yang sama. if gudang_id_cabang != 0: gudang_id = gudang_id_cabang gudang_nama = gudang_nama_cabang or gudang_nama_user else: gudang_id = gudang_id_user gudang_nama = gudang_nama_user cabang_nama = str( user.get("cabang_nama") or user.get("nama_cabang") or active_info.get("cabang_nama") or "" ).strip() if not cabang_nama and cabang_id > 0: cabang_nama = self._lookup_cabang_nama(cabang_id) toko_nama = str( user.get("toko_nama") or user.get("nama_toko") or active_info.get("toko_nama") or "" ).strip() if not toko_nama and toko_id > 0: toko_nama = self._lookup_toko_nama(toko_id) return { "machine_id": machine_id, "cabang_id": cabang_id, "cabang_nama": cabang_nama, "toko_id": toko_id, "toko_nama": toko_nama, "gudang_id": gudang_id, "gudang_nama": gudang_nama, "cpu_info": str(user.get("cpu_info") or "").strip(), "com_info": str(user.get("com_info") or "").strip(), } # edited by glg def build_detail_data( self, rows, on_row_error=None, strict_mode=False, error_log_limit=20, ): total_harga = 0 detail_data = [] errors_logged = 0 max_error_logs = max(1, int(error_log_limit or 1)) def _notify_row_error(row_index, reason, raw_row=None): nonlocal errors_logged errors_logged += 1 payload = { "stage": "build_detail_data", "row_index": int(row_index), "reason": str(reason or "invalid_row"), } if isinstance(raw_row, dict): payload["produk_id"] = str(raw_row.get("produk_id") or "").strip() payload["produk_nama"] = str(raw_row.get("produk_nama") or "").strip() if callable(on_row_error) and errors_logged <= max_error_logs: try: on_row_error(payload) except (TypeError, ValueError, KeyError, AttributeError, RuntimeError): pass if bool(strict_mode): raise ValueError( f"transaksi_payload_row_invalid stage=build_detail_data row={row_index} reason={payload['reason']}" ) for index, row in enumerate(list(rows or [])): try: produk_id = self._parse_int(row.get("produk_id")) if produk_id <= 0: _notify_row_error(index, "invalid_produk_id", row) continue produk_nama = str(row.get("produk_nama") or "") harga = self._normalisasi_harga(row.get("harga_text")) jumlah = self._parse_int(row.get("jumlah_text")) if jumlah <= 0: _notify_row_error(index, "invalid_jumlah", row) continue diskon = self._parse_float(row.get("diskon_text")) diskon_rp_per_unit = self._normalisasi_harga(row.get("diskon_rp_text")) harga_diskon = harga * (1 - (diskon / 100)) # edited by glg # Source-of-truth subtotal transaksi harus mengikuti nilai baris UI # (yang juga dipakai pada dialog pembayaran) untuk mencegah drift rupiah # akibat pembulatan persen diskon 2 desimal. subtotal_raw = row.get("subtotal_text") has_subtotal_ui = subtotal_raw is not None and str(subtotal_raw).strip() != "" if has_subtotal_ui: subtotal = int(round(self._normalisasi_harga(subtotal_raw))) else: subtotal = int(round(harga_diskon * jumlah)) total_harga += subtotal satuan = str(row.get("satuan") or "") produk_jenis = str(row.get("produk_jenis") or "invoice").strip().lower() or "invoice" # edited by glg # Jika promo free dipilih, pisahkan baris paid vs baris hadiah. # Baris utama tetap invoice (berbayar), baris hadiah dicatat sebagai free_produk. if produk_jenis == "free_produk": detail_data.append( DetailTransaksi( produk_id=produk_id, produk_nama=produk_nama, produk_ord_hrg=harga, produk_ord_jml=jumlah, produk_jenis="invoice", produk_ord_diskon=0, satuan=satuan, detail_tipe="items", ) ) free_qty = max(0, self._parse_int(row.get("free_qty"))) if free_qty > 0: free_produk_id = self._parse_int(row.get("free_produk_id")) if free_produk_id <= 0: free_produk_id = produk_id free_produk_nama = str(row.get("free_produk_nama") or "").strip() or produk_nama relasi_payload = { "source_produk_id": produk_id, "source_produk_nama": produk_nama, "source_qty": jumlah, "free_produk_id": free_produk_id, "free_produk_nama": free_produk_nama, "free_qty": free_qty, "reason": "promo_free_produk", } detail_data.append( DetailTransaksi( produk_id=free_produk_id, produk_nama=free_produk_nama, produk_ord_hrg=0, produk_ord_jml=free_qty, produk_jenis="free_produk", produk_ord_diskon=0, satuan=satuan, parent_id=produk_id, detail_tipe="free_item", produk_keterangan=f"FREE dari {produk_nama} (id:{produk_id})", ext_status="free_produk", ext_intext=json.dumps( {"free_relasi": relasi_payload}, ensure_ascii=False, separators=(",", ":"), ), free_source_produk_id=produk_id, free_source_produk_nama=produk_nama, ) ) continue detail_row = DetailTransaksi( produk_id=produk_id, produk_nama=produk_nama, produk_ord_hrg=harga, produk_ord_jml=jumlah, produk_jenis=produk_jenis, produk_ord_diskon=diskon, satuan=satuan, detail_tipe="items", ) # edited by glg # Simpan metadata diskon untuk menjaga konsistensi angka struk: # - produk_ord_diskon: persen (existing contract) # - produk_ord_diskon_nominal: nominal line discount (baru, untuk print akurat) diskon_nominal_line = 0.0 if diskon_rp_per_unit > 0: diskon_nominal_line = max(0.0, diskon_rp_per_unit * jumlah) elif diskon > 0: diskon_nominal_line = max(0.0, (harga * jumlah) * (diskon / 100.0)) setattr(detail_row, "produk_ord_diskon_persen", float(diskon or 0.0)) setattr(detail_row, "produk_ord_diskon_nominal", float(diskon_nominal_line)) detail_data.append(detail_row) except (TypeError, ValueError, KeyError, AttributeError) as exc: _notify_row_error(index, f"exception:{exc}", row) continue return detail_data, total_harga # edited by glg @staticmethod def _normalize_diskon_ctx(diskon_customer_ctx): return diskon_customer_ctx if isinstance(diskon_customer_ctx, dict) else {} # edited by glg @staticmethod def _normalize_ppn_mode(ppn_mode): mode = str(ppn_mode or "include").strip().lower() if mode not in {"include", "exclude"}: return "include" return mode # edited by glg def _build_financial_context( self, *, total_harga, diskon_input, diskon_customer_ctx, ppn_percent, ppn_mode, ): diskon_persen_raw = diskon_input if diskon_input is not None else 0 diskon_persen_calc = max(Decimal("0"), self._to_decimal(diskon_persen_raw, "0")) total_harga_dec = max(Decimal("0"), self._to_decimal(total_harga, "0")) diskon_customer_nilai_dec = max( Decimal("0"), self._to_decimal(diskon_customer_ctx.get("diskon_nilai", 0), "0"), ) base_after_customer = max(Decimal("0"), total_harga_dec - diskon_customer_nilai_dec) diskon_nilai_dec = (base_after_customer * diskon_persen_calc) / Decimal("100") mode = self._normalize_ppn_mode(ppn_mode) net_total = max(Decimal("0"), base_after_customer - diskon_nilai_dec) ppn_rate = max(Decimal("0"), self._to_decimal(ppn_percent, "0")) if mode == "exclude": ppn = (net_total * ppn_rate) / Decimal("100") total_bayar = net_total + ppn else: ppn = ( net_total * (ppn_rate / (Decimal("100") + ppn_rate)) if ppn_rate > 0 else Decimal("0") ) total_bayar = net_total diskon_nilai_total = diskon_customer_nilai_dec + diskon_nilai_dec return { "diskon_persen_raw": diskon_persen_raw, "diskon_customer_nilai_dec": diskon_customer_nilai_dec, "mode": mode, "total_bayar_int": self._to_rupiah_int(total_bayar), "ppn_int": self._to_rupiah_int(ppn), "diskon_nilai_total_int": self._to_rupiah_int(diskon_nilai_total), "total_harga_int": self._to_rupiah_int(total_harga_dec), } # edited by glg def _build_transaksi_identity(self, user_info): now = datetime.now() return { "nomer": now.strftime("%Y%m%d%H%M%S") + "-" + str(user_info.get("nama") or ""), "dtime": now.strftime("%Y-%m-%d %H:%M:%S"), "fulldate": now.strftime("%Y-%m-%d"), } # edited by glg @staticmethod def _build_transaksi_data_tuple(identity_ctx, financial_ctx, customer_id, customer_text, user_info): return ( identity_ctx["nomer"], identity_ctx["dtime"], financial_ctx["total_bayar_int"], financial_ctx["diskon_persen_raw"], financial_ctx["ppn_int"], financial_ctx["total_harga_int"], customer_id, customer_text, identity_ctx["fulldate"], user_info.get("id"), user_info.get("nama"), "invoice", "758", "1", financial_ctx["mode"], ) # edited by glg @staticmethod def _resolve_applied_rule_ids(diskon_customer_ctx): applied_rule_ids = [] for rid in (diskon_customer_ctx.get("applied_rule_ids") or []): try: rid_int = int(rid) except (TypeError, ValueError): continue if rid_int > 0 and rid_int not in applied_rule_ids: applied_rule_ids.append(rid_int) return applied_rule_ids # edited by glg @staticmethod def _resolve_rule_id_utama(diskon_customer_ctx): try: rule_obj = diskon_customer_ctx.get("rule") or {} if rule_obj: raw_rule_id = int(rule_obj.get("id") or 0) if raw_rule_id > 0: return str(raw_rule_id) except (TypeError, ValueError, AttributeError): return "" return "" # edited by glg def _build_rule_context(self, diskon_customer_ctx): applied_rule_ids = self._resolve_applied_rule_ids(diskon_customer_ctx) return { "applied_rule_ids_text": ",".join(str(v) for v in applied_rule_ids), "rule_id_utama": self._resolve_rule_id_utama(diskon_customer_ctx), "point_transaksi": self._to_rupiah_int(diskon_customer_ctx.get("point_nilai", 0) or 0), } # edited by glg @staticmethod def _build_diskon_log( *, diskon_customer_nilai_int, cashback_nilai, point_transaksi, rule_id_utama, applied_rule_ids_text, mode, ): return ( f"diskon_customer={diskon_customer_nilai_int};" f"cashback={int(round(cashback_nilai or 0))};" f"point={point_transaksi};" f"rule_id={rule_id_utama};" f"rule_ids={applied_rule_ids_text};" f"ppn_mode={mode}" ) # edited by glg def _build_transaksi_data_dict( self, *, identity_ctx, financial_ctx, customer_id, customer_nama_asli, user_info, diskon_customer_ctx, rule_ctx, trx_context, ): mode = financial_ctx["mode"] diskon_customer_nilai_dec = financial_ctx["diskon_customer_nilai_dec"] diskon_customer_nilai_int = self._to_rupiah_int(diskon_customer_nilai_dec) point_transaksi = rule_ctx["point_transaksi"] return { "nomer": identity_ctx["nomer"], "nomer2": identity_ctx["nomer"], "dtime": identity_ctx["dtime"], "transaksi_nilai": financial_ctx["total_bayar_int"], "diskon_persen": financial_ctx["diskon_persen_raw"], "diskon_nilai": financial_ctx["diskon_nilai_total_int"], "ppn_persen": financial_ctx["ppn_int"], "ppn_mode": mode, "transaksi_bulat": financial_ctx["total_harga_int"], "customers_id": customer_id, "customers_nama": customer_nama_asli, "fulldate": identity_ctx["fulldate"], "oleh_id": user_info.get("id"), "oleh_nama": user_info.get("nama"), "inv": identity_ctx["nomer"], "link_id": 0, "status": 1, "trash": 0, "jenis": 582, "jenis_label": "invoice", "transaksi_jenis": "758", "jumlah_bayar": 0, "kembalian": 0, "transaksi_dibayar": 0, "transaksi_dibayar_return": 0, "pembayaran": "", "pembayaran_sys": "tunai", "pembayaran_tunai": 0, "kasir_nama": user_info.get("nama"), "customer_nama": customer_nama_asli, "skip_logo": True, "diskon_jenis": "diskon_customer" if diskon_customer_nilai_dec > 0 else "", "diskon_log": self._build_diskon_log( diskon_customer_nilai_int=diskon_customer_nilai_int, cashback_nilai=diskon_customer_ctx.get("cashback_nilai", 0), point_transaksi=point_transaksi, rule_id_utama=rule_ctx["rule_id_utama"], applied_rule_ids_text=rule_ctx["applied_rule_ids_text"], mode=mode, ), "point_transaksi": point_transaksi, "cabang_id": trx_context["cabang_id"], "cabang_nama": trx_context["cabang_nama"], "toko_id": trx_context["toko_id"], "toko_nama": trx_context["toko_nama"], "gudang_id": trx_context["gudang_id"], "gudang_nama": trx_context["gudang_nama"], "reference_id": 0, "reference_jenis": 0, "reference_nomer": 0, "machine_id": trx_context["machine_id"], "cpu_info": trx_context["cpu_info"], "com_info": trx_context["com_info"], "detail_context": { "dtime": identity_ctx["dtime"], "cabang_id": trx_context["cabang_id"], "oleh_id": user_info.get("id"), "oleh_nama": user_info.get("nama"), }, } def build_transaksi_payload( self, *, detail_data, total_harga, diskon_input, customer_id, customer_text, customer_nama_asli, user_info, diskon_customer_ctx, ppn_percent, ppn_mode="include", ): user_info = user_info if isinstance(user_info, dict) else {} diskon_customer_ctx = self._normalize_diskon_ctx(diskon_customer_ctx) trx_context = self._resolve_transaksi_context(user_info) financial_ctx = self._build_financial_context( total_harga=total_harga, diskon_input=diskon_input, diskon_customer_ctx=diskon_customer_ctx, ppn_percent=ppn_percent, ppn_mode=ppn_mode, ) identity_ctx = self._build_transaksi_identity(user_info) transaksi_data = self._build_transaksi_data_tuple( identity_ctx=identity_ctx, financial_ctx=financial_ctx, customer_id=customer_id, customer_text=customer_text, user_info=user_info, ) rule_ctx = self._build_rule_context(diskon_customer_ctx) transaksi_data_dict = self._build_transaksi_data_dict( identity_ctx=identity_ctx, financial_ctx=financial_ctx, customer_id=customer_id, customer_nama_asli=customer_nama_asli, user_info=user_info, diskon_customer_ctx=diskon_customer_ctx, rule_ctx=rule_ctx, trx_context=trx_context, ) return transaksi_data, detail_data, transaksi_data_dict