type:
integration
system:
Python
name:
Python Datafeed Classes
description:
Python classes to decrypt and parse an XML datafeed.
tags:
python django
date:
2008-02-13
version:
0.5
developer:
http://www.themancan.com

python

Description

This is a work in progress implementation of a datafeed parser in Python. This code is not intended to be used in a production environment, but is a useful starting point for your own integrations and for testing.

Installation

- Download the Python modules. - Use them for good, not evil. - Test

Requirements

Code

view.py

"""
Example Django view for receiving and decrypting datafeed.
"""
 
import urllib
from django.http import *
 
from utils.foxycart import FoxyData		# I put foxycart.py in a 'utils' module
import settings	# Store datafeed key in your settings.py
 
def foxyfeed(request):
  if request.POST and 'FoxyData' in request.POST:
    try:
      data = FoxyData.from_crypted_str(urllib.unquote_plus(request.POST['FoxyData']), settings.FOXYCART_DATAFEED_KEY)	# IMPORTANT: unquote_plus is necessary for the non-ASCII binary that FoxyCart sends.
      for transaction in data.transactions:
        pass	# Process each transaction here
 
      return HttpResponse('foxy')
 
    except Exception, e:
	  # Something went wrong, handle the error...
      raise
 
  return HttpResponseForbidden('Unauthorized request.')  # No FoxyData?  Not a POST?  We don't speak that.

foxycart.py

"""
Utilities for decrypting and parsing a FoxyCart datafeed.
"""
from xml.dom.minidom import parseString
from datetime import datetime
 
# Thanks, Wikipedia: http://en.wikipedia.org/wiki/RC4#Implementation
class ARC4:
    def __init__(self, key = None):
        self.state = range(256) # Initialize state array with values 0 .. 255
        self.x = self.y = 0 # Our indexes. x, y instead of i, j
 
        if key is not None:
            self.init(key)
 
    # KSA
    def init(self, key):
        for i in range(256):
            self.x = (ord(key[i % len(key)]) + self.state[i] + self.x) & 0xFF
            self.state[i], self.state[self.x] = self.state[self.x], self.state[i]
        self.x = 0
 
    # PRGA
    def crypt(self, input):
        output = [None]*len(input)
        for i in xrange(len(input)):
            self.x = (self.x + 1) & 0xFF
            self.y = (self.state[self.x] + self.y) & 0xFF
            self.state[self.x], self.state[self.y] = self.state[self.y], self.state[self.x]
            r = self.state[(self.state[self.x] + self.state[self.y]) & 0xFF]
            output[i] = chr(ord(input[i]) ^ r)
        return ''.join(output)
 
 
class FoxyData:
  DateFmt = '%Y-%m-%d'
  DateTimeFmt = '%Y-%m-%d %H:%M:%S'
 
  class Transaction:
    def __init__(self, node):
      def extract_kv_node(node, key_name):
        el = node.getElementsByTagName(key_name)
        return len(el) > 0 and el[0].firstChild.data or ''
 
      self.id = extract_kv_node(node, 'id')
      self.date = datetime.strptime(
       extract_kv_node(node, 'transaction_date'), FoxyData.DateTimeFmt)
      self.customer_id = extract_kv_node(node, 'customer_id')
 
      self.attributes = attrs = {}
      self.items = items = attrs['items'] = []
 
      self.custom_fields = attrs['custom_fields'] = {}
      for custom_field in node.getElementsByTagName('custom_field'):
        self.custom_fields[extract_kv_node(custom_field, 'custom_field_name')] = \
         extract_kv_node(custom_field, 'custom_field_value')
 
      self.transaction_details = attrs['detail'] = []
      for details in node.getElementsByTagName('transaction_detail'):
        item = {'product_code': extract_kv_node(details, 'product_code')}
 
        for key in ['subscription_startdate', 'next_transaction_date']:
          date_str = extract_kv_node(details, key)
          try:
            item[key] = datetime.strptime(date_str, FoxyData.DateFmt)
          except ValueError:
            item[key] = date_str
 
        detail = item['detail'] = {}
        for detail_opt in details.getElementsByTagName('transaction_detail_option'):
          detail[extract_kv_node(detail_opt, 'product_option_name')] = \
           extract_kv_node(detail_opt, 'product_option_value')
 
        items.append(item)
 
  def __init__(self, markup):
    self.markup = markup
    self.doc = parseString(self.markup)
    self.transactions = []
 
    for transaction in self.doc.getElementsByTagName('transaction'):
      self.transactions.append(FoxyData.Transaction(transaction))
 
  def __str__(self):
    return str(self.markup)
 
 
  @classmethod
  def from_str(self, data_str):
    return FoxyData(data_str)
 
  """
  Given a string containing RC4-crypted FoxyCart datafeed XML and the
  cryptographic key, decrypt the contents and create a FoxyData object
  containing all of the Transactions in the data feed.
  """
  @classmethod
  def from_crypted_str(self, data_str, crypt_key):
    a = ARC4(crypt_key)
    return FoxyData.from_str(a.crypt(data_str))
 
  def __len__(self):
    return len(self.transactions)

foxycart_test.py

"""
Unit test for foxycart.py.
"""
 
import unittest
from foxycart import *
 
class Constants:
  pass
 
class FoxyDataVectorTest(unittest.TestCase):
  def _test_it_hard(self, vector):
    self.assertEqual(1, len(vector), 'expected one transaction')
 
    tx = vector.transactions[0]
    self.assertEqual('616', tx.id)
    self.assert_(tx.date)
    self.assertEqual("2007-05-04 20:53:57", tx.date.strftime(FoxyData.DateTimeFmt))
    self.assertEqual('122', tx.customer_id)
 
    self.assertEqual(1, len(tx.items), 'expected one item')
    self.assertEqual(2, len(tx.custom_fields), 'expected 2 custom fields')
    self.assert_('My_Cool_Text' in tx.custom_fields.keys(), 'missing custom field')
    self.assertEqual('Value123', tx.custom_fields['My_Cool_Text'], 'missing custom field value')
 
 
    self.assert_('Another_Custom_Field' in tx.custom_fields.keys(), 'missing custom field')
    self.assertEqual('10', tx.custom_fields['Another_Custom_Field'], 'missing custom field value')
 
    item = tx.items.pop()
    self.assertEqual('abc123', item['product_code'])
 
    detail = item['detail']
    self.assertEqual(1, len(detail))
    self.assertEqual('blue', detail['color'])
 
    self.assert_(item['subscription_start_date'])
    self.assertEqual('2007-07-07',
     item['subscription_start_date'].strftime(FoxyData.DateFmt))
    self.assert_(item['next_transaction_date'])
    self.assertEqual('2007-08-07',
     item['next_transaction_date'].strftime(FoxyData.DateFmt))
    self.assertEqual()
 
  def test_from_str(self):
    vector = FoxyData.from_str(Constants.subscription_xml)
    self._test_it_hard(vector)
 
  def test_from_crypted_str(self):
    crypted_str = ARC4(Constants.SECRET_KEY).crypt(Constants.subscription_xml)
    vector = FoxyData.from_crypted_str(crypted_str, Constants.SECRET_KEY)
    self._test_it_hard(vector)
 
 
Constants.SECRET_KEY = 'abc123akp8ak7898a,.aoeueaouaoeuaoeu'
Constants.subscription_xml = """<?xml version='1.0' standalone='yes'?>
<foxydata>
  <datafeed_version>XML FoxyCart Version 0.7</datafeed_version>
  <transactions>
    <transaction>
      <id>616</id>
      <transaction_date>2007-05-04 20:53:57</transaction_date>
      <customer_id>122</customer_id>
      <customer_first_name>John</customer_first_name>
      <customer_last_name>Doe</customer_last_name>
      <customer_address1>12345 Any Street</customer_address1>
      <customer_address2></customer_address2>
      <customer_city>Any City</customer_city>
      <customer_state>TN</customer_state>
      <customer_postal_code>37013</customer_postal_code>
      <customer_country>US</customer_country>
      <customer_phone>(123) 456-7890</customer_phone>
      <customer_email>someone@somewhere.com</customer_email>
      <customer_ip>71.228.237.177</customer_ip>
      <shipping_first_name>John</shipping_first_name>
      <shipping_last_name>Doe</shipping_last_name>
      <shipping_address1>1234 Any Street</shipping_address1>
      <shipping_address2></shipping_address2>
      <shipping_city>Some City</shipping_city>
      <shipping_state>TN</shipping_state>
      <shipping_postal_code>37013</shipping_postal_code>
      <shipping_country>US</shipping_country>
      <shipping_phone></shipping_phone>
      <shipping_service_description>UPS: Ground</shipping_service_description>
      <purchase_order></purchase_order>
      <product_total>20.00</product_total>
      <tax_total>0.00</tax_total>
      <shipping_total>4.38</shipping_total>
      <order_total>24.38</order_total>
      <order_total>24.38</order_total>
      <customer_password>1aab23051b24582c5dc8e23fc595d505</customer_password>
      <custom_fields>
        <custom_field>
          <custom_field_name>My_Cool_Text</custom_field_name>
          <custom_field_value>Value123</custom_field_value>
        </custom_field>
        <custom_field>
          <custom_field_name>Another_Custom_Field</custom_field_name>
          <custom_field_value>10</custom_field_value>
        </custom_field>
      </custom_fields>
      <transaction_details>
        <transaction_detail>
          <product_name>foo</product_name>
          <product_price>20.00</product_price>
          <product_quantity>1</product_quantity>
          <product_weight>0.10</product_weight>
          <product_code>abc123</product_code>
          <subscription_frequency>1m</subscription_frequency>
          <subscription_startdate>2007-07-07</subscription_startdate>
          <next_transaction_date>2007-08-07</next_transaction_date>
          <shipto>John Doe</shipto>
          <category_description>Default for all products</category_description>
          <category_code>DEFAULT</category_code>
          <product_delivery_type>shipped</product_delivery_type>
          <transaction_detail_options>
            <transaction_detail_option>
              <product_option_name>color</product_option_name>
              <product_option_value>blue</product_option_value>
              <price_mod></price_mod>
              <weight_mod></weight_mod>
            </transaction_detail_option>
          </transaction_detail_options>
        </transaction_detail>
      </transaction_details>
      <shipto_addresses>
        <shipto_address>
          <address_name>John Doe</address_name>
          <shipto_first_name>John</shipto_first_name>
          <shipto_last_name>Doe</shipto_last_name>
          <shipto_address1>2345 Some Address</shipto_address1>
          <shipto_address2></shipto_address2>
          <shipto_city>Some City</shipto_city>
          <shipto_state>TN</shipto_state>
          <shipto_postal_code>37013</shipto_postal_code>
          <shipto_country>US</shipto_country>
          <shipto_shipping_service_description>DHL: Next Afternoon</shipto_shipping_service_description>
          <shipto_subtotal>52.15</shipto_subtotal>
          <shipto_tax_total>6.31</shipto_tax_total>
          <shipto_shipping_total>15.76</shipto_shipping_total>
          <shipto_total>74.22</shipto_total>
          <shipto_custom_fields>
            <shipto_custom_field>
              <shipto_custom_field_name>My_Custom_Info</shipto_custom_field_name>
              <shipto_custom_field_value>john's stuff</shipto_custom_field_value>
            </shipto_custom_field>
            <shipto_custom_field>
              <shipto_custom_field_name>More_Custom_Info</shipto_custom_field_name>
              <shipto_custom_field_value>more of john's stuff</shipto_custom_field_value>
            </shipto_custom_field>
          </shipto_custom_fields>
        </shipto_address>
      </shipto_addresses>
    </transaction>
  </transactions>
</foxydata>
"""
 
if __name__ == '__main__':
  unittest.main()
 
integration/python.txt · Last modified: 2009/10/22 15:18 by 128.97.245.18
 
Recent changes RSS feed Creative Commons License Donate Powered by PHP Valid XHTML 1.0 Valid CSS Driven by DokuWiki