import requests import urllib import httplib import json from pprint import pprint from itertools import islice, izip, repeat import logging from xml.etree import ElementTree import random random.seed() import sys, os import json # a kludge to allow for isbn.py to be imported # and not just in the context of the regluit Django app try: from regluit.core import isbn as isbn_mod except: import isbn as isbn_mod try: import unittest from unittest import TestCase except: from django.test import TestCase from django.utils import unittest import re import freebase import logging logger = logging.getLogger(__name__) GOOGLE_BOOKS_KEY = "AIzaSyDsrHCUsUFNAf65cFPSF8MZTKj8C9oMuj8" MASHUPBOOK_ISBN_13 = '9781590598580' MASHUPBOOK_ISBN_10 = '159059858X' MASHUPBOOK_OLID = 'OL13439114M' RY_OLID = 'OL4264806A' SURFACING_WORK_OLID = 'OL675829W' SURFACING_EDITION_OLID = 'OL8075248M' SURFACING_ISBN = '9780446311076' SURFACING_LT_WORK_ID = '18997' USER_AGENT = "rdhyee@gluejar.com" # http://stackoverflow.com/questions/2348317/how-to-write-a-pager-for-python-iterators/2350904#2350904 def grouper(iterable, page_size): page= [] for item in iterable: page.append( item ) if len(page) == page_size: yield page page= [] yield page class FreebaseException(Exception): pass class OpenLibraryException(Exception): pass class HathiException(Exception): pass def filter_none(d): d2 = {} for (k,v) in d.iteritems(): if v is not None: d2[k] = v return d2 def to_list(s): """if input is not a list return a list with s""" if not isinstance(s,list): return [s] else: return s def thingisbn(isbn): """given an ISBN return a list of related edition ISBNs, according to Library Thing. (takes isbn_10 or isbn_13, returns isbn_10, except for 979 isbns, which come back as isbn_13') """ logger.info("looking up %s at ThingISBN" , isbn) url = "http://www.librarything.com/api/thingISBN/%s" % isbn xml = requests.get(url, headers={"User-Agent": USER_AGENT}).content doc = ElementTree.fromstring(xml) return [e.text for e in doc.findall('isbn')] def lt_whatwork(isbn=None, title=None, author=None): """ "What work?" takes an ISBN and/or a title-author and returns the LibraryThing work number. http://www.librarything.com/blogs/thingology/2009/03/new-api-what-work/ """ logger.info("looking up at lt_whatwork (isbn, title, author): %s %s %s" ,isbn, title, author) url = "http://www.librarything.com/api/whatwork.php" params=dict([(k,v) for (k,v) in {'isbn':isbn, 'title':title, 'author':author}.items() if v is not None]) xml = requests.get(url, params=params, headers={"User-Agent": USER_AGENT}).content doc = ElementTree.fromstring(xml) work = doc.find('work') if work is not None: return work.text else: return None def hathi_bib(id, id_type='isbn', detail_level='brief'): url = "http://catalog.hathitrust.org/api/volumes/brief/%s/%s.json" % (id_type, id) r = requests.get(url) if r.status_code == httplib.OK: return r.content else: raise Exception("Hathi Bib API response: %s " % (httplib.responses[r.status_code]) ) class GoogleBooks(object): def __init__(self, key): self.key = key def isbn (self, isbn, glossed=True): url = "https://www.googleapis.com/books/v1/volumes" try: results = self._get_json(url, {"q": "isbn:%s" % isbn}) except LookupFailure: logger.exception("lookup failure for %s", isbn) return None if not results.has_key('items') or len(results['items']) == 0: logger.warn("no google hits for %s" , isbn) return None else: if glossed: return self._parse_item(results['items'][0]) else: return results def query(self, q, glossed=True): url = "https://www.googleapis.com/books/v1/volumes" try: results = self._get_json(url, {'q':q}) except LookupFailure: logger.exception("lookup failure for %s", q) return None if not results.has_key('items') or len(results['items']) == 0: logger.warn("no google hits for %s", q) return None else: if glossed: return [self._parse_item(item) for item in results.get('items')] else: return results def volumeid(self, g_id, glossed=True): url = "https://www.googleapis.com/books/v1/volumes/{0}".format(g_id) try: item = self._get_json(url, {}) if glossed: return self._parse_item(item) else: return item except LookupFailure: logger.exception("lookup failure for %s", g_id) return None except Exception, e: logger.exception("other failure in volumeid %s", e) return None def _parse_item(self, item): d = item['volumeInfo'] google_books_id = item['id'] title = d.get('title') language = d.get('language') identifiers = d.get('industryIdentifiers', []) ratings_count = d.get('ratingsCount') # flip [{u'identifier': u'159059858X', u'type': u'ISBN_10'}, {u'identifier': u'9781590598580', u'type': u'ISBN_13'}] to # {u'ISBN_13': u'9781590598580', u'ISBN_10': u'159059858X'} identifiers = dict([(id["type"],id["identifier"]) for id in d.get('industryIdentifiers', [])]) isbn = identifiers.get('ISBN_13') or identifiers.get('ISBN_10') if isbn: isbn = isbn_mod.ISBN(isbn).to_string('13') published_date = d.get('publishedDate') publisher = d.get('publisher') data = {'title':title, 'language':language, 'isbn':isbn, 'google_books_id': google_books_id, 'ratings_count':ratings_count, 'published_date':published_date, 'publisher':publisher} return data def _get_item(self, results): if len(results): google_books_id = results['items'][0]['id'] item = results['items'][0] return self._parse_item(item) else: return None def _get_json(self, url, params={}, type='gb'): # lifted (with slight mod) from https://github.com/Gluejar/regluit/blob/master/core/bookloader.py # TODO: should X-Forwarded-For change based on the request from client? headers = {'User-Agent': 'raymond.yee@gmail.com', 'Accept': 'application/json', 'X-Forwarded-For': '69.174.114.214'} if type == 'gb': params['key'] = self.key response = requests.get(url, params=params, headers=headers) if response.status_code == 200: return json.loads(response.content) else: logger.error("unexpected HTTP response: %s" % response) if response.content: logger.error("response content: %s" % response.content) raise LookupFailure("GET failed: url=%s and params=%s" % (url, params)) class OpenLibrary(object): """http://openlibrary.org/developers/api""" @classmethod def books(cls, id, id_type="isbn", format="json", callback=None, jscmd=None): # http://openlibrary.org/dev/docs/api/books # bibkeys: one of isbn, oclc, lccn, olid # format: one of json, javascript # jscmd: one of viewapi, data, details (deprecated in favor of data) base_url = "http://openlibrary.org/api/books" params = filter_none({'bibkeys':'%s:%s' % (id_type,id), 'format':format, 'callback':callback, 'jscmd':jscmd}) r = requests.get(base_url,params=params) if r.status_code == httplib.OK: return json.loads(r.content) else: raise OpenLibraryException("OpenLibrary API response: %s " % (httplib.responses[r.status_code]) ) @classmethod def covers(cls, id, id_type='isbn', size='S'): # http://openlibrary.org/dev/docs/api/covers # id_type: one of 'id' (internal cover ID), olid (Open Library ID), isbn, oclc, lccn, goodreads, librarything # size: one of s, m, l # http://covers.openlibrary.org/b/$key/$value-$size.jpg if id_type.lower() not in ['id','isbn','oclc','lccn','goodreads','librarything']: raise OpenLibraryException("%s is an incorrect id_type for covers" % (id_type)) if size.upper() not in ['S','M','L']: raise OpenLibraryException("%s is an incorrect size for covers" % (size)) return "http://covers.openlibrary.org/b/%s/%s-%s.jpg" % (id_type.lower(),id,size.upper()) @classmethod def author_photos(cls, olid, size='S'): #http://covers.openlibrary.org/a/$key/$value-$size.jpg if size.upper() not in ['S','M','L']: raise OpenLibraryException("%s is an incorrect size for author" % (size)) return "http://covers.openlibrary.org/a/olid/%s-%s.jpg" % (olid,size.upper()) @classmethod def read(cls, queries): # http://openlibrary.org/dev/docs/api/read -- most of its value to be used in browser JS? # can do single or multiple requests # example of a single request: # http://openlibrary.org/api/volumes/brief/isbn/0596156715.json # multiple # http://openlibrary.org/api/volumes/brief/json/id:1;lccn:50006784|olid:OL6179000M;lccn:55011330 # the multiple format works, I think, for single requests #url = "http://openlibrary.org/api/volumes/brief/%s/%s.json" % (id_type, id) query_string = "|".join([ ";".join([ "%s:%s" % (id_type,id) for (id, id_type) in to_list(query)]) for query in to_list(queries)]) if query_string: url = "http://openlibrary.org/api/volumes/brief/json/%s" % (query_string) r = requests.get(url) if r.status_code == httplib.OK: return json.loads(r.content) else: raise OpenLibraryException("OpenLibrary API response: %s " % (httplib.responses[r.status_code]) ) else: return None @classmethod def lists(cls): # http://openlibrary.org/dev/docs/api/lists raise NotImplementedError @classmethod def query_iter(cls,**kwargs): # use limit for page size and offset as the starting point kwargs2 = kwargs.copy() kwargs2.setdefault('offset', 0) more_items = True while more_items: items = cls.query(**kwargs2) for item in items: yield item if len(items): kwargs2['offset'] += len(items) else: more_items = False @classmethod def query(cls, **kwargs): # limit and offset are special base_url = "http://openlibrary.org/query.json" r = requests.get(base_url, params=kwargs) if r.status_code == httplib.OK: return json.loads(r.content) else: raise OpenLibraryException("OpenLibrary API response: %s " % (httplib.responses[r.status_code]) ) @classmethod def editions(cls, work_olid): # http://openlibrary.org/query.json?type=/type/edition&works=/works/OL675829W&limit=5000 for item in cls.query_iter(type='/type/edition',works='/works/%s'%(work_olid), limit=10): try: yield re.match(r'^/books/(.*)$',item['key']).group(1) except Exception, e: raise OpenLibraryException("problem in editions: %s " % e) @classmethod def works0(cls, id, id_type='isbn'): # http://openlibrary.org/api/volumes/brief/olid/OL8075248M.json # can there be more than 1 work for a given edition? # will return a list of works # there is a bug in OL in which we have workless editions response = cls.read((id,id_type)) # resp['olid:OL8075248M']['records']['/books/OL8075248M']["details"]["details"]["works"][0]["key"] # resp.values()[0]['records'].values()[0]["details"]["details"]["works"][0]["key"] try: works_key = response.values()[0]['records'].values()[0]["details"]["details"]["works"] if (len(response.values()) == 1 and len(response.values()[0]['records'].values()) == 1): return [re.match(r'^/works/(.*)$',work_key["key"]).group(1) for work_key in works_key] else: raise OpenLibraryException("Assumption of 1 key in response invalid in OpenLibrary.works0") except Exception, e: return [] @classmethod def works(cls, ids, page_size=10): """generalize to handle more than one set of id at a time -- ids is an iterable""" for (i, page) in enumerate(grouper(ids, page_size=page_size)): response = cls.read(page) for (id, id_type) in page: key = "{1}:{0}".format(id, id_type) val = response.get(key) if val is not None: if (len(val['records'].values()) == 1): try: works_key = val['records'].values()[0]["details"]["details"]["works"] yield [re.match(r'^/works/(.*)$',work_key["key"]).group(1) for work_key in works_key] except Exception, e: pass else: raise OpenLibraryException("Assumption of 1 key in response invalid in OpenLibrary.works") else: yield [] @classmethod def json_for_olid(cls, olid, follow_redirect=True): ol_types = {'M': 'books', 'W':'works', 'A':'authors'} id_type = ol_types.get(str(olid)[-1].upper(), None) if id_type is not None: url = "http://openlibrary.org/{0}/{1}.json".format(id_type, olid.upper()) r = requests.get(url) if r.status_code == httplib.OK: # check to see whether type is redirect, retrieve that item it we are following redirects resp = json.loads(r.content) if resp["type"]["key"] == "/type/redirect" and follow_redirect: redir_olid = resp["location"].split("/")[-1] return OpenLibrary.json_for_olid(redir_olid) else: return resp else: raise OpenLibraryException("OpenLibrary API response: %s " % (httplib.responses[r.status_code]) ) else: return None @classmethod def xisbn(cls,isbn_val=None, work_id=None, page_size=5): logger.debug("isbn_val, work_id, page_size: %s %s %d", isbn_val, work_id, page_size) isbns = set() if isbn_val is None and work_id is None: raise Exception("One of isbn or work_id must be specified") elif isbn_val is not None and work_id is not None: raise Exception("Only one of isbn or work_id can be specified") if isbn_val is not None: # figure out the work_id and then pass back all the ISBNs from the manifestations of the work try: isbn_val = isbn_mod.ISBN(isbn_val).to_string('13') if isbn_val is not None: isbns.add(isbn_val) yield isbn_val work_ids = list(cls.works([(isbn_val,'isbn')])) if len(work_ids): work_id = work_ids[0][0] else: # can't find a work_id raise StopIteration() except isbn_mod.ISBNException: raise StopIteration() # by this point we have a work_id editions = cls.editions(work_olid=work_id) for page in grouper(editions, page_size): query = list(izip(page, repeat('olid'))) #print query k = cls.read(query) for edition in page: # k['olid:OL8075248M']['records'].values()[0]['data']['identifiers']['isbn_13'][0] identifiers = k['olid:{0}'.format(edition)]['records'].values()[0]['data']['identifiers'] isbn = identifiers.get('isbn_13',[None])[0] or identifiers.get('isbn_10',[None])[0] if isbn: try: isbn = isbn_mod.ISBN(isbn).to_string('13') if isbn is not None and isbn not in isbns: isbns.add(isbn) yield isbn except isbn_mod.ISBNException: print "Problem with isbn %s for edition %s " % (isbn, edition) except Exception as e: raise e class FreebaseBooks(object): def __init__(self, username=None, password=None, main_or_sandbox='main'): if main_or_sandbox == 'main': self.freebase = freebase else: self.freebase = freebase.sandbox if username is not None and password is not None: self.freebase.login(username,password) def books(self): MQL = u"""[{ "type": "/book/book", "id": null, "key": [{ "namespace": "/wikipedia/en", "value": null, "type": "/type/key" }] }]""".replace("\n"," ") query = json.loads(MQL) resp = self.freebase.mqlreaditer(query) for r in resp: yield r def book_editions(self): MQL = u"""[{ "type": "/book/book_edition", "id": null, "isbn": [{}], "ISBN": [{}], "LCCN": [{}], "OCLC_number": [{}], "openlibrary_id": [{}], "book": { "id": null, "name": null }}]""".replace("\n"," ") query = json.loads(MQL) resp = self.freebase.mqlreaditer(query) for r in resp: yield r def editions_for_book(self, book_id): MQL = u"""[{ "type": "/book/book_edition", "id": null, "isbn": [{}], "ISBN": [{}], "LCCN": [{}], "OCLC_number": [{}], "openlibrary_id": [{}], "book": { "id": null, "name": null }}]""".replace("\n"," ") query = json.loads(MQL) query[0]["book"]["id"] = book_id resp = self.freebase.mqlreaditer(query) for r in resp: yield r def book_edition_by_id(self,id,id_type): MQL = u"""[{ "type": "/book/book_edition", "id": null, "isbn": [{}], "ISBN": [{}], "LCCN": [{}], "OCLC_number": [{}], "openlibrary_id": [{}], "book": { "id": null, "name": null }}]""".replace("\n"," ") query = json.loads(MQL) if id_type == 'isbn': query[0][id_type][0].setdefault('name', id) elif id_type in ['LCCN', 'OCLC_number', 'openlibrary_id']: query[0][id_type][0].setdefault('value', id) if id_type in ['isbn', 'LCCN', 'OCLC_number', 'openlibrary_id']: resp = self.freebase.mqlreaditer(query) for r in resp: yield r else: raise FreebaseException('id_type must be one of ISBN, LCCN, OCLC_number or openlibrary_id, not %s' % (id_type)) def xisbn(self, isbn_val=None, book_id=None): """ pass in either isbn_val or book_id and xisbn returns related ISBNs in Freebase. Handles case in which either isbn or book_id is not None but not both """ if isbn_val is None and book_id is None: raise Exception("One of isbn or book_id must be specified") elif isbn_val is not None and book_id is not None: raise Exception("Only only of isbn or book_id can be specified") elif isbn_val is not None: isbn_val = isbn_mod.ISBN(isbn_val).to_string('13') MQL = """[{ "type": "/book/book_edition", "isbn": { "name": null }, "book": { "editions": [{ "isbn": { "name": null } }] } }]""".replace("\n"," ") query = json.loads(MQL) query[0]["book"]["editions"][0]["isbn"]["name"] = isbn_val resp = self.freebase.mqlreaditer(query) for r in resp: yield r["isbn"]["name"] elif book_id is not None: for edition in self.editions_for_book(book_id=book_id): for i in edition["isbn"]: yield i["name"] class WorkMapper(object): @classmethod def freebase_book_to_openlibrary_work(cls, fb_id, complete_search=False): """ Try to map a Freebase ID by taking the ISBNs of associated editions and asking OpenLibrary for the work id""" print "fb_id: ", fb_id fb = FreebaseBooks() work_ids = set() # grab all ISBNs correponding to Freebase fb_id and compute the OpenLibrary work ID # if complete_search is False, stop at first work id for work_id_list in OpenLibrary.works(izip(fb.xisbn(book_id=fb_id), repeat('isbn'))): for work_id in work_id_list: if work_id not in work_ids: work_ids.add(work_id) yield work_id if not complete_search: raise StopIteration() class LibraryThing(object): """ Provide cached access to thingisbn and LT whatwork interface. Allow for a cache file to be loaded and saved """ def __init__(self, fname=None): self.__isbn_to_work_id = {} self.fname = fname def __del__(self): self.save() def thingisbn(self, isbn, return_work_id=False): """ if return_work_id is True, we won't try to calculate all the relevant isbns""" # first, normalize the isbn isbn = isbn_mod.ISBN(isbn).to_string('13') if isbn is None: return [] # check to see whether we have isbn already if isbn in self.__isbn_to_work_id: # return all isbns with the work id # print "%s already cached" % (isbn) work_id = self.__isbn_to_work_id.get(isbn) if return_work_id: return work_id if work_id is not None: return [k for (k, v) in self.__isbn_to_work_id.items() if v == work_id] else: return [] else: # if isbn is not already cached, do look up and cache the results and return the results print "calling thingisbn for %s" % (isbn) results = [isbn_mod.ISBN(k).to_string('13') for k in thingisbn (isbn)] if len(results): # look up the librarything work id work_id = self.whatwork(isbn) if work_id is not None: # which should be the case since results is not zero-length self.__isbn_to_work_id.update(dict([(isbn_mod.ISBN(result).to_string('13'), work_id) for result in results])) else: logger.exception("work_id should not be None for isbn %s", isbn) return [] else: self.__isbn_to_work_id[isbn] = None # mark as not recognized by LT work_id = None if return_work_id: return work_id else: return results def whatwork(self, isbn=None, title=None, author=None): # if isbn is not None and title, author None then look up results, otherwise just pass along to lt_whatwork # first, normalize the isbn isbn = isbn_mod.ISBN(isbn).to_string('13') if isbn is not None and (title is None and author is None): if isbn in self.__isbn_to_work_id: work_id = self.__isbn_to_work_id.get(isbn) else: work_id = lt_whatwork(isbn=isbn) self.__isbn_to_work_id[isbn] = work_id return work_id else: return lt_whatwork(isbn=isbn, title=title, author=author) def load(self): try: f = open(self.fname) input_data = json.load(f) f.close() if isinstance(input_data, dict): self.__isbn_to_work_id = input_data return True else: return False except Exception, e: print e def save(self): if self.fname is not None: f = open(self.fname, "w") json.dump(self.__isbn_to_work_id, f) f.close() return True else: return False def look_up_my_zotero_books_in_hathi(): from regluit.experimental.zotero_books import MyZotero zot = MyZotero() for (i,b) in enumerate(zot.get_books(20)): try: print b, hathi_bib(b['isbn']) except Exception, e: print e def ol_practice(): print OpenLibrary.books(MASHUPBOOK_ISBN_10) pprint (OpenLibrary.books(MASHUPBOOK_ISBN_13, jscmd='data')) print OpenLibrary.books('0385472579', jscmd='data') print (OpenLibrary.covers(MASHUPBOOK_ISBN_10, size='M')) print (OpenLibrary.author_photos(RY_OLID,'S')) # can we status of a pd book oclc:03544699 The Log of a Cowboy - Andy Adams, 1903 # http://openlibrary.org/books/OL7173600M/The_log_of_a_cowboy -- not working? print OpenLibrary.books('OL7173600M' 'olid', jscmd='data') # http://openlibrary.org/books/OL6542070M/The_Montessori_method works print OpenLibrary.books('1181252','oclc',jscmd='data') print OpenLibrary.read([(MASHUPBOOK_ISBN_10,'isbn'),('1181252','oclc')]) # let's bring up the editions for Surfacing for (i,ed) in enumerate(islice(OpenLibrary.editions(SURFACING_WORK_OLID),100)): print i, ed # let's get the Work ID for one of the editions pprint(OpenLibrary.works(SURFACING_EDITION_OLID,id_type='olid')) class LookupFailure(Exception): pass class FreebaseBooksTest(TestCase): def test_books_iter(self): fb = FreebaseBooks() books = list(islice(fb.books(),4)) self.assertEqual(len(books),4) for book in books[0:1]: self.assertTrue(book["type"], "/book/book") def test_book_editions_iter(self): fb = FreebaseBooks() editions = list(islice(fb.book_editions(),4)) self.assertEqual(len(editions),4) for edition in editions[0:1]: self.assertTrue(edition["type"], "/book/book_edition") def test_book_edition_by_id(self): fb = FreebaseBooks() # http://www.amazon.com/New-Collected-Poems-Czeslaw-Milosz/dp/006019667X edition = list(fb.book_edition_by_id('9780060196677','isbn')) self.assertEqual(edition[0]['type'],'/book/book_edition') self.assertEqual(edition[0]['book']['id'],'/m/0c1t1yk') self.assertEqual(edition[0]['book']['name'],'New and collected poems 1931-2001') edition = list(fb.book_edition_by_id('76074298', 'OCLC_number')) self.assertEqual(edition[0]['type'],'/book/book_edition') self.assertEqual(edition[0]['book']['id'],'/m/021yncj') self.assertEqual(edition[0]['book']['name'],'Brave New Words: The Oxford Dictionary of Science Fiction') # test openlibary_id Moby Dick edition = list(fb.book_edition_by_id('9780486432151', 'isbn'))[0] self.assertEqual(edition['openlibrary_id'][0]['value'], 'OL3685847M') def test_editions_for_book(self): fb = FreebaseBooks() book_id = '/en/moby-dick' editions = fb.editions_for_book(book_id) for i, edition in enumerate(editions): pass def test_xisbn(self): isbn_val = '9780486432151' book_id = '/en/moby-dick' fb = FreebaseBooks() isbns = set(fb.xisbn(isbn_val)) isbns2 = set(fb.xisbn(book_id=book_id)) self.assertEqual(isbns, isbns2) class OpenLibraryTest(TestCase): def test_books(self): book = OpenLibrary.books(MASHUPBOOK_ISBN_10) self.assertEqual(book.values()[0]['info_url'], 'http://openlibrary.org/books/OL13439114M/Pro_Web_2.0_Mashups') book_data = OpenLibrary.books('0385472579', jscmd='data') self.assertEqual(book_data.values()[0]['title'], 'Zen Speaks') self.assertEqual(book_data.values()[0]['identifiers']['openlibrary'][0], 'OL7440033M') def test_books_olid(self): # can we status of a pd book oclc:03544699 The Log of a Cowboy - Andy Adams, 1903 # http://openlibrary.org/books/OL7173600M/The_log_of_a_cowboy book = OpenLibrary.books('OL7173600M', 'olid', jscmd='data') self.assertEqual(book.values()[0]['title'], 'The log of a cowboy') def test_books_oclc(self): # http://openlibrary.org/books/OL6542070M/The_Montessori_method works book = OpenLibrary.books('1181252','oclc',jscmd='data') self.assertEqual(book.values()[0]['title'], 'The Montessori method') def test_read(self): results = OpenLibrary.read([(MASHUPBOOK_ISBN_10,'isbn'),('1181252','oclc')]) self.assertEqual(results['oclc:1181252']['records'].values()[0]['data']['ebooks'][0]['formats']['epub']['url'], 'http://www.archive.org/download/cu31924032538500/cu31924032538500.epub') def test_covers(self): self.assertEqual(OpenLibrary.covers(MASHUPBOOK_ISBN_10, size='M'), 'http://covers.openlibrary.org/b/isbn/159059858X-M.jpg') def test_author_photos(self): self.assertEqual(OpenLibrary.author_photos(RY_OLID,'S'), 'http://covers.openlibrary.org/a/olid/OL4264806A-S.jpg') def test_editions(self): # let's bring up the editions for Surfacing for (i,ed) in enumerate(islice(OpenLibrary.editions(SURFACING_WORK_OLID),100)): self.assertTrue(re.match(r'^OL(\d+)M$',ed)) def test_works0(self): self.assertEqual(OpenLibrary.works0(SURFACING_EDITION_OLID,id_type='olid')[0], 'OL675829W') def test_works(self): ids =[(MASHUPBOOK_ISBN_10, 'isbn'), (SURFACING_EDITION_OLID,'olid'), ('233434','isbn')] resp = list(OpenLibrary.works(ids)) self.assertEqual(resp, [['OL10306321W'], ['OL675829W'], []]) def test_json_for_olid(self): # manifestation # http://openlibrary.org/books/OL13439114M.json id = "OL13439114M" edition = OpenLibrary.json_for_olid(id) self.assertEqual(edition["title"], "Pro Web 2.0 Mashups") self.assertEqual(edition["identifiers"]["librarything"], ['2771144']) self.assertEqual(edition["subjects"], ['Mashups (World Wide Web)']) # work # http://openlibrary.org/works/OL10306321W.json id = "OL10306321W" work = OpenLibrary.json_for_olid(id) self.assertEqual(work["title"], "Pro Web 2.0 Mashups") self.assertEqual(work["type"]["key"], "/type/work") self.assertEqual(work["authors"][0]["type"]["key"], "/type/author_role") self.assertEqual(work["authors"][0]["author"]["key"], "/authors/OL4264806A") # author # http://openlibrary.org/authors/OL4264806A.json id = "OL4264806A" author = OpenLibrary.json_for_olid(id) self.assertEqual(author["name"], "Raymond Yee") # redirect ok? # "OL14917149W" -> "OL362684W" id = "OL14917149W" work = OpenLibrary.json_for_olid(id,follow_redirect=True) self.assertEqual(work["title"], "King Richard III") self.assertEqual(work["key"], "/works/OL362684W") work = OpenLibrary.json_for_olid(id,follow_redirect=False) self.assertEqual(work["type"]["key"], "/type/redirect") def test_xisbn(self): work_id = SURFACING_WORK_OLID surfacing_fb_id = '/m/05p_vg' book_isbn = '9780446311076' #for isbn in islice(OpenLibrary.xisbn(work_id=work_id),5): # print isbn fb = FreebaseBooks() gb = GoogleBooks(key=GOOGLE_BOOKS_KEY) fb_isbn_set = set(fb.xisbn(book_id=surfacing_fb_id)) ol_isbn_set = set(OpenLibrary.xisbn(isbn_val=book_isbn)) lt_isbn_set = set(map(lambda x: isbn_mod.ISBN(x).to_string('13'), thingisbn(SURFACING_ISBN))) print "Freebase set: ", len(fb_isbn_set), fb_isbn_set print "OpenLibrary set: ", len(ol_isbn_set), ol_isbn_set print "in both", len(fb_isbn_set & ol_isbn_set), fb_isbn_set & ol_isbn_set print "in fb but not ol", len(fb_isbn_set - ol_isbn_set), fb_isbn_set - ol_isbn_set print "in ol but not fb", len(ol_isbn_set - fb_isbn_set), ol_isbn_set - fb_isbn_set # compare thingisbn with ol print "thingisbn set:", len(lt_isbn_set), lt_isbn_set print "in both ol and lt", len(lt_isbn_set & ol_isbn_set), lt_isbn_set & ol_isbn_set print "in lt but not ol", len(lt_isbn_set - ol_isbn_set), lt_isbn_set - ol_isbn_set print "in ol but not lt", len(ol_isbn_set - lt_isbn_set), ol_isbn_set - lt_isbn_set # run through the intersection set and query Google Books for (i, isbn) in enumerate(fb_isbn_set & ol_isbn_set & lt_isbn_set): print i, isbn, gb.isbn(isbn) class WorkMapperTest(TestCase): def test_freebase_book_to_openlibrary_work(self): id = '/en/moby-dick' #id = '/en/wuthering_heights' work_ids = list(WorkMapper.freebase_book_to_openlibrary_work(id, complete_search=True)) print work_ids def test_work_info_from_openlibrary(self): editions = list(OpenLibrary.editions(SURFACING_WORK_OLID)) print editions, len(editions) class GoogleBooksTest(TestCase): def test_isbn(self): isbn_num = MASHUPBOOK_ISBN_13 gb = GoogleBooks(key=GOOGLE_BOOKS_KEY) item = gb.isbn(isbn_num) self.assertEqual(item['isbn'], '9781590598580') self.assertEqual(item['language'], 'en') def test_query(self): q = 'Bach' gb = GoogleBooks(key=GOOGLE_BOOKS_KEY) results = gb.query(q, glossed=True) def test_volumeid(self): g_id = 'B0xbAAAAMAAJ' gb = GoogleBooks(key=GOOGLE_BOOKS_KEY) results = gb.volumeid(g_id, glossed=True) print results class LibraryThingTest(TestCase): def test_lt_isbn(self): isbns = thingisbn(SURFACING_ISBN) # convert to isbn-13 isbns = map(lambda x: isbn_mod.ISBN(x).to_string('13'), isbns) self.assertTrue(SURFACING_ISBN in isbns) # grab a random ISBN from the list, issue another call and then check that the new list is the same isbns1 = map(lambda x: isbn_mod.ISBN(x).to_string('13'), thingisbn(random.sample(isbns,1)[0])) self.assertEqual(set(isbns), set(isbns1)) def test_whatwork(self): work_id = lt_whatwork(isbn=SURFACING_ISBN) self.assertEqual(work_id, SURFACING_LT_WORK_ID) work_id = lt_whatwork(title='Hamlet', author='Shakespeare') self.assertEqual(work_id, '2199') def test_cache(self): lt = LibraryThing() res = lt.thingisbn(SURFACING_ISBN) res2 = lt.thingisbn(SURFACING_ISBN) self.assertEqual(set(res), set(res2)) self.assertEqual(lt.whatwork(SURFACING_ISBN), SURFACING_LT_WORK_ID) self.assertEqual(lt.thingisbn(SURFACING_ISBN, return_work_id=True), SURFACING_LT_WORK_ID) def suite(): #testcases = [WorkMapperTest,FreebaseBooksTest, OpenLibraryTest,GoogleBooksTest] testcases = [] suites = unittest.TestSuite([unittest.TestLoader().loadTestsFromTestCase(testcase) for testcase in testcases]) suites.addTest(LibraryThingTest('test_cache')) #suites.addTest(SettingsTest('test_dev_me_alignment')) # give option to test this alignment return suites if __name__ == '__main__': #look_up_my_zotero_books_in_hathi() #ol_practice() #print len(list(islice(parse_project_gutenberg_catalog(),100000))) #unittest.main() suites = suite() #suites = unittest.defaultTestLoader.loadTestsFromModule(__import__('__main__')) unittest.TextTestRunner().run(suites)