132 lines
4.6 KiB
Python
132 lines
4.6 KiB
Python
import copy
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
|
|
class CreditDebugService:
|
|
"""处理征信解析接口的调试标记、参数校验与响应封装。"""
|
|
|
|
VALID_MODEL = "LXCUSTALL"
|
|
|
|
def __init__(self, template_path: str):
|
|
self.template_path = template_path
|
|
self.templates = self._load_templates()
|
|
|
|
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")
|
|
|
|
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")
|
|
return None
|
|
|
|
def validate_result_request(
|
|
self,
|
|
serial_num: Optional[str],
|
|
org_code: Optional[str],
|
|
run_type: 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")
|
|
return None
|
|
|
|
def build_initiate_success_response(self, serial_num: str) -> dict:
|
|
message = f"文件写入成功，流水号为: {serial_num}"
|
|
return {
|
|
"success": True,
|
|
"code": 10000,
|
|
"data": {
|
|
"mappingOutputFields": {
|
|
"message": message,
|
|
},
|
|
"reasonMessage": "Running successfully",
|
|
"outputFields": {
|
|
"C_S_ZXJXMESSAGE": message,
|
|
},
|
|
"procCode": "999000",
|
|
"bizId": serial_num,
|
|
"reasonCode": 200,
|
|
"status": 1,
|
|
},
|
|
}
|
|
|
|
def build_success_response(self, payload: dict) -> dict:
|
|
response = copy.deepcopy(self.templates["success"])
|
|
response["payload"] = payload
|
|
return self.wrap_mapping_response(response, status=1, reason_code=200)
|
|
|
|
def build_result_not_found_response(self, serial_num: str) -> dict:
|
|
return self.wrap_mapping_response({
|
|
"message": f"征信解析结果未返回: {serial_num}",
|
|
"payload": None,
|
|
"status_code": "ERR_99999",
|
|
}, status=0, reason_code=500, reason_message=f"征信解析结果未返回: {serial_num}")
|
|
|
|
def build_missing_param_response(self, param_name: str) -> dict:
|
|
response = self.build_error_response("ERR_99999")
|
|
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 self.wrap_mapping_response(copy.deepcopy(self.templates["errors"][error_code]),
|
|
status=0, reason_code=500)
|
|
|
|
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 wrap_mapping_response(
|
|
self,
|
|
mapping_output_fields: dict,
|
|
status: int = 0,
|
|
reason_code: int = 500,
|
|
reason_message: Optional[str] = None,
|
|
) -> dict:
|
|
return {
|
|
"success": True,
|
|
"code": 10000,
|
|
"data": {
|
|
"mappingOutputFields": mapping_output_fields,
|
|
"reasonMessage": reason_message or mapping_output_fields.get("message"),
|
|
"reasonCode": reason_code,
|
|
"status": status,
|
|
},
|
|
}
|
|
|
|
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"))
|