Merge pull request #87 from akretion/feat/add_geodis_findLocalite

Feat/add geodis find localite

roulier/carriers/geodis/__init__.py

@@ -1,6 +1,8 @@
 # -*- coding: utf-8 -*-
 from . import geodis
+from . import geodis_common_ws
 from . import geodis_api_ws
+from . import geodis_api_find_localite_ws
 from . import geodis_api_edi
 from . import geodis_encoder_ws
 from . import geodis_encoder_edi

roulier/carriers/geodis/geodis.py

@@ -16,45 +16,53 @@ def api(self, action='label'):
         """Expose how to communicate with Geodis."""
         try:
             method = self.ACTIONS[action]
-        except:
+        except KeyError:
             raise InvalidAction("Action not supported")
         return method(self, None, api=True)
 
     def get(self, data, action):
         """."""
         try:
             method = self.ACTIONS[action]
-        except:
+        except KeyError:
             raise InvalidAction("Action not supported")
 
         return method(self, data)
 
+    def get_edi(self, data, api=False):
+        encoder = GeodisEncoderEdi()
+        transport = GeodisTransportEdi()
+        if api:
+            return encoder.api()
+        arr = encoder.encode(data)
+        return transport.send(arr)
+
     def get_label(self, data, api=False):
         """Genereate a demandeImpressionEtiquette."""
+        return self._get_ws(data, api, 'demandeImpressionEtiquette')
+
+    def address_validator(self, data, api=False):
+        return self._get_ws(data, api, 'findLocalite')
+
+    def _get_ws(self, data, api=False, action=None):
         encoder = GeodisEncoderWs()
         decoder = GeodisDecoder()
         transport = GeodisTransportWs()
 
         if api:
-            return encoder.api()
+            return encoder.api(action=action)
 
-        request = encoder.encode(data, "demandeImpressionEtiquette")
+        request = encoder.encode(data, action)
         response = transport.send(request)
         return decoder.decode(
             response['body'],
             response['parts'],
-            request['output_format'])
-
-    def get_edi(self, data, api=False):
-        encoder = GeodisEncoderEdi()
-        transport = GeodisTransportEdi()
-        if api:
-            return encoder.api()
-        arr = encoder.encode(data)
-        return transport.send(arr)
+            request['infos'],
+        )
 
     ACTIONS = {
         'label': get_label,
+        'findLocalite': address_validator,
         'demandeImpressionEtiquette': get_label,
         'edi': get_edi
     }

roulier/carriers/geodis/geodis_api_find_localite_ws.py

@@ -0,0 +1,35 @@
+# -*- coding: utf-8 -*-
+"""Implementation of Geodis Ws Api."""
+from roulier.api import Api
+
+
+class GeodisApiFindLocaliteWs(Api):
+
+    def _service(self):
+        schema = {}
+        schema['is_test'] = {
+            'type': 'boolean', 'default': True,
+            'description': 'Use test Ws'}
+        return schema
+
+    def _address(self):
+        schema = super(GeodisApiFindLocaliteWs, self)._address()
+        schema['country'].update({'required': True, 'empty': False})
+        return {
+            'country': schema['country'],
+            'zip': schema['zip'],
+            'city': schema['city'],
+        }
+
+    def _auth(self):
+        schema = super(GeodisApiFindLocaliteWs, self)._auth()
+        schema['login'].update({'required': True, 'empty': False})
+        schema['password']['required'] = False
+        return schema
+
+    def _schemas(self):
+        return {
+            'auth': self._auth(),
+            'to_address': self._address(),
+            'service': self._service(),
+        }

roulier/carriers/geodis/geodis_common_ws.py

@@ -0,0 +1,20 @@
+# -*- coding: utf-8 -*-
+"""Store infos about ws"""
+from .geodis_api_ws import GeodisApiWs
+from .geodis_api_find_localite_ws import GeodisApiFindLocaliteWs
+
+
+GEODIS_INFOS = {
+    'demandeImpressionEtiquette': {
+        'xmlns': 'http://impression.service.web.etiquette.geodis.com',
+        'endpoint': "http://espace.geodis.com/geolabel/services/ImpressionEtiquette",  # nopep8
+        'endpoint_test': "http://espace.recette.geodis.com/geolabel/services/ImpressionEtiquette",  # nopep8
+        'api': GeodisApiWs,
+    },
+    'findLocalite': {
+        'xmlns': 'http://localite.service.web.etiquette.geodis.com',
+        'endpoint': "http://espace.geodis.com/geolabel/services/RechercherLocalite",  # nopep8
+        'endpoint_test': "http://espace.recette.geodis.com/geolabel/services/RechercherLocalite",  # nopep8
+        'api': GeodisApiFindLocaliteWs,
+    }
+}

roulier/carriers/geodis/geodis_decoder.py

@@ -1,15 +1,17 @@
 # -*- coding: utf-8 -*-
 """Geodis XML -> Python."""
 from roulier.codec import Decoder
+from .geodis_common_ws import GEODIS_INFOS
 
 
 class GeodisDecoder(Decoder):
     """Geodis XML -> Python."""
 
-    def decode(self, body, parts, output_format):
+    def decode(self, body, parts, infos):
         """Geodis XML -> Python."""
 
         def reponse_impression_etiquette(msg, parts):
+            output_format = infos["output_format"]
             labels = '^XZ•^XA'.join(parts.split('^XZ\r\n^XA')).split('•')
             labels_idx = iter(range(len(labels)))
             labels_data = iter(labels)
@@ -43,9 +45,19 @@ def reponse_impression_etiquette(msg, parts):
                 }
             }
 
+        def response_find_localite(msg, parts):
+            return [{
+                "ordre": localite.numOrdre,
+                "region": localite.codeRegion,
+                "zip": localite.codePostal,
+                "city": localite.libelle,
+            } for localite in msg.infoLocalite]
+
         tag = body.tag
         lookup = {
             "{http://impression.service.web.etiquette.geodis.com}reponseImpressionEtiquette":
                 reponse_impression_etiquette,
+            "{http://localite.service.web.etiquette.geodis.com}findLocaliteResponse":
+                response_find_localite,
         }
         return lookup[tag](body, parts)

roulier/carriers/geodis/geodis_encoder_ws.py

@@ -3,53 +3,86 @@
 from jinja2 import Environment, PackageLoader
 from roulier.codec import Encoder
 from roulier.exception import InvalidApiInput
-from .geodis_api_ws import GeodisApiWs
+from .geodis_common_ws import GEODIS_INFOS
 
-GEODIS_ACTIONS = ('demandeImpressionEtiquette',)
+
+GEODIS_ACTIONS = GEODIS_INFOS.keys()
 
 
 class GeodisEncoderWs(Encoder):
     """Transform input to geodis compatible xml."""
 
-    def encode(self, api_input, action):
-        """Transform input to geodis compatible xml."""
+    def get_api(self, action):
+        """Get Api object based on action.
+
+        raise if action is not known"""
         if not (action in GEODIS_ACTIONS):
             raise InvalidApiInput(
-                'action %s not in %s' % (action, ', '.join(GEODIS_ACTIONS)))
+                'action %s not in %s' % (
+                    action, ', '.join(GEODIS_INFOS)))
+
+        return GEODIS_INFOS[action]['api']()
 
-        api = GeodisApiWs()
+    def encode(self, api_input, action):
+        """Transform input to geodis compatible xml."""
+        api = self.get_api(action)
         if not api.validate(api_input):
             raise InvalidApiInput(
                 'Input error : %s' % api.errors(api_input))
-        data = api.normalize(api_input)
-
-        data['service']['labelFormat'] = self.lookup_label_format(
-            data['service']['labelFormat'])
-        data['service']['shippingDate'] = data['service']['shippingDate'].replace('/','') 
-        data['from_address']['departement'] = data['from_address']['zip'][:2]
-
-        is_test = data['service']['is_test']
-
         env = Environment(
             loader=PackageLoader('roulier', '/carriers/geodis/templates'),
             extensions=['jinja2.ext.with_', 'jinja2.ext.autoescape'],
             autoescape=True)
-
         template = env.get_template("geodis_%s.xml" % action)
-        return {
-            "body": template.render(
-                service=data['service'],
-                parcels=data['parcels'],
-                sender_address=data['from_address'],
-                receiver_address=data['to_address']),
-            "headers": data['auth'],
-            "is_test": is_test,
-            "output_format": api_input['service']['labelFormat']
+
+        data = api.normalize(api_input)
+        infos = {
+            'xmlns': GEODIS_INFOS[action]['xmlns'],
+            'url': (
+                data['service']['is_test'] and
+                GEODIS_INFOS[action]['endpoint_test'] or
+                GEODIS_INFOS[action]['endpoint']),
+            'action': action,
         }
 
-    def api(self):
-        api = GeodisApiWs()
-        return api.api_values()
+        def request_impression_etiquette():
+            infos["output_format"] = api_input['service']['labelFormat'],
+            data['service']['labelFormat'] = self.lookup_label_format(
+                data['service']['labelFormat'])
+            data['service']['shippingDate'] = (
+                data['service']['shippingDate'].replace('/', ''))
+            data['from_address']['departement'] = (
+                data['from_address']['zip'][:2])
+            return {
+                "body": template.render(
+                    service=data['service'],
+                    parcels=data['parcels'],
+                    sender_address=data['from_address'],
+                    receiver_address=data['to_address'],
+                    xmlns=infos['xmlns'],
+                ),
+                "headers": data['auth'],
+                "infos": infos,
+            }
+
+        def request_find_localite():
+            return {
+                "body": template.render(
+                    receiver_address=data['to_address'],
+                    xmlns=infos['xmlns'],
+                ),
+                "headers": data['auth'],
+                "infos": infos,
+            }
+
+        if action == 'demandeImpressionEtiquette':
+            return request_impression_etiquette()
+        elif action == 'findLocalite':
+            return request_find_localite()
+
+    def api(self, action='demandeImpressionEtiquette'):
+        """Provide a dict prefilled with default values."""
+        return self.get_api(action).api_values()
 
     def lookup_label_format(self, label_format="ZPL"):
         """Get a Geodis compatible format of label.

roulier/carriers/geodis/geodis_transport_ws.py

@@ -14,15 +14,13 @@
 class GeodisTransportWs(Transport):
     """Implement Geodis WS communication."""
 
-    GEODIS_WS = "http://espace.geodis.com/geolabel/services/ImpressionEtiquette"  # nopep8
-    GEODIS_WS_TEST = "http://espace.recette.geodis.com/geolabel/services/ImpressionEtiquette"  # nopep8
-
     def send(self, payload):
         """Call this function.
 
         Args:
             payload.body: XML in a string
             payload.header : auth
+            payload.infos: { url: string, xmlns: string}
         Return:
             {
                 response: (Requests.response)
@@ -32,13 +30,13 @@ def send(self, payload):
         """
         body = payload['body']
         headers = payload['headers']
-        is_test = payload['is_test']
-        soap_message = self.soap_wrap(body, headers)
-        response = self.send_request(soap_message, is_test)
+        infos = payload['infos']
+        soap_message = self.soap_wrap(body, headers, infos)
+        response = self.send_request(soap_message, infos)
         log.info('WS response time %s' % response.elapsed.total_seconds())
         return self.handle_response(response)
 
-    def soap_wrap(self, body, auth):
+    def soap_wrap(self, body, auth, infos):
         """Wrap body in a soap:Enveloppe."""
         env = Environment(
             loader=PackageLoader('roulier', '/carriers/geodis/templates'),
@@ -47,13 +45,14 @@ def soap_wrap(self, body, auth):
         template = env.get_template("geodis_soap.xml")
         body_stripped = remove_empty_tags(body)
         header_template = env.get_template("geodis_header.xml")
-        header_xml = header_template.render(auth=auth)
-        data = template.render(body=body_stripped, header=header_xml)
+        header_xml = header_template.render(auth=auth, xmlns=infos['xmlns'])
+        data = template.render(
+            body=body_stripped, header=header_xml, xmlns=infos['xmlns'])
         return data.encode('utf8')
 
-    def send_request(self, body, is_test):
+    def send_request(self, body, infos):
         """Send body to geodis WS."""
-        ws_url = self.GEODIS_WS_TEST if is_test else self.GEODIS_WS
+        ws_url = infos['url']
         return requests.post(
             ws_url,
             headers={
@@ -65,7 +64,6 @@ def send_request(self, body, is_test):
     def handle_500(self, response):
         """Handle reponse in case of ERROR 500 type."""
         # TODO : put a try catch (like wrong server)
-        # no need to extract_body shit here
         log.warning('Geodis error 500')
         xml = get_parts(response)['start']
         obj = objectify.fromstring(xml)
@@ -92,8 +90,13 @@ def extract_soap(response_xml):
             return obj.Body.getchildren()[0]
 
         payload = extract_soap(xml)
-        attachement_cid = payload.codeAttachement.text[len('cid:'):]
-        attachement = parts[attachement_cid]
+        try:
+            # TODO : may be extract this elsewere
+            # rechercheLocalite has no attachment
+            attachement_cid = payload.codeAttachement.text[len('cid:'):]
+            attachement = parts[attachement_cid]
+        except AttributeError:
+            attachement = None
 
         return {
             "body": payload,

roulier/carriers/geodis/templates/geodis_demandeImpressionEtiquette.xml

@@ -1,4 +1,4 @@
-<imp:demandeImpressionEtiquette  xmlns:imp="http://impression.service.web.etiquette.geodis.com">
+<imp:demandeImpressionEtiquette  xmlns:imp="{{xmlns}}">
     <imp:csadep>{{ service.agencyId }}</imp:csadep>
     <imp:csadepteos>{{ service.hubId }}</imp:csadepteos>
     <imp:codclidep>{{ service.customerId }}</imp:codclidep>

roulier/carriers/geodis/templates/geodis_findLocalite.xml

@@ -0,0 +1,7 @@
+<imp:findLocalite  xmlns:imp="{{xmlns}}">
+    {% with address = receiver_address %}
+    <imp:codePays>{{ address.country }}</imp:codePays>
+    <imp:codePostal>{{ address.zip }}</imp:codePostal>
+    <imp:libelle>{{ address.city }}</imp:libelle>
+    {% endwith %}
+</imp:findLocalite>

roulier/carriers/geodis/templates/geodis_header.xml

@@ -1,4 +1,4 @@
-<imp:authenticationInfo xmlns:imp="http://impression.service.web.etiquette.geodis.com">
+<imp:authenticationInfo xmlns:imp="{{xmlns}}">
 	<imp:userName>{{ auth.login }}</imp:userName>
 	<imp:password>{{ auth.password }}</imp:password> 
 </imp:authenticationInfo>

roulier/carriers/geodis/templates/geodis_soap.xml

@@ -1,5 +1,5 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/" xmlns:imp="http://impression.service.web.etiquette.geodis.com"> 
+<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/" xmlns:imp="{{xmlns}}">
 	<soap:Header>
 		{{ header }}
 	</soap:Header>