# edited by glg import logging import sqlite3 from datetime import datetime from typing import Any, Dict, List from pypos.modules.sinkronisasi.services.transaction_export_legacy_rows_builder_service import ( TransactionExportLegacyRowsBuilderService, ) class TransactionExportHotspotUseCaseService: """ Use-case service untuk memecah hotspot TransactionExportService. Fokus: - normalisasi row export - backfill metadata transaksi/transaksi_data """ def __init__(self, export_service, logger=None): self._svc = export_service self._logger = logger or logging.getLogger(__name__) self._legacy_rows_builder_service = TransactionExportLegacyRowsBuilderService( export_service=export_service, logger=self._logger, ) def normalize_rows_for_export( self, *, table_name: str, rows: List[Dict[str, Any]], batch_end: str, config: Dict[str, Any], device_context: Dict[str, Any], machine_fallback: str, ) -> List[Dict[str, Any]]: svc = self._svc if not rows: return rows machine_id = str(device_context.get("machine_id") or machine_fallback or "").strip() cabang_id = svc._as_positive_int( device_context.get("cabang_id"), svc._as_positive_int((config or {}).get("cabang_id"), 0), ) toko_id = svc._as_positive_int( device_context.get("toko_id"), svc._as_positive_int((config or {}).get("toko_id"), 0), ) fallback_dtime = str(batch_end or datetime.now().strftime("%Y-%m-%d %H:%M:%S")) for row in rows: if not isinstance(row, dict): continue if table_name == "transaksi": # Sanitasi awal berbasis schema lokal untuk menghindari null pada kolom NOT NULL. row = svc._sanitize_row_by_notnull_schema("transaksi", row) # Endpoint compile terbaru hanya memproses row dengan key link_id. if "link_id" not in row or svc._is_blank_value(row.get("link_id")): row["link_id"] = 0 # Kontrak worker strict endpoint: field ini wajib string non-kosong. default_print_operator = ( str((config or {}).get("export_default_print_oleh_nama") or "system").strip() or "system" ) default_kontainer_size = ( str((config or {}).get("export_default_kontainer_size") or "none").strip() or "none" ) default_ekspedisi_status = ( str((config or {}).get("export_default_ekspedisi_status") or "none").strip() or "none" ) if svc._is_blank_value(row.get("print_oleh_nama")): row["print_oleh_nama"] = ( str(row.get("oleh_nama") or "").strip() or default_print_operator ) if svc._is_blank_value(row.get("kontainer_size")): row["kontainer_size"] = default_kontainer_size if svc._is_blank_value(row.get("ekspedisi_status")): row["ekspedisi_status"] = default_ekspedisi_status # Kontrak worker strict: field numerik ini tidak boleh null. strict_numeric_defaults = { "bank_rekening_id": 0, "bank_rekening_id_from": 0, "bank_id_from": 0, "ekspedisi_id": 0, "kontainer_size_id": 0, "kontainer_status": 0, "kontainer_approve_id": 0, "gudang_id": 0, "gudang2_id": 0, "cabang2_id": 0, "seller_id": 0, "referensi_id": 0, "referensi_id2": 0, "tpl_alamat_id": 0, "pembayaran_tunai": 0, "pembayaran_non_tunai": 0, "kontainer_harga": 0, "print_oleh_id": 0, "jatuh_tempo": 0, } # Hardening tambahan dari observasi endpoint. strict_presence_defaults = { "cli": 0, "inv_dollar": 0, "next_step_num": 0, "counters": 0, "returned": 0, "returns": 0, "deposit_persen_in": 0, "transaksi_nilai_ori": 0, "trash2": 0, "r_jenis": 0, "_temp": 0, "cli_loop": 0, "status_edit": 0, "print_counter": 0, "print_status": 0, "setor_status": 0, "sinkron_cache": 0, "auto_pemindahan": 0, "counters_customers_id_all": 0, "counters_gudang_id_all": 0, "batal_kirim": 0, "kontainer_valas_nilai": 0, "tambahan_nilai": 0, "cek": 0, "status_4": 0, "trash_4": 0, "r_sales": 0, "r_maju": 0, "r_mundur": 0, } # Hardening sekali jalan: kolom *_id yang kosong diset 0 (kecuali x_id). for field_name, field_val in list(row.items()): if not str(field_name).endswith("_id"): continue if field_name in {"x_id", "machine_id", "cabang_id", "toko_id"}: continue if svc._is_blank_value(field_val): row[field_name] = 0 for field_name, default_val in strict_numeric_defaults.items(): if svc._is_blank_value(row.get(field_name)): row[field_name] = default_val else: row[field_name] = svc._as_number_or_default(row.get(field_name), default_val) for field_name, default_val in strict_presence_defaults.items(): if field_name not in row: continue if not svc._is_blank_value(row.get(field_name)): continue row[field_name] = default_val if int(svc._as_number_or_default(row.get("print_oleh_id"), 0)) == 0: row["print_oleh_id"] = svc._as_number_or_default(row.get("oleh_id"), 0) if "jenis" not in row or svc._is_blank_value(row.get("jenis")): jenis_label = str(row.get("jenis_label") or "").strip().lower() if jenis_label in {"invoice", "penjualan", "jual", "transaksi"}: row["jenis"] = 582 elif jenis_label in {"settlement"}: row["jenis"] = 758 elif jenis_label in {"pembatalan", "batal"}: row["jenis"] = 982 else: row["jenis"] = 582 if "status" not in row or svc._is_blank_value(row.get("status")): row["status"] = 1 if "trash" not in row or svc._is_blank_value(row.get("trash")): row["trash"] = 0 if "nomer" not in row or svc._is_blank_value(row.get("nomer")): ref_nomer = row.get("x_id") if svc._is_blank_value(ref_nomer): ref_nomer = row.get("id") row["nomer"] = str(ref_nomer or "trx") if svc._is_blank_value(row.get("x_id")): ref_xid = row.get("id") if svc._is_blank_value(ref_xid): ref_xid = row.get("nomer") if not svc._is_blank_value(ref_xid): row["x_id"] = ref_xid if svc._is_blank_value(row.get("inv")): row["inv"] = row.get("nomer") if svc._is_blank_value(row.get("suppliers_id")): row["suppliers_id"] = 0 if svc._is_blank_value(row.get("suppliers_nama")): row["suppliers_nama"] = "" if table_name in { "transaksi", "transaksi_data", "transaksi_data_registry", "settlement_history", "transaksi_settlement", }: if machine_id and ("machine_id" not in row or svc._is_blank_value(row.get("machine_id"))): row["machine_id"] = machine_id resolved_cabang_id = svc._as_positive_int(row.get("cabang_id"), 0) if resolved_cabang_id <= 0: resolved_cabang_id = svc._as_positive_int(cabang_id, 0) if resolved_cabang_id > 0: row["cabang_id"] = resolved_cabang_id resolved_toko_id = svc._as_positive_int(row.get("toko_id"), 0) if resolved_toko_id <= 0: resolved_toko_id = svc._as_positive_int(toko_id, 0) if resolved_toko_id > 0: row["toko_id"] = resolved_toko_id if "dtime" not in row or svc._is_blank_value(row.get("dtime")): row["dtime"] = fallback_dtime if table_name in {"transaksi", "transaksi_data"}: if "x_id" not in row or svc._is_blank_value(row.get("x_id")): if not svc._is_blank_value(row.get("id")): row["x_id"] = row.get("id") if table_name == "transaksi": final_cabang_id = svc._as_positive_int(row.get("cabang_id"), 0) if final_cabang_id <= 0: ref_nomer = str(row.get("nomer") or row.get("x_id") or row.get("id") or "").strip() raise ValueError( "invalid_zero_components:cabang_id " f"table=transaksi nomer={ref_nomer} machine_id={machine_id}" ) row["cabang_id"] = final_cabang_id if "fulldate" not in row or svc._is_blank_value(row.get("fulldate")): ref_dtime = row.get("dtime") if svc._is_blank_value(ref_dtime): ref_dtime = fallback_dtime row["fulldate"] = str(ref_dtime)[:10] if table_name == "transaksi_data": if svc._is_blank_value(row.get("transaksi_id")): if not svc._is_blank_value(row.get("transaksiID")): row["transaksi_id"] = row.get("transaksiID") elif not svc._is_blank_value(row.get("x_id")): row["transaksi_id"] = row.get("x_id") if "x_id" not in row or svc._is_blank_value(row.get("x_id")): if not svc._is_blank_value(row.get("id")): row["x_id"] = row.get("id") elif not svc._is_blank_value(row.get("transaksi_id")) and not svc._is_blank_value( row.get("produk_id") ): row["x_id"] = f"{row.get('transaksi_id')}-{row.get('produk_id')}" elif not svc._is_blank_value(row.get("transaksi_id")): row["x_id"] = row.get("transaksi_id") if table_name == "transaksi_data_registry": if svc._is_blank_value(row.get("transaksi_id")) and not svc._is_blank_value( row.get("transaksiID") ): row["transaksi_id"] = row.get("transaksiID") if svc._is_blank_value(row.get("transaksi_id")) and not svc._is_blank_value(row.get("x_id")): row["transaksi_id"] = row.get("x_id") if svc._is_blank_value(row.get("x_id")) and not svc._is_blank_value(row.get("transaksi_id")): row["x_id"] = row.get("transaksi_id") return rows # edited by glg # edited by glg def build_legacy_return_rows_payload( self, *, return_rows: List[Dict[str, Any]], batch_end: str, ) -> List[Dict[str, Any]]: return self._legacy_rows_builder_service.build_legacy_return_rows_payload( return_rows=return_rows, batch_end=batch_end, ) # edited by glg # edited by glg def build_legacy_transaksi_rows( self, *, transaksi_rows: List[Dict[str, Any]], batch_end: str, extra_return_rows: List[Dict[str, Any]] = None, ) -> List[Dict[str, Any]]: return self._legacy_rows_builder_service.build_legacy_transaksi_rows( transaksi_rows=transaksi_rows, batch_end=batch_end, extra_return_rows=extra_return_rows, ) def backfill_transaksi_rows_in_db( self, *, rows: List[Dict[str, Any]], batch_end: str, config: Dict[str, Any], device_context: Dict[str, Any], ) -> List[Dict[str, Any]]: svc = self._svc if not rows: return rows transaksi_meta = svc._get_table_schema_meta("transaksi") detail_meta = svc._get_table_schema_meta("transaksi_data") if not transaksi_meta: return rows trans_cols = set(transaksi_meta.keys()) detail_cols = set(detail_meta.keys()) default_cabang_id = svc._as_positive_int(device_context.get("cabang_id"), 0) default_toko_id = svc._as_positive_int( device_context.get("toko_id"), svc._as_positive_int((config or {}).get("toko_id"), 0), ) fallback_dtime = str(batch_end or datetime.now().strftime("%Y-%m-%d %H:%M:%S")) out_rows = [dict(r) for r in rows if isinstance(r, dict)] if not out_rows: return rows updated_transaksi = 0 updated_detail = 0 transaksi_ids: List[int] = [] conn = None try: conn = svc._connect() conn.row_factory = sqlite3.Row cur = conn.cursor() default_cabang_nama = svc._lookup_cabang_nama_for_backfill(cur, default_cabang_id) default_toko_nama = svc._lookup_toko_nama_for_backfill(cur, default_toko_id) default_gudang_id, default_gudang_nama = svc._resolve_cabang_default_gudang_for_backfill( cur, default_cabang_id, ) for row in out_rows: trx_id = svc._as_positive_int(row.get("id"), 0) if trx_id <= 0: continue transaksi_ids.append(trx_id) update_map: Dict[str, Any] = {} cabang_id = svc._as_positive_int(row.get("cabang_id"), 0) if "cabang_id" in trans_cols and cabang_id <= 0 and default_cabang_id > 0: cabang_id = default_cabang_id update_map["cabang_id"] = cabang_id row["cabang_id"] = cabang_id cabang_nama = str(row.get("cabang_nama") or "").strip() if "cabang_nama" in trans_cols and svc._is_blank_value(cabang_nama): resolved_cabang_nama = ( svc._lookup_cabang_nama_for_backfill(cur, cabang_id) if cabang_id > 0 else "" ) if not resolved_cabang_nama: resolved_cabang_nama = default_cabang_nama if resolved_cabang_nama: update_map["cabang_nama"] = resolved_cabang_nama row["cabang_nama"] = resolved_cabang_nama toko_id = svc._as_positive_int(row.get("toko_id"), 0) if "toko_id" in trans_cols and toko_id <= 0 and default_toko_id > 0: toko_id = default_toko_id update_map["toko_id"] = toko_id row["toko_id"] = toko_id toko_nama = str(row.get("toko_nama") or "").strip() if "toko_nama" in trans_cols and svc._is_blank_value(toko_nama): resolved_toko_nama = svc._lookup_toko_nama_for_backfill(cur, toko_id) if toko_id > 0 else "" if not resolved_toko_nama: resolved_toko_nama = default_toko_nama if resolved_toko_nama: update_map["toko_nama"] = resolved_toko_nama row["toko_nama"] = resolved_toko_nama if "gudang_id" in trans_cols: gudang_val = int(svc._to_int(row.get("gudang_id"), 0)) resolved_gudang_id, resolved_gudang_nama = svc._resolve_cabang_default_gudang_for_backfill( cur, cabang_id if cabang_id > 0 else default_cabang_id, ) if resolved_gudang_id == 0: resolved_gudang_id = int(default_gudang_id or 0) if not resolved_gudang_nama: resolved_gudang_nama = str(default_gudang_nama or "").strip() if resolved_gudang_id != 0 and gudang_val != resolved_gudang_id: update_map["gudang_id"] = resolved_gudang_id row["gudang_id"] = resolved_gudang_id if ( "gudang_nama" in trans_cols and svc._is_blank_value(row.get("gudang_nama")) and str(resolved_gudang_nama or "").strip() ): update_map["gudang_nama"] = str(resolved_gudang_nama).strip() row["gudang_nama"] = str(resolved_gudang_nama).strip() for key_name, default_val in ( ("gudang_id", 0), ("reference_id", 0), ("reference_jenis", 0), ("reference_nomer", 0), ("status", 1), ("trash", 0), ): if key_name in trans_cols and svc._is_blank_value(row.get(key_name)): update_map[key_name] = default_val row[key_name] = default_val if "pembayaran_non_tunai" in trans_cols and svc._is_blank_value(row.get("pembayaran_non_tunai")): update_map["pembayaran_non_tunai"] = 0 row["pembayaran_non_tunai"] = 0 non_tunai_val = float(svc._as_number_or_default(row.get("pembayaran_non_tunai"), 0)) dibayar_val = float(svc._as_number_or_default(row.get("transaksi_dibayar"), 0)) tunai_calc = max(0.0, dibayar_val - max(0.0, non_tunai_val)) if "pembayaran_tunai" in trans_cols and svc._is_blank_value(row.get("pembayaran_tunai")): tunai_rounded = int(round(tunai_calc)) update_map["pembayaran_tunai"] = tunai_rounded row["pembayaran_tunai"] = tunai_rounded tunai_val = float(svc._as_number_or_default(row.get("pembayaran_tunai"), tunai_calc)) if "pembayaran_sys" in trans_cols and svc._is_blank_value(row.get("pembayaran_sys")): if non_tunai_val > 0 and tunai_val > 0: pembayaran_sys = "multi" elif non_tunai_val > 0: pembayaran_sys = "non_tunai" else: pembayaran_sys = "tunai" update_map["pembayaran_sys"] = pembayaran_sys row["pembayaran_sys"] = pembayaran_sys if "pembayaran" in trans_cols and svc._is_blank_value(row.get("pembayaran")): pembayaran_label = str(row.get("bank_nama") or row.get("pembayaran_sys") or "tunai").strip() update_map["pembayaran"] = pembayaran_label row["pembayaran"] = pembayaran_label for key_name in ("kartu_nomer", "nomer_ep"): if key_name in trans_cols and svc._is_blank_value(row.get(key_name)): update_map[key_name] = "" row[key_name] = "" need_rekening_id = "bank_rekening_id" in trans_cols and svc._is_blank_value( row.get("bank_rekening_id") ) need_rekening_nama = "bank_rekening_nama" in trans_cols and svc._is_blank_value( row.get("bank_rekening_nama") ) if need_rekening_id or need_rekening_nama: pembayaran_sys_norm = str(row.get("pembayaran_sys") or "").strip().lower() is_tunai = pembayaran_sys_norm in {"", "tunai"} and non_tunai_val <= 0 rekening_id, rekening_nama = svc._resolve_bank_rekening_for_backfill( cursor=cur, bank_id=svc._as_positive_int(row.get("bank_id"), 0), cabang_id=cabang_id, is_tunai=is_tunai, ) if need_rekening_id: update_map["bank_rekening_id"] = rekening_id row["bank_rekening_id"] = rekening_id if need_rekening_nama: update_map["bank_rekening_nama"] = rekening_nama row["bank_rekening_nama"] = rekening_nama if "rekening" in trans_cols and svc._is_blank_value(row.get("rekening")): update_map["rekening"] = rekening_nama row["rekening"] = rekening_nama if update_map: filtered = {k: v for k, v in update_map.items() if k in trans_cols} if filtered: set_clause = svc._build_update_set_clause(filtered) params = list(filtered.values()) + [trx_id] cur.execute(f"UPDATE transaksi SET {set_clause} WHERE id = ?", params) if cur.rowcount > 0: updated_transaksi += 1 if transaksi_ids and detail_cols: placeholders = ",".join(["?"] * len(transaksi_ids)) cur.execute( f""" SELECT td.id AS detail_id, td.transaksi_id, td.dtime, td.cabang_id, td.oleh_id, td.oleh_nama, td.status, td.trash, td.kirim, td.detail_tipe, td.transaksi_jenis, td.valid_qty, td.valid_qty_no_approve, td.produk_ord_jml, td.produk_ord_jml_return, td.produk_ord_stok, td.rekening_id_asal, td.rekening_id_tujuan, t.dtime AS trx_dtime, t.cabang_id AS trx_cabang_id, t.oleh_id AS trx_oleh_id, t.oleh_nama AS trx_oleh_nama, t.transaksi_jenis AS trx_jenis FROM transaksi_data td LEFT JOIN transaksi t ON t.id = td.transaksi_id WHERE td.transaksi_id IN ({placeholders}) """, tuple(transaksi_ids), ) for detail_row in cur.fetchall() or []: detail_id = svc._as_positive_int(detail_row["detail_id"], 0) if detail_id <= 0: continue detail_update = {} if "dtime" in detail_cols and svc._is_blank_value(detail_row["dtime"]): detail_update["dtime"] = str(detail_row["trx_dtime"] or fallback_dtime) if "cabang_id" in detail_cols: detail_cabang = svc._as_positive_int(detail_row["cabang_id"], 0) if detail_cabang <= 0: parent_cabang = svc._as_positive_int(detail_row["trx_cabang_id"], 0) if parent_cabang > 0: detail_update["cabang_id"] = parent_cabang if "oleh_id" in detail_cols and svc._is_blank_value(detail_row["oleh_id"]): detail_update["oleh_id"] = svc._as_positive_int(detail_row["trx_oleh_id"], 0) if "oleh_nama" in detail_cols and svc._is_blank_value(detail_row["oleh_nama"]): detail_update["oleh_nama"] = str(detail_row["trx_oleh_nama"] or "").strip() qty_val = int(svc._as_number_or_default(detail_row["produk_ord_jml"], 0) or 0) if qty_val < 0: qty_val = 0 transaksi_jenis = str(detail_row["trx_jenis"] or "758").strip() or "758" detail_defaults = { "status": 1, "trash": 0, "kirim": 0, "detail_tipe": "items", "transaksi_jenis": transaksi_jenis, "valid_qty": qty_val, "valid_qty_no_approve": qty_val, "produk_ord_jml_return": 0, "produk_ord_stok": qty_val, "rekening_id_asal": 0, "rekening_id_tujuan": 0, } for key_name, default_val in detail_defaults.items(): if key_name not in detail_cols: continue if not svc._is_blank_value(detail_row[key_name]): continue detail_update[key_name] = default_val detail_update = {k: v for k, v in detail_update.items() if k in detail_cols} if detail_update: set_clause = svc._build_update_set_clause(detail_update) params = list(detail_update.values()) + [detail_id] cur.execute(f"UPDATE transaksi_data SET {set_clause} WHERE id = ?", params) if cur.rowcount > 0: updated_detail += 1 if updated_transaksi or updated_detail: conn.commit() self._logger.info( "[DEBUG EXPORT] backfill transaksi rows=%s detail_rows=%s", updated_transaksi, updated_detail, ) else: conn.rollback() except (sqlite3.Error, ValueError, TypeError, RuntimeError) as exc: if conn is not None: try: conn.rollback() except sqlite3.Error: self._logger.debug("[DEBUG EXPORT] rollback backfill gagal") self._logger.warning("[DEBUG EXPORT] backfill transaksi gagal: %s", exc) finally: if conn is not None: try: conn.close() except sqlite3.Error: self._logger.debug("[DEBUG EXPORT] close koneksi backfill gagal") return out_rows