# -*- coding: utf-8 -*- import json import logging import requests import binascii from requests.auth import HTTPBasicAuth from odoo.exceptions import ValidationError from odoo import fields, models, api, _, exceptions from datetime import date, timedelta, datetime import holidays import pytz from timezonefinder import TimezoneFinder import phonenumbers from typing import List _logger = logging.getLogger('Gls Canada') class DeliveryCarrier(models.Model): _inherit = 'delivery.carrier' delivery_type = fields.Selection(selection_add=[("gls_canada", "GLS Canada")], ondelete={'gls_canada': 'set default'}) gls_canada_packaging_id = fields.Many2one('stock.package.type', string="Default Package Type") gls_canada_payment_type = fields.Selection([('Prepaid', 'Prepaid'), ('ThirdParty', 'ThirdParty'), ('Collect', 'Collect')], string='Gls canada payment type', help="Please select GLS Canada payment type") gls_canada_delivery_type = fields.Selection([('AIR', 'AIR'), ('GRD', 'GRD')], string='Gls canada delivery type', help="Please select GLS Canada delivery type") gls_canada_unit_measurement = fields.Selection([('K', 'kg (implies cm)'), ('L', 'lb (implies inch)'), ('KC', 'kg-cm'), ('KM', 'kg-meter'), ('LI', 'lb-inch'), ('LF', 'lb-feet'), ], string='Unit Of Measurement') parcel_type = fields.Selection([('Barrel', 'Barrel - freight'), ('Bundle', 'Bundle - freight'), ('Box', 'Box - freight'), ('Crate', 'Crate - freight'), ('FullLoad', 'FullLoad - freight'), ('Piece', 'Piece - freight'), ('Skid', 'Skid - freight'), ('Tube', 'Tube - freight'), ('Other', 'Other - freight'), ('Box', 'Box - Parcel Canada'), ('Envelope', 'Envelope - Parcel Canada'), ('Other', 'Other - Parcel Canada'), ('Mixed', 'Mixed - Parcel Canada')], string='Parcel Type') return_label = fields.Boolean("Required Return Label ?") category = fields.Selection([('Parcel', 'Parcel'), ('Freight', 'Freight'), ('Distribution', 'Distribution'), ('Logistics', 'Logistics'), ], string='Gls canada category type') appointment_type = fields.Selection([('Scheduled', 'Scheduled'), ('Required', 'Required')]) def gls_canada_rate_shipment(self, order): """ Estimate the cost of a shipment from a sales order. Called by rate_shipment in delivery/models/delivery_carrier.py. :param order: ResultSet of sale.order records, all for the same recipient and from the same warehouse :return: """ Estimate = self.env['gls.canada.shipping.estimate'] sender = order.warehouse_id.partner_id consignee = order.partner_shipping_id billing_account = order.delivery_billing_account weight = sum([ (line.product_id.weight * line.product_uom_qty) for line in order.order_line if not line.is_delivery ]) data = { "sender": { "postalCode": sender.zip or '', "provinceCode": sender.state_id and sender.state_id.code or '', "countryCode": sender.country_id and sender.country_id.code or '', }, "consignee": { "postalCode": consignee.zip or '', "provinceCode": consignee.state_id and consignee.state_id.code or '', "countryCode": consignee.country_id and consignee.country_id.code or '', }, "billing": self.company_id.gls_canada_billing_number or '', "paymentType": self.gls_canada_payment_type, "deliveryType": self.gls_canada_delivery_type, "unitOfMeasurement": self.gls_canada_unit_measurement, "parcels": [ { "id": "1", "parcelType": self.parcel_type, "quantity": "1", "weight": weight, "length": self.gls_canada_packaging_id.packaging_length or 0, "depth": self.gls_canada_packaging_id.height or 0, "width": self.gls_canada_packaging_id.width or 0, # "FCA_Class": "100.00", # "requestReturnLabel": False, } ], "category": self.category, } if self.category == "Freight": if self.appointment_type == 'Scheduled': data.update({"appointment": { "type": self.appointment_type, "date": order.commitment_date.strftime( "%Y-%m-%d") if self.appointment_type == "Scheduled" else '', "time": order.commitment_date.strftime( "%H:%M") if self.appointment_type == "Scheduled" else '', }}) # if self.appointment_type != 'Scheduled': else: data.update({"appointment": { "type": self.appointment_type, "phone": consignee.phone or '' }}) data = json.dumps(data) _logger.info("Rate Request Data %s" % data) username = self.company_id.gls_canada_username password = self.company_id.gls_canada_password headers = {"Content-Type": "application/json"} api_url = "%s/rate" % (self.company_id.gls_canada_api_url) response_data = requests.post(url=api_url, headers=headers, auth=HTTPBasicAuth(username, password), data=data) _logger.info("Rate response %s" % response_data.content) try: if response_data.status_code in [200, 201]: response_body = response_data.json() _logger.info(response_body) existing_records = Estimate.search( [('sale_order_id', '=', order and order.id)]) existing_records.sudo().unlink() if isinstance(response_body.get('rates'), dict): response_body = [response_body.get('rates')] for response_dict in response_body.get('rates'): Estimate.sudo().create( { 'account_type': response_dict.get('accountType'), 'rate_type': response_dict.get('rateType'), 'total_charge': response_dict.get('total'), 'sale_order_id': order and order.id } ) gls_canada_charge_id = Estimate.search( [('sale_order_id', '=', order and order.id)], order='total_charge', limit=1) order.gls_canada_shipping_charge_id = gls_canada_charge_id and gls_canada_charge_id.id return {'success': True, 'price': gls_canada_charge_id and gls_canada_charge_id.total_charge or 0.0, 'error_message': False, 'warning_message': False} else: return {'success': False, 'price': 0.0, 'error_message': "%s %s" % (response_data, response_data.text), 'warning_message': False} except Exception as e: raise ValidationError(e) def gls_canada_send_shipping(self, pickings): """ Send the package to the service provider :param pickings: A recordset of pickings :return list: A list of dictionaries (one per picking) containing of the form:: { 'exact_price': price, 'tracking_number': number } """ # See if there are available pickups shipments_dict = self._group_pickings_by_warehouse_and_recipient(pickings) pickups = self._gls_canada_get_pickups_for_senders(shipments_dict.keys()) picking_shipment_data_dict = {} for sender, recipient_dict in shipments_dict.items(): for recipient, picks in recipient_dict.items(): shipping_data, shipment_id = self._gls_canada_send_shipment(picks, pickups.get(sender)) picking_shipment_data_dict.update( {pick: shipping_data for pick in picks}) pickup_id = pickups.get(sender) if pickup_id: self._add_shipment_to_pickup(shipment_id, pickup_id) if not pickups.get(sender): pickup_time = self._get_pickup_time(sender) pickup_id = self._gls_canada_schedule_pickup(picks[0].shipper, picks.mapped('shipment_id'), pickup_time) picks.pickup_id = pickup_id return [picking_shipment_data_dict[picking] for picking in pickings] def _add_shipment_to_pickup(self, shipment_id, pickup_id): endpoint = f'/pickup/{pickup_id}/add-shipment/{shipment_id}' response = self._gls_canada_api_call(endpoint, "POST", {}) if response.status_code != 200: self._raise_gls_error(endpoint, response) def gls_canada_cancel_shipment(self, pickings): """This Method is used for cancel shipment from odoo to matkahuolto""" api_url = "%s/shipment/%s" % ( self.company_id.gls_canada_api_url, pickings.shipment_id) username = self.company_id.gls_canada_username password = self.company_id.gls_canada_password headers = {"Content-Type": "application/json"} response_body = requests.request(method="DELETE", url=api_url, headers=headers, auth=HTTPBasicAuth(username, password)) _logger.info("Cancel Shipment Response %s" % response_body.content) if response_body.status_code in [200, 201]: _logger.info("Shipment Cancel Successfully") return True else: raise ValidationError(response_body.content) def _gls_canada_shipping_request_data(self, pickings): parcel_data = [] counter = 0 for package in pickings.package_ids: counter += 1 parcel_data.append({ "id": counter, "parcelType": self.parcel_type, "quantity": "1", "weight": package.shipping_weight, "length": package.package_type_id.packaging_length or '', "depth": package.package_type_id.height or '', "width": package.package_type_id.width or '', # "requestReturnLabel": self.return_label, }) if pickings.weight_bulk: counter += 1 parcel_data.append({ "id": counter, "parcelType": self.parcel_type, "quantity": "1", "weight": pickings.shipping_weight, "length": self.gls_canada_packaging_id.packaging_length or '', "depth": self.gls_canada_packaging_id.height or '', "width": self.gls_canada_packaging_id.width or '', # "requestReturnLabel": self.return_label, }) receiver_address = pickings.partner_id shipper_address = pickings.picking_type_id.warehouse_id.partner_id sender_contact = pickings[0].shipper request_data = { "category": self.category, "paymentType": self.gls_canada_payment_type, "billingAccount": self.company_id.gls_canada_billing_number, "sender": self._gls_canada_make_sender_data(shipper_address, sender_contact), "consignee": self._gls_canada_make_consignee_data(receiver_address), "unitOfMeasurement": self.gls_canada_unit_measurement, "parcels": parcel_data, "deliveryType": self.gls_canada_delivery_type, } # if self.return_label == True: # request_data.update({"returnAddress": {"contact": { # "fullName": shipper_address.name or '', # "language": "en", # "email": "example@dicom.com", # "department": "", # "telephone": shipper_address.phone or '', # "extension": "" # } # }}) # if shipper_address.country_id and shipper_address.country_id.code != receiver_address.country_id.code: # request_data.update( # {"internationalDetails": {"descriptionOfGoods": 'Workspace123', "isDicomBroker": 'true', "products": [ # { # "id": "17451", # "Quantity": 1 # } # ] # }}) _logger.info("Shipping Request Data %s" % request_data) return request_data def _gls_canada_get_pickups_for_senders(self, senders) -> dict: """ Get a dict matching each sender to any already existing pickups today (or next business day). If no matching pickup exists, the sender is not added to the keys of the dict and a pickup will need to be created for it elsewhere. :param senders: The recordset of partner_ids matching the pickup locations :return: dict of the form {sender: "pickup ID"} """ endpoint = "/pickup/list" # First, group the senders by country and state so that holidays can be properly # handled for each location. locales = set() for sender in senders: locales.add((sender.country_id, sender.state_id)) sender_groups = [[sender for sender in senders if (sender.country_id, sender.state_id) == locale] for locale in locales] senders_to_pickups_dict = {} # For each group sharing a locale, get pickups for the next working day for sender_group in sender_groups: pickup_date = self._get_pickup_time(sender_group[0]) request_data = { 'category': self.category, 'pickupDate': pickup_date.strftime("%Y-%m-%d"), } response = self._gls_canada_api_call(endpoint, "GET", request_data) _logger.info("GLS Response status code:%s" % response.status_code) if response.status_code == 401: raise ValidationError( _("Username or password is incorrect for GLS Canada.")) elif response.status_code == 204: # No pickups scheduled return {} pickups = response.json() if response else [] for pickup in pickups: street = pickup['sender']['addressLine1'] zip = pickup['sender']['postalCode'].replace(' ', '') matching_senders = [sender for sender in sender_group if (sender.street == street and sender.zip.replace(' ', '') == zip)] senders_to_pickups_dict.update({ sender: pickup for sender in matching_senders }) return senders_to_pickups_dict @api.model def _group_pickings_by_warehouse_and_recipient(self, pickings) -> dict: """ Transform a list of pickings into a dict of the form:: { sender_partner : { recipient_partner : recordset(pickings) } } :param pickings: The recordset of pickings to be arranged :return: dict as described above """ shipments = {pick.picking_type_id.warehouse_id.partner_id: {} for pick in pickings} for sender in shipments.keys(): matching_pickings = pickings.filtered(lambda p: p.picking_type_id.warehouse_id.partner_id == sender) for pick in matching_pickings: recipient = pick.partner_id if recipient in shipments[sender]: shipments[sender][recipient] |= pick else: shipments[sender].update({recipient: pick}) return shipments def _gls_canada_schedule_pickup(self, sender_contact, shipments: List[str], pickup_time: datetime, office_close='16:00'): shipment = self._load_shipment(shipments[0]) sender = shipment['sender'] contact = shipment['sender']['contact'] endpoint = '/pickup' ready_time = pickup_time - timedelta(hours=1) request_data = { "shipments": [{'id': shipment} for shipment in shipments], "contact": self._gls_canada_make_contact_data(sender_contact)['contact'], "officeClose": office_close, "date": pickup_time.strftime("%Y-%m-%d"), "ready": pickup_time.strftime("%H:%M"), "category": self.category, "location": "SH", } """ Location is pickup location: * SS: Basement * RC: ground floor * PH: home * MR: mail room * MB: mailbox * BU: office * OT: other * SH: shipping """ response = self._gls_canada_api_call(endpoint, 'POST', request_data) if response.status_code == 201: # pickup created response_body = response.json() return response_body.get('ID') else: self._raise_gls_error(endpoint, response) def _load_shipment(self, shipment): endpoint = f"/shipment/{shipment}" response = self._gls_canada_api_call(endpoint, "GET", {}) return response.json() def _gls_canada_send_shipment(self, pickings, pickup_id): """ Create a new shipment for all the packages in the pickings and assign this shipment to the pickup_id submitted. If successful, the shipping label is retrieved from GLS and attached to each picking as a PDF. :param pickings: The iterable of pickings to be shipped. These must all be from the sender matching the pickup_id submitted and to the same recipient. :param pickups: The pickups that are already scheduled for the sender. The first will be used. :param pickup: The pickup_id of the pickup that is already scheduled with GLS. :return: Tuple containing The shipment data (see _gls_canada_send_shipping) and the shipment ID returned from GLS """ endpoint = "/shipment" request_data = self._gls_canada_shipping_request_data(pickings) response = self._gls_canada_api_call(endpoint, "POST", request_data) _logger.info("Shipping Response %s" % response.content) # response_body = self.shipment_response() if response.status_code in [200, 201]: response_body = response.json() _logger.info("Json Response %s " % response_body) parcel_tracking_numbers = [] for parcel_tracking_number in response_body.get( 'parcelTrackingNumbers'): parcel_tracking_numbers.append(parcel_tracking_number) pickings.parcel_tracking_numbers = ', '.join( parcel_tracking_numbers) pickings.shipment_id = response_body.get('ID') master_tracking_number = response_body.get('trackingNumber') # print label code self._gls_canada_get_label(master_tracking_number, pickings) return {'exact_price': 0.0, 'tracking_number': master_tracking_number}, response_body.get("ID") else: self._raise_gls_error(endpoint, response) def _gls_canada_get_label(self, master_tracking_number, pickings): endpoint = f"/shipment/label/{pickings.shipment_id}" headers = { 'Accept': 'application/pdf', # TODO: implement application/zpl here "Content-Type": "application/json", } response = self._gls_canada_api_call(endpoint, "GET", {}, headers=headers) _logger.info("Label Response %s" % response) log_message = f"Tracking Numbers: {master_tracking_number}" pickings.message_post(body=log_message, attachments=[(f"Label GLS Canada {pickings.name}.pdf", response.content)]) @classmethod def _gls_canada_make_sender_data(cls, sender, sender_contact): phone_parsed = phonenumbers.parse( sender_contact.phone) if sender_contact.phone else '' contact_data = cls._gls_canada_make_contact_data(sender_contact) return { "addressLine1": sender.street or '', "addressLine2": sender.street2 or '', "city": sender.city or '', "provinceCode": sender.state_id and sender.state_id.code or '', "postalCode": sender.zip or '', "countryCode": sender.country_id and sender.country_id.code or '', "customerName": sender.name or '', "contact": contact_data['contact'] } @classmethod def _gls_canada_make_contact_data(self, partner): """ Build a dict of the form { "contact": { ... GLS contact params ... } } from the given partner. :param partner: A single res_partner record. :return: """ phone_parsed = phonenumbers.parse(partner.phone) if partner.phone else '' return { "contact": { "fullName": partner.name or '', "language": partner.lang and partner.lang.split('_')[0] or '', "email": partner.email or '', "telephone": phone_parsed and phone_parsed.national_number or '', "extension": phone_parsed and phone_parsed.extension or '', } } @classmethod def _gls_canada_make_consignee_data(cls, partner_id): if partner_id.parent_id and partner_id.parent_id.name: parent = partner_id.parent_id receiver_company_name = parent.name receiver_contact_name = partner_id.name or '' else: receiver_company_name = partner_id.name or '' receiver_contact_name = '' return { "addressLine1": partner_id.street or '', "addressLine2": partner_id.street2 or '', # "streetType": "AVE", # "streetDirection": "N", # "suite": "10", "city": partner_id.city or '', "provinceCode": partner_id.state_id and partner_id.state_id.code or '', "postalCode": partner_id.zip or '', "countryCode": partner_id.country_id and partner_id.country_id.code or '', "customerName": receiver_company_name, "contact": { "fullName": receiver_contact_name, "language": "en", "email": partner_id.email or '', "department": "", "telephone": partner_id.phone or '', "extension": "" } } def _gls_canada_api_call(self, endpoint: str, method: str, request_data: dict, headers: dict = {"Content-Type": "application/json"}, ) -> requests.Response: """ Make a call to the GLS Canada API using the credentials set on the current company_id :param endpoint: The endpoint to use, such as "/pickup/list" :param method: The request method to use, typically "GET" or "POST" :param headers: Request headers :param request_data: :return: A list of dicts matching the GLS Canada API response specifications """ request_data = json.dumps(request_data) try: api_url = "%s%s" % ( self.company_id and self.company_id.gls_canada_api_url, endpoint) _logger.info("Sending GLS Canada request %s" % request_data) username = self.company_id.gls_canada_username password = self.company_id.gls_canada_password if method == "GET": response_body = requests.get(url=api_url, params=request_data, headers=headers, auth=HTTPBasicAuth(username, password)) elif method == "DELETE": response_body = requests.delete(url=api_url, params=request_data, headers=headers, auth=HTTPBasicAuth(username, password)) else: response_body = requests.request(method=method, url=api_url, data=request_data, headers=headers, auth=HTTPBasicAuth(username, password)) _logger.info(f"GLS Canada Response Status Code: {response_body.status_code}") _logger.info("GLS Canada API Response: %s" % response_body.content) return response_body except Exception as e: raise ValidationError(e) @api.model def gls_canada_get_tracking_link(self, picking): """This method is used for track you parcel""" return "https://www.gls-canada.com/en/express/tracking" @api.model def _get_pickup_time(cls, sender, closing_time='16:00', earliest_pickup='12:00') -> datetime: """ Generate a pickup time within working hours and not on holidays, based on the sender's localization (time zone) and closing and starting times. :param sender: res_partner representing the sender location, used for holidays :param closing_time: closing time to determine if datetime.now() is after the workday's closing time, as a string formatted '%H:%M'. :param earliest_pickup: earliest time of day for pickups at the sender location :return: Today if today is a business day, the next business day otherwise. """ # In theory all sender addresses are from Canada, but this is a more reusable # tool and doesn't introduce any extra dependencies for generalizing to all # locales. if not sender.partner_latitude or not sender.partner_longitude: if not sender.geo_localize(): _logger.warning(f'Sender address for {sender.name} could not be ' f'geolocated for time zone information. Server time zone' f' used instead.') localized = sender.partner_latitude and sender.partner_longitude now = datetime.now() + timedelta(minutes=30) if localized: tz_name = TimezoneFinder().timezone_at(lng=sender.partner_longitude, lat=sender.partner_latitude) tz = pytz.timezone(tz_name) now = now.astimezone(tz=tz) close_h, close_m = closing_time.split(':') start_h, start_m = earliest_pickup.split(':') closing = datetime(year=now.year, month=now.month, day=now.day, hour=int(close_h), minute=int(close_m)).astimezone(tz) starting = datetime(year=now.year, month=now.month, day=now.day, hour=int(start_h), minute=int(start_m)).astimezone(tz) # If past closing time, move to next day at starting time if now > closing: now = starting + timedelta(days=1) # If before starting time, move to current day starting time if now < starting: now = starting local_holidays = holidays.country_holidays(sender.country_id.code, sender.state_id.code) while now.date() in local_holidays or now.weekday() in (5, 6): now += timedelta(days=1) return now @api.model def _raise_gls_error(cls, endpoint, response): raise ValidationError(_(f"Got an error from GLS endpoint {endpoint}\n" f"Response: {response.content}"))