Files
Odoo-Modules/delivery_canadapost/models/delivery_carrier.py
gsinghpal f81e0cd918 changes
2026-03-09 23:45:00 -04:00

509 lines
31 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
import string
import random
from requests import request
import xml.etree.ElementTree as etree
import logging
_logger = logging.getLogger(__name__)
from odoo import models, fields, api, _
from odoo.exceptions import RedirectWarning,ValidationError
from odoo.addons.delivery_canadapost.canada_post_api.canada_post_response import Response
class DeliveryCarrier(models.Model):
_inherit = 'delivery.carrier'
delivery_type = fields.Selection(selection_add=[('canada_post', 'Canada Post')], ondelete={
'canada_post': lambda recs: recs.write({'delivery_type': 'fixed', 'fixed_price': 0})})
option_code = fields.Selection([('SO','SO - Signature'),
('COV','COV - Coverage'),
('COD','COD - Collect on delivery'),
('PA18','PA18 - Proof of Age Required - 18'),
('PA19','PA19 - Proof of Age Required - 19'),
('HFP','HFP - Card for pickup'),
('DNS','DNS - Do not safe drop'),
('LAD','LAD - Leave at door - do not card'),
],string="Option Code",
help="Required if the corresponding parent XML element option exists. This is the option code indicating which option applies to this shipment.\n Note: The D2PO option indicates that the parcel will be delivered directly to a nearby Post Office. For the D2PO option, the following XML elements are required: \n name (under destination) \n client-voice-number (under destination) \n notification \n option-qualifier-2 \n Note: If you select Collect on Delivery (COD), specify Card for Pickup (HFP) or Deliver to Post Office (D2PO). This is to facilitate the collection of COD funds at a post office. If not specified, the system will default to HFP. \n Non-delivery handling codes (required for some U.S.A. and international shipments)\n RASE - Return at Senders Expense \n RTS - Return to Sender \n ABAN - Abandon")
service_type = fields.Selection([('DOM.RP','DOM.RP - Regular Parcel'),
('DOM.EP','DOM.EP - Expedited Parcel'),
('DOM.XP','DOM.XP - Xpresspost'),
('DOM.PC','DOM.PC - Priority'),
('USA.XP','USA.XP - Xpresspost USA'),
('USA.EP','USA.EP - Expedited Parcel USA'),
('INT.IP.SURF','INT.IP.SURF - International Parcel Surface'),
('INT.PW.PARCEL','INT.PW.PARCEL - Priority Worldwide parcel Intl'),
('INT.XP','INT.XP - Xpresspost International'),
], string="Service Type",
help="Canada Post delivery service used for shipping the item")
product_packaging_id = fields.Many2one('stock.package.type', string="Default Package Type",help="Selected packaging type, used in the request parameter")
reason_for_export = fields.Selection([('DOC', 'DOC = document'),
('SAM', 'SAM = commercial sample'),
('REP', 'REP = repair or warranty'),
('SOG', 'SOG = sale of goods'),
('OTH', 'OTH = other')], string="Reason For Export",default="SOG",
help="This is a code that represents the reason for export, which assists with border crossing.")
username = fields.Char("Username", copy=False, help="UserName provided by canada post.")
password = fields.Char("Password", copy=False, help="Password provided by canada post.")
customer_number = fields.Char("Customer Number", copy=False, help="The mailed by customer, Customer number provided by canada post.")
tracking_link = fields.Char(string="Tracking Link",help="Tracking link(URL) useful to track the shipment or package from this URL.",size=256)
canadapost_type = fields.Selection(selection=[('commercial', 'Contract Shipping'), ('counter', 'Non-Contract Shipping')], string="Customer Type", required=False)
canadapost_contract_id = fields.Char(string="Contract ID")
canadapost_payment_method = fields.Selection([('CreditCard', 'Credit Card'), ('Account', 'Account')],
string="Payment Method", default='CreditCard',
help="This is the method of payment for the shipment. The default value is CreditCard.")
#set default weight_uom_id
def _default_uom_in_delive(self):
weight_uom_id = self.env.ref('uom.product_uom_kgm', raise_if_not_found=False)
if not weight_uom_id:
uom_categ_id = self.env.ref('uom.product_uom_categ_kgm').id
weight_uom_id = self.env['uom.uom'].search([('category_id', '=', uom_categ_id), ('factor', '=', 1)], limit=1)
return weight_uom_id
weight_uom_id = fields.Many2one('uom.uom', string='Shipping UoM according to API UoM',help="Set equivalent unit of measurement according to provider unit of measurement. For Example, if the provider unit of measurement is KG then you have to select KG unit of measurement in the Shipping Unit of Measurement field.",default=_default_uom_in_delive)
#Get Canada post URL
@api.model
def get_canadapost_url(self):
if self.prod_environment:
return "https://soa-gw.canadapost.ca/rs/"
else:
return "https://ct.soa-gw.canadapost.ca/rs/"
def _compute_can_generate_return(self):
super(DeliveryCarrier, self)._compute_can_generate_return()
for carrier in self:
if carrier.delivery_type == 'canada_post':
carrier.can_generate_return = True
#Check Address filed is validating or not and return boolean value
@api.model
def validating_address(self, partner, additional_fields=[]):
missing_value = []
mandatory_fields = ['country_id', 'city', 'zip']
mandatory_fields.extend(additional_fields)
if not partner.street and not partner.street2 :
mandatory_fields.append('street')
for field in mandatory_fields :
if not getattr(partner, field) :
missing_value.append(field)
return missing_value
#Check required value proper or not for Shipping
def check_required_value_to_ship(self, orders):
for order in orders :
if not order.order_line:
return _("You have not any item to ship. Please provide item first")
else :
order_lines_without_weight = order.order_line.filtered(lambda line_item: not line_item.product_id.type in ['service', 'digital'] and not line_item.product_id.weight and not line_item.is_delivery)
for order_line in order_lines_without_weight :
return _("Please define weight in product : \n %s") % order_line.product_id.name
# validating customer address
missing_value = self.validating_address(order.partner_shipping_id)
if missing_value :
fields = ", ".join(missing_value)
return (_("Missing the values of the Customer address. \n Missing field(s) : %s ") % fields)
# validation shipper address
missing_value = self.validating_address(order.warehouse_id.partner_id)
if missing_value :
fields = ", ".join(missing_value)
return (_("Missing the values of the Warehouse address. \n Missing field(s) : %s ") % fields)
return False
# Return Weight
def convert_weight(self,from_uom_unit ,to_uom_unit, weight):
if not from_uom_unit:
from_uom_unit = self.env[
"product.template"
]._get_weight_uom_id_from_ir_config_parameter()
return from_uom_unit._compute_quantity(weight, to_uom_unit)
#Check Validate weight or not
def check_max_weight(self, order, shipment_weight):
for order_line in order.order_line:
if order_line.product_id and order_line.product_id.weight > shipment_weight:
return (_("Product weight is more than maximum weight."))
return False
#Get Rate from API than set Rate
def canada_post_rate_shipment(self, order):
# check the address validation
check_value = self.check_required_value_to_ship(order)
# check the product weight is appropriate to maximum weight.
if check_value:
return {'success': False, 'price': 0.0, 'error_message': check_value, 'warning_message': False}
# check the product weight is appropriate to maximum weight.
shipment_weight = self.product_packaging_id.max_weight
check_weight = {}
if shipment_weight!=0.0:
check_weight = self.check_max_weight(order, shipment_weight)
if check_weight:
return {'success': False, 'price': 0.0, 'error_message': check_weight, 'warning_message': False}
shipper_address = order.warehouse_id.partner_id or order.company_id.partner_id
recipient_address = order.partner_shipping_id or order.partner_id
# Convert weight in to the delivery method's weight UOM
carrier_ctx = self.env.context
weight = carrier_ctx.get("order_weight", 0.0) or order._get_estimated_weight() or 0.0
total_weight = self.convert_weight(order.company_id and order.company_id.weight_unit_of_measurement_id,
self.weight_uom_id,
weight)
declared_value = round(order.amount_untaxed, 2)
declared_currency = order.currency_id.name
price=0.0
rate_dict = self.canada_post_get_shipping_rate(shipper_address, recipient_address, total_weight,
picking_bulk_weight=False, packages=False,
declared_value=declared_value, declared_currency=declared_currency,
company_id=order.company_id)
_logger.info("Rate Response Data : %s" % (rate_dict))
if rate_dict.get('messages', False):
return {'success': False, 'price':price, 'error_message': rate_dict['messages']['message']['description'],
'warning_message': False}
if rate_dict.get('price-quotes',False) and rate_dict.get('price-quotes').get('price-quote',False) :
cnt = 0;
quotes = rate_dict['price-quotes']['price-quote']
if isinstance(quotes, dict):
quotes = [quotes]
for quote in quotes:
if quote['service-code']==self.service_type:
price = quote['price-details']['due']
cnt+=1
if cnt==0:
return {'success': False, 'price': price, 'error_message': "Rate API dosen't provide this service type price",
'warning_message': False}
if self.canadapost_type == 'counter' and self.service_type in ['DOM.RP','DOM.EP','DOM.XP','DOM.PC'] and self.option_code == 'COD':
if order.amount_total + float(price)>= 999.99:
raise ValidationError(
_("Get Rate Request Fail : \n The COD amount cannot exceed 1000.00 in Non-Contract Shipping.")
)
if self.canadapost_type == 'commercial' and self.service_type in ['DOM.RP','DOM.EP','DOM.XP','DOM.PC'] and self.option_code == 'COD':
if order.amount_total + float(price) >= 5000.00:
raise ValidationError(
_("Get Rate Request Fail : \n The COD amount cannot exceed 5000.00 in Contract Shipping.")
)
return {'success': True, 'price': float(price), 'error_message': False, 'warning_message': False}
#Send required XML data and than response from Canada Post API
def canada_post_get_shipping_rate(self, shipper_address, recipient_address, total_weight, picking_bulk_weight,
packages=False, declared_value=False, declared_currency=False, company_id=False):
result = {}
# built request data
service_root = etree.Element("mailing-scenario")
if self.canadapost_type == 'commercial':
service_root.attrib["xmlns"] = "http://www.canadapost.ca/ws/ship/rate-v4"
else:
service_root.attrib["xmlns"] = "http://www.canadapost.ca/ws/ship/rate-v3"
etree.SubElement(service_root, "customer-number").text = self.customer_number
parcel = etree.SubElement(service_root, "parcel-characteristics")
etree.SubElement(parcel, "weight").text = str(total_weight)
etree.SubElement(service_root, "origin-postal-code").text = "%s" % (shipper_address.zip.replace(" ","").upper())
destination = etree.SubElement(service_root, "destination")
if str(self.service_type[:3])=='DOM':
domestic = etree.SubElement(destination, "domestic")
etree.SubElement(domestic, "postal-code").text = "%s" % (recipient_address.zip.replace(" ","").upper())
elif str(self.service_type[:3])=='USA':
united_states = etree.SubElement(destination, "united-states")
etree.SubElement(united_states, "zip-code").text = "%s" % (recipient_address.zip.upper())
elif str(self.service_type[:3])=='INT':
international = etree.SubElement(destination, "international")
etree.SubElement(international, "country-code").text = "%s" % (recipient_address.country_id and recipient_address.country_id.code)
url='%sship/price'%(self.get_canadapost_url())
base_data = etree.tostring(service_root).decode('utf-8')
if self.canadapost_type == 'commercial':
headers = {"Accept": "application/vnd.cpc.ship.rate-v4+xml","Content-Type":"application/vnd.cpc.ship.rate-v4+xml"}
else:
headers = {"Accept": "application/vnd.cpc.ship.rate-v3+xml","Content-Type":"application/vnd.cpc.ship.rate-v3+xml"}
_logger.info("Rate Request Data : %s" % (base_data))
try:
response_body = request(method='POST', url=url, data=base_data, headers=headers, auth=(self.username,self.password))
api = Response(response_body)
result = api.dict()
_logger.info("Rate Response Data : %s" % (result))
except Exception as e:
result['error_message'] = e.message
return result
return result
# Random generate string and return
def get_group_id(self):
size = 15
chars = string.ascii_uppercase
return ''.join(random.choice(chars) for _ in range(size))
# Create Shipment and return generate label and receipt from Canada post API
def canada_post_send_shipping(self, pickings):
response = []
for picking in pickings:
weight_get = picking.shipping_weight or picking.weight
from_unit = picking.company_id and picking.company_id.weight_unit_of_measurement_id or ""
total_weight = round(self.convert_weight(from_unit,
self.weight_uom_id,
weight_get), 2)
# get package type
package_type = self.product_packaging_id
for stock_quant in picking.move_line_ids.result_package_id:
if not stock_quant.package_type_id:
package_type = self.product_packaging_id
else:
package_type = stock_quant.package_type_id
break
package_info = self.get_canadapost_parcel(package_type)
# Get the address of the sender and recipient
destination_address = picking.partner_id
sender_address = picking.picking_type_id and picking.picking_type_id.warehouse_id and picking.picking_type_id.warehouse_id.partner_id
if self.canadapost_type == 'commercial':
root_node = etree.Element("shipment")
root_node.attrib["xmlns"] = "http://www.canadapost.ca/ws/shipment-v8"
etree.SubElement(root_node, "transmit-shipment").text ="true"
etree.SubElement(root_node, "provide-receipt-info").text ="true"
else:
root_node = etree.Element("non-contract-shipment")
root_node.attrib["xmlns"] = "http://www.canadapost.ca/ws/ncshipment-v4"
etree.SubElement(root_node, "requested-shipping-point").text="%s"%(sender_address.zip.replace(" ","").upper() or "")
delivery_spec_node=etree.SubElement(root_node, "delivery-spec")
etree.SubElement(delivery_spec_node, "service-code").text="%s"%(self.service_type)
sender_node = etree.SubElement(delivery_spec_node, "sender")
etree.SubElement(sender_node, "company").text=sender_address.name
etree.SubElement(sender_node, "contact-phone").text ="%s"%(sender_address.phone or "")
address_details=etree.SubElement(sender_node, "address-details")
etree.SubElement(address_details, "address-line-1").text =sender_address.street or ""
etree.SubElement(address_details, "city").text =sender_address.city or ""
etree.SubElement(address_details, "prov-state").text ="%s"%(sender_address.state_id and sender_address.state_id.code or "")
if self.canadapost_type == 'commercial':
etree.SubElement(address_details , "country-code").text = "%s" % (sender_address.country_id and sender_address.country_id.code or "")
etree.SubElement(address_details, "postal-zip-code").text ="%s"%(sender_address.zip.replace(" ","").upper() or "")
destination_node = etree.SubElement(delivery_spec_node, "destination")
etree.SubElement(destination_node, "name").text =destination_address.name
etree.SubElement(destination_node, "company").text =destination_address.name
etree.SubElement(destination_node, "client-voice-number").text = destination_address.phone
destination_address_details = etree.SubElement(destination_node, "address-details")
etree.SubElement(destination_address_details , "address-line-1").text =destination_address.street or ""
etree.SubElement(destination_address_details , "city").text =destination_address.city or ""
etree.SubElement(destination_address_details , "prov-state").text ="%s"%(destination_address.state_id and destination_address.state_id.code or "")
etree.SubElement(destination_address_details , "country-code").text = "%s" % (destination_address.country_id and destination_address.country_id.code or "")
etree.SubElement(destination_address_details , "postal-zip-code").text ="%s"%(destination_address.zip.replace(" ","").upper() or "")
if self.option_code and not self.service_type in ['USA.XP','USA.EP','INT.IP.SURF','INT.PW.PARCEL','INT.XP']:
options = etree.SubElement(delivery_spec_node, "options")
option = etree.SubElement(options, "option")
etree.SubElement(option , "option-code").text = str(self.option_code or "")
if self.option_code in ['PA18','PA19']:
option_pa = etree.SubElement(options, "option")
etree.SubElement(option_pa , "option-code").text = 'SO'
if self.option_code == 'COD' or self.option_code == 'COV':
etree.SubElement(option , "option-amount").text = str(picking.sale_id.amount_total or "")
etree.SubElement(option , "option-qualifier-1").text ="true"
else:
options = etree.SubElement(delivery_spec_node, "options")
option_usa = etree.SubElement(options, "option")
etree.SubElement(option_usa , "option-code").text = 'RASE'
parcel_characteristics= etree.SubElement(delivery_spec_node, "parcel-characteristics")
etree.SubElement(parcel_characteristics, "weight").text ="%s"%(total_weight)
dimensions= etree.SubElement(parcel_characteristics, "dimensions ")
etree.SubElement(dimensions, "length").text ="%s"%(package_info.get('length', 1))
etree.SubElement(dimensions, "width").text ="%s"%(package_info.get('width', 1))
etree.SubElement(dimensions, "height").text ="%s"%(package_info.get('height', 1))
preferences= etree.SubElement(delivery_spec_node, "preferences")
etree.SubElement(preferences, "show-packing-instructions").text ="true"
customs = etree.SubElement(delivery_spec_node, "customs")
etree.SubElement(customs, "currency").text = str(picking.sale_id.currency_id.name)
if picking.sale_id.currency_id.rate:
rate=picking.sale_id.currency_id.rate
rate=round(rate,2)
etree.SubElement(customs, "conversion-from-cad").text = str(rate or '')
etree.SubElement(customs, "reason-for-export").text = "%s"%(self.reason_for_export)
sku_list = etree.SubElement(customs, "sku-list")
for move_line in picking.move_ids:
item = etree.SubElement(sku_list,"item")
etree.SubElement(item, "customs-description").text = str(move_line.product_id.name)
etree.SubElement(item, "unit-weight").text = str(move_line.weight)
etree.SubElement(item, "customs-value-per-unit").text = str(move_line.product_id.lst_price)
etree.SubElement(item, "customs-number-of-units").text = str(int(move_line.product_uom_qty))
if self.canadapost_type == 'commercial':
settlement_info = etree.SubElement(delivery_spec_node, "settlement-info")
etree.SubElement(settlement_info, "contract-id").text = self.canadapost_contract_id
etree.SubElement(settlement_info, "intended-method-of-payment").text = self.canadapost_payment_method
api_url=self.get_canadapost_url()
if self.canadapost_type == 'commercial':
url="%s%s/%s/shipment"%(api_url,self.customer_number,self.customer_number)
else:
url="%s%s/ncshipment"%(api_url,self.customer_number)
base_data= etree.tostring(root_node).decode('utf-8')
if self.canadapost_type == 'commercial':
headers = {"Accept": "application/vnd.cpc.shipment-v8+xml","Content-Type":"application/vnd.cpc.shipment-v8+xml","Accept-language":"en-CA"}
else:
headers = {"Accept": "application/vnd.cpc.ncshipment-v4+xml","Content-Type":"application/vnd.cpc.ncshipment-v4+xml","Accept-language":"en-CA"}
#try:
_logger.info("Create Shipment Request Data : %s" % (base_data))
response_body = request(method='POST', url=url, data=base_data, headers=headers, auth=(self.username,self.password))
if response_body.status_code == 200:
api = Response(response_body)
result = api.dict()
_logger.info("Create Shipment Response Data : %s" % (result))
else:
error_code = "%s" % (response_body.status_code)
error_message = response_body.reason
message = error_code + " " + error_message
api = Response(response_body)
result = api.dict()
if result['messages']['message']['description']:
raise ValidationError(_("ShipmentRequest Fail : \n %s" % (result['messages']['message']['description'])))
else:
raise ValidationError(_("ShipmentRequest Fail : %s \n More Information \n %s" % (message, response_body.text)))
if self.canadapost_type == 'commercial':
if not result['shipment-info']['shipment-id'] or not result['shipment-info']['links']['link']:
raise RedirectWarning("ShipmentRequest Fail \n More Information \n %s" % (result))
shipment_id=str(result['shipment-info']['shipment-id'])
else:
if not result['non-contract-shipment-info']['shipment-id'] or not result['non-contract-shipment-info']['links']['link']:
raise RedirectWarning("ShipmentRequest Fail \n More Information \n %s" % (result))
shipment_id=str(result['non-contract-shipment-info']['shipment-id'])
commercial_invoice_url_attchment=""
commercial_invoice = False
url_attchment=""
if self.canadapost_type == 'commercial':
for link in result['shipment-info']['links']['link']:
if link['_rel'] == 'label':
url_attchment = link['_href']
if link['_rel'] =='commercialInvoice':
commercial_invoice_url_attchment = link['_href']
commercial_invoice=True
else:
for link in result['non-contract-shipment-info']['links']['link']:
if link['_rel'] == 'label':
url_attchment = link['_href']
if link['_rel'] =='commercialInvoice':
commercial_invoice_url_attchment = link['_href']
commercial_invoice=True
headers_attchment = {'Accept': 'application/pdf'}
try:
attachment_response = request(method='GET', url=url_attchment, headers=headers_attchment, auth=(
self.username, self.password))
_logger.info("Label Response Data : %s" % (attachment_response))
picking.message_post(attachments=[
('Shipment Label - %s.PDF' % (shipment_id),
attachment_response.content)])
if commercial_invoice:
commercial_invoice_attachment_response = request(method='GET', url=commercial_invoice_url_attchment, headers=headers_attchment, auth=(
self.username, self.password))
picking.message_post(attachments=[
('Shipment Commercial Invoice - %s.PDF' % (shipment_id),
commercial_invoice_attachment_response.content)])
except Exception as e:
raise RedirectWarning(e)
if self.canadapost_type == 'commercial':
headers = {"Accept": "application/vnd.cpc.shipment-v8+xml","Accept-language":"en-CA"}
url_receipt = "%s"%(self.get_canadapost_url())+ str(self.customer_number)+"/"+str(self.customer_number)+"/shipment/" + str(shipment_id) + "/receipt"
else:
url_receipt = "%s"%(self.get_canadapost_url())+ str(self.customer_number) +"/ncshipment/" + str(shipment_id) + "/receipt"
try:
receipt_response= request(method='GET', url=url_receipt, headers=headers, auth=(self.username,self.password))
if receipt_response.status_code == 200:
api_receipt = Response(receipt_response)
result_receipt = api_receipt.dict()
_logger.info("Get shipment Detail Response Data : %s" % (result_receipt))
else:
result_receipt = {}
error_code = "%s" % (receipt_response.status_code)
error_message = response_body.reason
message = error_code + " " + error_message
mesage="ShipmentAcceptRequest Fail : %s \n More Information \n %s" % (message, response_body.text)
picking.message_post(body=mesage)
except Exception as e:
picking.message_post(body=e)
if self.canadapost_type == 'commercial':
if result_receipt:
extra_price = result_receipt['shipment-receipt'] and result_receipt['shipment-receipt']['cc-receipt-details'] and result_receipt['shipment-receipt']['cc-receipt-details']['charge-amount'] or 0.0
else:
extra_price = 0.0
tracking_pin= result['shipment-info'].get('tracking-pin',False)
else:
if result_receipt:
extra_price = result_receipt['non-contract-shipment-receipt'] and result_receipt['non-contract-shipment-receipt']['cc-receipt-details'] and result_receipt['non-contract-shipment-receipt']['cc-receipt-details']['charge-amount'] or 0.0
else:
extra_price = 0.0
tracking_pin= result['non-contract-shipment-info'].get('tracking-pin',False)
if not tracking_pin:
_logger.info("This Service not provide the tracking no.Service is : %s"%self.service_type)
shipping_data = {
'exact_price': float(extra_price) or 0.0,
'tracking_number': tracking_pin}
response += [shipping_data]
return response
# Tracking link return
def canada_post_get_tracking_link(self, picking):
link = picking.carrier_id.tracking_link or 'https://www.canadapost.ca/trackweb/en#/resultList?searchFor='
res = '%s %s' % (link, picking.carrier_tracking_ref)
return res
def canada_post_cancel_shipment(self, picking):
raise ValidationError(_("Canada Post does not provide Shipment Cancel API!"))
def get_canadapost_parcel(self, package):
packaging_length = (
self.sudo()._canadapost_convert_dimension_to_uom(
package.packaging_length, package.length_uom_name
)
)
width = (
self.sudo()._canadapost_convert_dimension_to_uom(
package.width, package.length_uom_name
)
)
height = (
self.sudo()._canadapost_convert_dimension_to_uom(
package.height, package.length_uom_name
)
)
return {
"length": packaging_length,
"width": width,
"height": height
}
def _canadapost_convert_dimension_to_uom(self, dimension, length_uom_name):
target_uom = self.env.ref("uom.product_uom_cm")
from_uom = self.env["uom.uom"].sudo().search([("name", "=", length_uom_name)])
if not from_uom:
from_uom = self.env[
"product.template"
]._get_length_uom_id_from_ir_config_parameter()
# Convert dimensions
return from_uom._compute_quantity(dimension, target_uom)