Package: quodlibet-plugins
Version: 20080329-1
Severity: normal
albumart.py no longer works due to Amazon deprecating the old version
(3.x) of their API. Attached are a new version of _amazon.py (from
http://pyaws.sourceforge.net/) and a slightly modified version of
albumart.py to work with the new module.
-- System Information:
Debian Release: lenny/sid
APT prefers testing
APT policy: (600, 'testing'), (500, 'unstable'), (1, 'experimental')
Architecture: amd64 (x86_64)
Kernel: Linux 2.6.24-1-amd64 (SMP w/1 CPU core)
Locale: LANG=en_US.UTF-8, LC_CTYPE=en_US.UTF-8 (charmap=UTF-8)
Shell: /bin/sh linked to /bin/bash
Versions of packages quodlibet-plugins depends on:
ii exfalso 1.0-2 audio tag editor for GTK+
quodlibet-plugins recommends no packages.
-- no debconf information
#Copyright 2005 Eduardo Gonzalez, Niklas Janlert
#Amazon API code by Mark Pilgrim
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation
import os
import sys
import urllib
import threading
from cStringIO import StringIO
import gtk
import gobject
import pango
import util
import qltk
import config
from plugins.songsmenu import SongsMenuPlugin
if sys.version_info < (2, 4): from sets import Set as set
try:
import amazon
except ImportError:
import _amazon as amazon
class AlbumArtWindow(gtk.Window):
def __init__(self, songs):
gtk.Window.__init__(self)
self.set_border_width(12)
self.set_title("AlbumArt")
self.set_default_size(650, 350)
#TreeView stuff
self.liststore = liststore = gtk.ListStore(object, str)
treeview = gtk.TreeView(liststore)
treeview.set_headers_visible(False)
selection = treeview.get_selection()
selection.set_mode(gtk.SELECTION_SINGLE)
selection.connect("changed", self.__preview)
rend = gtk.CellRendererPixbuf()
def cell_data(column, cell, model, iter):
cell.set_property("pixbuf", model[iter][0]["thumb"])
tvcol1 = gtk.TreeViewColumn("Pixbuf", rend)
tvcol1.set_cell_data_func(rend, cell_data)
tvcol1.set_sizing(gtk.TREE_VIEW_COLUMN_GROW_ONLY)
rend.set_property('xpad', 2)
rend.set_property('ypad', 2)
rend.set_property('width', 56)
rend.set_property('height', 56)
treeview.append_column(tvcol1)
rend = gtk.CellRendererText()
rend.set_property("ellipsize", pango.ELLIPSIZE_END)
tvcol2 = gtk.TreeViewColumn("Info", rend, markup=1)
treeview.append_column(tvcol2)
sw = gtk.ScrolledWindow()
sw.set_policy(gtk.POLICY_NEVER, gtk.POLICY_AUTOMATIC)
sw.set_shadow_type(gtk.SHADOW_IN)
sw.add(treeview)
#Image frame and save button
self.image = image = gtk.Image()
frame = gtk.Frame()
frame.set_shadow_type(gtk.SHADOW_IN)
frame.add(image)
vbox = gtk.VBox(spacing=5)
vbox.pack_start(frame)
self.button = button = gtk.Button(stock=gtk.STOCK_SAVE)
button.set_sensitive(False)
def save_cb(button, combo):
model, path = selection.get_selected()
data = model[path][0]["cover_data"]
fname = self.__get_fname(songs, combo)
self.__save_cover(data, fname)
combo = gtk.combo_box_new_text()
try: set_fn = config.get("plugins", "cover_fn")
except: set_fn = ".folder.jpg"
active = -1
for i, fn in enumerate([".folder.jpg", "folder.jpg", "cover.jpg"]):
combo.append_text(fn)
if fn == set_fn: active = i
if active == -1:
combo.append_text(set_fn)
combo.set_active(len(combo.get_model()) - 1)
else: combo.set_active(active)
button.connect("clicked", save_cb, combo)
bbox = gtk.HButtonBox()
bbox.pack_start(combo)
bbox.pack_start(button, expand=False, fill=False)
bbox.set_layout(gtk.BUTTONBOX_SPREAD)
vbox.pack_start(bbox, expand=False, fill=False)
hpaned = gtk.HPaned()
hpaned.pack1(sw)
hpaned.pack2(vbox)
hpaned.set_position(300)
self.add(hpaned)
thread = threading.Thread(target=self.__search, args=(songs,))
thread.setDaemon(True)
thread.start()
self.show_all()
def __search(self, songs):
artist = songs[0]('artist')
album = songs[0]('album')
amazon.setLicenseKey("0RKH4ZH1JCFZHMND91G2")
try:
query = album + " " + artist
query = query.encode("latin1", 'replace')
bags = amazon.ItemSearch(
query, SearchIndex="Music", ResponseGroup="Medium")
except (amazon.AWSException, KeyError), msg:
dialog = qltk.Message(gtk.MESSAGE_ERROR, None, "Search error", msg)
dialog.connect('response', self.__destroy_cb)
gobject.idle_add(dialog.show)
except UnicodeEncodeError, msg:
dialog = qltk.Message(gtk.MESSAGE_ERROR, None, "Encoding error",
msg)
dialog.connect('response', self.__destroy_cb)
gobject.idle_add(dialog.show)
else:
# Just keep the top 10 matches
i = 0
for bag in bags:
i += 1
if i > 10: break
gobject.idle_add(self.__add_bag, self.liststore, bag)
def __destroy_cb(self, widget, *args):
widget.destroy()
self.destroy()
def __add_bag(self, model, bag):
# Text part
title = util.escape(getattr(bag, "Title", ""))
# print >>sys.stderr, bag.Artist, dir(bag.Artist)
artist = getattr(bag, "Artist", "")
# if isinstance(artist, list):
# artist = ", ".join(artist)
artist = util.escape(artist)
if hasattr(bag, "ReleaseDate"):
date = "(%s)" %util.escape(bag.ReleaseDate)
else:
date = ""
markup = "<i><b>%s</b></i> %s\n%s" %(title, date, artist)
item = {"bag": bag, "thumb": None, "thumb_data": ""}
iter = model.append([item, markup])
if hasattr(bag, "SmallImage"):
# Image part
urlinfo = urllib.urlopen(bag.SmallImage.URL)
sock = urlinfo.fp._sock
sock.setblocking(0)
data = StringIO()
loader = gtk.gdk.PixbufLoader()
loader.connect("closed", self.__got_thumb_cb, data, item, model, iter)
gobject.io_add_watch(
sock, gobject.IO_IN | gobject.IO_ERR | gobject.IO_HUP,
self.__copy_image, loader, data)
def __got_thumb_cb(self, loader, data, item, model, iter):
cover = loader.get_pixbuf()
if cover.get_width() > 1:
w = h = 48
cover = cover.scale_simple(w, h, gtk.gdk.INTERP_NEAREST)
thumb = gtk.gdk.Pixbuf(
gtk.gdk.COLORSPACE_RGB, True, 8, w + 2, h + 2)
thumb.fill(0x000000ff)
cover.copy_area(0, 0, w, h, thumb, 1, 1)
item["thumb"] = thumb
item["thumb_data"] = data.getvalue()
model.row_changed(model.get_path(iter), iter)
def __preview(self, selection):
model, path = selection.get_selected()
item = model[path][0]
self.image.hide()
self.button.set_sensitive(False)
if item["thumb"]: # If there exists no thumbnail, then nothing bigger.
if "cover" not in item:
self.__get_cover(item, item["bag"].LargeImage.URL)
else:
self.image.set_from_pixbuf(item["cover"])
self.image.show()
self.button.set_sensitive(True)
def __get_cover(self, item, url):
data = StringIO()
urlinfo = urllib.urlopen(url)
sock = urlinfo.fp._sock
sock.setblocking(0)
loader = gtk.gdk.PixbufLoader()
gobject.io_add_watch(
sock, gobject.IO_IN | gobject.IO_ERR | gobject.IO_HUP,
self.__copy_image, loader, data)
loader.connect("closed", self.__got_cover_cb, data, item, url)
def update(loader, x, y, w, h, image):
if (w, h) > (1, 1):
image.set_from_pixbuf(loader.get_pixbuf())
image.show()
loader.connect("area-updated", update, self.image)
def __got_cover_cb(self, loader, data, item, url):
cover = loader.get_pixbuf()
# For some reason we get a 1x1 image if the given size didn't exist
if cover.get_width() > 1:
item["cover"] = cover
item["cover_data"] = data.getvalue()
self.image.set_from_pixbuf(item["cover"])
self.button.set_sensitive(True)
elif url == item["bag"].LargeImage.URL:
self.__get_cover(item, item["bag"].MediumImage.URL)
else:
item["cover"] = item["thumb"]
item["cover_data"] = item["thumb_data"]
self.image.set_from_pixbuf(item["cover"])
self.button.set_sensitive(True)
def __copy_image(self, src, condition, loader, data):
if condition in (gobject.IO_ERR, gobject.IO_HUP):
loader.close()
src.close()
return False
else: # Read
buf = src.recv(1024)
if buf:
loader.write(buf)
data.write(buf)
return True # Run again
else:
loader.close()
src.close()
return False
def __save_cover(self, data, fname):
if os.path.exists(fname) and not qltk.ConfirmAction(None,
"File exists", "The file <b>%s</b> already exists."
"\n\nOverwrite?" %util.escape(fname)).run():
return
f = open(fname, "w")
f.write(data)
f.close()
self.destroy()
def __get_fname(self, songs, combo):
append = combo.get_model()[(combo.get_active(),)][0]
dirname = songs[0]("~dirname")
fname = os.path.join(dirname, append)
#print "Will save to", fname
config.set("plugins", "cover_fn", append)
return fname
class DownloadAlbumArt(SongsMenuPlugin):
PLUGIN_ID = "Download Album art"
PLUGIN_NAME = _("Download Album Art")
PLUGIN_DESC = "Downloads album covers from Amazon.com"
PLUGIN_ICON = gtk.STOCK_FIND
PLUGIN_VERSION = "0.25"
def PluginPreferences(parent):
b = gtk.Button("Visit Amazon.com")
b.connect('clicked', lambda s: util.website('http://www.amazon.com/'))
return b
PluginPreferences = staticmethod(PluginPreferences)
plugin_album = AlbumArtWindow
"""Python wrapper for AWS E-Commerce Serive APIs.
Based upon pyamazon (http://www.josephson.org/projects/pyamazon/) with
efforts to meet the latest AWS specification.
The Amazon's web APIs specication is described here:
http://www.amazon.com/webservices
You need a Amazon-provided license key to use these services.
Follow the link above to get one. These functions will look in
several places (in this order) for the license key:
- the "license_key" argument of each function
- the module-level LICENSE_KEY variable (call setLicense once to set it)
- an environment variable called AMAZON_LICENSE_KEY
- foo would return the python object, XMLfoo returns the DOM object
License: Python Software Foundation License
"""
import os, urllib, string, inspect
from xml.dom import minidom
__author__ = "Kun Xi < [EMAIL PROTECTED] >"
__version__ = "0.2.0"
__license__ = "Python Software Foundation"
"""Package-wide variables:
"""
LICENSE_KEY = None;
HTTP_PROXY = None
LOCALE = "us"
VERSION = "2007-02-22"
__supportedLocales = {
None : "webservices.amazon.com",
"us" : "webservices.amazon.com",
"uk" : "webservices.amazon.co.uk",
"de" : "webservices.amazon.de",
"jp" : "webservices.amazon.co.jp",
"fr" : "webservices.amazon.fr",
"ca" : "webservices.amazon.ca"
}
__licenseKeys = (
(lambda key: key),
(lambda key: LICENSE_KEY),
(lambda key: os.environ.get('AWS_LICENSE_KEY', None))
)
class AWSException(Exception) :
'''Base class for all AWS exceptions'''
pass
class NoLicenseKey(AWSException) : pass
class BadLocale(AWSException) : pass
# Runtime exception
class ExactParameterRequirement(AWSException): pass
class ExceededMaximumParameterValues(AWSException): pass
class InsufficientParameterValues(AWSException): pass
class InternalError(AWSException): pass
class InvalidEnumeratedParameter(AWSException): pass
class InvalidISO8601Time(AWSException): pass
class InvalidOperationForMarketplace(AWSException): pass
class InvalidOperationParameter(AWSException): pass
class InvalidParameterCombination(AWSException): pass
class InvalidParameterValue(AWSException): pass
class InvalidResponseGroup(AWSException): pass
class InvalidServiceParameter(AWSException): pass
class InvalidSubscriptionId(AWSException): pass
class InvalidXSLTAddress(AWSException): pass
class MaximumParameterRequirement(AWSException): pass
class MinimumParameterRequirement(AWSException): pass
class MissingOperationParameter(AWSException): pass
class MissingParameterCombination(AWSException): pass
class MissingParameters(AWSException): pass
class MissingParameterValueCombination(AWSException): pass
class MissingServiceParameter(AWSException): pass
class ParameterOutOfRange(AWSException): pass
class ParameterRepeatedInRequest(AWSException): pass
class RestrictedParameterValueCombination(AWSException): pass
class XSLTTransformationError(AWSException): pass
class Bag :
'''Wrapper class for DOM nodes'''
pass
# Utilities functions
def setLocale(locale):
"""set locale
if unsupported locale is set, BadLocale is raised."""
global LOCALE
if not __supportedLocales.has_key(locale):
raise BadLocale, ("Unsupported locale. Locale must be one of:
%s" %
', '.join([x for x in __supportedLocales.keys() if x]))
LOCALE = locale
def getLocale():
"""get locale"""
return LOCALE
def setLicenseKey(license_key=None):
"""set license key
license key can come from any number of locations;
see module docs for search order.
if no license key is specified, BadLocale is raised."""
global LICENSE_KEY
for get in __licenseKeys:
rc = get(license_key)
if rc:
LICENSE_KEY = rc;
return;
raise NoLicenseKey, ("Please get the license key from
http://www.amazon.com/webservices";)
def getLicenseKey():
"""get license key
if no license key is specified, BadLocale is raised."""
if not LICENSE_KEY:
raise NoLicenseKey, ("Please get the license key from
http://www.amazon.com/webservices";)
return LICENSE_KEY
def getVersion():
"""get version"""
return VERSION
def buildRequest(argv):
"""Build the REST request URL from argv,
all key, value pairs in argv are quoted."""
url = "http://"; + __supportedLocales[getLocale()] +
"/onca/xml?Service=AWSECommerceService&"
return url + '&'.join(['%s=%s' % (k,urllib.quote(str(v))) for (k,v) in
argv.items() if v])
def buildException(els):
"""Build the exception from the returned DOM node
Only the first exception is raised."""
error = els[0]
class_name = error.childNodes[0].firstChild.data[4:]
msg = error.childNodes[1].firstChild.data
e = globals()[ class_name ](msg)
return e
def query(url):
"""Send the query url and return the DOM
Exception is raised if there is errors"""
u = urllib.FancyURLopener(HTTP_PROXY)
usock = u.open(url)
dom = minidom.parse(usock)
usock.close()
errors = dom.getElementsByTagName('Error')
if errors:
e = buildException(errors)
raise e
return dom
def rawObject(XMLSearch, arguments, kwItem, plugins=None):
'''Return a unique object'''
dom = XMLSearch(** arguments)
return unmarshal(dom.getElementsByTagName(kwItem).item(0), plugins)
def rawIterator(XMLSearch, arguments, kwItems, plugins=None):
'''Return list of objects'''
dom = XMLSearch(** arguments)
items = unmarshal(dom.getElementsByTagName(kwItems).item(0), plugins,
wrappedIterator())
return items
class wrappedIterator(list):
'''Return inherited list object,
we may add more attributes later'''
pass
class pagedIterator:
'''Return a page-based iterator'''
def __init__(self, XMLSearch, arguments, kwPage, kwItems, plugins=None):
"""XMLSearch: the callback function that returns the DOM
arguments: the arguments for XMLSearch
kwPage, kwItems: Tag name of Page, Items to organize the object
plugins: please check unmarshal
"""
self.__search = XMLSearch
self.__arguments = arguments
self.__keywords ={'Page':kwPage, 'Items':kwItems}
self.__plugins = plugins
self.__page = arguments[kwPage] or 1
self.__index = 0
dom = self.__search(** self.__arguments)
self.__items =
unmarshal(dom.getElementsByTagName(kwItems).item(0), plugins, wrappedIterator())
try:
self.__len =
int(dom.getElementsByTagName("TotalResults").item(0).firstChild.data)
except AttributeError, e:
self.__len = len(self.__items)
def __len__(self):
return self.__len
def __iter__(self):
return self
def next(self):
if self.__index < self.__len:
self.__index = self.__index + 1
return self.__getitem__(self.__index-1)
else:
raise StopIteration
def __getitem__(self, key):
try:
num = int(key)
except TypeError, e:
raise e
if num >= self.__len:
raise IndexError
page = num / 10 + 1
index = num % 10
if page != self.__page:
self.__arguments[self.__keywords['Page']] = page
dom = self.__search(** self.__arguments)
self.__items =
unmarshal(dom.getElementsByTagName(self.__keywords['Items']).item(0),
self.__plugins, wrappedIterator())
self.__page = page
return self.__items[index]
def unmarshal(element, plugins=None, rc=None):
"""Return the Bag object with attributes populated using DOM element
element: the root of the DOM element we are interested in
plugins: callback functions to fine-tune the object structure
rc: parent object, used in the recursive call
This core function is inspired by Mark Pilgrim ([EMAIL PROTECTED])
with some enhancement. Each node.tagName is evalued by plugins' callback
functions:
if plugins['isBypassed'] is true:
this elment is ignored
if plugins['isPivoted'] is true:
this children of this elment is moved to grandparents
this object is ignored.
if plugins['isCollective'] is true:
this elment is mapped to []
if plugins['isCollected'] is true:
this children of elment is appended to grandparent
this object is ignored.
"""
if(rc == None):
rc = Bag()
if(plugins == None):
plugins = {}
childElements = [e for e in element.childNodes if isinstance(e,
minidom.Element)]
if childElements:
for child in childElements:
key = child.tagName
if hasattr(rc, key):
if type(getattr(rc, key)) <> type([]):
setattr(rc, key, [getattr(rc, key)])
setattr(rc, key, getattr(rc, key) +
[unmarshal(child, plugins)])
elif isinstance(child, minidom.Element):
if plugins.has_key('isPivoted') and
plugins['isPivoted'](child.tagName):
unmarshal(child, plugins, rc)
elif plugins.has_key('isBypassed') and
plugins['isBypassed'](child.tagName):
continue
elif plugins.has_key('isCollective') and
plugins['isCollective'](child.tagName):
setattr(rc, key, unmarshal(child,
plugins, wrappedIterator([])))
elif plugins.has_key('isCollected') and
plugins['isCollected'](child.tagName):
rc.append(unmarshal(child, plugins))
else:
setattr(rc, key, unmarshal(child,
plugins))
else:
rc = "".join([e.data for e in element.childNodes if
isinstance(e, minidom.Text)])
return rc
# User interfaces
def ItemLookup(ItemId, IdType=None, SearchIndex=None, MerchantId=None,
Condition=None, DeliveryMethod=None, ISPUPostalCode=None, OfferPage=None,
ReviewPage=None, VariationPage=None, ResponseGroup=None, AWSAccessKeyId=None):
'''ItemLookup in ECS'''
argv = inspect.getargvalues(inspect.currentframe())[-1]
plugins = {'isPivoted': lambda x: x == 'ItemAttributes',
'isCollective': lambda x: x == 'Items',
'isCollected': lambda x: x == 'Item'}
return pagedIterator(XMLItemLookup, argv, 'OfferPage', 'Items', plugins)
def XMLItemLookup(ItemId, IdType=None, SearchIndex=None, MerchantId=None,
Condition=None, DeliveryMethod=None, ISPUPostalCode=None, OfferPage=None,
ReviewPage=None, VariationPage=None, ResponseGroup=None, AWSAccessKeyId=None):
'''DOM representation of ItemLookup in ECS'''
Operation = "ItemLookup"
AWSAccessKeyId = AWSAccessKeyId or LICENSE_KEY
argv = inspect.getargvalues(inspect.currentframe())[-1]
return query(buildRequest(argv))
def ItemSearch(Keywords, SearchIndex="Blended", Availability=None, Title=None,
Power=None, BrowseNode=None, Artist=None, Author=None, Actor=None,
Director=None, AudienceRating=None, Manufacturer=None, MusicLabel=None,
Composer=None, Publisher=None, Brand=None, Conductor=None, Orchestra=None,
TextStream=None, ItemPage=None, Sort=None, City=None, Cuisine=None,
Neighborhood=None, MinimumPrice=None, MaximumPrice=None, MerchantId=None,
Condition=None, DeliveryMethod=None, ResponseGroup=None, AWSAccessKeyId=None):
'''ItemSearch in ECS'''
argv = inspect.getargvalues(inspect.currentframe())[-1]
plugins = {'isPivoted': lambda x: x == 'ItemAttributes',
'isCollective': lambda x: x == 'Items',
'isCollected': lambda x: x == 'Item'}
return pagedIterator(XMLItemSearch, argv, "ItemPage", 'Items', plugins)
def XMLItemSearch(Keywords, SearchIndex="Blended", Availability=None,
Title=None, Power=None, BrowseNode=None, Artist=None, Author=None, Actor=None,
Director=None, AudienceRating=None, Manufacturer=None, MusicLabel=None,
Composer=None, Publisher=None, Brand=None, Conductor=None, Orchestra=None,
TextStream=None, ItemPage=None, Sort=None, City=None, Cuisine=None,
Neighborhood=None, MinimumPrice=None, MaximumPrice=None, MerchantId=None,
Condition=None, DeliveryMethod=None, ResponseGroup=None, AWSAccessKeyId=None):
'''DOM representation of ItemSearch in ECS'''
Operation = "ItemSearch"
AWSAccessKeyId = AWSAccessKeyId or LICENSE_KEY
argv = inspect.getargvalues(inspect.currentframe())[-1]
return query(buildRequest(argv))
def SimilarityLookup(ItemId, SimilarityType=None, MerchantId=None,
Condition=None, DeliveryMethod=None, ResponseGroup=None, AWSAccessKeyId=None):
'''SimilarityLookup in ECS'''
argv = inspect.getargvalues(inspect.currentframe())[-1]
plugins = {'isPivoted': lambda x: x == 'ItemAttributes',
'isCollective': lambda x: x == 'Items',
'isCollected': lambda x: x == 'Item'}
return rawIterator(XMLSimilarityLookup, argv, 'Items', plugins)
def XMLSimilarityLookup(ItemId, SimilarityType=None, MerchantId=None,
Condition=None, DeliveryMethod=None, ResponseGroup=None, AWSAccessKeyId=None):
'''DOM representation of SimilarityLookup in ECS'''
Operation = "SimilarityLookup"
AWSAccessKeyId = AWSAccessKeyId or LICENSE_KEY
argv = inspect.getargvalues(inspect.currentframe())[-1]
return query(buildRequest(argv))
# List Operations
def ListLookup(ListType, ListId, ProductPage=None, ProductGroup=None,
Sort=None, MerchantId=None, Condition=None, DeliveryMethod=None,
ResponseGroup=None, AWSAccessKeyId=None):
'''ListLookup in ECS'''
argv = inspect.getargvalues(inspect.currentframe())[-1]
plugins = {'isPivoted': lambda x: x == 'ItemAttributes',
'isCollective': lambda x: x == 'Lists',
'isCollected': lambda x: x == 'List'}
return pagedIterator(XMLListLookup, argv, 'ProductPage', 'Lists',
plugins)
def XMLListLookup(ListType, ListId, ProductPage=None, ProductGroup=None,
Sort=None, MerchantId=None, Condition=None, DeliveryMethod=None,
ResponseGroup=None, AWSAccessKeyId=None):
'''DOM representation of ListLookup in ECS'''
Operation = "ListLookup"
AWSAccessKeyId = AWSAccessKeyId or LICENSE_KEY
argv = inspect.getargvalues(inspect.currentframe())[-1]
return query(buildRequest(argv))
def ListSearch(ListType, Name=None, FirstName=None, LastName=None, Email=None,
City=None, State=None, ListPage=None, ResponseGroup=None, AWSAccessKeyId=None):
'''ListSearch in ECS'''
argv = inspect.getargvalues(inspect.currentframe())[-1]
plugins = {'isPivoted': lambda x: x == 'ItemAttributes',
'isCollective': lambda x: x == 'Lists',
'isCollected': lambda x: x == 'List'}
return pagedIterator(XMLListSearch, argv, 'ListPage', 'Lists', plugins)
def XMLListSearch(ListType, Name=None, FirstName=None, LastName=None,
Email=None, City=None, State=None, ListPage=None, ResponseGroup=None,
AWSAccessKeyId=None):
'''DOM representation of ListSearch in ECS'''
Operation = "ListSearch"
AWSAccessKeyId = AWSAccessKeyId or LICENSE_KEY
argv = inspect.getargvalues(inspect.currentframe())[-1]
return query(buildRequest(argv))
#Remote Shopping Cart Operations
def CartCreate(Items, Quantities, ResponseGroup=None, AWSAccessKeyId=None):
'''CartCreate in ECS'''
argv = inspect.getargvalues(inspect.currentframe())[-1]
return __cartOperation(XMLCartCreate, argv)
def XMLCartCreate(Items, Quantities, ResponseGroup=None, AWSAccessKeyId=None):
'''DOM representation of CartCreate in ECS'''
Operation = "CartCreate"
AWSAccessKeyId = AWSAccessKeyId or LICENSE_KEY
argv = inspect.getargvalues(inspect.currentframe())[-1]
for x in ('Items', 'Quantities'):
del argv[x]
__fromListToItems(argv, Items, 'ASIN', Quantities)
return query(buildRequest(argv))
def CartAdd(Cart, Items, Quantities, ResponseGroup=None, AWSAccessKeyId=None):
'''CartAdd in ECS'''
argv = inspect.getargvalues(inspect.currentframe())[-1]
return __cartOperation(XMLCartAdd, argv)
def XMLCartAdd(Cart, Items, Quantities, ResponseGroup=None,
AWSAccessKeyId=None):
'''DOM representation of CartAdd in ECS'''
Operation = "CartAdd"
AWSAccessKeyId = AWSAccessKeyId or LICENSE_KEY
CartId = Cart.CartId
HMAC = Cart.HMAC
argv = inspect.getargvalues(inspect.currentframe())[-1]
for x in ('Items', 'Cart', 'Quantities'):
del argv[x]
__fromListToItems(argv, Items, 'ASIN', Quantities)
return query(buildRequest(argv))
def CartGet(Cart, ResponseGroup=None, AWSAccessKeyId=None):
'''CartGet in ECS'''
argv = inspect.getargvalues(inspect.currentframe())[-1]
return __cartOperation(XMLCartGet, argv)
def XMLCartGet(Cart, ResponseGroup=None, AWSAccessKeyId=None):
'''DOM representation of CartGet in ECS'''
Operation = "CartGet"
AWSAccessKeyId = AWSAccessKeyId or LICENSE_KEY
CartId = Cart.CartId
HMAC = Cart.HMAC
argv = inspect.getargvalues(inspect.currentframe())[-1]
del argv['Cart']
return query(buildRequest(argv))
def CartModify(Cart, Items, Actions, ResponseGroup=None, AWSAccessKeyId=None):
'''CartModify in ECS'''
argv = inspect.getargvalues(inspect.currentframe())[-1]
return __cartOperation(XMLCartModify, argv)
def XMLCartModify(Cart, Items, Actions, ResponseGroup=None,
AWSAccessKeyId=None):
'''DOM representation of CartModify in ECS'''
Operation = "CartModify"
AWSAccessKeyId = AWSAccessKeyId or LICENSE_KEY
CartId = Cart.CartId
HMAC = Cart.HMAC
argv = inspect.getargvalues(inspect.currentframe())[-1]
for x in ('Cart', 'Items', 'Actions'):
del argv[x]
__fromListToItems(argv, Items, 'CartItemId', Actions)
return query(buildRequest(argv))
def CartClear(Cart, ResponseGroup=None, AWSAccessKeyId=None):
'''CartClear in ECS'''
argv = inspect.getargvalues(inspect.currentframe())[-1]
return __cartOperation(XMLCartClear, argv)
def XMLCartClear(Cart, ResponseGroup=None, AWSAccessKeyId=None):
'''DOM representation of CartClear in ECS'''
Operation = "CartClear"
AWSAccessKeyId = AWSAccessKeyId or LICENSE_KEY
CartId = Cart.CartId
HMAC = Cart.HMAC
argv = inspect.getargvalues(inspect.currentframe())[-1]
del argv['Cart']
return query(buildRequest(argv))
def __fromListToItems(argv, items, id, actions):
'''Convert list to AWS REST arguments'''
for i in range(len(items)):
argv["Item.%d.%s" % (i+1, id)] = getattr(items[i], id);
action = actions[i]
if isinstance(action, int):
argv["Item.%d.Quantity" % (i+1)] = action
else:
argv["Item.%d.Action" % (i+1)] = action
def __cartOperation(XMLSearch, arguments):
'''Generic cart operation'''
plugins = {'isBypassed': lambda x: x == 'Request',
'isCollective': lambda x: x in ('CartItems',
'SavedForLaterItems'),
'isCollected': lambda x: x in ('CartItem', 'SavedForLaterItem')
}
return rawObject(XMLSearch, arguments, 'Cart', plugins)
# Seller Operation
def SellerLookup(Sellers, ResponseGroup=None, AWSAccessKeyId=None):
'''SellerLookup in AWS'''
argv = inspect.getargvalues(inspect.currentframe())[-1]
plugins = {'isBypassed': lambda x: x == 'Request',
'isCollective': lambda x: x == 'Sellers',
'isCollected': lambda x: x == 'Seller'}
return rawIterator(XMLSellerLookup, argv, 'Sellers', plugins)
def XMLSellerLookup(Sellers, ResponseGroup=None, AWSAccessKeyId=None):
'''DOM representation of SellerLookup in AWS'''
Operation = "SellerLookup"
AWSAccessKeyId = AWSAccessKeyId or LICENSE_KEY
SellerId = ",".join(Sellers)
argv = inspect.getargvalues(inspect.currentframe())[-1]
del argv['Sellers']
return query(buildRequest(argv))
def SellerListingLookup(SellerId, Id, IdType="Listing", ResponseGroup=None,
AWSAccessKeyId=None):
'''SellerListingLookup in AWS
Notice: although the repsonse includes TotalPage, TotalResults,
there is no ListingPage in the request, so we have to use rawIterator
instead of pagedIterator. Hope Amazaon would fix this inconsistance'''
argv = inspect.getargvalues(inspect.currentframe())[-1]
plugins = {'isBypassed': lambda x: x == 'Request',
'isCollective': lambda x: x == 'SellerListings',
'isCollected': lambda x: x == 'SellerListing'}
return rawIterator(XMLSellerListingLookup, argv, "SellerListings",
plugins)
def XMLSellerListingLookup(SellerId, Id, IdType="Listing", ResponseGroup=None,
AWSAccessKeyId=None):
'''DOM representation of SellerListingLookup in AWS'''
Operation = "SellerListingLookup"
AWSAccessKeyId = AWSAccessKeyId or LICENSE_KEY
argv = inspect.getargvalues(inspect.currentframe())[-1]
return query(buildRequest(argv))
def SellerListingSearch(SellerId, Title=None, Sort=None, ListingPage=None,
OfferStatus=None, ResponseGroup=None, AWSAccessKeyId=None):
'''SellerListingSearch in AWS'''
argv = inspect.getargvalues(inspect.currentframe())[-1]
plugins = {'isBypassed': lambda x: x == 'Request',
'isCollective': lambda x: x == 'SellerListings',
'isCollected': lambda x: x == 'SellerListing'}
return pagedIterator(XMLSellerListingSearch, argv, "ListingPage",
"SellerListings", plugins)
def XMLSellerListingSearch(SellerId, Title=None, Sort=None, ListingPage=None,
OfferStatus=None, ResponseGroup=None, AWSAccessKeyId=None):
'''DOM representation of SellerListingSearch in AWS'''
Operation = "SellerListingSearch"
AWSAccessKeyId = AWSAccessKeyId or LICENSE_KEY
argv = inspect.getargvalues(inspect.currentframe())[-1]
return query(buildRequest(argv))
def CustomerContentSearch(Name=None, Email=None, CustomerPage=1,
ResponseGroup=None, AWSAccessKeyId=None):
'''CustomerContentSearch in AWS'''
argv = inspect.getargvalues(inspect.currentframe())[-1]
plugins = {'isBypassed': lambda x: x == 'Request',
'isCollective': lambda x: x in ('Customers', 'CustomerReviews'),
'isCollected': lambda x: x in ('Customer', 'Review')}
return rawIterator(XMLCustomerContentSearch, argv, 'Customers', plugins)
def XMLCustomerContentSearch(Name=None, Email=None, CustomerPage=1,
ResponseGroup=None, AWSAccessKeyId=None):
'''DOM representation of CustomerContentSearch in AWS'''
Operation = "CustomerContentSearch"
AWSAccessKeyId = AWSAccessKeyId or LICENSE_KEY
argv = inspect.getargvalues(inspect.currentframe())[-1]
for x in ('Name', 'Email'):
if not argv[x]:
del argv[x]
return query(buildRequest(argv))
def CustomerContentLookup(CustomerId, ReviewPage=1, ResponseGroup=None,
AWSAccessKeyId=None):
'''CustomerContentLookup in AWS'''
argv = inspect.getargvalues(inspect.currentframe())[-1]
plugins = {'isBypassed': lambda x: x == 'Request',
'isCollective': lambda x: x == 'Customers',
'isCollected': lambda x: x == 'Customer'}
return rawIterator(XMLCustomerContentLookup, argv, 'Customers', plugins)
def XMLCustomerContentLookup(CustomerId, ReviewPage=1, ResponseGroup=None,
AWSAccessKeyId=None):
'''DOM representation of CustomerContentLookup in AWS'''
Operation = "CustomerContentLookup"
AWSAccessKeyId = AWSAccessKeyId or LICENSE_KEY
argv = inspect.getargvalues(inspect.currentframe())[-1]
return query(buildRequest(argv))
# BrowseNode
def BrowseNodeLookup(BrowseNodeId, ResponseGroup=None, AWSAccessKeyId=None):
'''BrowseNodeLookup in AWS'''
argv = inspect.getargvalues(inspect.currentframe())[-1]
plugins = {'isBypassed': lambda x: x == 'Request',
'isCollective': lambda x: x == 'Children',
'isCollected': lambda x: x == 'BrowseNode'}
return rawIterator(XMLBrowseNodeLookup, argv, 'BrowseNodes', plugins)
def XMLBrowseNodeLookup(BrowseNodeId, ResponseGroup=None, AWSAccessKeyId=None):
'''DOM representation of BrowseNodeLookup in AWS'''
Operation = "BrowseNodeLookup"
AWSAccessKeyId = AWSAccessKeyId or LICENSE_KEY
argv = inspect.getargvalues(inspect.currentframe())[-1]
return query(buildRequest(argv))
# Help
def Help(HelpType, About, ResponseGroup=None, AWSAccessKeyId=None):
'''Help in AWS'''
argv = inspect.getargvalues(inspect.currentframe())[-1]
plugins = {'isBypassed': lambda x: x == 'Request',
'isCollective': lambda x: x in ('RequiredParameters',
'AvailableParameters', 'DefaultResponseGroups',
'AvailableResponseGroups'),
'isCollected': lambda x: x in ('Parameter', 'ResponseGroup') }
return rawObject(XMLHelp, argv, 'Information', plugins)
def XMLHelp(HelpType, About, ResponseGroup=None, AWSAccessKeyId=None):
'''DOM representation of Help in AWS'''
Operation = "Help"
AWSAccessKeyId = AWSAccessKeyId or LICENSE_KEY
argv = inspect.getargvalues(inspect.currentframe())[-1]
return query(buildRequest(argv))
# Transaction
def TransactionLookup(TransactionId, ResponseGroup=None, AWSAccessKeyId=None):
'''TransactionLookup in AWS'''
argv = inspect.getargvalues(inspect.currentframe())[-1]
plugins = {'isBypassed': lambda x: x == 'Request',
'isCollective': lambda x: x in ('Transactions',
'TransactionItems', 'Shipments'),
'isCollected': lambda x: x in ('Transaction',
'TransactionItem', 'Shipment')}
return rawIterator(XMLTransactionLookup, argv, 'Transactions', plugins)
def XMLTransactionLookup(TransactionId, ResponseGroup=None,
AWSAccessKeyId=None):
'''DOM representation of TransactionLookup in AWS'''
Operation = "TransactionLookup"
AWSAccessKeyId = AWSAccessKeyId or LICENSE_KEY
argv = inspect.getargvalues(inspect.currentframe())[-1]
return query(buildRequest(argv))
if __name__ == "__main__" :
setLicenseKey("YOUR-LICENSE-HERE");
sll = SellerListingLookup("A3ENSIQ3ZA4FFN", "1106K206331")
print dir(sll[0])
