mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
277 lines
9.1 KiB
Python
277 lines
9.1 KiB
Python
import xml.etree.ElementTree as ET
|
|
from dataclasses import dataclass
|
|
from datetime import date
|
|
from typing import List, Optional, Tuple, Union
|
|
|
|
import requests
|
|
|
|
from domain.abri.models import (
|
|
AbriRequestRejected,
|
|
AmendJobRequest,
|
|
AmendJobResult,
|
|
AppointmentAmended,
|
|
GetTenantDataResult,
|
|
JobLogged,
|
|
LogJobRequest,
|
|
LogJobResult,
|
|
PlaceRef,
|
|
TenancyData,
|
|
Tenant,
|
|
)
|
|
from infrastructure.abri.config import AbriConfig
|
|
from infrastructure.abri.envelope import serialise_relay_request
|
|
from infrastructure.abri.errors import AbriResponseParseError, AbriTransportError
|
|
|
|
STD_JOB_CODE = "SCSEXT"
|
|
CLIENT_CODE = "HSG"
|
|
RESOURCE_GROUP = "Surveyors"
|
|
|
|
|
|
def _format_appointment_date(appointment_date: date) -> str:
|
|
return appointment_date.strftime("%d/%m/%Y")
|
|
|
|
|
|
def _phone_priority(entry: ET.Element) -> Tuple[int, int]:
|
|
"""Sort key: lowest numeric priority first; unusable priorities last."""
|
|
try:
|
|
return (0, int((entry.get("priority") or "").strip()))
|
|
except ValueError:
|
|
return (1, 0)
|
|
|
|
|
|
def _best_phone_number(entries: List[ET.Element]) -> Optional[ET.Element]:
|
|
ranked = sorted(entries, key=_phone_priority) # stable: ties keep document order
|
|
return ranked[0] if ranked else None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _SelectedPhoneNumbers:
|
|
mobile: Optional[str]
|
|
telephone: Optional[str]
|
|
secondary: Optional[str]
|
|
|
|
|
|
def _select_phone_numbers(tenant_element: ET.Element) -> _SelectedPhoneNumbers:
|
|
"""Best mobile, best landline, then the best of the leftovers of both kinds.
|
|
|
|
All three picks use the same rule: lowest numeric priority wins, entries
|
|
without a usable priority sort last, ties break by document order, blank
|
|
numbers are ignored.
|
|
"""
|
|
usable = [
|
|
entry
|
|
for entry in tenant_element
|
|
if entry.tag in ("Mobile", "Telephone") and (entry.text or "").strip()
|
|
]
|
|
|
|
best_mobile = _best_phone_number([e for e in usable if e.tag == "Mobile"])
|
|
best_telephone = _best_phone_number([e for e in usable if e.tag == "Telephone"])
|
|
runner_up = _best_phone_number(
|
|
[e for e in usable if e is not best_mobile and e is not best_telephone]
|
|
)
|
|
|
|
def _text(entry: Optional[ET.Element]) -> Optional[str]:
|
|
return (entry.text or "").strip() if entry is not None else None
|
|
|
|
return _SelectedPhoneNumbers(
|
|
mobile=_text(best_mobile),
|
|
telephone=_text(best_telephone),
|
|
secondary=_text(runner_up),
|
|
)
|
|
|
|
|
|
class AbriClient:
|
|
def __init__(self, config: AbriConfig) -> None:
|
|
self._config = config
|
|
self._session = requests.Session()
|
|
|
|
def log_job(self, request: LogJobRequest) -> LogJobResult:
|
|
outcome = self._exchange(
|
|
request_type="logjob",
|
|
parameters=[
|
|
("place_ref", request.place_ref),
|
|
("std_job_code", STD_JOB_CODE),
|
|
("client", CLIENT_CODE),
|
|
("short_description", request.short_description),
|
|
("long_description", request.long_description),
|
|
("client_ref", request.client_ref),
|
|
("appointment_date", _format_appointment_date(request.appointment_date)),
|
|
("appointment_time", request.appointment_time),
|
|
("resource", self._config.default_resource),
|
|
("resource_group", RESOURCE_GROUP),
|
|
],
|
|
)
|
|
|
|
if isinstance(outcome, AbriRequestRejected):
|
|
return outcome
|
|
|
|
return self._parse_job_logged(outcome)
|
|
|
|
def amend_job(self, request: AmendJobRequest) -> AmendJobResult:
|
|
parameters = [
|
|
("job_no", request.job_no),
|
|
("action", "amend"),
|
|
("appointment_date", _format_appointment_date(request.appointment_date)),
|
|
("appointment_time", request.appointment_time),
|
|
]
|
|
if request.resource is not None:
|
|
parameters.append(("resource", request.resource))
|
|
|
|
outcome = self._exchange(
|
|
request_type="amendoptiappt",
|
|
parameters=parameters,
|
|
)
|
|
|
|
if isinstance(outcome, AbriRequestRejected):
|
|
return outcome
|
|
|
|
return self._parse_appointment_amended(outcome)
|
|
|
|
def get_tenant_data(self, place_ref: PlaceRef) -> GetTenantDataResult:
|
|
outcome = self._exchange(
|
|
request_type="getSCSTenantData",
|
|
parameters=[("place_ref", place_ref)],
|
|
)
|
|
|
|
if isinstance(outcome, AbriRequestRejected):
|
|
return outcome
|
|
|
|
return self._parse_tenancy_data(outcome)
|
|
|
|
def _exchange(
|
|
self, request_type: str, parameters: List[Tuple[str, str]]
|
|
) -> Union[ET.Element, AbriRequestRejected]:
|
|
envelope = serialise_relay_request(
|
|
request_type=request_type,
|
|
parameters=parameters,
|
|
username=self._config.username,
|
|
password=self._config.password,
|
|
)
|
|
|
|
reply = self._parse_reply(self._post(envelope))
|
|
|
|
if self._is_failure_document(reply):
|
|
return self._parse_rejection(reply)
|
|
|
|
return reply
|
|
|
|
def _post(self, envelope: bytes) -> bytes:
|
|
try:
|
|
response = self._session.post(self._config.endpoint_url, data=envelope)
|
|
response.raise_for_status()
|
|
except requests.RequestException as error:
|
|
raise AbriTransportError(str(error)) from error
|
|
|
|
return response.content
|
|
|
|
@staticmethod
|
|
def _parse_reply(body: bytes) -> ET.Element:
|
|
try:
|
|
return ET.fromstring(body)
|
|
except ET.ParseError as error:
|
|
raise AbriResponseParseError(str(error)) from error
|
|
|
|
@staticmethod
|
|
def _is_failure_document(reply: ET.Element) -> bool:
|
|
return reply.tag == "response"
|
|
|
|
@staticmethod
|
|
def _parse_job_logged(root: ET.Element) -> JobLogged:
|
|
job_logged = root.find("Jobs/Job_logged")
|
|
|
|
if job_logged is None:
|
|
raise AbriResponseParseError(
|
|
"Job_logged element missing from relay response"
|
|
)
|
|
|
|
job_no = job_logged.get("job_no")
|
|
logged_info = job_logged.get("logged_info")
|
|
|
|
if job_no is None or logged_info is None:
|
|
raise AbriResponseParseError(
|
|
"Job_logged element missing job_no or logged_info"
|
|
)
|
|
|
|
return JobLogged(job_no=job_no, logged_info=logged_info)
|
|
|
|
@staticmethod
|
|
def _parse_appointment_amended(root: ET.Element) -> AppointmentAmended:
|
|
amended = root.find("AmendOptiAppt")
|
|
|
|
if amended is None:
|
|
raise AbriResponseParseError(
|
|
"AmendOptiAppt element missing from relay response"
|
|
)
|
|
|
|
if amended.get("success") != "Y":
|
|
raise AbriResponseParseError(
|
|
"AmendOptiAppt element did not confirm success"
|
|
)
|
|
|
|
job_no = amended.get("job_no")
|
|
appointment_date = amended.get("appointment_date")
|
|
appointment_time = amended.get("appointment_time")
|
|
|
|
if job_no is None or appointment_date is None or appointment_time is None:
|
|
raise AbriResponseParseError(
|
|
"AmendOptiAppt element missing job_no, appointment_date "
|
|
"or appointment_time"
|
|
)
|
|
|
|
return AppointmentAmended(
|
|
job_no=job_no,
|
|
appointment_date=appointment_date,
|
|
appointment_time=appointment_time,
|
|
)
|
|
|
|
@staticmethod
|
|
def _parse_tenancy_data(root: ET.Element) -> TenancyData:
|
|
tenancies = root.findall("Tenancy")
|
|
|
|
if len(tenancies) != 1:
|
|
raise AbriResponseParseError(
|
|
f"expected exactly one Tenancy element, found {len(tenancies)}"
|
|
)
|
|
|
|
tenancy = tenancies[0]
|
|
tenancy_reference = (tenancy.text or "").strip()
|
|
|
|
if not tenancy_reference:
|
|
raise AbriResponseParseError("Tenancy element missing its reference")
|
|
|
|
return TenancyData(
|
|
tenancy_reference=tenancy_reference,
|
|
tenants=tuple(
|
|
AbriClient._parse_tenant(tenant)
|
|
for tenant in tenancy.findall("Tenant")
|
|
),
|
|
)
|
|
|
|
@staticmethod
|
|
def _parse_tenant(element: ET.Element) -> Tenant:
|
|
# Data minimisation: person_title, main-contact and the per-person
|
|
# reference (the element text) are deliberately not read.
|
|
numbers = _select_phone_numbers(element)
|
|
return Tenant(
|
|
forenames=(element.get("forenames") or "").strip(),
|
|
surname=(element.get("surname") or "").strip(),
|
|
mobile=numbers.mobile,
|
|
telephone=numbers.telephone,
|
|
secondary_number=numbers.secondary,
|
|
vulnerabilities=tuple(
|
|
text
|
|
for vulnerability in element.findall("Vulnerability")
|
|
if (text := (vulnerability.text or "").strip())
|
|
),
|
|
)
|
|
|
|
@staticmethod
|
|
def _parse_rejection(root: ET.Element) -> AbriRequestRejected:
|
|
success = root.findtext("success")
|
|
code = root.findtext("code")
|
|
message = root.findtext("message")
|
|
|
|
if success != "false" or code is None or message is None:
|
|
raise AbriResponseParseError("malformed relay failure response")
|
|
|
|
return AbriRequestRejected(code=code, message=message)
|