diff --git a/src/calibre/ebooks/metadata/amazonfr.py b/src/calibre/ebooks/metadata/amazonfr.py new file mode 100644 index 0000000000..7091719f30 --- /dev/null +++ b/src/calibre/ebooks/metadata/amazonfr.py @@ -0,0 +1,457 @@ +from __future__ import with_statement +__license__ = 'GPL 3' +__copyright__ = '2010, sengian ' + +import sys, textwrap, re, traceback +from urllib import urlencode +from math import ceil + +from lxml import html +from lxml.html import soupparser + +from calibre.utils.date import parse_date, utcnow +from calibre import browser, preferred_encoding +from calibre.ebooks.chardet import xml_to_unicode +from calibre.ebooks.metadata import MetaInformation, check_isbn, \ + authors_to_sort_string +from calibre.ebooks.metadata.fetch import MetadataSource +from calibre.utils.config import OptionParser +from calibre.library.comments import sanitize_comments_html + + +class AmazonFr(MetadataSource): + + name = 'Amazon french' + description = _('Downloads social metadata from amazon.fr') + supported_platforms = ['windows', 'osx', 'linux'] + author = 'Sengian' + version = (1, 0, 0) + has_html_comments = True + + def fetch(self): + try: + self.results = search(self.title, self.book_author, self.publisher, + self.isbn, max_results=10, verbose=self.verbose, lang='fr') + except Exception, e: + self.exception = e + self.tb = traceback.format_exc() + +class Amazon(MetadataSource): + + name = 'Amazon' + description = _('Downloads social metadata from amazon.com') + supported_platforms = ['windows', 'osx', 'linux'] + author = 'Kovid Goyal & Sengian' + version = (1, 1, 0) + has_html_comments = True + + def fetch(self): + try: + self.results = search(self.title, self.book_author, self.publisher, + self.isbn, max_results=10, verbose=self.verbose, lang='en') + except Exception, e: + self.exception = e + self.tb = traceback.format_exc() + + +def report(verbose): + if verbose: + import traceback + traceback.print_exc() + +def replace_monthsfr(datefr): + # Replace french months by english equivalent for parse_date + frtoen = { + u'[jJ]anvier': u'jan', + u'[fF].vrier': u'feb', + u'[mM]ars': u'mar', + u'[aA]vril': u'apr', + u'[mM]ai': u'may', + u'[jJ]uin': u'jun', + u'[jJ]uillet': u'jul', + u'[aA]o.t': u'aug', + u'[sS]eptembre': u'sep', + u'[Oo]ctobre': u'oct', + u'[nN]ovembre': u'nov', + u'[dD].cembre': u'dec' } + for k in frtoen.iterkeys(): + tmp = re.sub(k, frtoen[k], datefr) + if tmp <> datefr: break + return tmp + +class Query(object): + + BASE_URL_FR = 'http://www.amazon.fr' + BASE_URL_EN = 'http://www.amazon.com' + + def __init__(self, title=None, author=None, publisher=None, isbn=None, keywords=None, + max_results=20, rlang='en'): + assert not(title is None and author is None and publisher is None \ + and isbn is None and keywords is None) + assert (max_results < 21) + + self.max_results = int(max_results) + self.renbres = re.compile(u'\s*(\d+)\s*') + + q = { 'search-alias' : 'stripbooks' , + 'unfiltered' : '1', + 'field-keywords' : '', + 'field-author' : '', + 'field-title' : '', + 'field-isbn' : '', + 'field-publisher' : '' + #get to amazon detailed search page to get all options + # 'node' : '', + # 'field-binding' : '', + #before, during, after + # 'field-dateop' : '', + #month as number + # 'field-datemod' : '', + # 'field-dateyear' : '', + #french only + # 'field-collection' : '', + #many options available + } + + if rlang =='en': + q['sort'] = 'relevanceexprank' + self.urldata = self.BASE_URL_EN + elif rlang =='fr': + q['sort'] = 'relevancerank' + self.urldata = self.BASE_URL_FR + self.baseurl = self.urldata + + if isbn is not None: + q['field-isbn'] = isbn.replace('-', '') + else: + if title is not None: + q['field-title'] = title + if author is not None: + q['field-author'] = author + if publisher is not None: + q['field-publisher'] = publisher + if keywords is not None: + q['field-keywords'] = keywords + + if isinstance(q, unicode): + q = q.encode('utf-8') + self.urldata += '/gp/search/ref=sr_adv_b/?' + urlencode(q) + + def __call__(self, browser, verbose, timeout = 5.): + if verbose: + print 'Query:', self.urldata + + try: + raw = browser.open_novisit(self.urldata, timeout=timeout).read() + except Exception, e: + report(verbose) + if callable(getattr(e, 'getcode', None)) and \ + e.getcode() == 404: + return + raise + if '404 - ' in raw: + return + raw = xml_to_unicode(raw, strip_encoding_pats=True, + resolve_entities=True)[0] + try: + feed = soupparser.fromstring(raw) + except: + return None, self.urldata + + #nb of page + try: + nbresults = self.renbres.findall(feed.xpath("//*[@class='resultCount']")[0].text) + rpp = 0 + if len(nbresults) > 1: + rpp = int(nbresults[1]) + nbresults = int(nbresults[2]) + except: + return None, self.urldata + + pages =[feed] + if rpp: + nbpagetoquery = int(ceil(float(min(nbresults, self.max_results))/ rpp)) + for i in xrange(2, nbpagetoquery + 1): + try: + urldata = self.urldata + '&page=' + str(i) + raw = browser.open_novisit(urldata, timeout=timeout).read() + except Exception, e: + continue + if '<title>404 - ' in raw: + continue + raw = xml_to_unicode(raw, strip_encoding_pats=True, + resolve_entities=True)[0] + try: + feed = soupparser.fromstring(raw) + except: + continue + pages.append(feed) + + results = [] + for x in pages: + results.extend([i.getparent().get('href') \ + for i in x.xpath("//a/span[@class='srTitle']")]) + return results[:self.max_results], self.baseurl + +class ResultList(list): + + def __init__(self, baseurl, lang = 'en'): + self.baseurl = baseurl + self.lang = lang + self.repub = re.compile(u'\((.*)\)') + self.rerat = re.compile(u'([0-9.]+)') + self.reattr = re.compile(r'<([a-zA-Z0-9]+)\s[^>]+>') + self.reoutp = re.compile(r'(?s)<em>--This text ref.*?</em>') + self.recom = re.compile(r'(?s)<!--.*?-->') + self.republi = re.compile(u'(Editeur|Publisher)', re.I) + self.reisbn = re.compile(u'(ISBN-10|ISBN-10|ASIN)', re.I) + self.relang = re.compile(u'(Language|Langue)', re.I) + self.reratelt = re.compile(u'(Average\s*Customer\s*Review|Moyenne\s*des\s*commentaires\s*client)', re.I) + self.reprod = re.compile(u'(Product\s*Details|D.tails\s*sur\s*le\s*produit)', re.I) + + def strip_tags_etree(self, etreeobj, invalid_tags): + for (itag, rmv) in invalid_tags.iteritems(): + if rmv: + for elts in etreeobj.getiterator(itag): + elts.drop_tree() + else: + for elts in etreeobj.getiterator(itag): + elts.drop_tag() + + def clean_entry(self, entry, invalid_tags = {'script': True}, + invalid_id = (), invalid_class=()): + #invalid_tags: remove tag and keep content if False else remove + #remove tags + if invalid_tags: + self.strip_tags_etree(entry, invalid_tags) + #remove id + if invalid_id: + for eltid in invalid_id: + elt = entry.get_element_by_id(eltid) + if elt is not None: + elt.drop_tree() + #remove class + if invalid_class: + for eltclass in invalid_class: + elts = entry.find_class(eltclass) + if elts is not None: + for elt in elts: + elt.drop_tree() + + def get_title(self, entry): + title = entry.get_element_by_id('btAsinTitle') + if title is not None: + title = title.text + return unicode(title.replace('\n', '').strip()) + + def get_authors(self, entry): + author = entry.get_element_by_id('btAsinTitle') + while author.getparent().tag != 'div': + author = author.getparent() + author = author.getparent() + authortext = [] + for x in author.getiterator('a'): + authortext.append(unicode(x.text_content().strip())) + return authortext + + def get_description(self, entry, verbose): + try: + description = entry.get_element_by_id("productDescription").find("div[@class='content']") + inv_class = ('seeAll', 'emptyClear') + inv_tags ={'img': True, 'a': False} + self.clean_entry(description, invalid_tags=inv_tags, invalid_class=inv_class) + description = html.tostring(description, method='html', encoding=unicode).strip() + # remove all attributes from tags + description = self.reattr.sub(r'<\1>', description) + # Remove the notice about text referring to out of print editions + description = self.reoutp.sub('', description) + # Remove comments + description = self.recom.sub('', description) + return unicode(sanitize_comments_html(description)) + except: + report(verbose) + return None + + def get_tags(self, entry, browser, verbose): + try: + tags = entry.get_element_by_id('tagContentHolder') + testptag = tags.find_class('see-all') + if testptag: + for x in testptag: + alink = x.xpath('descendant-or-self::a') + if alink: + if alink[0].get('class') == 'tgJsActive': + continue + link = self.baseurl + alink[0].get('href') + entry = self.get_individual_metadata(browser, link, verbose) + tags = entry.get_element_by_id('tagContentHolder') + break + tags = [a.text for a in tags.getiterator('a') if a.get('rel') == 'tag'] + except: + report(verbose) + tags = [] + return tags + + def get_book_info(self, entry, mi, verbose): + try: + entry = entry.get_element_by_id('SalesRank').getparent() + except: + try: + for z in entry.getiterator('h2'): + if self.reprod.search(z.text_content()): + entry = z.getparent().find("div[@class='content']/ul") + break + except: + report(verbose) + return mi + elts = entry.findall('li') + #pub & date + elt = filter(lambda x: self.republi.search(x.find('b').text), elts) + if elt: + pub = elt[0].find('b').tail + mi.publisher = unicode(self.repub.sub('', pub).strip()) + d = self.repub.search(pub) + if d is not None: + d = d.group(1) + try: + default = utcnow().replace(day=15) + if self.lang == 'fr': + d = replace_monthsfr(d) + d = parse_date(d, assume_utc=True, default=default) + mi.pubdate = d + except: + report(verbose) + #ISBN + elt = filter(lambda x: self.reisbn.search(x.find('b').text), elts) + if elt: + isbn = elt[0].find('b').tail.replace('-', '').strip() + if check_isbn(isbn): + mi.isbn = unicode(isbn) + elif len(elt) > 1: + isbn = elt[1].find('b').tail.replace('-', '').strip() + if check_isbn(isbn): + mi.isbn = unicode(isbn) + #Langue + elt = filter(lambda x: self.relang.search(x.find('b').text), elts) + if elt: + langue = elt[0].find('b').tail.strip() + if langue: + mi.language = unicode(langue) + #ratings + elt = filter(lambda x: self.reratelt.search(x.find('b').text), elts) + if elt: + ratings = elt[0].find_class('swSprite') + if ratings: + ratings = self.rerat.findall(ratings[0].get('title')) + if len(ratings) == 2: + mi.rating = float(ratings[0])/float(ratings[1]) * 5 + return mi + + def fill_MI(self, entry, title, authors, browser, verbose): + mi = MetaInformation(title, authors) + mi.author_sort = authors_to_sort_string(authors) + mi.comments = self.get_description(entry, verbose) + mi = self.get_book_info(entry, mi, verbose) + mi.tags = self.get_tags(entry, browser, verbose) + return mi + + def get_individual_metadata(self, browser, linkdata, verbose): + try: + raw = browser.open_novisit(linkdata).read() + except Exception, e: + report(verbose) + if callable(getattr(e, 'getcode', None)) and \ + e.getcode() == 404: + return + raise + if '<title>404 - ' in raw: + report(verbose) + return + raw = xml_to_unicode(raw, strip_encoding_pats=True, + resolve_entities=True)[0] + try: + return soupparser.fromstring(raw) + except: + return + + def populate(self, entries, browser, verbose=False): + for x in entries: + try: + entry = self.get_individual_metadata(browser, x, verbose) + # clean results + # inv_ids = ('divsinglecolumnminwidth', 'sims.purchase', 'AutoBuyXGetY', 'A9AdsMiddleBoxTop') + # inv_class = ('buyingDetailsGrid', 'productImageGrid') + # inv_tags ={'script': True, 'style': True, 'form': False} + # self.clean_entry(entry, invalid_id=inv_ids) + title = self.get_title(entry) + authors = self.get_authors(entry) + except Exception, e: + if verbose: + print 'Failed to get all details for an entry' + print e + continue + self.append(self.fill_MI(entry, title, authors, browser, verbose)) + + +def search(title=None, author=None, publisher=None, isbn=None, + max_results=5, verbose=False, keywords=None, lang='en'): + br = browser() + entries, baseurl = Query(title=title, author=author, isbn=isbn, publisher=publisher, + keywords=keywords, max_results=max_results,rlang=lang)(br, verbose) + + if entries is None or len(entries) == 0: + return + + #List of entry + ans = ResultList(baseurl, lang) + ans.populate(entries, br, verbose) + return ans + +def option_parser(): + parser = OptionParser(textwrap.dedent(\ + '''\ + %prog [options] + + Fetch book metadata from Amazon. You must specify one of title, author, + ISBN, publisher or keywords. Will fetch a maximum of 10 matches, + so you should make your query as specific as possible. + You can chose the language for metadata retrieval (french & american). + ''' + )) + parser.add_option('-t', '--title', help='Book title') + parser.add_option('-a', '--author', help='Book author(s)') + parser.add_option('-p', '--publisher', help='Book publisher') + parser.add_option('-i', '--isbn', help='Book ISBN') + parser.add_option('-k', '--keywords', help='Keywords') + parser.add_option('-c', '--covers', default=0, + help='Covers: 1-Check/ 2-Download') + parser.add_option('-p', '--coverspath', default='', + help='Covers files path') + parser.add_option('-m', '--max-results', default=10, + help='Maximum number of results to fetch') + parser.add_option('-l', '--lang', default='en', + help='Chosen language for metadata search') + parser.add_option('-v', '--verbose', default=0, action='count', + help='Be more verbose about errors') + return parser + +def main(args=sys.argv): + parser = option_parser() + opts, args = parser.parse_args(args) + try: + results = search(opts.title, opts.author, isbn=opts.isbn, publisher=opts.publisher, + keywords=opts.keywords, verbose=opts.verbose, max_results=opts.max_results, + lang=opts.lang) + except AssertionError: + report(True) + parser.print_help() + return 1 + if results is None or len(results) == 0: + print 'No result found for this search!' + return 0 + for result in results: + print unicode(result).encode(preferred_encoding, 'replace') + print + +if __name__ == '__main__': + sys.exit(main()) \ No newline at end of file