import json from urllib.parse import urljoin import requests from pypos.core.base_service import BaseService from pypos.core.utils.config_utils import normalize_base_url from pypos.core.utils.http_retry import _execute_request from pypos.modules.auth.config.auth_config import ( get_auth_endpoint_config, get_login_network_policy_config, ) # edited by glg class NetworkProbeService(BaseService): RAW_SERVER_OK = "SERVER_OK" RAW_INTERNET_ONLY = "INTERNET_ONLY" RAW_OFFLINE = "OFFLINE" def __init__( self, http_client=None, endpoint_config_getter=get_auth_endpoint_config, policy_config_getter=get_login_network_policy_config, ): super().__init__(http_client=http_client or requests) self.endpoint_config_getter = endpoint_config_getter self.policy_config_getter = policy_config_getter def _request_once(self, method, url, timeout): if not self.http: raise RuntimeError("HTTP client belum tersedia untuk network probe.") # edited by glg # Bungkus call HTTP agar tetap aman jika dipanggil dari UI thread. response = _execute_request( self.http.request, { "method": str(method or "GET").upper(), "url": str(url or "").strip(), "timeout": float(timeout), "allow_redirects": True, "stream": True, }, ) status_code = int(getattr(response, "status_code", 0) or 0) try: response.close() except (AttributeError, RuntimeError, OSError): pass return status_code, response def _probe_url(self, url, timeout): clean_url = str(url or "").strip() if not clean_url: return { "ok": False, "status_code": 0, "method": "", "url": clean_url, "error": "empty_url", } for method in ("HEAD", "GET"): try: status_code, _ = self._request_once(method, clean_url, timeout=timeout) if status_code > 0 and status_code not in (405, 501): return { "ok": True, "status_code": status_code, "method": method, "url": clean_url, "error": "", } if status_code in (405, 501): continue return { "ok": status_code > 0, "status_code": status_code, "method": method, "url": clean_url, "error": "", } except requests.RequestException as exc: last_error = str(exc or "").strip() or exc.__class__.__name__ continue except (RuntimeError, TypeError, ValueError, OSError) as exc: last_error = str(exc or "").strip() or exc.__class__.__name__ continue return { "ok": False, "status_code": 0, "method": "", "url": clean_url, "error": str(locals().get("last_error") or "request_failed"), } def _build_health_url(self, base_url, health_endpoint): base = normalize_base_url(str(base_url or "").strip()) endpoint = str(health_endpoint or "").strip() if not base or not endpoint: return "" if endpoint.startswith(("http://", "https://")): return endpoint if not endpoint.startswith("/"): endpoint = "/" + endpoint return urljoin(base + "/", endpoint.lstrip("/")) def probe_server(self, timeout_sec=None, policy=None): cfg = self.endpoint_config_getter() or {} policy_cfg = policy if isinstance(policy, dict) else (self.policy_config_getter() or {}) base_url = normalize_base_url(str(cfg.get("api_base_url") or "").strip()) timeout = float(timeout_sec or policy_cfg.get("server_probe_timeout_sec") or cfg.get("request_timeout") or 1.0) result = { "ok": False, "base_url": base_url, "status_code": 0, "method": "", "error": "", "health_supported": False, "health_ok": False, "health_status_code": 0, "health_error": "", "capabilities": {}, "health_payload": {}, } if not base_url: result["error"] = "missing_api_base_url" return result health_enabled = bool(policy_cfg.get("health_check_enabled")) health_endpoint = str(policy_cfg.get("health_endpoint") or "").strip() health_timeout = float(policy_cfg.get("health_timeout_sec") or timeout) if health_enabled and health_endpoint: health_url = self._build_health_url(base_url, health_endpoint) if health_url: health_probe = self._probe_url(health_url, timeout=health_timeout) health_status_code = int(health_probe.get("status_code") or 0) result["health_status_code"] = health_status_code health_http_ok = bool(health_probe.get("ok")) and (0 < health_status_code < 400) if health_http_ok: result["health_supported"] = True result["health_ok"] = True result["ok"] = True result["status_code"] = health_status_code result["method"] = str(health_probe.get("method") or "") result["error"] = "" # edited by glg # Future-ready parser kontrak health endpoint; aman jika payload belum tersedia. try: status_code, response = self._request_once("GET", health_url, timeout=health_timeout) text = str(getattr(response, "text", "") or "") payload = json.loads(text) if text else {} if isinstance(payload, dict): data = payload.get("data") if isinstance(payload.get("data"), dict) else {} caps = data.get("capabilities") if isinstance(data.get("capabilities"), dict) else {} result["health_payload"] = payload result["capabilities"] = caps result["health_status_code"] = int(status_code or result["health_status_code"] or 0) except (requests.RequestException, ValueError, TypeError, RuntimeError, OSError): pass return result result["health_error"] = str(health_probe.get("error") or "") # 404/405 berarti server reachable, tapi kontrak health belum tersedia. if health_status_code in (401, 403, 404, 405, 501): result["health_supported"] = False elif health_status_code > 0: result["health_supported"] = True base_probe = self._probe_url(base_url, timeout=timeout) result["ok"] = bool(base_probe.get("ok")) result["status_code"] = int(base_probe.get("status_code") or 0) result["method"] = str(base_probe.get("method") or "") result["error"] = str(base_probe.get("error") or "") return result def probe_internet(self, timeout_sec=None, policy=None): policy_cfg = policy if isinstance(policy, dict) else (self.policy_config_getter() or {}) timeout = float(timeout_sec or policy_cfg.get("internet_probe_timeout_sec") or 0.8) probe_urls = policy_cfg.get("internet_probe_urls") if isinstance(policy_cfg.get("internet_probe_urls"), list) else [] result = { "ok": False, "checked_url": "", "status_code": 0, "method": "", "error": "", } for raw_url in probe_urls: probe = self._probe_url(raw_url, timeout=timeout) if probe.get("ok"): result["ok"] = True result["checked_url"] = str(probe.get("url") or "") result["status_code"] = int(probe.get("status_code") or 0) result["method"] = str(probe.get("method") or "") result["error"] = "" return result result["checked_url"] = str(probe.get("url") or result["checked_url"] or "") result["error"] = str(probe.get("error") or result["error"] or "") return result def probe_network_snapshot(self, policy=None): policy_cfg = policy if isinstance(policy, dict) else (self.policy_config_getter() or {}) server_result = self.probe_server(policy=policy_cfg) internet_result = { "ok": False, "checked_url": "", "status_code": 0, "method": "", "error": "", } raw_state = self.RAW_OFFLINE if bool(server_result.get("ok")): raw_state = self.RAW_SERVER_OK else: internet_result = self.probe_internet(policy=policy_cfg) raw_state = self.RAW_INTERNET_ONLY if bool(internet_result.get("ok")) else self.RAW_OFFLINE return { "raw_state": raw_state, "server": server_result, "internet": internet_result, "server_online": bool(server_result.get("ok")), "internet_online": bool(internet_result.get("ok")), }