Files
ccdi/lsfx-mock-server/services/credit_debug_service.py

63 lines
2.2 KiB
Python

import copy
import json
import re
from pathlib import Path
from typing import Optional
class CreditDebugService:
"""处理征信解析接口的调试标记、参数校验与响应封装。"""
VALID_MODEL = "LXCUSTALL"
VALID_HTYPES = {"PERSON", "ENTERPRISE"}
def __init__(self, template_path: str):
self.template_path = template_path
self.templates = self._load_templates()
def validate_request(self, model: Optional[str], h_type: Optional[str], file_present: bool):
if not model:
return self.build_missing_param_response("model")
if not file_present:
return self.build_missing_param_response("file")
if not h_type:
return self.build_missing_param_response("hType")
error_code = self.detect_error_marker(model)
if error_code:
return self.build_error_response(error_code)
if model != self.VALID_MODEL:
return self.build_error_response("ERR_10002")
if h_type not in self.VALID_HTYPES:
return self.build_error_response("ERR_10003")
return None
def build_success_response(self, payload: dict) -> dict:
response = copy.deepcopy(self.templates["success"])
response["payload"] = payload
return response
def build_missing_param_response(self, param_name: str) -> dict:
response = self.build_error_response("ERR_99999")
response["message"] = response["message"].replace("XX", param_name)
return response
def build_error_response(self, error_code: str) -> dict:
return copy.deepcopy(self.templates["errors"][error_code])
def detect_error_marker(self, model: str) -> Optional[str]:
matched = re.search(r"error_(ERR_\d+)", model)
if not matched:
return None
error_code = matched.group(1)
if error_code in self.templates["errors"]:
return error_code
return None
def _load_templates(self) -> dict:
template_file = Path(self.template_path)
if not template_file.is_absolute():
template_file = Path(__file__).resolve().parent.parent / template_file
return json.loads(template_file.read_text(encoding="utf-8"))