# edited by glg import hashlib import json import logging import re from typing import Any, Dict, List 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, sanitize_response_text from pypos.modules.sinkronisasi.config import ( get_settlement_direct_retry_backoff_sec, get_settlement_direct_retry_max_attempt, get_settlement_direct_timeout_sec, get_sync_circuit_retryable_statuses, is_settlement_direct_auth_required, is_settlement_direct_enabled, ) from pypos.modules.sinkronisasi.services.transaction_export_service import TransactionExportService LOGGER = logging.getLogger(__name__) _SUCCESS_REASON_CODES = {"success", "duplicate", "debug_no_commit"} _UNAUTHORIZED_REASON_CODES = {"unauthorized", "unauthorized_user_context"} _INVALID_REASON_CODES = {"invalid_payload"} _CONFLICT_REASON_CODES = { "idempotency_in_progress", "idempotency_key_conflict", "idempotency_replay_rejected", } _SERVER_REASON_CODES = {"server_error", "db_transaction_failed", "idempotency_error"} class SettlementDirectService(BaseService): def __init__( self, http_client=None, export_service=None, endpoint_config_reader=read_endpoint_config, enabled_getter=is_settlement_direct_enabled, timeout_getter=get_settlement_direct_timeout_sec, retry_max_getter=get_settlement_direct_retry_max_attempt, retry_backoff_getter=get_settlement_direct_retry_backoff_sec, auth_required_getter=is_settlement_direct_auth_required, retryable_statuses_getter=get_sync_circuit_retryable_statuses, ): super().__init__(http_client=http_client or requests) self.export_service = export_service or TransactionExportService() self.endpoint_config_reader = endpoint_config_reader self.enabled_getter = enabled_getter self.timeout_getter = timeout_getter self.retry_max_getter = retry_max_getter self.retry_backoff_getter = retry_backoff_getter self.auth_required_getter = auth_required_getter self.retryable_statuses_getter = retryable_statuses_getter def _to_int(self, value, default=0): try: return int(value) except (TypeError, ValueError, OverflowError): return int(default) def _normalize_transaksi_ids(self, raw_ids) -> List[int]: values = raw_ids if isinstance(raw_ids, (list, tuple, set)) else [] normalized = [] seen = set() for item in values: try: parsed = int(item) except (TypeError, ValueError, OverflowError): continue if parsed <= 0 or parsed in seen: continue seen.add(parsed) normalized.append(parsed) return normalized def _resolve_url(self) -> str: cfg = self.endpoint_config_reader() if callable(self.endpoint_config_reader) else {} cfg = cfg if isinstance(cfg, dict) else {} endpoint = str(cfg.get("ep_settlement_direct") or "").strip() if not endpoint: raise ValueError("Endpoint ep_settlement_direct belum diatur") if endpoint.startswith(("http://", "https://")): return endpoint base_url = normalize_base_url(str(cfg.get("api_base_url") or "")) if not base_url: raise ValueError("api_base_url belum diatur") if not endpoint.startswith("/"): endpoint = "/" + endpoint return f"{base_url}{endpoint}" def _resolve_retry_attempt(self) -> int: try: return max(1, int(self.retry_max_getter() or 1)) except (TypeError, ValueError, OverflowError): return 1 def _resolve_timeout(self) -> int: try: return max(1, int(self.timeout_getter() or 8)) except (TypeError, ValueError, OverflowError): return 8 def _resolve_backoff(self) -> float: try: return max(0.0, float(self.retry_backoff_getter() or 0.0)) except (TypeError, ValueError, OverflowError): return 0.0 def _extract_status_code(self, error) -> int: try: response = getattr(error, "response", None) status_code = getattr(response, "status_code", None) return int(status_code or 0) except (TypeError, ValueError, OverflowError, AttributeError): return 0 def _is_retryable_status(self, status_code: int) -> bool: try: retryable_statuses = set(int(v) for v in (self.retryable_statuses_getter() or [])) except (TypeError, ValueError, OverflowError): retryable_statuses = {408, 429, 500, 502, 503, 504} return int(status_code or 0) in retryable_statuses def _build_idempotency_key( self, payload: Dict[str, Any], settlement_counter: str, transaksi_ids: List[int], ) -> str: data = payload if isinstance(payload, dict) else {} machine_id = str(data.get("device_id") or "").strip() dtime = str(data.get("dtime") or "").strip() counter = str(settlement_counter or "").strip() trx_part = ",".join([str(v) for v in sorted(self._normalize_transaksi_ids(transaksi_ids))]) seed = f"settlement-direct|{machine_id}|{counter}|{dtime}|{trx_part}" return hashlib.sha256(seed.encode("utf-8")).hexdigest() def _is_success_marker_payload(self, payload_json) -> bool: if not isinstance(payload_json, dict): return False success_value = payload_json.get("success") status_value = payload_json.get("status") if success_value in {True, 1, "1", "true", "ok", "success", 200, "200", 201, "201"}: return True if status_value in {True, 1, "1", "true", "ok", "success", 200, "200", 201, "201"}: return True if str(success_value).strip().lower() in {"true", "ok", "success", "200", "201"}: return True if str(status_value).strip().lower() in {"true", "ok", "success", "200", "201"}: return True return False # edited by glg # Beberapa endpoint legacy mengirimkan indikator sukses via text/message # meski field reason/status tidak konsisten. def _is_success_text_marker(self, raw_text) -> bool: text = str(raw_text or "").strip().lower() if not text: return False negative_hints = ("tidak", "gagal", "failed", "error", "invalid", "reject", "rejected", "ditolak") if any(hint in text for hint in negative_hints): return False positive_hints = ("success", "sukses", "berhasil", "duplicate", "ok", "stored", "accepted") return any(hint in text for hint in positive_hints) # edited by glg # Salvage JSON ketika body response tercampur warning/notice sebelum/after JSON. def _try_salvage_json_dict(self, response_text): text = str(response_text or "").strip() if not text: return None if text.startswith("{") and text.endswith("}"): try: parsed = json.loads(text) if isinstance(parsed, dict): return parsed except (ValueError, TypeError, json.JSONDecodeError): pass match = re.search(r"\{[\s\S]*\}", text) if not match: return None candidate = str(match.group(0) or "").strip() if not candidate: return None try: parsed = json.loads(candidate) if isinstance(parsed, dict): return parsed except (ValueError, TypeError, json.JSONDecodeError): return None return None # edited by glg # Kontrak reason-code settlement direct (server-side) diprioritaskan. # Fallback tetap kompatibel dengan endpoint lama yang hanya mengirim success/status. def _normalize_reason_code(self, payload_json) -> str: if not isinstance(payload_json, dict): return "" raw_reason = str(payload_json.get("reason") or "").strip().lower() if raw_reason in {"-", "null", "none"}: raw_reason = "" if raw_reason: return raw_reason raw_error = str(payload_json.get("error") or "").strip().lower() if raw_error in ( _SUCCESS_REASON_CODES | _UNAUTHORIZED_REASON_CODES | _INVALID_REASON_CODES | _CONFLICT_REASON_CODES | _SERVER_REASON_CODES ): return raw_error raw_status = payload_json.get("status") if isinstance(raw_status, str): normalized = raw_status.strip().lower() if normalized: return normalized if self._is_success_marker_payload(payload_json): return "success" return "" def _extract_message(self, payload_json, fallback="") -> str: if not isinstance(payload_json, dict): return str(fallback or "").strip() msg = str( payload_json.get("message") or payload_json.get("reason") or payload_json.get("error") or fallback or "" ).strip() return msg def _classify_response(self, response, payload_json, response_text="") -> Dict[str, Any]: status_code = self._to_int(getattr(response, "status_code", 0), 0) reason_code = self._normalize_reason_code(payload_json) retryable = False # edited by glg # HTTP policy kontrak settlement: # 200 -> success/duplicate/debug_no_commit # 400 -> invalid_payload # 401 -> unauthorized/unauthorized_user_context # 409 -> idempotency_* conflict/in_progress/replay_rejected # 500 -> server_error/db_transaction_failed/idempotency_error if status_code == 200: if reason_code in _SUCCESS_REASON_CODES: return {"ok": True, "reason_code": reason_code, "retryable": False} if reason_code in (_UNAUTHORIZED_REASON_CODES | _INVALID_REASON_CODES | _CONFLICT_REASON_CODES | _SERVER_REASON_CODES): retryable = reason_code in {"idempotency_in_progress"} | _SERVER_REASON_CODES return {"ok": False, "reason_code": reason_code, "retryable": retryable} if not isinstance(payload_json, dict): return { "ok": bool(self._is_success_text_marker(response_text)), "reason_code": "", "retryable": False, } if not payload_json and self._is_success_text_marker(response_text): return {"ok": True, "reason_code": "", "retryable": False} message_hint = self._extract_message(payload_json, fallback="") ok_by_text = self._is_success_text_marker(message_hint) return { "ok": bool(self._is_success_marker_payload(payload_json) or ok_by_text), "reason_code": reason_code, "retryable": False, } if status_code == 400: return { "ok": False, "reason_code": reason_code or "invalid_payload", "retryable": False, } if status_code == 401: return { "ok": False, "reason_code": reason_code or "unauthorized", "retryable": False, } if status_code == 409: normalized_reason = reason_code or "idempotency_key_conflict" return { "ok": False, "reason_code": normalized_reason, "retryable": normalized_reason == "idempotency_in_progress", } if status_code >= 500: normalized_reason = reason_code or "server_error" return { "ok": False, "reason_code": normalized_reason, "retryable": True, } if status_code in {201, 202}: if isinstance(payload_json, dict): ok = self._is_success_marker_payload(payload_json) else: ok = True return { "ok": bool(ok), "reason_code": reason_code, "retryable": False, } if status_code == 0: return { "ok": False, "reason_code": reason_code, "retryable": True, } if not isinstance(payload_json, dict): return {"ok": False, "reason_code": reason_code, "retryable": False} return { "ok": bool(self._is_success_marker_payload(payload_json)), "reason_code": reason_code, "retryable": False, } def _is_success_response(self, response, payload_json, response_text="") -> bool: classification = self._classify_response(response, payload_json, response_text=response_text) return bool(classification.get("ok")) def _extract_response_tuple(self, response): status_code = self._to_int(getattr(response, "status_code", 0), 0) response_text = sanitize_response_text(getattr(response, "text", "") or "")[:2000] payload_json = {} parse_error = "" try: payload_json = parse_json_response(response, label="send_settlement_direct") except (ValueError, TypeError, RuntimeError, AttributeError) as exc: payload_json = {} parse_error = str(exc or "").strip() salvaged_payload = self._try_salvage_json_dict(response_text) if isinstance(salvaged_payload, dict): payload_json = salvaged_payload parse_error = "" if not isinstance(payload_json, dict): payload_json = {} return status_code, response_text, payload_json, parse_error def _build_payload(self, settlement_result: Dict[str, Any]) -> Dict[str, Any]: result = settlement_result if isinstance(settlement_result, dict) else {} transaksi_ids = self._normalize_transaksi_ids(result.get("transaksi_ids")) counter = str(result.get("counter") or "").strip() payload = self.export_service.build_direct_settlement_payload( transaksi_ids=transaksi_ids, settlement_counter=counter, ) if isinstance(payload, dict): return payload return {} def send_settlement(self, settlement_result: Dict[str, Any]) -> Dict[str, Any]: if not bool(self.enabled_getter()): return { "attempted": False, "ok": False, "skipped": "disabled", "error": "", "status_code": 0, "response_text": "", "idempotency_key": "", } try: url = self._resolve_url() except (TypeError, ValueError, RuntimeError, AttributeError) as exc: return { "attempted": False, "ok": False, "skipped": "invalid_endpoint", "error": str(exc), "status_code": 0, "response_text": "", "idempotency_key": "", } result = settlement_result if isinstance(settlement_result, dict) else {} transaksi_ids = self._normalize_transaksi_ids(result.get("transaksi_ids")) counter = str(result.get("counter") or "").strip() payload = self._build_payload(result) if not payload or not isinstance(payload.get("data"), dict) or not payload.get("data"): # edited by glg LOGGER.warning( "[SettlementDirect] skip payload_empty counter=%s transaksi_ids=%s", counter or "-", ",".join([str(item) for item in transaksi_ids]) or "-", ) return { "attempted": False, "ok": False, "skipped": "payload_empty", "error": "Payload settlement direct tidak tersedia.", "status_code": 0, "response_text": "", "idempotency_key": "", "context": { "counter": counter, "transaksi_ids": list(transaksi_ids), }, } idempotency_key = self._build_idempotency_key(payload, counter, transaksi_ids) timeout = self._resolve_timeout() total_attempt = self._resolve_retry_attempt() retries = max(0, total_attempt - 1) backoff = self._resolve_backoff() auth_required = bool(self.auth_required_getter()) headers = { "Content-Type": "application/json", "X-Idempotency-Key": idempotency_key, } try: response = self.request_with_retry( "POST", url, json=payload, headers=headers, timeout=timeout, retries=retries, backoff_seconds=backoff, retry_on=(requests.RequestException,), auth_required=auth_required, ) except requests.exceptions.RequestException as exc: status_code = self._extract_status_code(exc) response = getattr(exc, "response", None) response_text = "" response_json = {} reason_code = "" message = str(exc) retryable = self._is_retryable_status(status_code) or status_code == 0 if response is not None: status_code, response_text, response_json, _ = self._extract_response_tuple(response) classification = self._classify_response( response, response_json, response_text=response_text, ) reason_code = str(classification.get("reason_code") or "").strip() retryable = bool(classification.get("retryable")) or retryable message = self._extract_message(response_json, fallback=message) return { "attempted": True, "ok": False, "skipped": "", "error": message, "status_code": status_code, "reason_code": reason_code, "response_text": response_text, "response_json": response_json if isinstance(response_json, dict) else {}, "retryable": retryable, "idempotency_key": idempotency_key, } except (TypeError, ValueError, RuntimeError, KeyError, AttributeError) as exc: return { "attempted": True, "ok": False, "skipped": "", "error": str(exc), "status_code": 0, "reason_code": "", "response_text": "", "response_json": {}, "retryable": True, "idempotency_key": idempotency_key, } status_code, response_text, response_json, parse_error = self._extract_response_tuple(response) classification = self._classify_response( response, response_json, response_text=response_text, ) ok = bool(classification.get("ok")) reason_code = str(classification.get("reason_code") or "").strip() retryable = bool(classification.get("retryable")) message = self._extract_message(response_json, fallback="") if not ok and not message: message = "Direct settlement ditolak oleh endpoint." if ok and message: LOGGER.info( "[SettlementDirect] success response message=%s reason=%s status=%s", message, reason_code or "-", status_code, ) elif ok: LOGGER.info("[SettlementDirect] success status=%s reason=%s", status_code, reason_code or "-") else: response_excerpt = str(response_text or "").replace("\n", " ").strip()[:300] LOGGER.warning( "[SettlementDirect] rejected status=%s reason=%s retryable=%s message=%s parse_error=%s response=%s", status_code, reason_code or "-", int(retryable), message, parse_error or "-", response_excerpt or "-", ) return { "attempted": True, "ok": bool(ok), "skipped": "", "error": "" if ok else message, "status_code": status_code, "reason_code": reason_code, "response_text": response_text, "response_json": response_json if isinstance(response_json, dict) else {}, "retryable": retryable, "idempotency_key": idempotency_key, "payload_preview": json.dumps( { "user_id": payload.get("user_id"), "dtime": payload.get("dtime"), "cabang_id": payload.get("cabang_id"), "device_id": payload.get("device_id"), "sales_dates": sorted(list((payload.get("data") or {}).keys()))[:5], }, ensure_ascii=False, ), }