# edited by glg import threading from typing import Any, Dict import requests from pypos.core.utils.app_state_utils import get_bool, set_bool from pypos.core.utils.config_utils import CONFIG_FILE, read_endpoint_config from pypos.core.utils.db_helper import connect_sqlite from pypos.core.utils.path_utils import get_db_path from pypos.core.utils.ui_message_utils import sanitize_ui_message from pypos.core.utils.device_utils import get_active_device_info from pypos.modules.sinkronisasi.models.sinkron_model import SinkronModel from pypos.modules.sinkronisasi.services.sync_api_service import SyncApiService class EmployeeSeedUseCaseError(RuntimeError): pass class EmployeeSeedConfigurationError(EmployeeSeedUseCaseError): pass class EmployeeSeedContextError(EmployeeSeedUseCaseError): pass class EmployeeSeedTimeoutError(EmployeeSeedUseCaseError): pass class EmployeeSeedSyncError(EmployeeSeedUseCaseError): pass class EmployeeSeedUseCaseService: def __init__( self, *, logger, seed_pending_key: str, seed_done_key: str, default_wait_seconds: float = 25.0, ): self.logger = logger self.seed_pending_key = str(seed_pending_key or "per_employee_seed_pending").strip() self.seed_done_key = str(seed_done_key or "per_employee_seed_done").strip() self.default_wait_seconds = max(0.1, float(default_wait_seconds or 25.0)) @staticmethod def _as_int(value: Any, default: int = 0) -> int: try: return int(value) except (TypeError, ValueError): return int(default) def _set_seed_flags(self, *, pending: bool, done: bool) -> None: set_bool(self.seed_pending_key, bool(pending)) set_bool(self.seed_done_key, bool(done)) def mark_seed_pending(self) -> None: self._set_seed_flags(pending=True, done=False) def _count_local_per_employee_rows(self, db_path: str = None) -> int: conn = None try: conn = connect_sqlite(db_path or get_db_path()) cur = conn.cursor() cur.execute( "SELECT name FROM sqlite_master WHERE type='table' AND name='per_employee'" ) if not cur.fetchone(): return 0 cur.execute("SELECT COUNT(1) FROM per_employee") row = cur.fetchone() return int((row or [0])[0] or 0) except (RuntimeError, OSError, TypeError, ValueError) as exc: self.logger.warning("Gagal hitung data per_employee lokal: %s", exc) return 0 finally: try: if conn is not None: conn.close() except (RuntimeError, OSError, TypeError, ValueError): pass @staticmethod def _build_server_sync_url(endpoint_cfg: Dict[str, Any]) -> str: cfg = endpoint_cfg if isinstance(endpoint_cfg, dict) else {} base_url = str(cfg.get("api_base_url") or "").strip().rstrip("/") server_sync_ep = str(cfg.get("ep_server_sync") or "").strip() if server_sync_ep and not server_sync_ep.startswith(("http://", "https://")): if not server_sync_ep.startswith("/"): server_sync_ep = "/" + server_sync_ep return f"{base_url}{server_sync_ep}" if base_url else server_sync_ep return server_sync_ep def _sync_per_employee_once_impl(self, *, machine_id: str, cabang_id: int) -> int: model_sync = SinkronModel() sync_api = SyncApiService() try: response = sync_api.sync_data_server( model=model_sync, machine_id=machine_id, cabang_id=cabang_id, tables=["per_employee"], force_full_tables=["per_employee"], auth_required=False, ) except requests.exceptions.HTTPError as exc: response_obj = getattr(exc, "response", None) status_code = self._as_int(getattr(response_obj, "status_code", 0), default=0) if status_code != 401: raise EmployeeSeedSyncError(str(exc or "HTTP error sinkron seed.")) from exc endpoint_cfg = read_endpoint_config() or {} bootstrap_ep = str(endpoint_cfg.get("ep_seed_per_employee_bootstrap") or "").strip() if not bootstrap_ep: server_sync_url = self._build_server_sync_url(endpoint_cfg) raise EmployeeSeedConfigurationError( "Endpoint seed akun awal terblokir JWT (401). " "Set config `ep_seed_per_employee_bootstrap` ke endpoint bootstrap non-JWT " "untuk device yang sudah approved.\n" f"Runtime endpoint seed: {server_sync_url or '-'}\n" f"Sumber config runtime: {CONFIG_FILE}" ) from exc model_sync._log_sync( "sync_data_server.bootstrap_fallback", level="WARNING", context={ "machine_id": machine_id, "cabang_id": cabang_id, "endpoint": bootstrap_ep, }, ) try: response = sync_api.sync_data_server_bootstrap( model=model_sync, machine_id=machine_id, cabang_id=cabang_id, tables=["per_employee"], force_full_tables=["per_employee"], ) except requests.RequestException as bootstrap_exc: raise EmployeeSeedSyncError( str(bootstrap_exc or "Sinkronisasi bootstrap akun gagal.") ) from bootstrap_exc except requests.RequestException as exc: raise EmployeeSeedSyncError(str(exc or "Sinkronisasi akun karyawan gagal.")) from exc data_block = (response.get("data") or {}).get("per_employee") if isinstance(response, dict) else [] if isinstance(data_block, dict) and "new" in data_block: rows = data_block.get("new") or [] else: rows = data_block or [] if rows: model_sync.apply_sync_result("per_employee", rows, full_refresh=True) return int(len(rows or [])) def _sync_per_employee_once(self, *, machine_id: str, cabang_id: int) -> int: state = {"rows": 0, "error": None} done_event = threading.Event() def _worker(): try: state["rows"] = int( self._sync_per_employee_once_impl(machine_id=machine_id, cabang_id=cabang_id) or 0 ) except EmployeeSeedUseCaseError as typed_exc: state["error"] = typed_exc except (RuntimeError, OSError, TypeError, ValueError, requests.RequestException) as exc: state["error"] = EmployeeSeedSyncError( str(exc or "Sinkronisasi akun karyawan gagal.") ) finally: done_event.set() thread = threading.Thread( target=_worker, daemon=True, name="bootstrap-per-employee-sync", ) thread.start() finished = done_event.wait(timeout=float(self.default_wait_seconds)) if not finished: raise EmployeeSeedTimeoutError( "Sinkronisasi akun karyawan timeout. Cek koneksi jaringan/server lalu coba lagi." ) state_error = state.get("error") if state_error is not None: if isinstance(state_error, BaseException): raise state_error raise EmployeeSeedSyncError(str(state_error)) return int(state.get("rows") or 0) def ensure_seed(self, *, device_id: str, strict: bool = False) -> Dict[str, Any]: local_rows_before = self._count_local_per_employee_rows() seed_pending = get_bool(self.seed_pending_key, False) seed_done = get_bool(self.seed_done_key, False) if local_rows_before <= 0: try: self._set_seed_flags(pending=True, done=False) except (RuntimeError, OSError, TypeError, ValueError): pass seed_pending = True seed_done = False needs_seed = bool(seed_pending or (not seed_done and local_rows_before <= 0)) if not needs_seed: if local_rows_before > 0 and not seed_done: self._set_seed_flags(pending=False, done=True) return {"ok": True, "rows_synced": 0, "local_rows": local_rows_before, "seeded": False} device_info = get_active_device_info(device_id) or {} machine_id = str(device_info.get("machine_id") or device_id or "").strip() cabang_id = self._as_int(device_info.get("cabang_id"), default=0) if not machine_id or cabang_id <= 0: if local_rows_before > 0: return {"ok": True, "rows_synced": 0, "local_rows": local_rows_before, "seeded": False} raise EmployeeSeedContextError("Device/cabang belum valid untuk sinkronisasi akun karyawan.") rows_synced = self._sync_per_employee_once(machine_id=machine_id, cabang_id=cabang_id) local_rows_after = self._count_local_per_employee_rows() if local_rows_after <= 0 and bool(strict): raise EmployeeSeedSyncError( "Sinkronisasi akun karyawan belum berhasil. " "Login pertama belum bisa dilanjutkan sebelum data akun tersedia." ) if local_rows_after > 0: self._set_seed_flags(pending=False, done=True) return { "ok": True, "rows_synced": int(rows_synced or 0), "local_rows": int(local_rows_after or 0), "seeded": True, } @staticmethod def format_error_message(exc: Exception) -> str: raw_message = str(exc or "").strip() lower = raw_message.lower() if ( "ep_seed_per_employee_bootstrap" in lower or "endpoint seed akun awal terblokir jwt" in lower ): return ( "Data akun karyawan belum siap untuk login pertama.\n" "Endpoint ini mewajibkan JWT untuk sinkronisasi awal.\n" "Atur endpoint bootstrap non-JWT pada config `ep_seed_per_employee_bootstrap`." ) safe_message, _ = sanitize_ui_message("critical", raw_message) return ( "Data akun karyawan belum siap untuk login pertama.\n" f"{safe_message}" )