82 lines
2.5 KiB
Python
82 lines
2.5 KiB
Python
from typing import Optional
|
|
from urllib.parse import urlparse
|
|
from urllib.request import urlopen
|
|
|
|
from fastapi import APIRouter, Form
|
|
|
|
from services.credit_debug_service import CreditDebugService
|
|
from services.credit_html_identity_service import CreditHtmlIdentityService
|
|
from services.credit_payload_service import CreditPayloadService
|
|
|
|
router = APIRouter()
|
|
payload_service = CreditPayloadService("config/credit_feature_schema.json")
|
|
debug_service = CreditDebugService("config/credit_response_examples.json")
|
|
identity_service = CreditHtmlIdentityService()
|
|
result_cache = {}
|
|
|
|
|
|
@router.post("/api/service/interface/invokeService/xfeature")
|
|
async def html_eval(
|
|
serialNum: Optional[str] = Form(None),
|
|
orgCode: Optional[str] = Form(None),
|
|
runType: Optional[str] = Form(None),
|
|
remotePath: Optional[str] = Form(None),
|
|
model: Optional[str] = Form(None),
|
|
):
|
|
error_response = debug_service.validate_request(
|
|
serial_num=serialNum,
|
|
org_code=orgCode,
|
|
run_type=runType,
|
|
remote_path=remotePath,
|
|
model=model,
|
|
)
|
|
if error_response:
|
|
return error_response
|
|
|
|
html_content = fetch_remote_html(remotePath)
|
|
subject_identity = identity_service.extract_identity(html_content)
|
|
payload = payload_service.generate_payload(
|
|
model=model,
|
|
h_type="PERSON",
|
|
filename=remote_filename(remotePath),
|
|
subject_identity=subject_identity,
|
|
)
|
|
result_cache[serialNum] = payload
|
|
return debug_service.build_initiate_success_response(serialNum)
|
|
|
|
|
|
@router.post("/api/service/interface/invokeService/xfeatureResult")
|
|
async def html_eval_result(
|
|
serialNum: Optional[str] = Form(None),
|
|
orgCode: Optional[str] = Form(None),
|
|
runType: Optional[str] = Form(None),
|
|
):
|
|
error_response = debug_service.validate_result_request(
|
|
serial_num=serialNum,
|
|
org_code=orgCode,
|
|
run_type=runType,
|
|
)
|
|
if error_response:
|
|
return error_response
|
|
|
|
payload = result_cache.get(serialNum)
|
|
if payload is None:
|
|
return debug_service.build_result_not_found_response(serialNum)
|
|
return debug_service.build_success_response(payload)
|
|
|
|
|
|
@router.get("/credit/health")
|
|
async def credit_health():
|
|
return {"status": "healthy", "service": "credit-mock"}
|
|
|
|
|
|
def fetch_remote_html(remote_path: str) -> bytes:
|
|
with urlopen(remote_path, timeout=5) as response:
|
|
return response.read()
|
|
|
|
|
|
def remote_filename(remote_path: str) -> str:
|
|
path = urlparse(remote_path).path
|
|
filename = path.rsplit("/", 1)[-1]
|
|
return filename or "credit.html"
|