import json import uuid import requests from pypos.core.base_service import BaseService from pypos.core.utils.config_utils import normalize_base_url, read_endpoint_config from pypos.core.utils.http_json_utils import parse_json_response from pypos.modules.sinkronisasi.config.sync_config import ( get_sync_check_timeout, get_sync_data_timeout, ) SENTINEL_TS = "1990-01-01 23:59:59" SYNC_API_NON_FATAL_EXCEPTIONS = ( TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError, ) class SyncApiService(BaseService): def __init__(self, http_client=None): super().__init__(http_client=http_client or requests) # edited by glg @staticmethod def _extract_http_status(exc): try: response = getattr(exc, "response", None) status_code = getattr(response, "status_code", None) return int(status_code or 0) except (TypeError, ValueError, AttributeError): return 0 # edited by glg @staticmethod def _build_error_contract(*, error_code: str, reason: str, message: str, status: int = 0): return { "status": int(status or 0), "error_code": str(error_code or "SYNC_API_ERROR"), "reason": str(reason or "sync_api_error"), "trace_id": uuid.uuid4().hex[:12], "message": str(message or "").strip(), } # edited by glg def _raise_contract_error(self, *, error_code: str, reason: str, message: str, status: int = 0): contract = self._build_error_contract( error_code=error_code, reason=reason, message=message, status=status, ) raise RuntimeError(json.dumps(contract, ensure_ascii=False)) def _build_url(self, endpoint_key: str) -> str: cfg = read_endpoint_config() base_url = normalize_base_url(str(cfg.get("api_base_url") or "")) endpoint = str(cfg.get(endpoint_key) or "").strip() if not base_url: raise ValueError("api_base_url belum diatur") if not endpoint: raise ValueError(f"Endpoint {endpoint_key} belum diatur") if endpoint.startswith(("http://", "https://")): return endpoint if not endpoint.startswith("/"): endpoint = "/" + endpoint return f"{base_url}{endpoint}" def _prepare_payload(self, model, machine_id, cabang_id, tables, force_full_tables=None): force_tables = set(force_full_tables or []) last_info = model.get_last_sync_info() resolved_cabang_id = model._resolve_cabang_id(machine_id, cabang_id) payload = { "machine_id": machine_id, "cabang_id": resolved_cabang_id, } for table_name in list(tables or []): info = last_info.get(table_name, {"last_update": "2000-01-01 00:00:00", "last_id": 0}) if table_name in force_tables: payload[f"date_last[{table_name}][dtime]"] = SENTINEL_TS payload[f"date_last[{table_name}][lastID]"] = 0 continue payload[f"date_last[{table_name}][dtime]"] = info["last_update"] payload[f"date_last[{table_name}][lastID]"] = info["last_id"] return payload, resolved_cabang_id def check_update_server( self, model, machine_id, cabang_id, tables=None, force_full_tables=None, ): payload, resolved_cabang_id = self._prepare_payload( model=model, machine_id=machine_id, cabang_id=cabang_id, tables=tables or [], force_full_tables=force_full_tables or [], ) model._log_sync( "check_update_server.request", context={ "machine_id": machine_id, "cabang_id": resolved_cabang_id, "tables_len": len(list(tables or [])), "tables": ",".join(list(tables or [])[:20]), }, ) url = self._build_url("ep_update_cek") try: resp = self.request_with_retry( "POST", url, data=payload, timeout=get_sync_check_timeout(), retry_on=(requests.RequestException,), auth_required=True, ) except requests.exceptions.HTTPError as exc: response = getattr(exc, "response", None) status_code = getattr(response, "status_code", None) snippet = "" if response is not None: snippet = (response.text or "")[:300] model._log_sync( "check_update_server.http_error", level="ERROR", context={ "status": status_code or "", "machine_id": machine_id, "cabang_id": resolved_cabang_id, "snippet": snippet.replace("\n", " ")[:200], }, ) if status_code == 400: self._raise_contract_error( error_code="SYNC_CHECK_VALIDATION", reason="http_400", message=( "Validasi server gagal (400) saat cek update. " "Periksa Web Admin, registrasi device, dan cabang aktif." ), status=400, ) raise try: data = parse_json_response(resp, label="check_update_server") if isinstance(data, dict): keys = ",".join(list(data.keys())[:20]) else: keys = type(data).__name__ model._log_sync( "check_update_server.response", context={ "status": resp.status_code, "keys": keys, }, ) return data except (requests.RequestException, ValueError, TypeError, RuntimeError, AttributeError) as exc: content_type = resp.headers.get("Content-Type", "") snippet = (resp.text or "")[:300] model._log_sync( "check_update_server.response_error", level="ERROR", context={ "status": resp.status_code, "content_type": content_type, "snippet": snippet.replace("\n", " ")[:200], }, ) self._raise_contract_error( error_code="SYNC_CHECK_RESPONSE_NOT_JSON", reason="invalid_response_envelope", message=( "check_update_server: response bukan JSON " f"(status={resp.status_code}, content-type={content_type}, body={snippet})" ), status=int(resp.status_code or 0), ) def sync_data_server( self, model, machine_id, cabang_id, tables, force_full_tables=None, auth_required=True, endpoint_key="ep_server_sync", ): payload, resolved_cabang_id = self._prepare_payload( model=model, machine_id=machine_id, cabang_id=cabang_id, tables=tables or [], force_full_tables=force_full_tables or [], ) # edited by glg # Backend bootstrap non-JWT membaca flag auth_required dari request payload. # Tanpa field ini, endpoint baru tetap menegakkan JWT dan mengembalikan 401. payload["auth_required"] = 0 if not bool(auth_required) else 1 tables_list = list(tables or []) url = self._build_url(str(endpoint_key or "ep_server_sync")) # edited by glg # Debug runtime wajib jelas: endpoint aktual + marker seed per_employee. # Ini membantu bedakan bug payload klien vs endpoint server yang aktif. seed_marker_dtime = str(payload.get("date_last[per_employee][dtime]") or "") seed_marker_last_id = str(payload.get("date_last[per_employee][lastID]") or "") model._log_sync( "sync_data_server.request", context={ "machine_id": machine_id, "cabang_id": resolved_cabang_id, "tables_len": len(tables_list), "tables": ",".join(tables_list[:20]), "auth_required": 1 if bool(auth_required) else 0, "endpoint_key": str(endpoint_key or "ep_server_sync"), "url": url, "payload_has_auth_required": 1 if "auth_required" in payload else 0, "payload_auth_required": payload.get("auth_required"), "seed_dtime_per_employee": seed_marker_dtime, "seed_last_id_per_employee": seed_marker_last_id, }, ) try: resp = self.request_with_retry( "POST", url, data=payload, timeout=get_sync_data_timeout(), retry_on=(requests.RequestException,), auth_required=bool(auth_required), ) except requests.exceptions.HTTPError as exc: status_code = self._extract_http_status(exc) model._log_sync( "sync_data_server.http_error", level="ERROR", context={ "status": status_code or "", "machine_id": machine_id, "cabang_id": resolved_cabang_id, "auth_required": 1 if bool(auth_required) else 0, "endpoint_key": str(endpoint_key or "ep_server_sync"), "url": url, "payload_has_auth_required": 1 if "auth_required" in payload else 0, "payload_auth_required": payload.get("auth_required"), "seed_dtime_per_employee": seed_marker_dtime, "seed_last_id_per_employee": seed_marker_last_id, }, ) raise try: data = parse_json_response(resp, label="sync_data_server") if isinstance(data, dict): keys = ",".join(list(data.keys())[:20]) else: keys = type(data).__name__ model._log_sync( "sync_data_server.response", context={ "status": resp.status_code, "keys": keys, "auth_required": 1 if bool(auth_required) else 0, "endpoint_key": str(endpoint_key or "ep_server_sync"), "url": url, }, ) return data except (requests.RequestException, ValueError, TypeError, RuntimeError, AttributeError) as exc: content_type = resp.headers.get("Content-Type", "") snippet = (resp.text or "")[:300] model._log_sync( "sync_data_server.response_error", level="ERROR", context={ "status": resp.status_code, "content_type": content_type, "snippet": snippet.replace("\n", " ")[:200], "auth_required": 1 if bool(auth_required) else 0, "endpoint_key": str(endpoint_key or "ep_server_sync"), "url": url, }, ) self._raise_contract_error( error_code="SYNC_DATA_RESPONSE_NOT_JSON", reason="invalid_response_envelope", message=( "sync_data_server: response bukan JSON " f"(status={resp.status_code}, content-type={content_type}, body={snippet})" ), status=int(resp.status_code or 0), ) # edited by glg def sync_data_server_bootstrap(self, model, machine_id, cabang_id, tables, force_full_tables=None): return self.sync_data_server( model=model, machine_id=machine_id, cabang_id=cabang_id, tables=tables, force_full_tables=force_full_tables, auth_required=False, endpoint_key="ep_seed_per_employee_bootstrap", ) def sync_data_server_table( self, model, machine_id, cabang_id, table_name, dtime, last_id=0, partial=0, page_size=500, ): resolved_cabang_id = model._resolve_cabang_id(machine_id, cabang_id) payload = { "machine_id": machine_id, "cabang_id": resolved_cabang_id, f"date_last[{table_name}][dtime]": dtime, f"date_last[{table_name}][lastID]": last_id, f"date_last[{table_name}][partial]": partial, f"date_last[{table_name}][page_size]": page_size, } model._log_sync( "sync_data_server_table.request", context={ "machine_id": machine_id, "cabang_id": resolved_cabang_id, "table": table_name, "last_id": last_id, "partial": partial, "page_size": page_size, }, ) url = self._build_url("ep_server_sync") resp = self.request_with_retry( "POST", url, data=payload, timeout=get_sync_data_timeout(), retry_on=(requests.RequestException,), auth_required=True, ) try: data = parse_json_response(resp, label="sync_data_server_table") if isinstance(data, dict): keys = ",".join(list(data.keys())[:20]) else: keys = type(data).__name__ model._log_sync( "sync_data_server_table.response", context={ "status": resp.status_code, "table": table_name, "keys": keys, }, ) return data except (requests.RequestException, ValueError, TypeError, RuntimeError, AttributeError) as exc: content_type = resp.headers.get("Content-Type", "") snippet = (resp.text or "")[:300] model._log_sync( "sync_data_server_table.response_error", level="ERROR", context={ "status": resp.status_code, "table": table_name, "content_type": content_type, "snippet": snippet.replace("\n", " ")[:200], }, ) self._raise_contract_error( error_code="SYNC_TABLE_RESPONSE_NOT_JSON", reason="invalid_response_envelope", message=( "sync_data_server_table: response bukan JSON " f"(status={resp.status_code}, content-type={content_type}, body={snippet})" ), status=int(resp.status_code or 0), )