import string import random import base64 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.fusion_canada_post.fusion_cp_api.fusion_cp_response import Response class DeliveryCarrier(models.Model): _inherit = 'delivery.carrier' delivery_type = fields.Selection(selection_add=[('fusion_canada_post', 'Canada Post')], ondelete={ 'fusion_canada_post': lambda recs: recs.write({'delivery_type': 'fixed', 'fixed_price': 0})}) option_code = fields.Selection([ ('SO', 'Signature Required'), ('COV', 'Coverage (Insurance)'), ('COD', 'Collect on Delivery'), ('PA18', 'Proof of Age – 18+'), ('PA19', 'Proof of Age – 19+'), ('HFP', 'Card for Pickup'), ('DNS', 'Do Not Safe Drop'), ('LAD', 'Leave at Door'), ], string="Shipping Option", help="Optional delivery option applied to all shipments using this carrier.") service_type = fields.Selection([ ('DOM.RP', 'Regular Parcel'), ('DOM.EP', 'Expedited Parcel'), ('DOM.XP', 'Xpresspost'), ('DOM.PC', 'Priority'), ('USA.XP', 'Xpresspost USA'), ('USA.EP', 'Expedited Parcel USA'), ('INT.IP.SURF', 'International Parcel – Surface'), ('INT.PW.PARCEL', 'Priority Worldwide Parcel'), ('INT.XP', 'Xpresspost International'), ], string="Default Service", help="Default Canada Post service. Users can select a different service when adding shipping to a sale order.") product_packaging_id = fields.Many2one('stock.package.type', string="Default Package Type",help="Optional default package type. Dimensions will pre-fill in the shipping wizard but can be overridden per shipment.") reason_for_export = fields.Selection([ ('DOC', 'Document'), ('SAM', 'Commercial Sample'), ('REP', 'Repair or Warranty'), ('SOG', 'Sale of Goods'), ('OTH', 'Other'), ], string="Reason for Export", default="SOG", help="Required for international and cross-border shipments.") username = fields.Char("Username", copy=False, help="API key from Canada Post Developer Program.") password = fields.Char("Password", copy=False, help="API password from Canada Post Developer Program.") customer_number = fields.Char("Customer Number", copy=False, help="Your Canada Post customer number (mailed-by number).") tracking_link = fields.Char(string="Tracking Link",help="Base URL for tracking shipments. Leave blank to use the default Canada Post tracking page.",size=256) fusion_cp_type = fields.Selection(selection=[('commercial', 'Contract Shipping'), ('counter', 'Non-Contract Shipping')], string="Customer Type", required=False) fusion_cp_contract_id = fields.Char(string="Contract ID") fusion_cp_payment_method = fields.Selection([('CreditCard', 'Credit Card'), ('Account', 'Account')], string="Payment Method", default='CreditCard', help="How Canada Post charges for shipments on this account.") fusion_cp_output_format = fields.Selection( [('4x6', '4×6 Thermal Label'), ('8.5x11', '8.5×11 Full Page')], string="Label Output Format", default='4x6', help="Label format returned by Canada Post. " "4×6 is for thermal label printers, " "8.5×11 is the full page format.", ) fusion_cp_dimension_unit = fields.Selection( [('cm', 'Centimeters (cm)'), ('in', 'Inches (in)')], string="Package Dimension Unit", default='cm', help="Unit of measurement for package dimensions (length, width, height). " "Choose the unit you use when entering dimensions on your package types. " "Dimensions will be automatically converted to centimeters for the Canada Post API.", ) 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='API Weight Unit',help="Weight unit expected by the Canada Post API (kg). Used internally for converting weights before sending requests.",default=_default_uom_in_delive) @api.model def get_fusion_cp_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 == 'fusion_canada_post': carrier.can_generate_return = True @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 def check_required_value_to_ship(self, orders): for order in orders : if not order.order_line: return _("There are no items to ship. Please add products to the order.") 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 _("Product '%s' has no weight defined. Please set the weight before shipping.") % 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 (_("Incomplete customer address. 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 (_("Incomplete warehouse address. Missing field(s): %s") % fields) return False 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) 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 exceeds the maximum allowed weight for this package type.")) return False def fusion_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. if self.product_packaging_id: 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 # Build package_info from SO dimensions or default package type package_info = None if (order.fusion_cp_package_length and order.fusion_cp_package_width and order.fusion_cp_package_height): package_info = { 'length': round(self._fusion_cp_convert_dimension_to_cm( order.fusion_cp_package_length), 1), 'width': round(self._fusion_cp_convert_dimension_to_cm( order.fusion_cp_package_width), 1), 'height': round(self._fusion_cp_convert_dimension_to_cm( order.fusion_cp_package_height), 1), } elif self.product_packaging_id: package_info = self.get_fusion_cp_parcel( self.product_packaging_id) price = 0.0 rate_dict = self.fusion_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, package_info=package_info) _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': "Canada Post does not offer pricing for the selected service type to this destination.", 'warning_message': False} if self.fusion_cp_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( _("The COD amount cannot exceed $1,000.00 for non-contract shipping.") ) if self.fusion_cp_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( _("The COD amount cannot exceed $5,000.00 for contract shipping.") ) return {'success': True, 'price': float(price), 'error_message': False, 'warning_message': False} def fusion_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, package_info=None): result = {} # built request data service_root = etree.Element("mailing-scenario") if self.fusion_cp_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) # Include dimensions if provided (improves cubing/volumetric accuracy) if package_info: dims = etree.SubElement(parcel, "dimensions") etree.SubElement(dims, "length").text = str( package_info.get('length', 1)) etree.SubElement(dims, "width").text = str( package_info.get('width', 1)) etree.SubElement(dims, "height").text = str( package_info.get('height', 1)) etree.SubElement(service_root, "origin-postal-code").text = "%s" % (shipper_address.zip.replace(" ","").upper()) # Determine destination type from recipient country (not service_type) destination = etree.SubElement(service_root, "destination") country_code = recipient_address.country_id and recipient_address.country_id.code or '' if country_code == 'CA': domestic = etree.SubElement(destination, "domestic") etree.SubElement(domestic, "postal-code").text = "%s" % (recipient_address.zip.replace(" ","").upper()) elif country_code == 'US': united_states = etree.SubElement(destination, "united-states") etree.SubElement(united_states, "zip-code").text = "%s" % (recipient_address.zip.upper()) else: international = etree.SubElement(destination, "international") etree.SubElement(international, "country-code").text = "%s" % (country_code) url='%sship/price'%(self.get_fusion_cp_url()) base_data = etree.tostring(service_root).decode('utf-8') if self.fusion_cp_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 def fusion_canada_post_rate_shipment_all(self, order): """Return ALL available Canada Post service rates for the order.""" check_value = self.check_required_value_to_ship(order) if check_value: return {'success': False, 'error_message': check_value} shipper_address = order.warehouse_id.partner_id or order.company_id.partner_id recipient_address = order.partner_shipping_id or order.partner_id 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 # Get package dimensions from context (set by wizard) package_info = self.env.context.get('cp_package_info') rate_dict = self.fusion_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, package_info=package_info) if rate_dict.get('messages', False): return { 'success': False, 'error_message': rate_dict['messages']['message']['description'], } rates = [] if rate_dict.get('price-quotes') and rate_dict['price-quotes'].get('price-quote'): quotes = rate_dict['price-quotes']['price-quote'] if isinstance(quotes, dict): quotes = [quotes] for quote in quotes: expected_delivery = '' service_standard = quote.get('service-standard', {}) if isinstance(service_standard, dict): expected_delivery = service_standard.get( 'expected-delivery-date', '') rates.append({ 'service_code': quote.get('service-code', ''), 'service_name': quote.get('service-name', ''), 'price': float(quote.get('price-details', {}).get('due', 0)), 'expected_delivery': expected_delivery, }) if not rates: return { 'success': False, 'error_message': _("No shipping prices available for this destination."), } return rates def get_group_id(self): size = 15 chars = string.ascii_uppercase return ''.join(random.choice(chars) for _ in range(size)) def fusion_canada_post_send_shipping(self, pickings): response = [] for picking in pickings: # Determine service code (order-level selection or carrier default) service_code = self.service_type if picking.sale_id and picking.sale_id.fusion_cp_service_code: service_code = picking.sale_id.fusion_cp_service_code sale_order = picking.sale_id so_packages = ( sale_order.fusion_cp_package_ids.sorted('sequence') if sale_order else False) from_unit = (picking.company_id and picking.company_id .weight_unit_of_measurement_id or "") all_tracking = [] all_results = [] total_price = 0.0 # ── Decide what drives the shipment loop ── # If the SO has a package list (from the wizard), use it so we # create one shipment per SO-defined package — even when the # picking itself has only one "bulk" package. if so_packages: for so_pkg in so_packages: pkg_weight = round(self.convert_weight( from_unit, self.weight_uom_id, so_pkg.weight), 2) package_info = { 'length': round( self._fusion_cp_convert_dimension_to_cm( so_pkg.package_length), 1), 'width': round( self._fusion_cp_convert_dimension_to_cm( so_pkg.package_width), 1), 'height': round( self._fusion_cp_convert_dimension_to_cm( so_pkg.package_height), 1), } svc = so_pkg.service_code or service_code self._fusion_cp_validate_package( pkg_weight, package_info) result = self._fusion_cp_create_single_shipment( picking, pkg_weight, package_info, svc) self._fusion_cp_record_shipment( result, picking, sale_order, svc, pkg_weight, pkg_name="Package %d" % ( so_packages.ids.index(so_pkg.id) + 1)) all_tracking.append( result.get('tracking_number', '')) total_price += result.get('exact_price', 0.0) all_results.append(result) else: # Fall back to physical packages from the picking packages = self._get_packages_from_picking( picking, self.product_packaging_id) for idx, pkg in enumerate(packages): pkg_weight = round(self.convert_weight( from_unit, self.weight_uom_id, pkg.weight), 2) # Dimensions: legacy SO fields → package type → default package_info = None if sale_order: if (sale_order.fusion_cp_package_length and sale_order.fusion_cp_package_width and sale_order.fusion_cp_package_height): package_info = { 'length': round( self._fusion_cp_convert_dimension_to_cm( sale_order.fusion_cp_package_length ), 1), 'width': round( self._fusion_cp_convert_dimension_to_cm( sale_order.fusion_cp_package_width ), 1), 'height': round( self._fusion_cp_convert_dimension_to_cm( sale_order.fusion_cp_package_height ), 1), } if not package_info: pkg_type = ( pkg.package_type if hasattr(pkg, 'package_type') and pkg.package_type else self.product_packaging_id) if pkg_type: package_info = self.get_fusion_cp_parcel( pkg_type) else: raise ValidationError(_( "No package dimensions available for " "'%s'. Please set dimensions when " "adding shipping to the sale order.") % (pkg.name or 'package')) self._fusion_cp_validate_package( pkg_weight, package_info) result = self._fusion_cp_create_single_shipment( picking, pkg_weight, package_info, service_code) self._fusion_cp_record_shipment( result, picking, sale_order, service_code, pkg_weight, pkg_name=(pkg.name if hasattr(pkg, 'name') else '')) all_tracking.append( result.get('tracking_number', '')) total_price += result.get('exact_price', 0.0) all_results.append(result) # ── Post combined chatter message with ALL labels ── self._fusion_cp_post_shipping_documents( picking, all_results) shipping_data = { 'exact_price': total_price, 'tracking_number': ','.join( filter(None, all_tracking)), } response.append(shipping_data) return response def _fusion_cp_create_single_shipment( self, picking, pkg_weight, package_info, service_code): """Create one CP shipment for a single package. Returns dict with tracking_number, shipment_id, exact_price, and attachment records. """ 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) # --- Build XML --- if self.fusion_cp_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" % service_code # Sender 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.fusion_cp_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 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 dest_addr = etree.SubElement( destination_node, "address-details") etree.SubElement( dest_addr, "address-line-1" ).text = destination_address.street or "" etree.SubElement( dest_addr, "city").text = destination_address.city or "" etree.SubElement( dest_addr, "prov-state" ).text = "%s" % (destination_address.state_id and destination_address.state_id.code or "") etree.SubElement( dest_addr, "country-code" ).text = "%s" % (destination_address.country_id and destination_address.country_id.code or "") etree.SubElement( dest_addr, "postal-zip-code" ).text = "%s" % (destination_address.zip.replace( " ", "").upper() or "") # Options if (self.option_code and self.service_type not 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 in ('COD', '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 parcel_chars = etree.SubElement( delivery_spec_node, "parcel-characteristics") etree.SubElement( parcel_chars, "weight").text = "%s" % pkg_weight dimensions = etree.SubElement(parcel_chars, "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 preferences = etree.SubElement(delivery_spec_node, "preferences") etree.SubElement( preferences, "show-packing-instructions").text = "true" if self.fusion_cp_type != 'commercial': etree.SubElement( preferences, "output-format" ).text = self.fusion_cp_output_format or "4x6" else: print_prefs = etree.SubElement( delivery_spec_node, "print-preferences") etree.SubElement( print_prefs, "output-format" ).text = self.fusion_cp_output_format or "4x6" # Customs customs = etree.SubElement(delivery_spec_node, "customs") etree.SubElement( customs, "currency" ).text = str(picking.sale_id.currency_id.name) rate = 1.0 if picking.sale_id.currency_id.rate: rate = round(picking.sale_id.currency_id.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)) # Settlement info (commercial only) if self.fusion_cp_type == 'commercial': settlement_info = etree.SubElement( delivery_spec_node, "settlement-info") etree.SubElement( settlement_info, "contract-id" ).text = self.fusion_cp_contract_id etree.SubElement( settlement_info, "intended-method-of-payment" ).text = self.fusion_cp_payment_method # --- Make API call --- api_url = self.get_fusion_cp_url() if self.fusion_cp_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.fusion_cp_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", } _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( _("Canada Post: %s") % result['messages']['message']['description']) else: raise ValidationError( _("Canada Post shipment request failed (%s).\n%s") % (message, response_body.text[:500])) # --- Extract shipment ID --- if self.fusion_cp_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']) # --- Extract artifact URLs --- label_url = "" receipt_link_url = "" commercial_invoice_url = "" commercial_invoice = False links = (result['shipment-info']['links']['link'] if self.fusion_cp_type == 'commercial' else result['non-contract-shipment-info'] ['links']['link']) if isinstance(links, dict): links = [links] for link in links: if link['_rel'] == 'label': label_url = link['_href'] elif link['_rel'] == 'receipt': receipt_link_url = link['_href'] elif link['_rel'] == 'commercialInvoice': commercial_invoice_url = link['_href'] commercial_invoice = True # --- Download artifacts --- pdf_headers = {'Accept': 'application/pdf'} label_attachment = False receipt_attachment = False full_label_attachment = False ci_attachment = False label_content = b'' receipt_pdf_content = b'' try: if label_url: label_resp = request( method='GET', url=label_url, headers=pdf_headers, auth=(self.username, self.password)) _logger.info( "Label Response Status: %s", label_resp.status_code) if label_resp.status_code == 200: label_content = label_resp.content label_attachment = self.env['ir.attachment'].create({ 'name': 'Label-%s.pdf' % shipment_id, 'type': 'binary', 'datas': base64.b64encode(label_content), 'res_model': 'stock.picking', 'res_id': picking.id, 'mimetype': 'application/pdf', }) if not receipt_link_url: if self.fusion_cp_type == 'commercial': receipt_link_url = ( "%s%s/%s/shipment/%s/receipt" % (self.get_fusion_cp_url(), self.customer_number, self.customer_number, shipment_id)) else: receipt_link_url = ( "%s%s/ncshipment/%s/receipt" % (self.get_fusion_cp_url(), self.customer_number, shipment_id)) if receipt_link_url: receipt_pdf_resp = request( method='GET', url=receipt_link_url, headers=pdf_headers, auth=(self.username, self.password)) if receipt_pdf_resp.status_code == 200: receipt_pdf_content = receipt_pdf_resp.content receipt_attachment = ( self.env['ir.attachment'].create({ 'name': 'Receipt-%s.pdf' % shipment_id, 'type': 'binary', 'datas': base64.b64encode( receipt_pdf_content), 'res_model': 'stock.picking', 'res_id': picking.id, 'mimetype': 'application/pdf', })) if label_content and receipt_pdf_content: try: from odoo.tools import pdf as pdf_tools full_content = pdf_tools.merge_pdf( [label_content, receipt_pdf_content]) except Exception: full_content = label_content full_label_attachment = ( self.env['ir.attachment'].create({ 'name': 'FullLabel-%s.pdf' % shipment_id, 'type': 'binary', 'datas': base64.b64encode(full_content), 'res_model': 'stock.picking', 'res_id': picking.id, 'mimetype': 'application/pdf', })) if commercial_invoice and commercial_invoice_url: ci_resp = request( method='GET', url=commercial_invoice_url, headers=pdf_headers, auth=(self.username, self.password)) if ci_resp.status_code == 200: ci_attachment = self.env['ir.attachment'].create({ 'name': 'CommercialInvoice-%s.pdf' % shipment_id, 'type': 'binary', 'datas': base64.b64encode(ci_resp.content), 'res_model': 'stock.picking', 'res_id': picking.id, 'mimetype': 'application/pdf', }) except Exception as e: _logger.error( "Error downloading Canada Post artifacts: %s", e) picking.message_post( body=_('Error downloading shipping documents ' 'for shipment %s: %s') % (shipment_id, str(e))) # --- Fetch receipt XML for pricing info --- if self.fusion_cp_type == 'commercial': receipt_xml_headers = { "Accept": "application/vnd.cpc.shipment-v8+xml", "Accept-language": "en-CA", } url_receipt = "%s%s/%s/shipment/%s/receipt" % ( self.get_fusion_cp_url(), self.customer_number, self.customer_number, shipment_id) else: receipt_xml_headers = { "Accept": "application/vnd.cpc.ncshipment-v4+xml", "Accept-language": "en-CA", } url_receipt = "%s%s/ncshipment/%s/receipt" % ( self.get_fusion_cp_url(), self.customer_number, shipment_id) result_receipt = {} try: receipt_response = request( method='GET', url=url_receipt, headers=receipt_xml_headers, auth=(self.username, self.password)) if receipt_response.status_code == 200: api_receipt = Response(receipt_response) result_receipt = api_receipt.dict() _logger.info("Receipt XML Data: %s", result_receipt) else: _logger.warning( "Receipt fetch: %s %s", receipt_response.status_code, receipt_response.reason) except Exception as e: _logger.warning("Error fetching receipt XML: %s", e) # Extract pricing and tracking extra_price = 0.0 tracking_pin = False if self.fusion_cp_type == 'commercial': if result_receipt: try: extra_price = float( result_receipt.get('shipment-receipt', {}) .get('cc-receipt-details', {}) .get('charge-amount', 0.0)) except (TypeError, ValueError): extra_price = 0.0 tracking_pin = result['shipment-info'].get( 'tracking-pin', False) else: if result_receipt: try: extra_price = float( result_receipt .get('non-contract-shipment-receipt', {}) .get('cc-receipt-details', {}) .get('charge-amount', 0.0)) except (TypeError, ValueError): extra_price = 0.0 tracking_pin = ( result['non-contract-shipment-info'] .get('tracking-pin', False)) if not tracking_pin: _logger.info( "This service does not provide tracking. Service: %s", service_code) return { 'tracking_number': tracking_pin or '', 'shipment_id': shipment_id, 'exact_price': float(extra_price) or 0.0, 'label_attachment': label_attachment, 'full_label_attachment': full_label_attachment, 'receipt_attachment': receipt_attachment, 'ci_attachment': ci_attachment, 'label_content': label_content, 'receipt_content': receipt_pdf_content, } def _fusion_cp_post_shipping_documents(self, picking, all_results): """Post a single chatter message with combined shipping documents. Merges all label PDFs into one printable file so the user can print every label at once. Individual labels remain on the ``fusion.cp.shipment`` records for per-shipment access. """ from odoo.tools import pdf as pdf_tools shipment_ids = [ r.get('shipment_id', '') for r in all_results if r.get('shipment_id')] # ── Merge all label PDFs ── label_pages = [ r['label_content'] for r in all_results if r.get('label_content')] receipt_pages = [ r['receipt_content'] for r in all_results if r.get('receipt_content')] attach_ids = [] if label_pages: try: if len(label_pages) > 1: merged = pdf_tools.merge_pdf(label_pages) else: merged = label_pages[0] att = self.env['ir.attachment'].create({ 'name': 'Labels-All.pdf', 'type': 'binary', 'datas': base64.b64encode(merged), 'res_model': 'stock.picking', 'res_id': picking.id, 'mimetype': 'application/pdf', }) attach_ids.append(att.id) except Exception as e: _logger.warning( "Could not merge label PDFs: %s", e) # Fall back: attach each label individually for r in all_results: if r.get('label_attachment'): attach_ids.append( r['label_attachment'].id) if receipt_pages: try: if len(receipt_pages) > 1: merged_rcpt = pdf_tools.merge_pdf(receipt_pages) else: merged_rcpt = receipt_pages[0] att_r = self.env['ir.attachment'].create({ 'name': 'Receipts-All.pdf', 'type': 'binary', 'datas': base64.b64encode(merged_rcpt), 'res_model': 'stock.picking', 'res_id': picking.id, 'mimetype': 'application/pdf', }) attach_ids.append(att_r.id) except Exception as e: _logger.warning( "Could not merge receipt PDFs: %s", e) for r in all_results: if r.get('receipt_attachment'): attach_ids.append( r['receipt_attachment'].id) # Commercial invoices (one per shipment, not merged) for r in all_results: if r.get('ci_attachment'): attach_ids.append(r['ci_attachment'].id) # ── Post to chatter ── if len(shipment_ids) == 1: body = _('Canada Post shipment %s created.') % ( shipment_ids[0]) elif shipment_ids: body = _('Canada Post — %d shipments created: %s') % ( len(shipment_ids), ', '.join(shipment_ids)) else: body = _('Canada Post shipment created.') if attach_ids: picking.message_post( body=body, attachment_ids=attach_ids) else: picking.message_post(body=body) def _fusion_cp_record_shipment( self, result, picking, sale_order, service_code, pkg_weight, pkg_name=''): """Create a ``fusion.cp.shipment`` record from shipment result dict. Called once per package by ``fusion_canada_post_send_shipping`` after ``_fusion_cp_create_single_shipment`` returns successfully. """ shipment_vals = { 'tracking_number': result.get('tracking_number', ''), 'shipment_id': result.get('shipment_id', ''), 'carrier_id': self.id, 'sale_order_id': (sale_order.id if sale_order else False), 'picking_id': picking.id, 'service_type': service_code, 'shipment_date': fields.Datetime.now(), 'shipping_cost': result.get('exact_price', 0.0), 'weight': pkg_weight, 'status': 'confirmed', 'company_id': picking.company_id.id, 'package_name': pkg_name, } # Attach label / receipt / CI documents if result.get('label_attachment'): shipment_vals['label_attachment_id'] = ( result['label_attachment'].id) if result.get('full_label_attachment'): shipment_vals['full_label_attachment_id'] = ( result['full_label_attachment'].id) if result.get('receipt_attachment'): shipment_vals['receipt_attachment_id'] = ( result['receipt_attachment'].id) if result.get('ci_attachment'): shipment_vals['commercial_invoice_attachment_id'] = ( result['ci_attachment'].id) self.env['fusion.cp.shipment'].create(shipment_vals) def fusion_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 fusion_canada_post_cancel_shipment(self, picking): """Cancel all CP shipments linked to this picking via void API.""" shipments = self.env['fusion.cp.shipment'].search([ ('picking_id', '=', picking.id), ('status', '!=', 'cancelled'), ]) for shipment in shipments: try: shipment.action_void_shipment() except ValidationError as e: _logger.warning( "Failed to void CP shipment %s: %s", shipment.name, e) raise def _fusion_cp_validate_package(self, weight_kg, package_info): """Validate package dimensions and weight against Canada Post limits. ``package_info`` values are already in **cm** (converted by ``get_fusion_cp_parcel``). CP limits: - Max weight: 30 kg - Max longest dimension: 100 cm - Max second longest dimension: 76 cm - Volumetric weight: L × W × H (cm) / 5000 - Billable weight = max(actual, volumetric) Raises ValidationError if hard limits exceeded. Logs warning if volumetric weight exceeds actual weight. """ errors = [] length_cm = float(package_info.get('length', 0)) width_cm = float(package_info.get('width', 0)) height_cm = float(package_info.get('height', 0)) # Sort dimensions to find longest and second longest dims = sorted([length_cm, width_cm, height_cm], reverse=True) # Determine user's preferred display unit dim_unit = self.fusion_cp_dimension_unit or 'cm' if dim_unit == 'in': unit_label = 'in' # Convert cm limits to inches for the error message max_longest = 39.4 # 100 cm ≈ 39.4 in max_second = 29.9 # 76 cm ≈ 29.9 in factor = 1 / 2.54 # cm → inches else: unit_label = 'cm' max_longest = 100 max_second = 76 factor = 1 if weight_kg > 30: errors.append( _("Weight %.2f kg exceeds Canada Post maximum of 30 kg.") % weight_kg) if dims[0] > 100: errors.append( _("Longest dimension %.1f %s exceeds Canada Post " "maximum of %.1f %s.") % ( dims[0] * factor, unit_label, max_longest, unit_label)) if dims[1] > 76: errors.append( _("Second longest dimension %.1f %s exceeds Canada Post " "maximum of %.1f %s.") % ( dims[1] * factor, unit_label, max_second, unit_label)) if errors: raise ValidationError('\n'.join(errors)) # Volumetric weight warning (CP cubing) if length_cm and width_cm and height_cm: vol_weight = (length_cm * width_cm * height_cm) / 5000 if vol_weight > weight_kg: _logger.warning( "CP cubing: volumetric weight %.2f kg > actual %.2f kg " "for package (%.1f × %.1f × %.1f cm). Canada Post will " "bill at the higher volumetric weight.", vol_weight, weight_kg, length_cm, width_cm, height_cm) def get_fusion_cp_parcel(self, package): """Get package dimensions in cm for Canada Post API. Uses the carrier's ``fusion_cp_dimension_unit`` setting to determine the source unit (cm or inches) instead of relying on the company's default length UOM, which may be set to feet or another unit. """ packaging_length = self.sudo()._fusion_cp_convert_dimension_to_cm( package.packaging_length) width = self.sudo()._fusion_cp_convert_dimension_to_cm( package.width) height = self.sudo()._fusion_cp_convert_dimension_to_cm( package.height) return { "length": round(packaging_length, 1), "width": round(width, 1), "height": round(height, 1), } def _fusion_cp_convert_dimension_to_cm(self, dimension): """Convert a dimension value to centimeters using the carrier's configured dimension unit (``fusion_cp_dimension_unit``). If the carrier is set to *cm*, the value is returned as-is. If set to *in* (inches), it is converted from inches → cm. """ if not dimension: return 0.0 dim_unit = self.fusion_cp_dimension_unit or 'cm' if dim_unit == 'cm': return dimension # inches → cm target_uom = self.env.ref("uom.product_uom_cm") from_uom = self.env.ref("uom.product_uom_inch") return from_uom._compute_quantity(dimension, target_uom)