[MOV] delivery_stamps: from hibou-suite-enterprise:12.0

This commit is contained in:
Jared Kipe
2020-10-29 14:00:04 -07:00
parent f692dc2df1
commit c42f8e7eac
12 changed files with 11635 additions and 0 deletions

View File

@@ -0,0 +1 @@
from . import models

View File

@@ -0,0 +1,26 @@
{
'name': 'Stamps.com (USPS) Shipping',
'summary': 'Send your shippings through Stamps.com and track them online.',
'version': '12.0.1.0.0',
'author': "Hibou Corp.",
'category': 'Warehouse',
'license': 'AGPL-3',
'images': [],
'website': "https://hibou.io",
'description': """
Stamps.com (USPS) Shipping
==========================
Send your shippings through Stamps.com and track them online.
""",
'depends': [
'delivery_hibou',
],
'demo': [],
'data': [
'views/delivery_stamps_view.xml',
],
'auto_install': False,
'installable': True,
}

View File

@@ -0,0 +1 @@
from . import delivery_stamps

View File

@@ -0,0 +1,32 @@
Copyright (c) 2019 by Hibou Corp.
Copyright (c) 2014 by Jonathan Zempel.
Some rights reserved.
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are
met:
* Redistributions of source code must retain the above copyright
notice, this list of conditions and the following disclaimer.
* Redistributions in binary form must reproduce the above
copyright notice, this list of conditions and the following
disclaimer in the documentation and/or other materials provided
with the distribution.
* The names of the contributors may not be used to endorse or
promote products derived from this software without specific
prior written permission.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

View File

@@ -0,0 +1,14 @@
# -*- coding: utf-8 -*-
"""
stamps
~~~~~~
Stamps.com API.
:copyright: 2014 by Jonathan Zempel.
:license: BSD, see LICENSE for more details.
"""
__author__ = "Jonathan Zempel"
__license__ = "BSD"
__version__ = "0.9.1"

View File

@@ -0,0 +1,102 @@
# -*- coding: utf-8 -*-
"""
stamps.config
~~~~~~~~~~~~~
Stamps.com configuration.
:copyright: 2014 by Jonathan Zempel.
:license: BSD, see LICENSE for more details.
"""
from configparser import NoOptionError, NoSectionError, SafeConfigParser
from urllib.request import pathname2url
from urllib.parse import urljoin
import os
VERSION = 84
class StampsConfiguration(object):
"""Stamps service configuration. The service configuration may be provided
directly via parameter values, or it can be read from a configuration file.
If no parameters are given, the configuration will attempt to read from a
``'.stamps.cfg'`` file in the user's HOME directory. Alternately, a
configuration filename can be passed to the constructor.
Here is a sample configuration (by default the constructor reads from a
``'default'`` section)::
[default]
integration_id = XXXXXXXX-1111-2222-3333-YYYYYYYYYYYY
username = stampy
password = secret
:param integration_id: Default `None`. Unique ID, provided by Stamps.com,
that represents your application.
:param username: Default `None`. Stamps.com account username.
:param password: Default `None`. Stamps.com password.
:param wsdl: Default `None`. WSDL URI. Use ``'testing'`` to use the test
server WSDL.
:param port: Default `None`. The name of the WSDL port to use.
:param file_name: Default `None`. Optional configuration file name.
:param section: Default ``'default'``. The configuration section to use.
"""
def __init__(self, integration_id=None, username=None, password=None,
wsdl=None, port=None, file_name=None, section="default"):
parser = SafeConfigParser()
if file_name:
parser.read([file_name])
else:
parser.read([os.path.expanduser("~/.stamps.cfg")])
self.integration_id = self.__get(parser, section, "integration_id",
integration_id)
self.username = self.__get(parser, section, "username", username)
self.password = self.__get(parser, section, "password", password)
self.wsdl = self.__get(parser, section, "wsdl", wsdl)
self.port = self.__get(parser, section, "port", port)
if self.wsdl is None or wsdl == "testing":
file_path = os.path.abspath(__file__)
directory_path = os.path.dirname(file_path)
if wsdl == "testing":
file_name = "stamps_v{0}.test.wsdl".format(VERSION)
else:
file_name = "stamps_v{0}.wsdl".format(VERSION)
wsdl = os.path.join(directory_path, "wsdls", file_name)
self.wsdl = urljoin("file:", pathname2url(wsdl))
if self.port is None:
self.port = "SwsimV{0}Soap12".format(VERSION)
assert self.integration_id
assert self.username
assert self.password
assert self.wsdl
assert self.port
@staticmethod
def __get(parser, section, name, default):
"""Get a configuration value for the named section.
:param parser: The configuration parser.
:param section: The section for the given name.
:param name: The name of the value to retrieve.
"""
if default:
vars = {name: default}
else:
vars = None
try:
ret_val = parser.get(section, name, vars=vars)
except (NoSectionError, NoOptionError):
ret_val = default
return ret_val

View File

@@ -0,0 +1,301 @@
# -*- coding: utf-8 -*-
"""
stamps.services
~~~~~~~~~~~~~~~
Stamps.com services.
:copyright: 2014 by Jonathan Zempel.
:license: BSD, see LICENSE for more details.
"""
from decimal import Decimal
from logging import getLogger
from re import compile
from suds import WebFault
from suds.bindings.document import Document
from suds.client import Client
from suds.plugin import MessagePlugin
from suds.sax.element import Element
from suds.sudsobject import asdict
from suds.xsd.sxbase import XBuiltin
from suds.xsd.sxbuiltin import Factory
PATTERN_HEX = r"[0-9a-fA-F]"
PATTERN_ID = r"{hex}{{8}}-{hex}{{4}}-{hex}{{4}}-{hex}{{4}}-{hex}{{12}}".format(
hex=PATTERN_HEX)
RE_TRANSACTION_ID = compile(PATTERN_ID)
class AuthenticatorPlugin(MessagePlugin):
"""Handle message authentication.
:param credentials: Stamps API credentials.
:param wsdl: Configured service client.
"""
def __init__(self, credentials, client):
self.credentials = credentials
self.client = client
self.authenticator = None
def marshalled(self, context):
"""Add an authenticator token to the document before it is sent.
:param context: The current message context.
"""
body = context.envelope.getChild("Body")
operation = body[0]
if operation.name in ("AuthenticateUser", "RegisterAccount"):
pass
elif self.authenticator:
namespace = operation.namespace()
element = Element("Authenticator", ns=namespace)
element.setText(self.authenticator)
operation.insert(element)
else:
document = Document(self.client.wsdl)
method = self.client.service.AuthenticateUser.method
parameter = document.param_defs(method)[0]
element = document.mkparam(method, parameter, self.credentials)
operation.insert(element)
def unmarshalled(self, context):
"""Store the authenticator token for the next call.
:param context: The current message context.
"""
if hasattr(context.reply, "Authenticator"):
self.authenticator = context.reply.Authenticator
del context.reply.Authenticator
else:
self.authenticator = None
return context
class BaseService(object):
"""Base service.
:param configuration: API configuration.
"""
def __init__(self, configuration):
Factory.maptag("decimal", XDecimal)
self.client = Client(configuration.wsdl)
credentials = self.create("Credentials")
credentials.IntegrationID = configuration.integration_id
credentials.Username = configuration.username
credentials.Password = configuration.password
self.plugin = AuthenticatorPlugin(credentials, self.client)
self.client.set_options(plugins=[self.plugin], port=configuration.port)
self.logger = getLogger("stamps")
def call(self, method, **kwargs):
"""Call the given web service method.
:param method: The name of the web service operation to call.
:param kwargs: Method keyword-argument parameters.
"""
self.logger.debug("%s(%s)", method, kwargs)
instance = getattr(self.client.service, method)
try:
ret_val = instance(**kwargs)
except WebFault as error:
self.logger.warning("Retry %s", method, exc_info=True)
self.plugin.authenticator = None
try: # retry with a re-authenticated user.
ret_val = instance(**kwargs)
except WebFault as error:
self.logger.exception("%s retry failed", method)
self.plugin.authenticator = None
raise error
return ret_val
def create(self, wsdl_type):
"""Create an object of the given WSDL type.
:param wsdl_type: The WSDL type to create an object for.
"""
return self.client.factory.create(wsdl_type)
class StampsService(BaseService):
"""Stamps.com service.
"""
def add_postage(self, amount, transaction_id=None):
"""Add postage to the account.
:param amount: The amount of postage to purchase.
:param transaction_id: Default `None`. ID that may be used to retry the
purchase of this postage.
"""
account = self.get_account()
control = account.AccountInfo.PostageBalance.ControlTotal
return self.call("PurchasePostage", PurchaseAmount=amount,
ControlTotal=control, IntegratorTxID=transaction_id)
def create_add_on(self):
"""Create a new add-on object.
"""
return self.create("AddOnV15")
def create_customs(self):
"""Create a new customs object.
"""
return self.create("CustomsV3")
def create_array_of_customs_lines(self):
"""Create a new array of customs objects.
"""
return self.create("ArrayOfCustomsLine")
def create_customs_lines(self):
"""Create new customs lines.
"""
return self.create("CustomsLine")
def create_address(self):
"""Create a new address object.
"""
return self.create("Address")
def create_purchase_status(self):
"""Create a new purchase status object.
"""
return self.create("PurchaseStatus")
def create_registration(self):
"""Create a new registration object.
"""
ret_val = self.create("RegisterAccount")
ret_val.IntegrationID = self.plugin.credentials.IntegrationID
ret_val.UserName = self.plugin.credentials.Username
ret_val.Password = self.plugin.credentials.Password
return ret_val
def create_extended_postage_info(self):
return self.create("ExtendedPostageInfoV1")
def create_shipping(self):
"""Create a new shipping object.
"""
return self.create("RateV31")
def get_address(self, address):
"""Get a shipping address.
:param address: Address instance to get a clean shipping address for.
"""
return self.call("CleanseAddress", Address=address)
def get_account(self):
"""Get account information.
"""
return self.call("GetAccountInfo")
def get_label(self, from_address, to_address, rate, transaction_id, image_type=None,
customs=None, sample=False, extended_postage_info=False):
"""Get a shipping label.
:param from_address: The shipping 'from' address.
:param to_address: The shipping 'to' address.
:param rate: A rate instance for the shipment.
:param transaction_id: ID that may be used to retry/rollback the
purchase of this label.
:param customs: A customs instance for international shipments.
:param sample: Default ``False``. Get a sample label without postage.
"""
return self.call("CreateIndicium", IntegratorTxID=transaction_id,
Rate=rate, From=from_address, To=to_address, ImageType=image_type, Customs=customs,
SampleOnly=sample, ExtendedPostageInfo=extended_postage_info)
def get_postage_status(self, transaction_id):
"""Get postage purchase status.
:param transaction_id: The transaction ID returned by
:meth:`add_postage`.
"""
return self.call("GetPurchaseStatus", TransactionID=transaction_id)
def get_rates(self, shipping):
"""Get shipping rates.
:param shipping: Shipping instance to get rates for.
"""
rates = self.call("GetRates", Rate=shipping)
if rates.Rates:
ret_val = [rate for rate in rates.Rates.Rate]
else:
ret_val = []
return ret_val
def get_tracking(self, transaction_id):
"""Get tracking events for a shipment.
:param transaction_id: The transaction ID (or tracking number) returned
by :meth:`get_label`.
"""
if RE_TRANSACTION_ID.match(transaction_id):
arguments = dict(StampsTxID=transaction_id)
else:
arguments = dict(TrackingNumber=transaction_id)
return self.call("TrackShipment", **arguments)
def register_account(self, registration):
"""Register a new account.
:param registration: Registration instance.
"""
arguments = asdict(registration)
return self.call("RegisterAccount", **arguments)
def remove_label(self, transaction_id):
"""Cancel a shipping label.
:param transaction_id: The transaction ID (or tracking number) returned
by :meth:`get_label`.
"""
if RE_TRANSACTION_ID.match(transaction_id):
arguments = dict(StampsTxID=transaction_id)
else:
arguments = dict(TrackingNumber=transaction_id)
return self.call("CancelIndicium", **arguments)
class XDecimal(XBuiltin):
"""Represents an XSD decimal type.
"""
def translate(self, value, topython=True):
"""Translate between string and decimal values.
:param value: The value to translate.
:param topython: Default `True`. Determine whether to translate the
value for python.
"""
if topython:
if isinstance(value, str) and len(value):
ret_val = Decimal(value)
else:
ret_val = None
else:
if isinstance(value, (int, float, Decimal)):
ret_val = str(value)
else:
ret_val = value
return ret_val

View File

@@ -0,0 +1,149 @@
# -*- coding: utf-8 -*-
"""
stamps.tests
~~~~~~~~~~~~
Stamps.com API tests.
:copyright: 2014 by Jonathan Zempel.
:license: BSD, see LICENSE for more details.
"""
from .config import StampsConfiguration
from .services import StampsService
from datetime import date, datetime
from time import sleep
from unittest import TestCase
import logging
import os
logging.basicConfig()
logging.getLogger("suds.client").setLevel(logging.DEBUG)
file_path = os.path.abspath(__file__)
directory_path = os.path.dirname(file_path)
file_name = os.path.join(directory_path, "tests.cfg")
CONFIGURATION = StampsConfiguration(wsdl="testing", file_name=file_name)
def get_rate(service):
"""Get a test rate.
:param service: Instance of the stamps service.
"""
ret_val = service.create_shipping()
ret_val.ShipDate = date.today().isoformat()
ret_val.FromZIPCode = "94107"
ret_val.ToZIPCode = "20500"
ret_val.PackageType = "Package"
rate = service.get_rates(ret_val)[0]
ret_val.Amount = rate.Amount
ret_val.ServiceType = rate.ServiceType
ret_val.DeliverDays = rate.DeliverDays
ret_val.DimWeighting = rate.DimWeighting
ret_val.Zone = rate.Zone
ret_val.RateCategory = rate.RateCategory
ret_val.ToState = rate.ToState
add_on = service.create_add_on()
add_on.AddOnType = "US-A-DC"
ret_val.AddOns.AddOnV15.append(add_on)
return ret_val
def get_from_address(service):
"""Get a test 'from' address.
:param service: Instance of the stamps service.
"""
address = service.create_address()
address.FullName = "Pickwick & Weller"
address.Address1 = "300 Brannan St."
address.Address2 = "Suite 405"
address.City = "San Francisco"
address.State = "CA"
return service.get_address(address).Address
def get_to_address(service):
"""Get a test 'to' address.
:param service: Instance of the stamps service.
"""
address = service.create_address()
address.FullName = "POTUS"
address.Address1 = "1600 Pennsylvania Avenue NW"
address.City = "Washington"
address.State = "DC"
return service.get_address(address).Address
class StampsTestCase(TestCase):
initialized = False
def setUp(self):
if not StampsTestCase.initialized:
self.service = StampsService(CONFIGURATION)
StampsTestCase.initalized = True
def _test_0(self):
"""Test account registration.
"""
registration = self.service.create_registration()
type = self.service.create("CodewordType")
registration.Codeword1Type = type.Last4SocialSecurityNumber
registration.Codeword1 = 1234
registration.Codeword2Type = type.Last4DriversLicense
registration.Codeword2 = 1234
registration.PhysicalAddress = get_from_address(self.service)
registration.MachineInfo.IPAddress = "127.0.0.1"
registration.Email = "sws-support@stamps.com"
type = self.service.create("AccountType")
registration.AccountType = type.OfficeBasedBusiness
result = self.service.register_account(registration)
print result
def _test_1(self):
"""Test postage purchase.
"""
transaction_id = datetime.now().isoformat()
result = self.service.add_postage(10, transaction_id=transaction_id)
transaction_id = result.TransactionID
status = self.service.create_purchase_status()
seconds = 4
while result.PurchaseStatus in (status.Pending, status.Processing):
seconds = 32 if seconds * 2 >= 32 else seconds * 2
print "Waiting {0:d} seconds to get status...".format(seconds)
sleep(seconds)
result = self.service.get_postage_status(transaction_id)
print result
def test_2(self):
"""Test label generation.
"""
self.service = StampsService(CONFIGURATION)
rate = get_rate(self.service)
from_address = get_from_address(self.service)
to_address = get_to_address(self.service)
transaction_id = datetime.now().isoformat()
label = self.service.get_label(from_address, to_address, rate,
transaction_id=transaction_id)
self.service.get_tracking(label.StampsTxID)
self.service.get_tracking(label.TrackingNumber)
self.service.remove_label(label.StampsTxID)
print label
def test_3(self):
"""Test authentication retry.
"""
self.service.get_account()
authenticator = self.service.plugin.authenticator
self.service.get_account()
self.service.plugin.authenticator = authenticator
result = self.service.get_account()
print result

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,329 @@
from datetime import date
from logging import getLogger
from urllib.request import urlopen
from suds import WebFault
from odoo import api, fields, models, _
from odoo.exceptions import ValidationError
from .api.config import StampsConfiguration
from .api.services import StampsService
_logger = getLogger(__name__)
STAMPS_PACKAGE_TYPES = [
'Unknown',
'Postcard',
'Letter',
'Large Envelope or Flat',
'Thick Envelope',
'Package',
'Flat Rate Box',
'Small Flat Rate Box',
'Large Flat Rate Box',
'Flat Rate Envelope',
'Flat Rate Padded Envelope',
'Large Package',
'Oversized Package',
'Regional Rate Box A',
'Regional Rate Box B',
'Legal Flat Rate Envelope',
'Regional Rate Box C',
]
STAMPS_INTEGRATION_ID = 'f62cb4f0-aa07-4701-a1dd-f0e7c853ed3c'
class ProductPackaging(models.Model):
_inherit = 'product.packaging'
package_carrier_type = fields.Selection(selection_add=[('stamps', 'Stamps.com')])
stamps_cubic_pricing = fields.Boolean(string="Stamps.com Use Cubic Pricing")
class ProviderStamps(models.Model):
_inherit = 'delivery.carrier'
delivery_type = fields.Selection(selection_add=[('stamps', 'Stamps.com (USPS)')])
stamps_username = fields.Char(string='Stamps.com Username', groups='base.group_system')
stamps_password = fields.Char(string='Stamps.com Password', groups='base.group_system')
stamps_service_type = fields.Selection([('US-FC', 'First-Class'),
('US-PM', 'Priority'),
],
required=True, string="Service Type", default="US-PM")
stamps_default_packaging_id = fields.Many2one('product.packaging', string='Default Package Type')
stamps_image_type = fields.Selection([('Auto', 'Auto'),
('Png', 'PNG'),
('Gif', 'GIF'),
('Pdf', 'PDF'),
('Epl', 'EPL'),
('Jpg', 'JPG'),
('PrintOncePdf', 'Print Once PDF'),
('EncryptedPngUrl', 'Encrypted PNG URL'),
('Zpl', 'ZPL'),
('AZpl', 'AZPL'),
('BZpl', 'BZPL'),
],
required=True, string="Image Type", default="Pdf")
def _stamps_package_type(self, package=None):
if not package:
return self.stamps_default_packaging_id.shipper_package_code
return package.packaging_id.shipper_package_code if package.packaging_id.shipper_package_code in STAMPS_PACKAGE_TYPES else 'Package'
def _stamps_package_is_cubic_pricing(self, package=None):
if not package:
return self.stamps_default_packaging_id.stamps_cubic_pricing
return package.packaging_id.stamps_cubic_pricing
def _stamps_package_dimensions(self, package=None):
if not package:
package_type = self.stamps_default_packaging_id
else:
package_type = package.packaging_id
return package_type.length, package_type.width, package_type.height
def _get_stamps_service(self):
sudoself = self.sudo()
config = StampsConfiguration(integration_id=STAMPS_INTEGRATION_ID,
username=sudoself.stamps_username,
password=sudoself.stamps_password,
wsdl=('testing' if not sudoself.prod_environment else None))
return StampsService(configuration=config)
def _stamps_convert_weight(self, weight):
""" weight always expressed in database units (KG/LBS) """
if self.stamps_default_packaging_id.max_weight and self.stamps_default_packaging_id.max_weight < weight:
raise ValidationError('Stamps cannot ship for weight: ' + str(weight) + ' kgs/lbs.')
get_param = self.env['ir.config_parameter'].sudo().get_param
product_weight_in_lbs_param = get_param('product.weight_in_lbs')
if product_weight_in_lbs_param == '1':
return weight
weight_in_pounds = weight * 2.20462
return weight_in_pounds
def _get_stamps_shipping_for_order(self, service, order, date_planned):
weight = sum([(line.product_id.weight * line.product_qty) for line in order.order_line]) or 0.0
weight = self._stamps_convert_weight(weight)
if not all((order.warehouse_id.partner_id.zip, order.partner_shipping_id.zip)):
raise ValidationError('Stamps needs ZIP. From: ' + str(order.warehouse_id.partner_id.zip) + ' To: ' + str(order.partner_shipping_id.zip))
ret_val = service.create_shipping()
ret_val.ShipDate = date_planned.strftime('%Y-%m-%d') if date_planned else date.today().isoformat()
ret_val.FromZIPCode = self.get_shipper_warehouse(order=order).zip
ret_val.ToZIPCode = order.partner_shipping_id.zip
ret_val.PackageType = self._stamps_package_type()
ret_val.ServiceType = self.stamps_service_type
ret_val.WeightLb = weight
ret_val.ContentType = 'Merchandise'
return ret_val
def _stamps_get_addresses_for_picking(self, picking):
company = self.get_shipper_company(picking=picking)
from_ = self.get_shipper_warehouse(picking=picking)
to = self.get_recipient(picking=picking)
return company, from_, to
def _stamps_get_shippings_for_picking(self, service, picking):
ret = []
company, from_partner, to_partner = self._stamps_get_addresses_for_picking(picking)
if not all((from_partner.zip, to_partner.zip)):
raise ValidationError('Stamps needs ZIP. From: ' + str(from_partner.zip) + ' To: ' + str(to_partner.zip))
for package in picking.package_ids:
weight = self._stamps_convert_weight(package.shipping_weight)
l, w, h = self._stamps_package_dimensions(package=package)
ret_val = service.create_shipping()
ret_val.ShipDate = date.today().isoformat()
ret_val.FromZIPCode = from_partner.zip
ret_val.ToZIPCode = to_partner.zip
ret_val.PackageType = self._stamps_package_type(package=package)
ret_val.CubicPricing = self._stamps_package_is_cubic_pricing(package=package)
ret_val.Length = l
ret_val.Width = w
ret_val.Height = h
ret_val.ServiceType = self.stamps_service_type
ret_val.WeightLb = weight
ret_val.ContentType = 'Merchandise'
ret.append((package.name + ret_val.ShipDate + str(ret_val.WeightLb), ret_val))
if not ret:
weight = self._stamps_convert_weight(picking.shipping_weight)
l, w, h = self._stamps_package_dimensions()
ret_val = service.create_shipping()
ret_val.ShipDate = date.today().isoformat()
ret_val.FromZIPCode = from_partner.zip
ret_val.ToZIPCode = to_partner.zip
ret_val.PackageType = self._stamps_package_type()
ret_val.CubicPricing = self._stamps_package_is_cubic_pricing()
ret_val.Length = l
ret_val.Width = w
ret_val.Height = h
ret_val.ServiceType = self.stamps_service_type
ret_val.WeightLb = weight
ret_val.ContentType = 'Merchandise'
ret.append((picking.name + ret_val.ShipDate + str(ret_val.WeightLb), ret_val))
return ret
def stamps_get_shipping_price_from_so(self, orders):
res = self.stamps_get_shipping_price_for_plan(orders, date.today().isoformat())
return map(lambda r: r[0] if r else 0.0, res)
def stamps_get_shipping_price_for_plan(self, orders, date_planned):
res = []
service = self._get_stamps_service()
for order in orders:
shipping = self._get_stamps_shipping_for_order(service, order, date_planned)
rates = service.get_rates(shipping)
if rates and len(rates) >= 1:
rate = rates[0]
price = float(rate.Amount)
if order.currency_id.name != 'USD':
quote_currency = self.env['res.currency'].search([('name', '=', 'USD')], limit=1)
price = quote_currency.compute(rate.Amount, order.currency_id)
delivery_days = rate.DeliverDays
if delivery_days.find('-') >= 0:
delivery_days = delivery_days.split('-')
transit_days = int(delivery_days[-1])
else:
transit_days = int(delivery_days)
date_delivered = None
if date_planned and transit_days > 0:
date_delivered = self.calculate_date_delivered(date_planned, transit_days)
res = res + [(price, transit_days, date_delivered)]
continue
res = res + [(0.0, 0, None)]
return res
def stamps_rate_shipment(self, order):
self.ensure_one()
result = {
'success': False,
'price': 0.0,
'error_message': 'Error Retrieving Response from Stamps.com',
'warning_message': False
}
date_planned = None
if self.env.context.get('date_planned'):
date_planned = self.env.context.get('date_planned')
rate = self.stamps_get_shipping_price_for_plan(order, date_planned)
if rate:
price, transit_time, date_delivered = rate[0]
result.update({
'success': True,
'price': price,
'error_message': False,
'transit_time': transit_time,
'date_delivered': date_delivered,
})
return result
return result
def stamps_send_shipping(self, pickings):
res = []
service = self._get_stamps_service()
for picking in pickings:
package_labels = []
shippings = self._stamps_get_shippings_for_picking(service, picking)
company, from_partner, to_partner = self._stamps_get_addresses_for_picking(picking)
from_address = service.create_address()
from_address.FullName = company.name
from_address.Address1 = from_partner.street
if from_partner.street2:
from_address.Address2 = from_partner.street2
from_address.City = from_partner.city
from_address.State = from_partner.state_id.code
from_address = service.get_address(from_address).Address
to_address = service.create_address()
to_address.FullName = to_partner.name
to_address.Address1 = to_partner.street
if to_partner.street2:
to_address.Address2 = to_partner.street2
to_address.City = to_partner.city
to_address.State = to_partner.state_id.code
to_address = service.get_address(to_address).Address
try:
for txn_id, shipping in shippings:
rates = service.get_rates(shipping)
if rates and len(rates) >= 1:
rate = rates[0]
shipping.Amount = rate.Amount
shipping.ServiceType = rate.ServiceType
shipping.DeliverDays = rate.DeliverDays
shipping.DimWeighting = rate.DimWeighting
shipping.Zone = rate.Zone
shipping.RateCategory = rate.RateCategory
shipping.ToState = rate.ToState
add_on = service.create_add_on()
add_on.AddOnType = 'US-A-DC'
add_on2 = service.create_add_on()
add_on2.AddOnType = 'SC-A-HP'
shipping.AddOns.AddOnV15 = [add_on, add_on2]
extended_postage_info = service.create_extended_postage_info()
if self.is_amazon(picking=picking):
extended_postage_info.bridgeProfileType = 'Amazon MWS'
label = service.get_label(from_address, to_address, shipping,
transaction_id=txn_id, image_type=self.stamps_image_type,
extended_postage_info=extended_postage_info)
package_labels.append((txn_id, label))
except WebFault as e:
_logger.warn(e)
if package_labels:
for name, label in package_labels:
body = 'Cancelling due to error: ' + str(label.TrackingNumber)
try:
service.remove_label(label.TrackingNumber)
except WebFault as e:
raise ValidationError(e)
else:
picking.message_post(body=body)
raise ValidationError('Error on full shipment. Attempted to cancel any previously shipped.')
raise ValidationError('Error on shipment. ' + str(e))
else:
carrier_price = 0.0
tracking_numbers = []
for name, label in package_labels:
body = 'Shipment created into Stamps.com <br/> <b>Tracking Number : <br/>' + label.TrackingNumber + '</b>'
tracking_numbers.append(label.TrackingNumber)
carrier_price += float(label.Rate.Amount)
url = label.URL
response = urlopen(url)
attachment = response.read()
picking.message_post(body=body, attachments=[('LabelStamps-%s.%s' % (label.TrackingNumber, self.stamps_image_type), attachment)])
shipping_data = {'exact_price': carrier_price, 'tracking_number': ','.join(tracking_numbers)}
res = res + [shipping_data]
return res
def stamps_get_tracking_link(self, pickings):
res = []
for picking in pickings:
ref = picking.carrier_tracking_ref
res = res + ['https://tools.usps.com/go/TrackConfirmAction_input?qtc_tLabels1=%s' % ref]
return res
def stamps_cancel_shipment(self, picking):
service = self._get_stamps_service()
try:
service.remove_label(picking.carrier_tracking_ref)
picking.message_post(body=_(u'Shipment N° %s has been cancelled' % picking.carrier_tracking_ref))
picking.write({'carrier_tracking_ref': '',
'carrier_price': 0.0})
except WebFault as e:
raise ValidationError(e)

View File

@@ -0,0 +1,38 @@
<?xml version="1.0" encoding="utf-8"?>
<odoo>
<record id="view_delivery_carrier_form_with_provider_stamps" model="ir.ui.view">
<field name="name">delivery.carrier.form.provider.stamps</field>
<field name="model">delivery.carrier</field>
<field name="inherit_id" ref="delivery.view_delivery_carrier_form"/>
<field name="arch" type="xml">
<xpath expr="//page[@name='destination']" position='before'>
<page string="Stamps.com Configuration" attrs="{'invisible': [('delivery_type', '!=', 'stamps')]}">
<group>
<group>
<field name="stamps_username" attrs="{'required': [('delivery_type', '=', 'stamps')]}" />
<field name="stamps_password" attrs="{'required': [('delivery_type', '=', 'stamps')]}" password="True"/>
</group>
<group>
<field name="stamps_service_type" attrs="{'required': [('delivery_type', '==', 'stamps')]}"/>
<field name="stamps_default_packaging_id" attrs="{'required': [('delivery_type', '==', 'stamps')]}"/>
<field name="stamps_image_type" attrs="{'required': [('delivery_type', '==', 'stamps')]}"/>
</group>
</group>
</page>
</xpath>
</field>
</record>
<record id="product_packaging_delivery_form" model="ir.ui.view">
<field name="name">stamps.product.packaging.form.delivery</field>
<field name="model">product.packaging</field>
<field name="inherit_id" ref="delivery.product_packaging_delivery_form"/>
<field name="arch" type="xml">
<xpath expr="//field[@name='max_weight']" position='after'>
<field name="stamps_cubic_pricing"/>
</xpath>
</field>
</record>
</odoo>