import logging from pypos.core.utils.config_utils import read_config from pypos.core.utils.device_utils import get_active_device_info, get_device_id from pypos.modules.penjualan.services.ppn_server_settings_service import PpnServerSettingsService from pypos.modules.sinkronisasi.config.sync_config import get_sync_page_size from pypos.modules.sinkronisasi.models.sinkron_model import SENTINEL_TS, SinkronModel from pypos.modules.sinkronisasi.services.network_orchestrator_service import NetworkOrchestratorService from pypos.modules.sinkronisasi.services.sync_circuit_breaker_service import SyncCircuitBreakerService LOGGER = logging.getLogger(__name__) class SyncExecutionService: def __init__( self, model_factory=SinkronModel, config_reader=read_config, device_id_getter=get_device_id, active_device_getter=get_active_device_info, page_size_getter=get_sync_page_size, circuit_breaker_factory=SyncCircuitBreakerService, network_orchestrator_factory=NetworkOrchestratorService, ): self.model_factory = model_factory self.config_reader = config_reader self.device_id_getter = device_id_getter self.active_device_getter = active_device_getter self.page_size_getter = page_size_getter self.circuit_breaker = circuit_breaker_factory() self.network_orchestrator = network_orchestrator_factory() def _emit_progress(self, progress_callback, percent, status, detail_log): if callable(progress_callback): progress_callback(int(percent or 0), str(status or ""), str(detail_log or "")) def _resolve_device_context(self): config = self.config_reader() device_id = self.device_id_getter() device_info = self.active_device_getter(device_id) machine_id = device_info.get("machine_id") if device_info else (config.get("machine_id") or device_id) cabang_id = device_info.get("cabang_id") if device_info else config.get("cabang_id", 0) def _valid_positive_int(value): try: return int(value) > 0 except (TypeError, ValueError): return False if not device_info or not str(machine_id or "").strip() or not _valid_positive_int(cabang_id): raise RuntimeError("Device/cabang belum terdaftar. Registrasikan device terlebih dahulu.") return machine_id, int(cabang_id) def _extract_check_row(self, check_response): if not isinstance(check_response, dict): return 0 try: return int(check_response.get("row") or 0) except (TypeError, ValueError): return 0 def _extract_sync_data(self, sync_response): if not isinstance(sync_response, dict): return {} data_field = sync_response.get("data", {}) if isinstance(data_field, dict): return data_field return {} # edited by glg @staticmethod def _extract_table_rows(table_payload): """ Kompatibilitas payload sinkronisasi: - format baru: {"new": [..]} - format legacy: [..] """ if isinstance(table_payload, list): return [row for row in table_payload if isinstance(row, dict)] if not isinstance(table_payload, dict): return [] for key in ("new", "rows", "data", "items"): rows = table_payload.get(key) if isinstance(rows, list): return [row for row in rows if isinstance(row, dict)] return [] def _calc_progress_percent(self, idx, total_tables): if total_tables <= 0: return 100 return 20 + int((idx / total_tables) * 80) def _sync_ppn_server_setting_safely(self): # edited by glg # Non-blocking untuk alur utama sinkronisasi: # gagal fetch setting pajak tidak boleh membuat sinkronisasi master gagal. try: result = PpnServerSettingsService().sync_if_configured() status = int(result.get("status") or 0) reason = str(result.get("reason") or "").strip() source = str(result.get("source") or "").strip() if status == 1: if source == "server": LOGGER.info( "[SyncExecution] PPN setting dari server: mode=%s percent=%s changed=%s", str(result.get("mode") or "include"), int(result.get("ppn_percent") or 0), int(bool(result.get("changed"))), ) else: LOGGER.info( "[SyncExecution] PPN setting fallback aktif: mode=%s percent=%s reason=%s", str(result.get("mode") or "include"), int(result.get("ppn_percent") or 0), str(result.get("fallback_reason") or reason or "-"), ) return if reason not in {"missing_api_base_url", "missing_ep_ppn_settings"}: LOGGER.warning("[SyncExecution] Sinkron setting PPN dilewati: %s", reason or "unknown") except (RuntimeError, TypeError, ValueError) as exc: LOGGER.warning("[SyncExecution] Sinkron setting PPN gagal non-fatal: %s", exc) def execute(self, table_list, force_full_tables=None, full_refresh_tables=None, progress_callback=None): model = None try: # edited by glg # Single orchestrator gate untuk operasi sinkronisasi. # Jika jaringan/server belum layak, sinkronisasi ditunda lebih awal # agar tidak memicu retry berantai yang tidak perlu. gate = self.network_orchestrator.can_run( process=NetworkOrchestratorService.PROCESS_SYNC, force_probe=True, ) if not bool(gate.get("allow")): raise RuntimeError( "Sinkronisasi ditunda karena jaringan belum stabil. " f"(state={gate.get('state')}, reason={gate.get('reason')})" ) self.circuit_breaker.guard_or_raise() model = self.model_factory() updated_total = 0 machine_id, cabang_id = self._resolve_device_context() tables = list(table_list or []) force_tables = set(force_full_tables or []) refresh_tables = set(full_refresh_tables or []) request_force_full = force_tables.union(refresh_tables) self._emit_progress(progress_callback, 10, "Memeriksa update server...", "Menghubungi server") check = model.check_update_server( machine_id=machine_id, cabang_id=cabang_id, tables=tables, force_full_tables=list(request_force_full), ) check_row = self._extract_check_row(check) if check_row == 0 and not force_tables and not refresh_tables: self._emit_progress(progress_callback, 100, "Up-to-date", "Tidak ada data baru") return 0 self._emit_progress(progress_callback, 20, "Mengambil data dari server...", "Memproses permintaan") tables_to_sync = tables if check_row == 0 and request_force_full: tables_to_sync = [name for name in tables if name in request_force_full] sync_data = {} non_refresh_tables = [name for name in tables_to_sync if name not in refresh_tables] if non_refresh_tables: sync_response = model.sync_data_server( machine_id, cabang_id, non_refresh_tables, force_full_tables=list(request_force_full), ) sync_data = self._extract_sync_data(sync_response) total_tables = len(tables_to_sync) for idx, table_name in enumerate(tables_to_sync, 1): if table_name in refresh_tables: page_size = int(self.page_size_getter() or 500) offset = 0 first_page = True while True: paged = model.sync_data_server_table( machine_id, cabang_id, table_name, SENTINEL_TS, last_id=0, partial=offset, page_size=page_size, ) data_field = paged.get("data", {}) if isinstance(paged, dict) else {} table_payload = data_field.get(table_name, {}) if isinstance(data_field, dict) else {} new_data = self._extract_table_rows(table_payload) if not new_data: break updated = model.apply_sync_result( table_name, new_data, full_refresh=first_page, cabang_id=cabang_id, ) updated_total += updated first_page = False if len(new_data) < page_size: break offset += page_size progress_percent = self._calc_progress_percent(idx, total_tables) self._emit_progress( progress_callback, progress_percent, f"Sinkron {table_name}", "paged sync selesai", ) continue lookup_key = table_name if isinstance(sync_data, dict) and table_name not in sync_data: if table_name == "per_employee" and "employee" in sync_data: lookup_key = "employee" elif table_name == "per_customers" and "customer" in sync_data: lookup_key = "customer" table_payload = sync_data.get(lookup_key, {}) if isinstance(sync_data, dict) else {} new_data = self._extract_table_rows(table_payload) updated = model.apply_sync_result( table_name, new_data, full_refresh=(table_name in refresh_tables), cabang_id=cabang_id, ) updated_total += updated progress_percent = self._calc_progress_percent(idx, total_tables) self._emit_progress( progress_callback, progress_percent, f"Sinkron {table_name}", f"{updated} baris diperbarui", ) self._sync_ppn_server_setting_safely() self._emit_progress(progress_callback, 100, "Sinkronisasi selesai", f"Total {updated_total} baris diperbarui") self.circuit_breaker.mark_success() return updated_total except (RuntimeError, ValueError, TypeError) as exc: self.circuit_breaker.mark_failure(exc) LOGGER.exception("Sinkronisasi gagal: %s", str(exc)) raise finally: if model: try: model.close_connections() except (RuntimeError, AttributeError) as close_exc: LOGGER.warning("Gagal menutup koneksi sinkron model: %s", close_exc)