1# Copyright (C) 2006 Adam Olsen 2# 3# This program is free software; you can redistribute it and/or modify 4# it under the terms of the GNU General Public License as published by 5# the Free Software Foundation; either version 1, or (at your option) 6# any later version. 7# 8# This program is distributed in the hope that it will be useful, 9# but WITHOUT ANY WARRANTY; without even the implied warranty of 10# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 11# GNU General Public License for more details. 12# 13# You should have received a copy of the GNU General Public License 14# along with this program; if not, write to the Free Software 15# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. 16 17import urllib.parse 18import urllib.request 19import hmac 20import hashlib 21import base64 22import datetime 23import re 24from xl import common 25import logging 26 27logger = logging.getLogger(__name__) 28 29 30class AmazonSearchError(Exception): 31 pass 32 33 34def generate_timestamp(): 35 ret = datetime.datetime.utcnow() 36 return ret.isoformat() + 'Z' 37 38 39# make a valid RESTful AWS query, that is signed, from a dictionary 40# https://docs.aws.amazon.com/AWSECommerceService/latest/DG/RequestAuthenticationArticle.html 41# code by Robert Wallis: SmilingRob@gmail.com, your hourly software contractor 42 43 44def get_aws_query_string(aws_access_key_id, secret, query_dictionary): 45 query_dictionary["AWSAccessKeyId"] = aws_access_key_id 46 query_dictionary["Timestamp"] = generate_timestamp() 47 query_pairs = sorted( 48 map(lambda k, v: (k + "=" + urllib.parse.quote(v)), query_dictionary.items()) 49 ) 50 # The Amazon specs require a sorted list of arguments 51 query_string = "&".join(query_pairs) 52 hm = hmac.new( 53 secret, 54 "GET\nwebservices.amazon.com\n/onca/xml\n" + query_string, 55 hashlib.sha256, 56 ) 57 signature = urllib.parse.quote(base64.b64encode(hm.digest())) 58 query_string = "https://webservices.amazon.com/onca/xml?%s&Signature=%s" % ( 59 query_string, 60 signature, 61 ) 62 return query_string 63 64 65def search_covers(search, api_key, secret_key, user_agent): 66 params = { 67 'Operation': 'ItemSearch', 68 'Keywords': str(search), 69 'AssociateTag': 'InvalidTag', # now required for AWS cover search API 70 'Version': '2009-01-06', 71 'SearchIndex': 'Music', 72 'Service': 'AWSECommerceService', 73 'ResponseGroup': 'ItemAttributes,Images', 74 } 75 76 query_string = get_aws_query_string( 77 str(api_key).strip(), str(secret_key).strip(), params 78 ) 79 80 headers = {'User-Agent': user_agent} 81 req = urllib.request.Request(query_string, None, headers) 82 data = urllib.request.urlopen(req).read() 83 84 data = common.get_url_contents(query_string, user_agent) 85 86 # check for an error message 87 m = re.search(r'<Message>(.*)</Message>', data, re.DOTALL) 88 if m: 89 logger.warning('Amazon Covers Search Error: %s', m.group(1)) 90 raise AmazonSearchError(m.group(1)) 91 92 # check for large images 93 regex = re.compile(r'<LargeImage><URL>([^<]*)', re.DOTALL) 94 items = regex.findall(data) 95 96 return items 97