import sqlite3, math from datetime import datetime from dataclasses import dataclass from pypos.core.base_model import BaseModel from pypos.core.utils.db_helper import connect_sqlite, connect_sqlite_read_fast from pypos.core.utils.device_utils import get_active_device_info, get_device_id from pypos.core.utils.sql_identifier_utils import quote_sql_identifier from pypos.core.database.schema_migrator import run_schema_migrations_once from pypos.modules.penjualan.config.penjualan_config import ( get_autocomplete_contains_min_keyword, get_toko_id_from_config, is_lookup_require_active_harga_enabled, ) from pypos.modules.penjualan.errors import TransaksiSaveError from pypos.modules.penjualan.models.transaksi_value_utils import TransaksiValueUtils # upgraded: inherit base class def generate_nomer2(counter: int, kasir_username: str) -> str: """ Generator nomor transaksi kedua (nomer2) dengan format timestamp + counter + username. """ timestamp = datetime.now().strftime('%Y%m%d%H%M%S') no_urut = f"{counter:08d}" return f"{timestamp}{no_urut}-{kasir_username}" # upgraded: inherit base class @dataclass class DetailTransaksi(BaseModel): produk_id: int produk_nama: str produk_ord_hrg: float produk_ord_jml: int produk_jenis: str produk_ord_diskon: float satuan: str = "" def __iter__(self): return iter(( self.produk_id, self.produk_nama, self.produk_ord_hrg, self.produk_ord_jml, self.produk_jenis, self.produk_ord_diskon, self.satuan, )) # upgraded: inherit base class class TransaksiModel(BaseModel): def __init__( self, db_path, counter_service=None, lookup_harga_service=None, payment_payload_service=None, simpan_use_case_service=None, ): super().__init__() self.db_path = db_path # edited by glg # Guard lookup produk agar item tanpa harga aktif tidak masuk alur scan/manual input. try: self._lookup_require_active_harga = bool( is_lookup_require_active_harga_enabled(default=1) ) except (TypeError, ValueError): self._lookup_require_active_harga = True # edited by glg # Ambang fallback LIKE untuk mencari nama/barcode yang tidak persis sama. try: self._autocomplete_contains_min_keyword = int( get_autocomplete_contains_min_keyword(default=2) ) except (TypeError, ValueError): self._autocomplete_contains_min_keyword = 2 # edited by glg # HMVC strict: domain logic lookup harga dan counter dipisah ke service tersendiri. self.counter_service = counter_service or self._build_default_counter_service() self.lookup_harga_service = lookup_harga_service or self._build_default_lookup_harga_service() self.payment_payload_service = ( payment_payload_service or self._build_default_payment_payload_service() ) self.simpan_use_case_service = simpan_use_case_service or self._build_default_simpan_use_case_service() # edited by glg def _build_default_counter_service(self): module = __import__( "pypos.modules.penjualan.services.transaksi_counter_service", fromlist=["TransaksiCounterService"], ) return module.TransaksiCounterService( db_path=self.db_path, log_warning=self.log_warning, ) # edited by glg def _build_default_lookup_harga_service(self): module = __import__( "pypos.modules.penjualan.services.transaksi_lookup_harga_service", fromlist=["TransaksiLookupHargaService"], ) return module.TransaksiLookupHargaService(self) # edited by glg def _build_default_payment_payload_service(self): module = __import__( "pypos.modules.penjualan.services.transaksi_payment_payload_service", fromlist=["TransaksiPaymentPayloadService"], ) return module.TransaksiPaymentPayloadService() # edited by glg def _build_default_simpan_use_case_service(self): module = __import__( "pypos.modules.penjualan.services.transaksi_simpan_use_case_service", fromlist=["TransaksiSimpanUseCaseService"], ) return module.TransaksiSimpanUseCaseService(self) # edited by glg def _connect_read_fast(self): return connect_sqlite_read_fast(self.db_path) # edited by glg def _is_lookup_require_active_harga_enabled(self): return bool(getattr(self, "_lookup_require_active_harga", True)) # edited by glg def _get_autocomplete_contains_min_keyword(self): value = int(getattr(self, "_autocomplete_contains_min_keyword", 2) or 2) return max(1, min(5, value)) # edited by glg def _ensure_schema_ready_or_fail(self, reason_code="TRANSAKSI_SCHEMA_MIGRATION_FAILED"): ok = run_schema_migrations_once(self.db_path, strict=False) if bool(ok): return True message = ( f"{reason_code}: run_schema_migrations_once(strict=False) mengembalikan False " f"untuk db_path={self.db_path}" ) self.log_error(message) raise RuntimeError(message) # edited by glg def _build_lookup_price_exists_clause(self, produk_alias="p", include_price_filters=True): # edited by glg # Bandit B608: query ini dibentuk dari fragmen internal terkontrol (bukan input bebas user). parts = [ f"pr.produk_id = {produk_alias}.id", "pr.jenis_value = 'harga_list'", "COALESCE(pr.nilai, 0) > 0", ] if include_price_filters: parts.append("COALESCE(pr.status, 1) = 1") parts.append("COALESCE(pr.trash, 0) = 0") where_sql = " AND ".join(parts) return f"EXISTS (SELECT 1 FROM price pr WHERE {where_sql})" # nosec B608 def _ensure_pembayaran_non_tunai_column(self, cursor): self._ensure_schema_ready_or_fail("TRANSAKSI_SCHEMA_COLUMN_MIGRATION_FAILED") cursor.execute("PRAGMA table_info(transaksi)") cols = [row[1] for row in cursor.fetchall()] added_cols = False if "pembayaran_non_tunai" not in cols: cursor.execute( "ALTER TABLE transaksi ADD COLUMN pembayaran_non_tunai REAL DEFAULT 0" ) added_cols = True if "point_transaksi" not in cols: cursor.execute( "ALTER TABLE transaksi ADD COLUMN point_transaksi INTEGER DEFAULT 0" ) added_cols = True # edited by glg # Kolom eksplisit mode pajak untuk sinkronisasi/settlement/reprint lintas versi. if "ppn_mode" not in cols: cursor.execute( "ALTER TABLE transaksi ADD COLUMN ppn_mode TEXT DEFAULT 'include'" ) added_cols = True if added_cols: cache = getattr(self, "_table_columns_cache", None) if isinstance(cache, dict): cache.pop("transaksi", None) # edited by glg def _get_table_columns(self, cursor, table_name): cache_attr = "_table_columns_cache" cache = getattr(self, cache_attr, None) if cache is None: cache = {} setattr(self, cache_attr, cache) table_key = str(table_name or "").strip() if not table_key: return set() if table_key in cache: return cache[table_key] try: table_sql = quote_sql_identifier(table_key) except ValueError: self.log_warning(f"Nama tabel tidak valid saat baca schema: {table_key}") cache[table_key] = set() return set() cursor.execute(f"PRAGMA table_info({table_sql})") columns = {str(row[1]) for row in cursor.fetchall() if row and len(row) > 1} cache[table_key] = columns return columns # edited by glg def _filter_insert_dict_by_table(self, cursor, table_name, payload): if not isinstance(payload, dict): return {} columns = self._get_table_columns(cursor, table_name) if not columns: return {} filtered = {} dropped = [] for key, val in payload.items(): key_name = str(key or "").strip() if not key_name: continue if key_name in columns: filtered[key_name] = val else: dropped.append(key_name) if dropped: self.log_debug( f"skip kolom tidak dikenal table={table_name}: {','.join(sorted(set(dropped)))}" ) return filtered # edited by glg def _as_positive_int(self, value, default=0): return TransaksiValueUtils.as_positive_int(value, default) # edited by glg def _extract_point_from_diskon_log(self, diskon_log): return TransaksiValueUtils.extract_point_from_diskon_log(diskon_log) # edited by glg def _extract_diskon_customer_from_diskon_log(self, diskon_log): return TransaksiValueUtils.extract_diskon_customer_from_diskon_log(diskon_log) # edited by glg def _extract_ppn_mode_from_diskon_log(self, diskon_log): return TransaksiValueUtils.extract_ppn_mode_from_diskon_log(diskon_log) # edited by glg def _normalize_ppn_mode(self, value, default="include"): mode = str(value or "").strip().lower() if mode in {"include", "exclude"}: return mode fallback = str(default or "include").strip().lower() return fallback if fallback in {"include", "exclude"} else "include" # edited by glg def _upsert_ppn_mode_in_diskon_log(self, diskon_log, ppn_mode): mode = self._normalize_ppn_mode(ppn_mode, default="include") text = str(diskon_log or "").strip() if not text: return f"ppn_mode={mode}" parts = [] replaced = False for raw_part in text.split(";"): part = str(raw_part or "").strip() if not part: continue if "=" not in part: parts.append(part) continue key, value = part.split("=", 1) if str(key or "").strip().lower() == "ppn_mode": parts.append(f"ppn_mode={mode}") replaced = True continue parts.append(f"{str(key).strip()}={str(value).strip()}") if not replaced: parts.append(f"ppn_mode={mode}") return ";".join(parts) # edited by glg def _normalize_transaksi_master_payload(self, payload): insert_dict = dict(payload or {}) mode = self._normalize_ppn_mode( insert_dict.get("ppn_mode"), default=self._extract_ppn_mode_from_diskon_log(insert_dict.get("diskon_log", "")) or "include", ) insert_dict["ppn_mode"] = mode insert_dict["diskon_log"] = self._upsert_ppn_mode_in_diskon_log( insert_dict.get("diskon_log", ""), mode, ) return insert_dict # edited by glg def _build_default_transaksi_context(self): machine_id = str(get_device_id() or "").strip() active_info = get_active_device_info(machine_id) or {} cabang_id = self._as_positive_int(active_info.get("cabang_id"), 0) 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) default_gudang = self._lookup_cabang_default_gudang(cabang_id=cabang_id) return { "machine_id": machine_id, "cabang_id": cabang_id, "cabang_nama": str(active_info.get("cabang_nama") or "").strip(), "toko_id": toko_id, "toko_nama": str(active_info.get("toko_nama") or "").strip(), "gudang_id": int(default_gudang.get("gudang_id") or 0), "gudang_nama": str(default_gudang.get("gudang_nama") or "").strip(), } # edited by glg def _to_int(self, value, default=0): try: if value is None: return int(default) if isinstance(value, bool): return int(value) if isinstance(value, (int, float)): return int(round(float(value))) return int(round(float(str(value).strip()))) except (TypeError, ValueError, OverflowError): return int(default) # edited by glg # Source of truth gudang transaksi: default gudang dari tabel per_cabang. def _lookup_cabang_default_gudang(self, cabang_id, cursor=None): 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._to_int(cached.get("gudang_id"), 0), "gudang_nama": str(cached.get("gudang_nama") or "").strip(), } own_conn = None cur = cursor result = {"gudang_id": 0, "gudang_nama": ""} try: if cur is None: own_conn = connect_sqlite(self.db_path) cur = own_conn.cursor() cur.execute("PRAGMA table_info(per_cabang)") columns = { str(row[1]) for row in cur.fetchall() if row and len(row) > 1 and row[1] } if not columns: 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((name for name in gudang_id_candidates if name in columns), "") gudang_nama_col = next((name for name in gudang_nama_candidates if name in columns), "") 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) where_status = " AND COALESCE(status, 1) = 1" if "status" in columns else "" where_trash = " AND COALESCE(trash, 0) = 0" if "trash" in columns else "" query = ( f"SELECT {', '.join(selected_cols)} " # nosec B608 f"FROM per_cabang " f"WHERE id = ?{where_status}{where_trash} " f"LIMIT 1" ) cur.execute(query, (cabang_int,)) row = cur.fetchone() if not row: cache[cabang_int] = dict(result) return dict(result) row_map = self._row_to_dict(cur, row) result = { "gudang_id": self._to_int( row_map.get(gudang_id_col) if gudang_id_col else 0, 0, ), "gudang_nama": str( row_map.get(gudang_nama_col) if gudang_nama_col else "" ).strip(), } cache[cabang_int] = dict(result) return dict(result) except (sqlite3.Error, TypeError, ValueError): cache[cabang_int] = dict(result) return dict(result) finally: if own_conn is not None: try: own_conn.close() except sqlite3.Error: pass # edited by glg def _apply_cabang_default_gudang_to_insert(self, cursor, insert_dict): payload = dict(insert_dict or {}) cabang_id = self._as_positive_int(payload.get("cabang_id"), 0) if cabang_id <= 0: return payload lookup = self._lookup_cabang_default_gudang(cabang_id=cabang_id, cursor=cursor) resolved_gudang_id = self._to_int(lookup.get("gudang_id"), 0) resolved_gudang_nama = str(lookup.get("gudang_nama") or "").strip() current_gudang_id = self._to_int(payload.get("gudang_id"), 0) if current_gudang_id == 0 and resolved_gudang_id != 0: payload["gudang_id"] = resolved_gudang_id if ( self._is_blank_value(payload.get("gudang_nama")) and resolved_gudang_nama ): payload["gudang_nama"] = resolved_gudang_nama return payload # edited by glg def _build_detail_insert_dict(self, detail, transaksi_id, detail_context): if hasattr(detail, "to_dict_with_transaksi_id"): insert_dict = detail.to_dict_with_transaksi_id(transaksi_id) else: insert_dict = { "transaksi_id": transaksi_id, "produk_id": getattr(detail, "produk_id", 0), "produk_nama": getattr(detail, "produk_nama", ""), "produk_ord_hrg": getattr(detail, "produk_ord_hrg", 0), "produk_ord_jml": getattr(detail, "produk_ord_jml", 0), "produk_jenis": getattr(detail, "produk_jenis", "invoice"), "produk_ord_diskon": getattr(detail, "produk_ord_diskon", 0), "satuan": getattr(detail, "satuan", ""), } context = detail_context if isinstance(detail_context, dict) else {} insert_dict["dtime"] = context.get("dtime") insert_dict["cabang_id"] = context.get("cabang_id") insert_dict["oleh_id"] = context.get("oleh_id") insert_dict["oleh_nama"] = context.get("oleh_nama") transaksi_jenis_ctx = str(context.get("transaksi_jenis") or "758").strip() or "758" def _is_blank(val): return val is None or str(val).strip() == "" def _to_int(val, default=0): try: return int(float(val)) except (TypeError, ValueError, OverflowError): return int(default) def _to_float(val, default=0.0): try: return float(val) except (TypeError, ValueError, OverflowError): return float(default) def _set_default(key, value): if _is_blank(insert_dict.get(key)): insert_dict[key] = value qty = max(0, _to_int(insert_dict.get("produk_ord_jml"), 0)) diskon_persen = max(0.0, _to_float(insert_dict.get("produk_ord_diskon"), 0.0)) hrg_ori = _to_float(insert_dict.get("produk_ord_hrg"), 0.0) # edited by glg # Simpan metadata diskon agar jalur print/export membaca nilai nominal real. diskon_persen_meta = max( 0.0, _to_float( getattr( detail, "produk_ord_diskon_persen", insert_dict.get("produk_ord_diskon_persen", diskon_persen), ), diskon_persen, ), ) diskon_nominal_meta = max( 0.0, _to_float( getattr( detail, "produk_ord_diskon_nominal", insert_dict.get("produk_ord_diskon_khusus", 0), ), 0.0, ), ) if diskon_persen_meta > 0: insert_dict["produk_ord_diskon_persen"] = diskon_persen_meta if diskon_nominal_meta > 0: insert_dict["produk_ord_diskon_khusus"] = diskon_nominal_meta dtime_val = str(insert_dict.get("dtime") or "") tahun = _to_int(dtime_val[:4], datetime.now().year) bulan = _to_int(dtime_val[5:7], datetime.now().month) # edited by glg # Isi default operasional detail transaksi agar kolom turunan tidak kosong. _set_default("valid_qty", qty) _set_default("valid_qty_no_approve", qty) _set_default("produk_nama_2", str(insert_dict.get("produk_nama") or "").strip()) _set_default("produk_label", "") _set_default("produk_keterangan", "") _set_default("produk_folders", "") _set_default("produk_kode", "") _set_default("produk_ord_jml_return", 0) _set_default("produk_ord_stok", qty) _set_default("produk_ord_diskon_persen", diskon_persen_meta) _set_default("produk_ord_diskon_khusus", 0) _set_default("produk_ord_diterima", 0) _set_default("produk_ord_kurang", 0) _set_default("produk_hrg_ori", hrg_ori) _set_default("produk_hrg_gap", 0) _set_default("produk_berat", 0) _set_default("produk_berat_gross", 0) _set_default("produk_volume", 0) _set_default("produk_volume_gross", 0) _set_default("sorting", 0) _set_default("parent_id", 0) _set_default("trash", 0) _set_default("transaksi_jenis", transaksi_jenis_ctx) _set_default("rekening_id_asal", 0) _set_default("rekening_id_tujuan", 0) _set_default("status", 1) _set_default("kirim", 0) _set_default("keterangan", "") _set_default("no_part", "") _set_default("sub_step_number", 0) _set_default("sub_step_current", 0) _set_default("sub_step_avail", 0) _set_default("next_substep_num", 0) _set_default("next_substep_code", "") _set_default("next_substep_label", "") _set_default("next_subgroup_code", "") _set_default("sub_tail_number", 0) _set_default("sub_tail_code", "") _set_default("detail_tipe", "items") _set_default("ppn", 0) _set_default("ppn_persen", 0) _set_default("hpp", 0) _set_default("hpp_ori", 0) _set_default("produk_diskon_persen_ori", 0) _set_default("produk_diskon_ori", 0) _set_default("produk_diskon_khusus_ori", 0) _set_default("komisi", 0) _set_default("bunga", 0) _set_default("uang_muka", 0) _set_default("administrasi", 0) _set_default("angsuran", 0) _set_default("bulan", bulan) _set_default("tahun", tahun) _set_default("produk_ord_valas_nilai", 0) _set_default("produk_ord_valas_nama", "") _set_default("produk_ord_batal", 0) _set_default("valas_ord_nama", "") _set_default("valas_ord_hrg", 0) _set_default("valas_produk_diskon", 0) _set_default("valas_produk_diskon_khusus", 0) _set_default("cancel_qty", 0) _set_default("cancel_id", 0) _set_default("status_cancel", 0) _set_default("cancel_name", "") _set_default("req_cancel_qty", 0) _set_default("mongo", "") return insert_dict # edited by glg def _as_float(self, value, default=0.0): try: return float(value) except (TypeError, ValueError, OverflowError): return float(default or 0.0) # edited by glg def _resolve_bank_rekening_fields(self, bank_id, cabang_id, is_tunai): preferred_jenis = "account_cash" if is_tunai else "account_in" fallback_id = 108 if is_tunai else 110 fallback_name = "Tunai" if is_tunai else "Non Tunai" conn = None try: conn = connect_sqlite(self.db_path) conn.row_factory = sqlite3.Row cur = conn.cursor() cabang_int = self._as_positive_int(cabang_id, 0) params = [preferred_jenis] where_cabang = "" if cabang_int > 0: where_cabang = "AND (cabang_id = ? OR cabang_id = -1 OR cabang_id = 0 OR cabang_id IS NULL)" params.append(cabang_int) query = ( "SELECT id, nama " # nosec B608 "FROM bank " "WHERE jenis = ? " "AND COALESCE(status, 1) = 1 " "AND COALESCE(trash, 0) = 0 " f"{where_cabang} " "ORDER BY CASE WHEN id = ? THEN 0 ELSE 1 END, id ASC " "LIMIT 1" ) cur.execute(query, tuple(params + [self._as_positive_int(bank_id, 0)])) row = cur.fetchone() if row: return int(row["id"]), str(row["nama"] or fallback_name) return fallback_id, fallback_name except (sqlite3.Error, TypeError, ValueError): return fallback_id, fallback_name finally: try: conn.close() except sqlite3.Error: pass # edited by glg def _is_blank_value(self, value): if value is None: return True return str(value).strip() == "" # edited by glg def _row_to_dict(self, cursor, row): if row is None: return {} try: return dict(row) except (TypeError, ValueError): pass try: description = getattr(cursor, "description", None) or [] if not description: return {} col_names = [str(col[0]) for col in description] return {col_names[idx]: row[idx] for idx in range(min(len(col_names), len(row)))} except (TypeError, ValueError, KeyError, IndexError): return {} # edited by glg def _build_transaksi_copy_insert_dict(self, transaksi_row, copy_columns): source = transaksi_row if isinstance(transaksi_row, dict) else {} cols = set(copy_columns or set()) payload = {} for col in cols: if col in source and source.get(col) is not None: payload[col] = source.get(col) alias_map = { "id_cabang": "cabang_id", "nama_cabang": "cabang_nama", "id_gudang": "gudang_id", "nama_gudang": "gudang_nama", "id_pegawai": "oleh_id", "nama_pegawai": "oleh_nama", "id_pelanggan": "customers_id", "nama_pelanggan": "customers_nama", "no_invoice": "nomer", "total": "transaksi_bulat", "grand_total": "transaksi_nilai", "diskon": "diskon_nilai", "pajak": "ppn_persen", "uang_dibayar": "transaksi_dibayar", "uang_kembali": "transaksi_dibayar_return", "metode_pembayaran": "pembayaran_sys", "total_harga": "transaksi_bulat", "ppn": "ppn_persen", "jenis_transaksi": "jenis_label", "kode_transaksi": "nomer", } for target_col, source_col in alias_map.items(): if target_col not in cols: continue if not self._is_blank_value(payload.get(target_col)): continue src_val = source.get(source_col) if src_val is None: continue payload[target_col] = src_val dtime_val = str(source.get("dtime") or "").strip() if "tanggal" in cols and self._is_blank_value(payload.get("tanggal")): payload["tanggal"] = dtime_val[:10] if len(dtime_val) >= 10 else "" if "jam" in cols and self._is_blank_value(payload.get("jam")): payload["jam"] = dtime_val[11:19] if len(dtime_val) >= 19 else "" if "metode_pembayaran" in cols and self._is_blank_value(payload.get("metode_pembayaran")): payload["metode_pembayaran"] = str( source.get("bank_nama") or source.get("pembayaran") or source.get("pembayaran_sys") or "tunai" ).strip() if "is_deleted" in cols and self._is_blank_value(payload.get("is_deleted")): payload["is_deleted"] = 0 if "status_pembayaran" in cols and self._is_blank_value(payload.get("status_pembayaran")): total_nilai = self._as_float(source.get("transaksi_nilai"), 0.0) total_bayar = self._as_float(source.get("transaksi_dibayar"), 0.0) payload["status_pembayaran"] = "lunas" if total_bayar >= total_nilai else "belum_lunas" if "waktu_dibuat" in cols and self._is_blank_value(payload.get("waktu_dibuat")) and dtime_val: payload["waktu_dibuat"] = dtime_val return payload # edited by glg def _build_transaksi_data_copy_insert_dict(self, detail_row, copy_columns): source = detail_row if isinstance(detail_row, dict) else {} cols = set(copy_columns or set()) payload = {} for col in cols: if col in source and source.get(col) is not None: payload[col] = source.get(col) if "produk_ord_hpp" in cols and self._is_blank_value(payload.get("produk_ord_hpp")): hpp_val = source.get("hpp") if hpp_val is None: hpp_val = source.get("hpp_ori") if hpp_val is not None: payload["produk_ord_hpp"] = hpp_val if "produk_ord_laba" in cols and self._is_blank_value(payload.get("produk_ord_laba")): payload["produk_ord_laba"] = 0 return payload # edited by glg # Mirror transaksi aktif ke tabel copy agar modul legacy tetap kompatibel. def _sync_copy_tables_for_transaksi(self, cursor, transaksi_id): trx_id = self._as_positive_int(transaksi_id, 0) if trx_id <= 0: return False transaksi_copy_cols = self._get_table_columns(cursor, "transaksi_copy") transaksi_data_copy_cols = self._get_table_columns(cursor, "transaksi_data_copy") if not transaksi_copy_cols and not transaksi_data_copy_cols: return False try: cursor.execute("SELECT * FROM transaksi WHERE id = ? LIMIT 1", (trx_id,)) master_row = self._row_to_dict(cursor, cursor.fetchone()) if not master_row: return False if transaksi_copy_cols: master_payload = self._build_transaksi_copy_insert_dict(master_row, transaksi_copy_cols) master_payload = self._filter_insert_dict_by_table(cursor, "transaksi_copy", master_payload) if master_payload: if "id" in transaksi_copy_cols: cursor.execute("DELETE FROM transaksi_copy WHERE id = ?", (trx_id,)) columns = ", ".join(master_payload.keys()) placeholders = ", ".join(["?"] * len(master_payload)) values = list(master_payload.values()) cursor.execute( f"INSERT INTO transaksi_copy ({columns}) VALUES ({placeholders})", # nosec B608 values, ) if transaksi_data_copy_cols: if "transaksi_id" in transaksi_data_copy_cols: cursor.execute("DELETE FROM transaksi_data_copy WHERE transaksi_id = ?", (trx_id,)) cursor.execute( "SELECT * FROM transaksi_data WHERE transaksi_id = ? ORDER BY id ASC", (trx_id,), ) detail_rows = cursor.fetchall() or [] for raw_detail in detail_rows: detail_row = self._row_to_dict(cursor, raw_detail) if not detail_row: continue detail_payload = self._build_transaksi_data_copy_insert_dict( detail_row, transaksi_data_copy_cols, ) detail_payload = self._filter_insert_dict_by_table( cursor, "transaksi_data_copy", detail_payload, ) if not detail_payload: continue detail_cols = ", ".join(detail_payload.keys()) detail_placeholders = ", ".join(["?"] * len(detail_payload)) detail_values = list(detail_payload.values()) cursor.execute( f"INSERT INTO transaksi_data_copy ({detail_cols}) VALUES ({detail_placeholders})", # nosec B608 detail_values, ) return True except (sqlite3.Error, TypeError, ValueError, KeyError, IndexError) as exc: self.log_warning(f"Gagal mirror transaksi_copy/transaksi_data_copy id={trx_id}: {exc}") return False # edited by glg def mirror_all_transaksi_to_copy_tables(self): conn = connect_sqlite(self.db_path) cursor = conn.cursor() mirrored = 0 try: cursor.execute("SELECT id FROM transaksi ORDER BY id ASC") rows = cursor.fetchall() or [] for row in rows: try: trx_id = int(row[0]) except (TypeError, ValueError): continue if self._sync_copy_tables_for_transaksi(cursor, trx_id): mirrored += 1 conn.commit() return mirrored except sqlite3.Error: conn.rollback() raise finally: conn.close() def _build_autocomplete_sql( self, include_trash=True, include_price_filters=True, include_saleable_only=False, ): produk_filter = "COALESCE(p.trash, 0) = 0 AND " if include_trash else "" price_filter_parts = ["pr.produk_id = p.id", "pr.jenis_value = 'harga_list'"] if include_price_filters: price_filter_parts.append("COALESCE(pr.status, 1) = 1") price_filter_parts.append("COALESCE(pr.trash, 0) = 0") price_filter = " AND ".join(price_filter_parts) saleable_filter = "" if include_saleable_only: saleable_filter = ( self._build_lookup_price_exists_clause( produk_alias="p", include_price_filters=include_price_filters, ) + " AND " ) return ( "SELECT " # nosec B608 "p.id, " "p.nama, " "p.barcode, " "COALESCE(( " "SELECT pr.nilai " "FROM price pr " f"WHERE {price_filter} " "ORDER BY pr.id DESC " "LIMIT 1 " "), 0) AS harga_jual " "FROM produk p " f"WHERE {produk_filter}{saleable_filter}{{where_clause}} " "ORDER BY {order_clause} " "LIMIT ?" ) def _fetch_autocomplete_rows(self, where_clause, order_clause, params): conn = self._connect_read_fast() conn.row_factory = sqlite3.Row cursor = conn.cursor() base_params = tuple(params or ()) saleable_only = self._is_lookup_require_active_harga_enabled() sql_variants = [ self._build_autocomplete_sql( include_trash=True, include_price_filters=True, include_saleable_only=saleable_only, ), self._build_autocomplete_sql( include_trash=False, include_price_filters=True, include_saleable_only=saleable_only, ), self._build_autocomplete_sql( include_trash=False, include_price_filters=False, include_saleable_only=saleable_only, ), ] try: for sql_tpl in sql_variants: sql = sql_tpl.format(where_clause=where_clause, order_clause=order_clause) try: cursor.execute(sql, base_params) return cursor.fetchall() except sqlite3.OperationalError: continue return [] finally: conn.close() def _to_autocomplete_payload(self, rows): barang_list = [] mapping = {} for row in rows: harga_jual = row["harga_jual"] if row["harga_jual"] is not None else 0 text = f"{row['nama']} - {harga_jual}" if text in mapping: continue barang_list.append(text) mapping[text] = { "id": row["id"], "nama": row["nama"], "barcode": row["barcode"], "harga_jual": harga_jual, } return barang_list, mapping def get_produk_autocomplete(self, keyword="", limit=50): return self.lookup_harga_service.get_produk_autocomplete(keyword=keyword, limit=limit) # edited by glg def _get_active_cabang_id_for_pricing(self): try: device_info = get_active_device_info(get_device_id()) or {} return self._as_positive_int(device_info.get("cabang_id"), 0) except (TypeError, ValueError): return 0 # edited by glg def _fetch_latest_harga_list(self, cursor, id_produk, cabang_id=0): try: if int(cabang_id or 0) > 0: cursor.execute( """ SELECT nilai FROM price WHERE produk_id = ? AND jenis_value = 'harga_list' AND COALESCE(status, 1) = 1 AND COALESCE(trash, 0) = 0 AND (cabang_id = ? OR cabang_id IN (-1, 0) OR cabang_id IS NULL) ORDER BY CASE WHEN cabang_id = ? THEN 0 WHEN cabang_id IN (-1, 0) OR cabang_id IS NULL THEN 1 ELSE 2 END, id DESC LIMIT 1 """, (id_produk, int(cabang_id), int(cabang_id)), ) else: cursor.execute( """ SELECT nilai FROM price WHERE produk_id = ? AND jenis_value = 'harga_list' AND COALESCE(status, 1) = 1 AND COALESCE(trash, 0) = 0 ORDER BY id DESC LIMIT 1 """, (id_produk,), ) except sqlite3.OperationalError: cursor.execute( """ SELECT nilai FROM price WHERE produk_id = ? AND jenis_value = 'harga_list' ORDER BY id DESC LIMIT 1 """, (id_produk,), ) row = cursor.fetchone() if not row: return 0.0 try: return self._as_float(row["nilai"], 0.0) except (TypeError, ValueError, KeyError, IndexError): try: return self._as_float(row[0], 0.0) except (TypeError, ValueError, KeyError, IndexError): return 0.0 # edited by glg def _is_reasonable_grosir_base(self, harga_normal, harga_setelah_diskon): normal = self._as_float(harga_normal, 0.0) kandidat = self._as_float(harga_setelah_diskon, 0.0) if normal <= 0 or kandidat <= 0: return True ratio = kandidat / normal return 0.5 <= ratio <= 2.0 def cari_barang_by_id(self, id_produk, jumlah_beli): return self.lookup_harga_service.cari_barang_by_id(id_produk=id_produk, jumlah_beli=jumlah_beli) # edited by glg def _build_lookup_base_filters(self): filters = [ "COALESCE(p.status, 1) = 1", "COALESCE(p.trash, 0) = 0", ] if self._is_lookup_require_active_harga_enabled(): filters.append( self._build_lookup_price_exists_clause( produk_alias="p", include_price_filters=True, ) ) return " AND ".join(filters) # edited by glg def _legacy_lookup_by_nama(self, cursor, key): cursor.execute("SELECT * FROM produk WHERE nama = ? LIMIT 1", (key,)) row = cursor.fetchone() if not row: cursor.execute( "SELECT * FROM produk WHERE nama >= ? AND nama < ? LIMIT 1", (key, f"{key}\uffff"), ) row = cursor.fetchone() if not row and len(key) >= self._get_autocomplete_contains_min_keyword(): cursor.execute("SELECT * FROM produk WHERE nama LIKE ? LIMIT 1", (f"%{key}%",)) row = cursor.fetchone() return row def cari_barang_by_nama(self, nama_barang): conn = self._connect_read_fast() conn.row_factory = sqlite3.Row cursor = conn.cursor() key = str(nama_barang or "").strip() if not key: conn.close() return None try: lookup_filters = self._build_lookup_base_filters() cursor.execute( f"SELECT p.* FROM produk p WHERE {lookup_filters} AND p.nama = ? LIMIT 1", # nosec B608 (key,), ) row = cursor.fetchone() if not row: cursor.execute( f"SELECT p.* FROM produk p WHERE {lookup_filters} " # nosec B608 "AND p.nama >= ? AND p.nama < ? LIMIT 1", (key, f"{key}\uffff"), ) row = cursor.fetchone() if not row and len(key) >= self._get_autocomplete_contains_min_keyword(): cursor.execute( f"SELECT p.* FROM produk p WHERE {lookup_filters} AND p.nama LIKE ? LIMIT 1", # nosec B608 (f"%{key}%",), ) row = cursor.fetchone() except sqlite3.OperationalError: row = self._legacy_lookup_by_nama(cursor, key) conn.close() return dict(row) if row else None # edited by glg def _legacy_lookup_by_barcode(self, cursor, barcode): cursor.execute("SELECT * FROM produk WHERE barcode = ?", (barcode,)) return cursor.fetchone() # edited by glg def _lookup_raw_by_barcode(self, cursor, barcode): try: cursor.execute( """ SELECT p.* FROM produk p WHERE COALESCE(p.status, 1) = 1 AND COALESCE(p.trash, 0) = 0 AND p.barcode = ? LIMIT 1 """, (barcode,), ) return cursor.fetchone() except sqlite3.OperationalError: return self._legacy_lookup_by_barcode(cursor, barcode) # edited by glg def cari_barang_by_barcode_raw(self, barcode): conn = self._connect_read_fast() conn.row_factory = sqlite3.Row cursor = conn.cursor() row = self._lookup_raw_by_barcode(cursor, barcode) conn.close() return dict(row) if row else None def cari_barang_by_barcode(self, barcode): conn = self._connect_read_fast() conn.row_factory = sqlite3.Row cursor = conn.cursor() try: lookup_filters = self._build_lookup_base_filters() cursor.execute( f"SELECT p.* FROM produk p WHERE {lookup_filters} AND p.barcode = ? LIMIT 1", # nosec B608 (barcode,), ) row = cursor.fetchone() except sqlite3.OperationalError: row = self._legacy_lookup_by_barcode(cursor, barcode) conn.close() return dict(row) if row else None # edited by glg def _lookup_raw_by_nama(self, cursor, key): try: cursor.execute( """ SELECT p.* FROM produk p WHERE COALESCE(p.status, 1) = 1 AND COALESCE(p.trash, 0) = 0 AND p.nama = ? LIMIT 1 """, (key,), ) row = cursor.fetchone() if row: return row cursor.execute( """ SELECT p.* FROM produk p WHERE COALESCE(p.status, 1) = 1 AND COALESCE(p.trash, 0) = 0 AND p.nama >= ? AND p.nama < ? LIMIT 1 """, (key, f"{key}\uffff"), ) row = cursor.fetchone() if row or len(key) < self._get_autocomplete_contains_min_keyword(): return row cursor.execute( """ SELECT p.* FROM produk p WHERE COALESCE(p.status, 1) = 1 AND COALESCE(p.trash, 0) = 0 AND p.nama LIKE ? LIMIT 1 """, (f"%{key}%",), ) return cursor.fetchone() except sqlite3.OperationalError: return self._legacy_lookup_by_nama(cursor, key) # edited by glg def cari_barang_by_nama_raw(self, nama_barang): conn = self._connect_read_fast() conn.row_factory = sqlite3.Row cursor = conn.cursor() key = str(nama_barang or "").strip() if not key: conn.close() return None row = self._lookup_raw_by_nama(cursor, key) conn.close() return dict(row) if row else None def get_lookup_query_plan(self, sample_keyword="A", sample_barcode="12345"): keyword = str(sample_keyword or "A").strip() or "A" barcode = str(sample_barcode or "12345").strip() or "12345" conn = connect_sqlite(self.db_path) cursor = conn.cursor() query_map = { "barcode_exact": ( "EXPLAIN QUERY PLAN SELECT id FROM produk WHERE barcode = ? LIMIT 1", (barcode,), ), "nama_exact": ( "EXPLAIN QUERY PLAN SELECT id FROM produk WHERE nama = ? LIMIT 1", (keyword,), ), "nama_prefix": ( "EXPLAIN QUERY PLAN SELECT id FROM produk WHERE nama >= ? AND nama < ? LIMIT 1", (keyword, f"{keyword}\uffff"), ), "nama_contains": ( "EXPLAIN QUERY PLAN SELECT id FROM produk WHERE nama LIKE ? LIMIT 1", (f"%{keyword}%",), ), } result = {} try: for key, payload in query_map.items(): sql, params = payload try: cursor.execute(sql, params) rows = cursor.fetchall() details = [str(row[3]) for row in rows if len(row) > 3] result[key] = details except sqlite3.Error as exc: result[key] = [f"error: {exc}"] return result finally: conn.close() def get_and_increment_counter(self, nama="transaksi") -> int: try: return int(self.counter_service.get_and_increment_counter(nama=nama)) except TransaksiSaveError: raise except (sqlite3.Error, RuntimeError, TypeError, ValueError, KeyError) as exc: raise TransaksiSaveError( "TRX_COUNTER_UNEXPECTED_ERROR", "Terjadi kesalahan saat memperbarui counter transaksi.", cause=exc, ) from exc def ambil_diskon_by_produk(self, db_path, produk_id): """Ambil data diskon untuk 1 produk - method legacy""" conn = connect_sqlite(db_path) conn.row_factory = sqlite3.Row cursor = conn.cursor() cursor.execute(""" SELECT nomer_diskon, produk_id, free_produk_id, free_produk_nama, kelipatan, quota_global, quota_used FROM diskon WHERE produk_id = ? AND jenis = 'free_produk' ORDER BY id DESC LIMIT 1 """, (produk_id,)) row = cursor.fetchone() conn.close() if row: return { "nomer_diskon": row["nomer_diskon"], "produk_id": row["produk_id"], "free_produk_id": row["free_produk_id"], "free_produk_nama": row["free_produk_nama"], "kelipatan": row["kelipatan"], "quota_global": row["quota_global"], "quota_used": row["quota_used"], } else: return None def ambil_diskon_batch(self, db_path, produk_ids): """ Ambil diskon untuk multiple produk sekaligus (batch query). """ if not produk_ids: return {} conn = connect_sqlite(db_path) conn.row_factory = sqlite3.Row cursor = conn.cursor() try: placeholders = ",".join(["?"] * len(produk_ids)) query = ( "SELECT " # nosec B608 "nomer_diskon, " "produk_id, " "free_produk_id, " "free_produk_nama, " "kelipatan, " "quota_global, " "quota_used " "FROM diskon " f"WHERE produk_id IN ({placeholders}) AND jenis = 'free_produk' " "ORDER BY produk_id, id DESC" ) cursor.execute(query, produk_ids) rows = cursor.fetchall() diskon_map = {} for row in rows: prod_id = row["produk_id"] if prod_id not in diskon_map: diskon_map[prod_id] = { "nomer_diskon": row["nomer_diskon"], "produk_id": row["produk_id"], "free_produk_id": row["free_produk_id"], "free_produk_nama": row["free_produk_nama"], "kelipatan": row["kelipatan"], "quota_global": row["quota_global"], "quota_used": row["quota_used"], } return diskon_map finally: conn.close() # edited by glg def _insert_transaksi_detail_rows(self, cursor, transaksi_id, detail_data, detail_context): for detail in detail_data: detail_insert = self._build_detail_insert_dict( detail=detail, transaksi_id=transaksi_id, detail_context=detail_context, ) detail_insert = self._filter_insert_dict_by_table( cursor, "transaksi_data", detail_insert, ) if not detail_insert: continue detail_columns = ", ".join(detail_insert.keys()) detail_placeholders = ", ".join(["?"] * len(detail_insert)) detail_values = list(detail_insert.values()) cursor.execute( f"INSERT INTO transaksi_data ({detail_columns}) VALUES ({detail_placeholders})", # nosec B608 detail_values, ) # edited by glg def _build_free_produk_payload(self, detail_data, transaksi_data_dict, transaksi_id): arr_free_produk = [] free_produk_details = [ d for d in detail_data if d.produk_jenis == "free_produk" and d.produk_ord_jml > 0 ] if not free_produk_details: return arr_free_produk self.log_debug(f"[OPTIMIZED] Proses {len(free_produk_details)} free produk dengan batch query") # edited by glg # Diskon free harus di-resolve dari produk pemicu (source), bukan dari produk hadiah. source_produk_ids = [] for detail in free_produk_details: source_id = self._as_positive_int( getattr(detail, "free_source_produk_id", 0), self._as_positive_int(getattr(detail, "produk_id", 0), 0), ) if source_id > 0 and source_id not in source_produk_ids: source_produk_ids.append(source_id) diskon_map = self.ambil_diskon_batch(self.db_path, source_produk_ids) for detail in free_produk_details: source_produk_id = self._as_positive_int( getattr(detail, "free_source_produk_id", 0), self._as_positive_int(getattr(detail, "produk_id", 0), 0), ) source_produk_nama = str( getattr(detail, "free_source_produk_nama", "") or getattr(detail, "produk_nama", "") or "" ).strip() diskon_data = ( diskon_map.get(source_produk_id) or diskon_map.get(self._as_positive_int(getattr(detail, "produk_id", 0), 0)) or {} ) try: jumlah_free = int(getattr(detail, "produk_ord_jml", 0) or 0) except (TypeError, ValueError): jumlah_free = 0 if jumlah_free <= 0: continue free_produk_id = self._as_positive_int( getattr(detail, "produk_id", 0), self._as_positive_int(diskon_data.get("free_produk_id"), 0), ) free_produk_nama = str( getattr(detail, "produk_nama", "") or diskon_data.get("free_produk_nama") or source_produk_nama or "" ).strip() if source_produk_id <= 0: source_produk_id = free_produk_id if not source_produk_nama: source_produk_nama = free_produk_nama free_item = { "diskon_id": diskon_data.get("nomer_diskon"), "produk_id": source_produk_id, "produk_nama": source_produk_nama, "free_produk_id": free_produk_id, "free_produk_nama": free_produk_nama, "free_qty": jumlah_free, "kelipatan": diskon_data.get("kelipatan", 1), "quota_global": diskon_data.get("quota_global", 0), "quota_used": diskon_data.get("quota_used", 0), # edited by glg # Gunakan waktu transaksi real agar payload saveFreeProduk konsisten audit. "date": transaksi_data_dict.get("dtime"), "transaksi_id": transaksi_id, "transaksi_no": transaksi_data_dict["nomer"], "oleh_id": transaksi_data_dict["oleh_id"], "oleh_nama": transaksi_data_dict["oleh_nama"], "customer_id": transaksi_data_dict.get("customers_id", 1), "customer_nama": transaksi_data_dict.get("customers_nama", "Tunai"), "toko_id": transaksi_data_dict.get("toko_id", 0), # edited by glg # Relasi eksplisit source -> free untuk audit. "source_produk_id": source_produk_id, "source_produk_nama": source_produk_nama, } arr_free_produk.append(free_item) self.log_info( f"Free produk item: {free_item['produk_nama']} -> {free_item['free_produk_nama']} qty={free_item['free_qty']}" ) return arr_free_produk # edited by glg def _prepare_master_insert_payload(self, cursor, transaksi_data_dict): insert_dict = dict(transaksi_data_dict or {}) detail_context = insert_dict.pop("detail_context", {}) for field in ["jumlah_bayar", "kembalian", "kasir_nama", "customer_nama", "skip_logo"]: insert_dict.pop(field, None) insert_dict = self._normalize_transaksi_master_payload(insert_dict) insert_dict = self._filter_insert_dict_by_table(cursor, "transaksi", insert_dict) if not insert_dict: raise ValueError("Payload transaksi kosong setelah validasi schema.") if self._as_positive_int(insert_dict.get("cabang_id"), 0) <= 0: raise ValueError("invalid_cabang_id: cabang_id wajib terisi pada transaksi.") insert_dict = self._apply_cabang_default_gudang_to_insert(cursor, insert_dict) ctx = detail_context if isinstance(detail_context, dict) else {} if not ctx.get("dtime"): ctx["dtime"] = insert_dict.get("dtime") if ctx.get("cabang_id") in (None, "", 0): ctx["cabang_id"] = insert_dict.get("cabang_id") if ctx.get("oleh_id") in (None, "", 0): ctx["oleh_id"] = insert_dict.get("oleh_id") if not str(ctx.get("oleh_nama") or "").strip(): ctx["oleh_nama"] = insert_dict.get("oleh_nama") if not str(ctx.get("transaksi_jenis") or "").strip(): ctx["transaksi_jenis"] = insert_dict.get("transaksi_jenis") return insert_dict, ctx # edited by glg @staticmethod def _insert_master_row(cursor, insert_dict): columns = ", ".join(insert_dict.keys()) placeholders = ", ".join(["?"] * len(insert_dict)) values = list(insert_dict.values()) cursor.execute(f"INSERT INTO transaksi ({columns}) VALUES ({placeholders})", values) # nosec B608 return int(cursor.lastrowid or 0) # edited by glg def _execute_simpan_transaksi_flow(self, cursor, conn, detail_data, transaksi_data_dict, precommit_hook=None): insert_dict, detail_context = self._prepare_master_insert_payload(cursor, transaksi_data_dict) transaksi_id = self._insert_master_row(cursor, insert_dict) self._insert_transaksi_detail_rows( cursor=cursor, transaksi_id=transaksi_id, detail_data=detail_data, detail_context=detail_context, ) self._sync_copy_tables_for_transaksi(cursor, transaksi_id) arr_free_produk = self._build_free_produk_payload( detail_data=detail_data, transaksi_data_dict=transaksi_data_dict, transaksi_id=transaksi_id, ) if callable(precommit_hook): precommit_hook(transaksi_id, conn) return transaksi_id, arr_free_produk # edited by glg def _prepare_simpan_transaksi_identity(self, transaksi_data_dict): counter = self.get_and_increment_counter("transaksi") nomer2 = generate_nomer2(counter, transaksi_data_dict["oleh_nama"]) transaksi_data_dict["nomer2"] = nomer2 return counter, nomer2 # edited by glg def _log_simpan_transaksi_context(self, counter, nomer2, transaksi_data_dict, trace_id=""): trace_text = str(trace_id or "").strip() or "-" self.log_debug(f"[TRX_SAVE][{trace_text}] simpan_transaksi counter={counter}, nomer2={nomer2}") self.log_debug( f"[TRX_SAVE][{trace_text}] simpan_transaksi transaksi_nilai={transaksi_data_dict['transaksi_nilai']}" ) self.log_debug( f"[TRX_SAVE][{trace_text}] transaksi_dibayar={transaksi_data_dict.get('transaksi_dibayar', 'NOT SET')}" ) self.log_debug( f"[TRX_SAVE][{trace_text}] transaksi_dibayar_return=" f"{transaksi_data_dict.get('transaksi_dibayar_return', 'NOT SET')}" ) # edited by glg @staticmethod def _rollback_connection(conn): try: conn.rollback() except sqlite3.Error: pass # edited by glg # precommit_hook dipanggil sebelum commit agar side-effect (contoh: voucher usage) # bisa ikut atomik dalam transaksi yang sama. def simpan_transaksi( self, transaksi_data, detail_data, transaksi_data_dict, precommit_hook=None, trace_id="", ): return self.simpan_use_case_service.execute( transaksi_data=transaksi_data, detail_data=detail_data, transaksi_data_dict=transaksi_data_dict, precommit_hook=precommit_hook, trace_id=trace_id, ) def simpan_transaksi_f9(self, transaksi_data, detail_data): conn = connect_sqlite(self.db_path) cursor = conn.cursor() try: self._ensure_pembayaran_non_tunai_column(cursor) transaksi_list = list(transaksi_data or []) base_keys = [ "nomer", "dtime", "transaksi_nilai", "diskon_persen", "ppn_persen", "transaksi_bulat", "customers_id", "customers_nama", "fulldate", "oleh_id", "oleh_nama", "jenis_label", "transaksi_jenis", "settlement_id", # edited by glg "ppn_mode", ] insert_dict = {} for idx, key in enumerate(base_keys): insert_dict[key] = transaksi_list[idx] if idx < len(transaksi_list) else None insert_dict["nomer2"] = insert_dict.get("nomer") insert_dict["transaksi_dibayar"] = 0 insert_dict["transaksi_dibayar_return"] = 0 insert_dict["pembayaran_non_tunai"] = 0 insert_dict["point_transaksi"] = 0 insert_dict["reference_id"] = 0 insert_dict["reference_jenis"] = 0 insert_dict["reference_nomer"] = 0 default_ctx = self._build_default_transaksi_context() if self._as_positive_int(insert_dict.get("cabang_id"), 0) <= 0: insert_dict["cabang_id"] = default_ctx["cabang_id"] if self._as_positive_int(insert_dict.get("toko_id"), 0) <= 0: insert_dict["toko_id"] = default_ctx["toko_id"] if self._to_int(insert_dict.get("gudang_id"), 0) == 0: insert_dict["gudang_id"] = default_ctx.get("gudang_id", 0) if not str(insert_dict.get("gudang_nama") or "").strip(): insert_dict["gudang_nama"] = default_ctx.get("gudang_nama", "") if not str(insert_dict.get("cabang_nama") or "").strip(): insert_dict["cabang_nama"] = default_ctx["cabang_nama"] if not str(insert_dict.get("toko_nama") or "").strip(): insert_dict["toko_nama"] = default_ctx["toko_nama"] insert_dict["machine_id"] = default_ctx["machine_id"] # edited by glg # Simpan transaksi pre-order/draft tetap menyertakan metadata mode PPN. insert_dict = self._normalize_transaksi_master_payload(insert_dict) insert_dict = self._filter_insert_dict_by_table(cursor, "transaksi", insert_dict) if not insert_dict: raise ValueError("Payload transaksi F9 kosong setelah validasi schema.") if self._as_positive_int(insert_dict.get("cabang_id"), 0) <= 0: raise ValueError("invalid_cabang_id: cabang_id wajib terisi pada transaksi F9.") insert_dict = self._apply_cabang_default_gudang_to_insert(cursor, insert_dict) columns = ", ".join(insert_dict.keys()) placeholders = ", ".join(["?"] * len(insert_dict)) values = list(insert_dict.values()) cursor.execute( f"INSERT INTO transaksi ({columns}) VALUES ({placeholders})", # nosec B608 values, ) transaksi_id = cursor.lastrowid detail_context = { "dtime": insert_dict.get("dtime"), "cabang_id": insert_dict.get("cabang_id"), "oleh_id": insert_dict.get("oleh_id"), "oleh_nama": insert_dict.get("oleh_nama"), "transaksi_jenis": insert_dict.get("transaksi_jenis"), } for detail in detail_data: detail_insert = self._build_detail_insert_dict( detail=detail, transaksi_id=transaksi_id, detail_context=detail_context, ) detail_insert = self._filter_insert_dict_by_table( cursor, "transaksi_data", detail_insert, ) if not detail_insert: continue detail_columns = ", ".join(detail_insert.keys()) detail_placeholders = ", ".join(["?"] * len(detail_insert)) detail_values = list(detail_insert.values()) cursor.execute( f"INSERT INTO transaksi_data ({detail_columns}) VALUES ({detail_placeholders})", # nosec B608 detail_values, ) # edited by glg self._sync_copy_tables_for_transaksi(cursor, transaksi_id) conn.commit() return transaksi_id except TransaksiSaveError: conn.rollback() raise except sqlite3.Error as exc: conn.rollback() raise TransaksiSaveError( "TRX_SAVE_F9_DB_ERROR", "Gagal menyimpan transaksi F9 ke database.", cause=exc, ) from exc except (TypeError, ValueError, KeyError) as exc: conn.rollback() raise TransaksiSaveError( "TRX_SAVE_F9_DATA_ERROR", "Data transaksi F9 tidak valid.", cause=exc, ) from exc except (AttributeError, RuntimeError) as exc: conn.rollback() raise TransaksiSaveError( "TRX_SAVE_F9_UNEXPECTED_ERROR", "Terjadi kesalahan saat menyimpan transaksi F9.", cause=exc, ) from exc finally: conn.close() def preorder_exists(self, nomer): if not nomer: return False conn = None cursor = None try: conn = connect_sqlite(self.db_path) cursor = conn.cursor() cursor.execute( "SELECT 1 FROM transaksi WHERE nomer = ? AND jenis_label = 'simpan_transaksi' AND COALESCE(trash, 0) = 0 LIMIT 1", (nomer,) ) return cursor.fetchone() is not None except sqlite3.Error: return False finally: try: if cursor is not None: cursor.close() except sqlite3.Error: pass try: if conn is not None: conn.close() except sqlite3.Error: pass def get_transaksi_terakhir(self): """ Ambil transaksi terakhir yang SUDAH SELESAI dibayar (bukan simpan_transaksi/F9). Hanya ambil transaksi dengan jenis_label = 'invoice' (transaksi selesai). """ conn = connect_sqlite(self.db_path) conn.row_factory = sqlite3.Row cursor = conn.cursor() cursor.execute(""" SELECT * FROM transaksi WHERE jenis_label = 'invoice' AND COALESCE(trash, 0) = 0 ORDER BY id DESC LIMIT 1 """) transaksi_row = cursor.fetchone() conn.close() return dict(transaksi_row) if transaksi_row else None def get_detail_transaksi(self, transaksi_id): conn = connect_sqlite(self.db_path) conn.row_factory = sqlite3.Row cursor = conn.cursor() cursor.execute("SELECT * FROM transaksi_data WHERE transaksi_id = ? AND COALESCE(trash, 0) = 0", (transaksi_id,)) detail_rows = cursor.fetchall() conn.close() return detail_rows # DATA HELPERS (dipanggil controller) def map_detail_rows(self, detail_rows): """Konversi rows detail ke list DetailTransaksi.""" mapped = [] for row in detail_rows: try: satuan_val = row["satuan"] if "satuan" in row.keys() else "" except (TypeError, KeyError, AttributeError): satuan_val = "" detail_obj = DetailTransaksi( produk_id=row["produk_id"], produk_nama=row["produk_nama"], produk_ord_hrg=row["produk_ord_hrg"], produk_ord_jml=row["produk_ord_jml"], produk_jenis=row["produk_jenis"], produk_ord_diskon=row["produk_ord_diskon"], satuan=satuan_val ) # edited by glg # Metadata ini dipakai printer agar diskon item terbaca nominal (Rp) secara konsisten. try: diskon_persen = float( row["produk_ord_diskon_persen"] if "produk_ord_diskon_persen" in row.keys() else row["produk_ord_diskon"] ) except (TypeError, ValueError, KeyError, AttributeError): diskon_persen = 0.0 try: diskon_nominal = float( row["produk_ord_diskon_khusus"] if "produk_ord_diskon_khusus" in row.keys() else 0.0 ) except (TypeError, ValueError, KeyError, AttributeError): diskon_nominal = 0.0 if diskon_nominal <= 0 and 0 < diskon_persen <= 100: try: gross = float(row["produk_ord_hrg"] or 0) * int(float(row["produk_ord_jml"] or 0)) diskon_nominal = max(0.0, gross * (diskon_persen / 100.0)) except (TypeError, ValueError, KeyError, AttributeError): diskon_nominal = 0.0 setattr(detail_obj, "produk_ord_diskon_persen", diskon_persen) setattr(detail_obj, "produk_ord_diskon_nominal", diskon_nominal) mapped.append(detail_obj) return mapped def build_transaksi_dict_from_row(self, transaksi_row): """ Bangun transaksi_data_dict dari row transaksi (dict/sqlite Row). """ get_val = transaksi_row.get if hasattr(transaksi_row, "get") else lambda k, default=None: transaksi_row[k] if k in transaksi_row.keys() else default # PATCH[FrontEndAgent|PrintUnification]: Tambah field untuk print (samakan dengan history + payment) transaksi_dibayar = get_val("transaksi_dibayar", 0) transaksi_dibayar_return = get_val("transaksi_dibayar_return", 0) point_transaksi = self._as_positive_int( get_val("point_transaksi", 0), self._extract_point_from_diskon_log(get_val("diskon_log", "")), ) ppn_mode = str(get_val("ppn_mode", "") or "").strip().lower() if ppn_mode not in {"include", "exclude"}: ppn_mode = self._extract_ppn_mode_from_diskon_log(get_val("diskon_log", "")) return { "id": get_val("id"), "nomer": get_val("nomer"), "dtime": get_val("dtime"), "customers_id": get_val("customers_id", 1), "customers_nama": get_val("customers_nama", "Tunai"), "transaksi_nilai": get_val("transaksi_nilai", 0), "transaksi_bulat": get_val("transaksi_bulat", 0), "ppn_persen": get_val("ppn_persen", 0), "diskon_persen": get_val("diskon_persen", 0), "settlement_id": get_val("settlement_id", 1), "oleh_id": get_val("oleh_id"), "oleh_nama": get_val("oleh_nama", ""), "transaksi_dibayar": transaksi_dibayar, "transaksi_dibayar_return": transaksi_dibayar_return, "point_transaksi": point_transaksi, "ppn_mode": ppn_mode, "diskon_log": get_val("diskon_log", ""), # PATCH: Duplicate untuk kompatibilitas print "jumlah_bayar": transaksi_dibayar, "kembalian": transaksi_dibayar_return, "bank_nama": get_val("bank_nama", ""), "bank_rekening_nama": get_val("bank_rekening_nama", ""), "rekening": get_val("rekening", ""), # PATCH: Field tambahan untuk print "kasir_nama": get_val("oleh_nama", ""), "customer_nama": get_val("customers_nama", "Tunai"), "skip_logo": True, # Skip logo (too large) } def apply_single_payment(self, transaksi_data_dict, hasil_pembayaran): """ Lengkapi transaksi_data_dict dengan info pembayaran tunggal. """ # edited by glg # Pindahkan mapping pembayaran ke use-case service agar file model tetap ringkas. return self.payment_payload_service.apply_single_payment( transaksi_data_dict=transaksi_data_dict, hasil_pembayaran=hasil_pembayaran, extract_diskon_customer_callback=self._extract_diskon_customer_from_diskon_log, resolve_bank_rekening_callback=self._resolve_bank_rekening_fields, ) def apply_multi_payment(self, transaksi_data_dict, payment_list): """ Lengkapi transaksi_data_dict untuk multi pembayaran (split). """ # edited by glg # Multi-payment dipusatkan ke use-case service agar konsisten dengan single-payment. return self.payment_payload_service.apply_multi_payment( transaksi_data_dict=transaksi_data_dict, payment_list=payment_list, extract_diskon_customer_callback=self._extract_diskon_customer_from_diskon_log, resolve_bank_rekening_callback=self._resolve_bank_rekening_fields, )