Refactor credit parse to use remote HTML paths

This commit is contained in:
wkc
2026-05-13 14:20:42 +08:00
parent b822cc202e
commit be443d1b31
18 changed files with 473 additions and 171 deletions

View File

@@ -9,19 +9,29 @@ 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):
def validate_request(
self,
serial_num: Optional[str],
org_code: Optional[str],
run_type: Optional[str],
remote_path: Optional[str],
model: Optional[str],
):
if not serial_num:
return self.build_missing_param_response("serialNum")
if not org_code:
return self.build_missing_param_response("orgCode")
if not run_type:
return self.build_missing_param_response("runType")
if not remote_path:
return self.build_missing_param_response("remotePath")
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:
@@ -29,22 +39,21 @@ class CreditDebugService:
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
return self.wrap_mapping_response(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)
mapping_output_fields = response["data"]["mappingOutputFields"]
mapping_output_fields["message"] = mapping_output_fields["message"].replace("XX", param_name)
return response
def build_error_response(self, error_code: str) -> dict:
return copy.deepcopy(self.templates["errors"][error_code])
return self.wrap_mapping_response(copy.deepcopy(self.templates["errors"][error_code]))
def detect_error_marker(self, model: str) -> Optional[str]:
matched = re.search(r"error_(ERR_\d+)", model)
@@ -55,6 +64,15 @@ class CreditDebugService:
return error_code
return None
def wrap_mapping_response(self, mapping_output_fields: dict) -> dict:
return {
"success": True,
"code": 1000,
"data": {
"mappingOutputFields": mapping_output_fields,
},
}
def _load_templates(self) -> dict:
template_file = Path(self.template_path)
if not template_file.is_absolute():