create 'utils.py' for bits shared between SOAP and REST API

This commit is contained in:
Harald Welte 2021-03-07 10:32:26 +01:00
parent 1da49c7763
commit ef3dd757be
3 changed files with 31 additions and 51 deletions

View File

@ -1,9 +1,7 @@
#!/usr/bin/python #!/usr/bin/python
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
from datetime import datetime, date from datetime import date
from pytz import timezone
import hashlib
import json import json
from lxml import etree from lxml import etree
from zeep import Client from zeep import Client
@ -15,6 +13,8 @@ import io
import logging import logging
import decimal import decimal
from .utils import compute_1c4a_hash, gen_timestamp
__version__ = pkg_resources.get_distribution(__package__).version __version__ = pkg_resources.get_distribution(__package__).version
_logger = logging.getLogger(__name__) _logger = logging.getLogger(__name__)
@ -61,25 +61,6 @@ class Internetmarke(object):
# generate a 1C4A SOAP header # generate a 1C4A SOAP header
def gen_1c4a_hdr(self, partner_id, key_phase, key): def gen_1c4a_hdr(self, partner_id, key_phase, key):
# Compute 1C4A request hash accordig to Section 4 of service description
def compute_1c4a_hash(partner_id, req_ts, key_phase, key):
# trim leading and trailing spaces of each argument
partner_id = partner_id.strip()
req_ts = req_ts.strip()
key_phase = key_phase.strip()
key = key.strip()
# concatenate with "::" separator
inp = "%s::%s::%s::%s" % (partner_id, req_ts, key_phase, key)
# compute MD5 hash as 32 hex nibbles
md5_hex = hashlib.md5(inp.encode('utf8')).hexdigest()
# return the first 8 characters
return md5_hex[:8]
def gen_timestamp():
de_zone = timezone("Europe/Berlin")
de_time = datetime.now(de_zone)
return de_time.strftime("%d%m%Y-%H%M%S")
nsmap={'soapenv': 'http://schemas.xmlsoap.org/soap/envelope/', nsmap={'soapenv': 'http://schemas.xmlsoap.org/soap/envelope/',
'v3':'http://oneclickforpartner.dpag.de'} 'v3':'http://oneclickforpartner.dpag.de'}
r = etree.Element("{http://schemas.xmlsoap.org/soap/envelope/}Header", nsmap = nsmap) r = etree.Element("{http://schemas.xmlsoap.org/soap/envelope/}Header", nsmap = nsmap)

24
inema/utils.py Normal file
View File

@ -0,0 +1,24 @@
import hashlib
from datetime import datetime
from pytz import timezone
def compute_1c4a_hash(partner_id, req_ts, key_phase, key):
"""Compute 1C4A request hash according to Section 4 of service description."""
# trim leading and trailing spaces of each argument
partner_id = partner_id.strip()
req_ts = req_ts.strip()
key_phase = key_phase.strip()
key = key.strip()
# concatenate with "::" separator
inp = "%s::%s::%s::%s" % (partner_id, req_ts, key_phase, key)
# compute MD5 hash as 32 hex nibbles
md5_hex = hashlib.md5(inp.encode('utf8')).hexdigest()
# return the first 8 characters
return md5_hex[:8]
def gen_timestamp():
"""Compute a timestamp as used in the 1C4A and WaPoInt APIs."""
de_zone = timezone("Europe/Berlin")
de_time = datetime.now(de_zone)
return de_time.strftime("%d%m%Y-%H%M%S")

View File

@ -11,12 +11,11 @@
import json import json
import logging import logging
import hashlib
from datetime import datetime
import requests import requests
from lxml import etree from lxml import etree
from pytz import timezone
from .utils import compute_1c4a_hash, gen_timestamp
_logger = logging.getLogger(__name__) _logger = logging.getLogger(__name__)
@ -39,30 +38,6 @@ class WarenpostInt(object):
self.user_token = None self.user_token = None
self.wallet_balance = None self.wallet_balance = None
# FIXME: merge with inema?
@staticmethod
def compute_1c4a_hash(partner_id, req_ts, key_phase, key):
""" Compute 1C4A request hash accordig to Section 4 of service description. """
# trim leading and trailing spaces of each argument
partner_id = partner_id.strip()
req_ts = req_ts.strip()
key_phase = key_phase.strip()
key = key.strip()
# concatenate with "::" separator
inp = "%s::%s::%s::%s" % (partner_id, req_ts, key_phase, key)
# compute MD5 hash as 32 hex nibbles
md5_hex = hashlib.md5(inp.encode('utf8')).hexdigest()
# return the first 8 characters
return md5_hex[:8]
# FIXME: merge with inema?
@staticmethod
def gen_timestamp():
"""Generate a timestamp as used in the Warenpsost International API."""
de_zone = timezone("Europe/Berlin")
de_time = datetime.now(de_zone)
return de_time.strftime("%d%m%Y-%H%M%S")
def gen_headers(self): def gen_headers(self):
"""Generate the HTTP headers required for the API.""" """Generate the HTTP headers required for the API."""
ret = { ret = {
@ -74,8 +49,8 @@ class WarenpostInt(object):
ret['REQUEST_TIMESTAMP'] = '16082018-122210' ret['REQUEST_TIMESTAMP'] = '16082018-122210'
ret['PARTNER_SIGNATURE'] = '9d7c35be' ret['PARTNER_SIGNATURE'] = '9d7c35be'
else: else:
timestamp = self.gen_timestamp() timestamp = gen_timestamp()
sig = self.compute_1c4a_hash(self.partner_id, timestamp, self.key_phase, self.key) sig = compute_1c4a_hash(self.partner_id, timestamp, self.key_phase, self.key)
ret['REQUEST_TIMESTAMP'] = timestamp ret['REQUEST_TIMESTAMP'] = timestamp
ret['PARTNER_SIGNATURE'] = sig ret['PARTNER_SIGNATURE'] = sig
return ret return ret