|
import logging |
|
import requests |
|
import sentry_sdk |
|
from contextlib import suppress |
|
from json import JSONDecodeError |
|
|
|
from django.conf import settings |
|
from rest_framework import status |
|
|
|
from cloud.surfaces.api import make_request, poll_job_result |
|
|
|
logger = logging.getLogger(__name__) |
|
|
|
def get_surfaces(address: Address) -> list[dict]: |
|
response = make_request( |
|
settings.SURFACES_SERVICE_URL, |
|
cadastral_unit_id=address.cadastral_unit_id, |
|
cadastral_address_id=address.cadastral_address_id, |
|
) |
|
|
|
if 'application/json' in response.headers.get('Content-Type'): |
|
data = response.json() |
|
response_status = data.get('ResponseStatus', {}) |
|
if response_status: |
|
logger.info('Found response status (test 1)') |
|
message = {} |
|
if 'Message' in data and data['Message']: |
|
with suppress(JSONDecodeError): |
|
message = json.loads(data['Message']) |
|
|
|
response_status = message.get('ResponseStatus', {}) |
|
if response_status: |
|
logger.info('Found response status (test 2)') |
|
|
|
if 'Cannot find any buildings for' in message: |
|
logger.info(f'No buildings found on address {address} (test 1)') |
|
return [] |
|
|
|
error_code = None |
|
if 'ErrorCode' in response_status: |
|
logger.info('Found ErrorCode in response_status') |
|
error_code = response_status.get('ErrorCode', '') |
|
elif 'Message' in data and 'ErrorCode' in data['Message']: |
|
error_code = message.get('ErrorCode', '') |
|
|
|
if error_code: |
|
if 'no-buildings' in error_code: |
|
logger.info(f'No buildings found on address {address} (test 2)') |
|
return [] |
|
|
|
logger.error( |
|
f'Unable to get surface data. Error code: {error_code}', |
|
exc_info=True, |
|
) |
|
return [] |
|
|
|
if response.status_code == status.HTTP_404_NOT_FOUND: |
|
return [] |
|
|
|
try: |
|
response.raise_for_status() |
|
except requests.exceptions.RequestException: |
|
logger.info('Unable to refresh surfaces') |
|
sentry_sdk.capture_exception() |
|
return [] |
|
|
|
data = response.json() |
|
|
|
if not data['Success']: |
|
logger.error( |
|
'Unable to refresh surfaces', |
|
exc_info=True, |
|
extra={'data': {'url': settings.SURFACES_SERVICE_URL, 'response': data}}, |
|
) |
|
return [] |
|
|
|
if not data['Completed']: |
|
response = poll_job_result(data['JobId']) |
|
else: |
|
url = f'{settings.SURFACE_JOB_DOWNLOAD_URL}?JobId={data['JobId']}' |
|
response = session.get( |
|
url, |
|
headers={ |
|
'Content-Type': 'application/json', |
|
'Accept': 'application/json', |
|
'X-WAAPI-TOKEN': settings.SURFACE_API_KEY, |
|
}, |
|
) |
|
|
|
response.raise_for_status() |
|
|
|
return response.json()['roofs'] |