# edited by glg import json import sqlite3 from datetime import datetime, timedelta from pypos.core.base_model import BaseModel from pypos.core.utils.db_helper import connect_sqlite, connect_sqlite_read_fast from pypos.core.utils.path_utils import get_db_path from pypos.core.utils.sql_query_builder import render_sql_template from pypos.core.database.schema_migrator import run_schema_migrations_once from pypos.modules.penjualan.errors import PembatalanProcessError # upgraded: inherit base class class PembatalanTransaksiModel(BaseModel): def __init__(self, db_path: str = None, settlement_lock_service=None): super().__init__() self.db_path = db_path or get_db_path() # edited by glg # Validasi lock settlement pembatalan dipusatkan pada shared service. self.settlement_lock_service = settlement_lock_service or self._build_default_settlement_lock_service() self._ensure_history_table() # edited by glg def _build_default_settlement_lock_service(self): module = __import__( "pypos.modules.penjualan.services.settlement_mutation_lock_service", fromlist=["SettlementMutationLockService"], ) return module.SettlementMutationLockService( db_path=self.db_path, log_warning=self.log_warning, ) # edited by glg def _connect_read(self): return connect_sqlite_read_fast(self.db_path) # edited by glg def _connect_write(self): return connect_sqlite(self.db_path) def _ensure_history_table(self): self._ensure_schema_ready_or_fail("CANCEL_SCHEMA_MIGRATION_FAILED") conn = self._connect_write() try: cursor = conn.cursor() conn.commit() finally: conn.close() # edited by glg def _ensure_schema_ready_or_fail(self, reason_code="CANCEL_SCHEMA_MIGRATION_FAILED"): ok = run_schema_migrations_once(self.db_path, strict=False) # Kompatibilitas: beberapa jalur migrator lama mengembalikan None saat sukses/no-op. if ok is not False: 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 _to_non_negative_int(self, value, default=0): try: parsed = int(float(value)) except (TypeError, ValueError): parsed = int(default or 0) return parsed if parsed >= 0 else int(default or 0) # edited by glg # Gunakan range datetime agar filter tanggal tetap index-friendly. def _build_datetime_range(self, start_date: str, end_date: str): try: start_dt = datetime.strptime(str(start_date or "").strip(), "%Y-%m-%d") end_dt = datetime.strptime(str(end_date or "").strip(), "%Y-%m-%d") except (TypeError, ValueError): return None if end_dt < start_dt: return None range_start = start_dt.strftime("%Y-%m-%d 00:00:00") range_end = (end_dt + timedelta(days=1)).strftime("%Y-%m-%d 00:00:00") return range_start, range_end # edited by glg def _is_transaksi_in_settlement_history(self, cursor, transaksi_id: str): _ = cursor return bool(self.settlement_lock_service.is_transaksi_locked(transaksi_id)) def search_transaksi(self, keyword: str, start_date: str, end_date: str, limit: int = 200): kw = f"%{keyword}%" params = [] date_clause = "" if start_date and end_date: date_range = self._build_datetime_range(start_date, end_date) if date_range: date_clause = " AND t.dtime >= ? AND t.dtime < ?" params.extend([date_range[0], date_range[1]]) else: date_clause = " AND date(t.dtime) BETWEEN ? AND ?" params.extend([start_date, end_date]) params.extend([kw, kw, kw, int(limit)]) conn = self._connect_read() conn.row_factory = sqlite3.Row try: cursor = conn.cursor() # edited by glg # Optimasi query pencarian pembatalan: # - kandidat transaksi dibatasi dulu lewat CTE base (LIMIT) # - agregasi return/item dihitung hanya untuk kandidat tersebut. query = render_sql_template( """ WITH base AS ( SELECT t.id, t.nomer, t.dtime, t.customers_nama, t.oleh_nama AS kasir_nama, COALESCE(t.transaksi_nilai, 0) AS transaksi_nilai FROM transaksi t WHERE t.jenis_label = 'invoice' AND COALESCE(t.trash, 0) = 0 AND COALESCE(t.settlement_id, 1) = 1 {date_clause} AND (t.nomer LIKE ? OR t.dtime LIKE ? OR t.customers_nama LIKE ?) ORDER BY t.dtime DESC LIMIT ? ), return_cash AS ( SELECT r.transaksi_id, SUM(COALESCE(dr.harga, 0) * COALESCE(dr.jumlah, 0)) AS total_refund_cash FROM return_transaksi_penjualan r JOIN detail_return_transaksi_penjualan dr ON dr.return_id = r.id WHERE r.transaksi_id IN (SELECT id FROM base) AND COALESCE(r.refund_method, 'cash') = 'cash' GROUP BY r.transaksi_id ), sold_rows AS ( SELECT td.transaksi_id, td.produk_id, COALESCE(td.produk_ord_jml, 0) AS qty_jual FROM transaksi_data td WHERE td.transaksi_id IN (SELECT id FROM base) AND COALESCE(td.trash, 0) = 0 ), returned_qty AS ( SELECT r.transaksi_id, dr.produk_id, SUM(COALESCE(dr.jumlah, 0)) AS qty_returned FROM return_transaksi_penjualan r JOIN detail_return_transaksi_penjualan dr ON dr.return_id = r.id WHERE r.transaksi_id IN (SELECT id FROM base) AND COALESCE(r.refund_method, 'cash') = 'cash' GROUP BY r.transaksi_id, dr.produk_id ), item_sisa AS ( SELECT s.transaksi_id, COUNT(1) AS jumlah_item FROM sold_rows s LEFT JOIN returned_qty rt ON rt.transaksi_id = s.transaksi_id AND rt.produk_id = s.produk_id WHERE (COALESCE(s.qty_jual, 0) - COALESCE(rt.qty_returned, 0)) > 0 GROUP BY s.transaksi_id ) SELECT b.id, b.nomer, b.dtime, b.customers_nama, b.kasir_nama, (COALESCE(b.transaksi_nilai, 0) - COALESCE(rc.total_refund_cash, 0)) AS transaksi_nilai, COALESCE(i.jumlah_item, 0) AS jumlah_item FROM base b LEFT JOIN return_cash rc ON rc.transaksi_id = b.id LEFT JOIN item_sisa i ON i.transaksi_id = b.id ORDER BY b.dtime DESC """, date_clause=date_clause, ) cursor.execute(query, params) return cursor.fetchall() finally: conn.close() # edited by glg def get_transaksi_cancel_context(self, transaksi_id: str): conn = self._connect_read() conn.row_factory = sqlite3.Row try: cursor = conn.cursor() cursor.execute( """ SELECT t.id, t.nomer, t.dtime, COALESCE(t.jenis_label, '') AS jenis_label, COALESCE(t.trash, 0) AS trash, COALESCE(t.settlement_id, 1) AS settlement_id, ( SELECT COUNT(1) FROM return_transaksi_penjualan r WHERE r.transaksi_id = t.id ) AS return_count FROM transaksi t WHERE t.id = ? LIMIT 1 """, (transaksi_id,), ) row = cursor.fetchone() if not row: return None payload = dict(row) payload["trash"] = self._to_non_negative_int(payload.get("trash"), 0) payload["settlement_id"] = self._to_non_negative_int(payload.get("settlement_id"), 1) payload["return_count"] = self._to_non_negative_int(payload.get("return_count"), 0) return payload finally: conn.close() def get_transaksi_date(self, transaksi_id: str): conn = self._connect_read() conn.row_factory = sqlite3.Row try: cursor = conn.cursor() cursor.execute( "SELECT dtime FROM transaksi WHERE id = ? AND COALESCE(trash, 0) = 0", (transaksi_id,) ) row = cursor.fetchone() return row["dtime"] if row else None finally: conn.close() def delete_transaksi(self, transaksi_id: str, admin_name: str = "", dibatalkan_oleh_id=None, dibatalkan_oleh_nama=None): conn = self._connect_write() conn.row_factory = sqlite3.Row try: cursor = conn.cursor() trx_id = str(transaksi_id or "").strip() if not trx_id: raise PembatalanProcessError( "CANCEL_INVALID_TRANSAKSI_ID", "ID transaksi tidak valid.", ) # Ambil snapshot transaksi untuk history cursor.execute( """ SELECT nomer, dtime, customers_nama, oleh_nama, transaksi_nilai, COALESCE(trash, 0) AS trash, COALESCE(settlement_id, 1) AS settlement_id, COALESCE(jenis_label, '') AS jenis_label FROM transaksi WHERE id = ? """, (trx_id,) ) row = cursor.fetchone() if not row: raise PembatalanProcessError( "CANCEL_TRANSAKSI_NOT_FOUND", "Transaksi tidak ditemukan.", ) if self._to_non_negative_int(row["trash"], 0) == 1: raise PembatalanProcessError( "CANCEL_ALREADY_DONE", "Transaksi sudah dibatalkan.", ) if self._to_non_negative_int(row["settlement_id"], 1) == 0: raise PembatalanProcessError( "CANCEL_SETTLEMENT_LOCKED", "Transaksi sudah disettle dan tidak dapat dibatalkan.", ) if self._is_transaksi_in_settlement_history(cursor, trx_id): raise PembatalanProcessError( "CANCEL_SETTLEMENT_LOCKED", "Transaksi sudah disettle dan tidak dapat dibatalkan.", ) if str(row["jenis_label"] or "").strip().lower() != "invoice": raise PembatalanProcessError( "CANCEL_INVALID_JENIS_LABEL", "Hanya transaksi invoice yang dapat dibatalkan.", ) cancel_dtime = datetime.now().strftime("%Y-%m-%d %H:%M:%S") cursor.execute( """ UPDATE transaksi SET trash = 1, cancel_dtime = ? WHERE id = ? AND COALESCE(trash, 0) = 0 AND COALESCE(settlement_id, 1) = 1 AND COALESCE(jenis_label, 'invoice') = 'invoice' """, (cancel_dtime, trx_id), ) if int(cursor.rowcount or 0) <= 0: raise PembatalanProcessError( "CANCEL_UPDATE_REJECTED", "Transaksi tidak dapat dibatalkan.", ) cursor.execute( "UPDATE transaksi_data SET trash = 1 WHERE transaksi_id = ? AND COALESCE(trash, 0) = 0", (trx_id,), ) # Simpan history pembatalan nomer = row["nomer"] if row else None transaksi_dtime = row["dtime"] if row else None customers_nama = row["customers_nama"] if row else None kasir_nama = row["oleh_nama"] if row else None transaksi_nilai = row["transaksi_nilai"] if row else None cursor.execute( """ INSERT INTO pembatalan_transaksi_history (transaksi_id, nomer, transaksi_dtime, customers_nama, kasir_nama, transaksi_nilai, admin_verifikasi, dibatalkan_oleh_id, dibatalkan_oleh_nama, cancel_dtime) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( trx_id, nomer, transaksi_dtime, customers_nama, kasir_nama, transaksi_nilai, admin_name or "-", str(dibatalkan_oleh_id) if dibatalkan_oleh_id is not None else None, dibatalkan_oleh_nama, cancel_dtime ) ) conn.commit() return True except PembatalanProcessError: conn.rollback() raise except sqlite3.Error as exc: conn.rollback() raise PembatalanProcessError( "CANCEL_DB_ERROR", "Gagal membatalkan transaksi.", cause=exc, ) from exc except (TypeError, ValueError, KeyError) as exc: conn.rollback() raise PembatalanProcessError( "CANCEL_DATA_ERROR", "Data pembatalan transaksi tidak valid.", cause=exc, ) from exc except Exception as exc: conn.rollback() raise PembatalanProcessError( "CANCEL_UNEXPECTED_ERROR", "Terjadi kesalahan saat membatalkan transaksi.", cause=exc, ) from exc finally: conn.close() def get_history(self, keyword: str = "", start_date: str = None, end_date: str = None, limit: int = 200): kw = f"%{keyword}%" params = [] date_clause = "" if start_date and end_date: date_range = self._build_datetime_range(start_date, end_date) if date_range: date_clause = " AND cancel_dtime >= ? AND cancel_dtime < ?" params.extend([date_range[0], date_range[1]]) else: date_clause = " AND date(cancel_dtime) BETWEEN ? AND ?" params.extend([start_date, end_date]) params.extend([kw, kw, kw, kw, kw, kw]) params.append(int(limit)) conn = self._connect_read() conn.row_factory = sqlite3.Row try: cursor = conn.cursor() query = render_sql_template( """ SELECT transaksi_id, nomer, transaksi_dtime, customers_nama, kasir_nama, transaksi_nilai, admin_verifikasi, dibatalkan_oleh_id, dibatalkan_oleh_nama, cancel_dtime FROM pembatalan_transaksi_history WHERE 1=1 {date_clause} AND ( COALESCE(nomer, '') LIKE ? OR COALESCE(customers_nama, '') LIKE ? OR COALESCE(kasir_nama, '') LIKE ? OR COALESCE(admin_verifikasi, '') LIKE ? OR COALESCE(dibatalkan_oleh_nama, '') LIKE ? OR COALESCE(transaksi_id, '') LIKE ? ) ORDER BY cancel_dtime DESC LIMIT ? """, date_clause=date_clause, ) cursor.execute(query, params) return cursor.fetchall() finally: conn.close() def get_history_items(self, transaksi_id: str): conn = self._connect_read() conn.row_factory = sqlite3.Row try: cursor = conn.cursor() cursor.execute( """ SELECT produk_id, produk_nama, produk_ord_hrg, produk_ord_jml, produk_ord_diskon, satuan FROM transaksi_data WHERE transaksi_id = ? ORDER BY id ASC """, (transaksi_id,) ) return cursor.fetchall() finally: conn.close()