import time from pypos.core.base_service import BaseService from pypos.modules.auth.config.auth_config import get_login_network_policy_config from pypos.modules.auth.services.network_probe_service import NetworkProbeService # edited by glg class LoginNetworkPolicyService(BaseService): STATE_DISABLED = "DISABLED" STATE_ONLINE_SERVER_STABLE = "ONLINE_SERVER_STABLE" STATE_ONLINE_SERVER_UNSTABLE = "ONLINE_SERVER_UNSTABLE" STATE_ONLINE_INTERNET_ONLY = "ONLINE_INTERNET_ONLY" STATE_OFFLINE = "OFFLINE" def __init__( self, probe_service=None, policy_config_getter=get_login_network_policy_config, now_fn=None, ): super().__init__() self.probe_service = probe_service or NetworkProbeService() self.policy_config_getter = policy_config_getter self.now_fn = now_fn or time.monotonic self._consecutive_fail = 0 self._consecutive_success = 0 self._last_good_ts = None def _now(self): return float(self.now_fn()) @staticmethod def _normalize_policy(raw): cfg = raw if isinstance(raw, dict) else {} def _to_bool(v, default=False): if isinstance(v, bool): return v if v is None: return bool(default) text = str(v).strip().lower() if text in {"1", "true", "yes", "on"}: return True if text in {"0", "false", "no", "off"}: return False return bool(default) def _to_int(v, default, minimum): try: parsed = int(v) except (TypeError, ValueError): parsed = int(default) return max(int(minimum), parsed) return { "enabled": _to_bool(cfg.get("enabled"), default=True), "require_online": _to_bool(cfg.get("require_online"), default=True), "show_startup_notice": _to_bool(cfg.get("show_startup_notice"), default=True), "disable_button_when_blocked": _to_bool( cfg.get("disable_button_when_blocked"), default=True ), "recheck_interval_ms": _to_int( cfg.get("recheck_interval_ms"), default=3000, minimum=500, ), "fail_threshold": _to_int(cfg.get("fail_threshold"), default=3, minimum=1), "recover_threshold": _to_int(cfg.get("recover_threshold"), default=2, minimum=1), "grace_seconds": _to_int(cfg.get("grace_seconds"), default=45, minimum=0), "allow_server_down_but_internet_up": _to_bool( cfg.get("allow_server_down_but_internet_up"), default=True, ), "allow_network_unstable": _to_bool( cfg.get("allow_network_unstable"), default=True, ), "offline_emergency_admin_enabled": _to_bool( cfg.get("offline_emergency_admin_enabled"), default=False, ), } def _build_result( self, state, reason, message, allow_login, policy, snapshot, within_grace, ): probe_snapshot = snapshot if isinstance(snapshot, dict) else {} server_online = bool(probe_snapshot.get("server_online")) internet_online = bool(probe_snapshot.get("internet_online")) return { "state": str(state or self.STATE_OFFLINE), "raw_state": str(probe_snapshot.get("raw_state") or ""), "reason": str(reason or ""), "message": str(message or ""), "allow_login": bool(allow_login), "block_login": not bool(allow_login), "server_online": server_online, "internet_online": internet_online, "within_grace": bool(within_grace), "fail_streak": int(self._consecutive_fail), "success_streak": int(self._consecutive_success), "policy": dict(policy or {}), "probe": probe_snapshot, "timestamp": self._now(), } def evaluate(self, snapshot=None, policy=None): policy_cfg = self._normalize_policy( policy if isinstance(policy, dict) else self.policy_config_getter() ) snapshot_data = ( snapshot if isinstance(snapshot, dict) else self.probe_service.probe_network_snapshot(policy=policy_cfg) ) if not bool(policy_cfg.get("enabled")): self._consecutive_fail = 0 self._consecutive_success = 0 self._last_good_ts = self._now() return self._build_result( state=self.STATE_DISABLED, reason="policy_disabled", message="Kebijakan jaringan login dinonaktifkan oleh konfigurasi.", allow_login=True, policy=policy_cfg, snapshot=snapshot_data, within_grace=True, ) raw_state = str(snapshot_data.get("raw_state") or "") now_ts = self._now() fail_threshold = int(policy_cfg.get("fail_threshold") or 3) recover_threshold = int(policy_cfg.get("recover_threshold") or 2) grace_seconds = int(policy_cfg.get("grace_seconds") or 0) if raw_state == NetworkProbeService.RAW_SERVER_OK: self._consecutive_success += 1 self._consecutive_fail = 0 self._last_good_ts = now_ts if self._consecutive_success >= recover_threshold: return self._build_result( state=self.STATE_ONLINE_SERVER_STABLE, reason="server_ok_stable", message="Koneksi server stabil. Login siap digunakan.", allow_login=True, policy=policy_cfg, snapshot=snapshot_data, within_grace=True, ) return self._build_result( state=self.STATE_ONLINE_SERVER_UNSTABLE, reason="server_recovering", message="Koneksi mulai pulih. Anda dapat melanjutkan login.", allow_login=True, policy=policy_cfg, snapshot=snapshot_data, within_grace=True, ) self._consecutive_success = 0 self._consecutive_fail += 1 within_grace = False if grace_seconds > 0 and self._last_good_ts is not None: within_grace = (now_ts - float(self._last_good_ts)) <= float(grace_seconds) if raw_state == NetworkProbeService.RAW_INTERNET_ONLY: if bool(policy_cfg.get("allow_server_down_but_internet_up")): if self._consecutive_fail >= fail_threshold: return self._build_result( state=self.STATE_ONLINE_INTERNET_ONLY, reason="server_down_internet_ok", message=( "Server POS belum merespons, tetapi internet tersedia. " "Login lokal tetap diizinkan." ), allow_login=True, policy=policy_cfg, snapshot=snapshot_data, within_grace=within_grace, ) allow_unstable = bool(policy_cfg.get("allow_network_unstable")) return self._build_result( state=self.STATE_ONLINE_SERVER_UNSTABLE, reason="server_unstable_internet_ok", message=( "Koneksi ke server belum stabil. " "Silakan lanjut login atau tunggu koneksi stabil." ), allow_login=allow_unstable, policy=policy_cfg, snapshot=snapshot_data, within_grace=within_grace, ) allow_grace = bool(policy_cfg.get("allow_network_unstable")) and within_grace return self._build_result( state=self.STATE_ONLINE_INTERNET_ONLY, reason="server_required_but_down", message=( "Internet tersedia, tetapi server POS tidak dapat dijangkau. " "Silakan coba lagi." ), allow_login=allow_grace, policy=policy_cfg, snapshot=snapshot_data, within_grace=within_grace, ) require_online = bool(policy_cfg.get("require_online")) if not require_online: return self._build_result( state=self.STATE_OFFLINE, reason="offline_allowed_by_config", message="Perangkat sedang offline. Login diizinkan oleh konfigurasi.", allow_login=True, policy=policy_cfg, snapshot=snapshot_data, within_grace=within_grace, ) if ( bool(policy_cfg.get("allow_network_unstable")) and within_grace and self._consecutive_fail < fail_threshold ): return self._build_result( state=self.STATE_ONLINE_SERVER_UNSTABLE, reason="offline_grace_window", message="Koneksi sedang tidak stabil. Coba login kembali.", allow_login=True, policy=policy_cfg, snapshot=snapshot_data, within_grace=True, ) return self._build_result( state=self.STATE_OFFLINE, reason="offline_blocked", message="Perangkat offline. Aplikasi harus online untuk login.", allow_login=False, policy=policy_cfg, snapshot=snapshot_data, within_grace=within_grace, )