~martijnbraam/shipments

5c2b206515333caea4888ee3b74a5a950e5f019e — Martijn Braam 4 months ago 9f22cca master
shipments: carriers: add Austrain Post
1 files changed, 121 insertions(+), 3 deletions(-)

M shipments/carrier.py
M shipments/carrier.py => shipments/carrier.py +121 -3
@@ 241,7 241,6 @@ class UPS(Carrier):
        if details.get('errorCode') and details['errorCode'] == '504':
            result.status = details['errorText']
            return result
            

        result.tracking_url = 'https://www.ups.com/track?loc=en_EN&requester=DAN&tracknum=' + code
        result.status = details['packageStatus'] or ''


@@ 251,7 250,6 @@ class UPS(Carrier):
        elif details['progressBarType'] == 'Delivered':
            result.status_category = StatusCategory.DELIVERED


        # Avoid TypeError: 'NoneType' object is not iterable
        if details['shipmentProgressActivities']:
            for event in details['shipmentProgressActivities']:


@@ 269,7 267,6 @@ class UPS(Carrier):
                e = PackageEvent(pdt, event['location'], label)
                result.events.append(e)


        if details['shipToAddress'] is not None:
            a = details['shipToAddress']
            sa = Address()


@@ 614,3 611,124 @@ class Track4PX(Carrier):
        result.destination.country = payload['ctEndName']

        return result


class AustrianPost(Carrier):
    DISPLAYNAME = "Austrian Post"

    @classmethod
    def identify(cls, code):
        # Extrapolated from a single tracking codes
        if len(code) != 13:
            return False
        if not code.startswith('CA'):
            return False
        if not code.endswith('AT'):
            return False
        return True

    def get_info(self, code, extra):
        url = 'https://api.post.at/sendungen/sv/graphqlPublic'

        query = """
        query { 
          einzelsendung(sendungsnummer: "%CODE%") { 
            sendungsnummer
            branchkey
            estimatedDelivery {
              startDate
              endDate
              startTime
              endTime
            }
            dimensions {
              height
              width
              length
            }
            status
            weight
            sendungsEvents {
              timestamp
              status
              reasontypecode
              text
              textEn
              eventpostalcode
              eventcountry
            }
            customsInformation {
              customsDocumentAvailable,
              userDocumentNeeded
            }
          }
        }
        """

        data = {
            'query': query.replace('%CODE%', code).strip(),
        }
        response = requests.post(url, json=data, headers={
            'User-Agent': 'Mozilla/5.0 Firefox/102.0',
            'Accept': '*',
            'Accept-Language': 'en-US,en;q=0.5',
            'Referer': 'https://www.post.at/',
            'Origin': 'https://www.post.at',
            'DNT': '1',
            'Connection': 'keep-alive',
            'Sec-Fetch-Dest': 'empty',
            'Sec-Fetch-Mode': 'cors',
            'Sec-Fetch-Site': 'same-site',
            'Pragma': 'no-cache',
            'Cache-Control': 'no-cache',
        })
        payload = response.json()
        result = PackageInfo(code, self.DISPLAYNAME)
        result.carrier_name = self.DISPLAYNAME
        result.status_category = StatusCategory.LABEL_CREATED
        result.tracking_url = 'https://www.post.at/en/sv/item-details?snr=' + code

        if payload['data']['einzelsendung'] is None:
            result.status = 'unknown'
            return result

        statuscodes = {
            'EB': "Your item is ready for pickup",
            'AV': "The sender has transmitted additional information",
            'AN': "Accepted",
            'IV': "En route",
            'NichtAngetroffen': 'A delivery attempt was made. The recipient was not home.',
            'RE': "Your item has been returned",
            'ZU': "Delivered",
            'ZH': "Delivery was stopped",
            'IZ': "Delivered"
        }

        payload = payload['data']['einzelsendung']

        if payload['status'] in statuscodes:
            result.status = statuscodes[payload['status']]
        else:
            result.status = f'Unknown {payload["status"]}'

        if payload['status'] == 'IV':
            result.status_category = StatusCategory.IN_TRANSIT
        elif payload['status'] in ['ZU', 'RE']:
            result.status_category = StatusCategory.DELIVERED
        elif payload['status'] in ['ZH']:
            result.status_category = StatusCategory.ERROR

        if 'weight' in payload and payload['weight']:
            result.weight = payload['weight'] * 1000

        for event in payload['sendungsEvents']:
            timestamp, zone = event['timestamp'].split('+', maxsplit=1)
            timestamp += '+' + zone.replace(':', '')
            if '.' in timestamp:
                stamp = datetime.strptime(timestamp, '%Y-%m-%dT%H:%M:%S.%f%z')
            else:
                stamp = datetime.strptime(timestamp, '%Y-%m-%dT%H:%M:%S%z')
            e = PackageEvent(stamp, f'{event["eventpostalcode"]}, {event["eventcountry"]}', event['textEn'])
            result.events.append(e)

        return result