dinislage/CouchPotatoServer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

readded tvdb_api

Browse Source
This commit is contained in:
Jason Mehring
2013-08-20 02:11:54 -04:00
parent 36950993f1
commit ff782669f6
15 changed files with 3812 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/tvdb_api/.gitignore
+4
View File
@@ -0,0 +1,4 @@
.DS_Store
*.pyc
*.egg-info/*
dist/*.tar.gz
libs/tvdb_api/.travis.yml
+9
View File
@@ -0,0 +1,9 @@
language: python
python:
- 2.5
- 2.6
- 2.7
install: pip install nose
script: nosetests
libs/tvdb_api/MANIFEST.in
+4
View File
@@ -0,0 +1,4 @@
include UNLICENSE
include readme.md
include tests/*.py
include Rakefile
libs/tvdb_api/Rakefile
+103
View File
@@ -0,0 +1,103 @@
require 'fileutils'
task :default => [:clean]
task :clean do
[".", "tests"].each do |cd|
puts "Cleaning directory #{cd}"
Dir.new(cd).each do |t|
if t =~ /.*\.pyc$/
puts "Removing #{File.join(cd, t)}"
File.delete(File.join(cd, t))
end
end
end
end
desc "Upversion files"
task :upversion do
puts "Upversioning"
Dir.glob("*.py").each do |filename|
f = File.new(filename, File::RDWR)
contents = f.read()
contents.gsub!(/__version__ = ".+?"/){|m|
cur_version = m.scan(/\d+\.\d+/)[0].to_f
new_version = cur_version + 0.1
puts "Current version: #{cur_version}"
puts "New version: #{new_version}"
new_line = "__version__ = \"#{new_version}\""
puts "Old line: #{m}"
puts "New line: #{new_line}"
m = new_line
}
puts contents[0]
f.truncate(0) # empty the existing file
f.seek(0)
f.write(contents.to_s) # write modified file
f.close()
end
end
desc "Upload current version to PyPi"
task :topypi => :test do
cur_file = File.open("tvdb_api.py").read()
tvdb_api_version = cur_file.scan(/__version__ = "(.*)"/)
tvdb_api_version = tvdb_api_version[0][0].to_f
puts "Build sdist and send tvdb_api v#{tvdb_api_version} to PyPi?"
if $stdin.gets.chomp == "y"
puts "Sending source-dist (sdist) to PyPi"
if system("python setup.py sdist register upload")
puts "tvdb_api uploaded!"
end
else
puts "Cancelled"
end
end
desc "Profile by running unittests"
task :profile do
cd "tests"
puts "Profiling.."
`python -m cProfile -o prof_runtest.prof runtests.py`
puts "Converting prof to dot"
`python gprof2dot.py -o prof_runtest.dot -f pstats prof_runtest.prof`
puts "Generating graph"
`~/Applications/dev/graphviz.app/Contents/macOS/dot -Tpng -o profile.png prof_runtest.dot -Gbgcolor=black`
puts "Cleanup"
rm "prof_runtest.dot"
rm "prof_runtest.prof"
end
task :test do
puts "Nosetest'ing"
if not system("nosetests -v --with-doctest")
raise "Test failed!"
end
puts "Doctesting *.py (excluding setup.py)"
Dir.glob("*.py").select{|e| ! e.match(/setup.py/)}.each do |filename|
if filename =~ /^setup\.py/
skip
end
puts "Doctesting #{filename}"
if not system("python", "-m", "doctest", filename)
raise "Failed doctest"
end
end
puts "Doctesting readme.md"
if not system("python", "-m", "doctest", "readme.md")
raise "Doctest"
end
end
libs/tvdb_api/UNLICENSE
+26
View File
@@ -0,0 +1,26 @@
Copyright 2011-2012 Ben Dickson (dbr)
This is free and unencumbered software released into the public domain.
Anyone is free to copy, modify, publish, use, compile, sell, or
distribute this software, either in source code form or as a compiled
binary, for any purpose, commercial or non-commercial, and by any
means.
In jurisdictions that recognize copyright laws, the author or authors
of this software dedicate any and all copyright interest in the
software to the public domain. We make this dedication for the benefit
of the public at large and to the detriment of our heirs and
successors. We intend this dedication to be an overt act of
relinquishment in perpetuity of all present and future rights to this
software under copyright law.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR
OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
OTHER DEALINGS IN THE SOFTWARE.
For more information, please refer to <http://unlicense.org/>
libs/tvdb_api/__init__.py
View File
libs/tvdb_api/readme.md
+109
View File
@@ -0,0 +1,109 @@
# `tvdb_api`
`tvdb_api` is an easy to use interface to [thetvdb.com][tvdb]
`tvnamer` has moved to a separate repository: [github.com/dbr/tvnamer][tvnamer] - it is a utility which uses `tvdb_api` to rename files from `some.show.s01e03.blah.abc.avi` to `Some Show - [01x03] - The Episode Name.avi` (which works by getting the episode name from `tvdb_api`)
[![Build Status](https://secure.travis-ci.org/dbr/tvdb_api.png?branch=master)](http://travis-ci.org/dbr/tvdb_api)
## To install
You can easily install `tvdb_api` via `easy_install`
easy_install tvdb_api
You may need to use sudo, depending on your setup:
sudo easy_install tvdb_api
The [`tvnamer`][tvnamer] command-line tool can also be installed via `easy_install`, this installs `tvdb_api` as a dependancy:
easy_install tvnamer
## Basic usage
import tvdb_api
t = tvdb_api.Tvdb()
episode = t['My Name Is Earl'][1][3] # get season 1, episode 3 of show
print episode['episodename'] # Print episode name
## Advanced usage
Most of the documentation is in docstrings. The examples are tested (using doctest) so will always be up to date and working.
The docstring for `Tvdb.__init__` lists all initialisation arguments, including support for non-English searches, custom "Select Series" interfaces and enabling the retrieval of banners and extended actor information. You can also override the default API key using `apikey`, recommended if you're using `tvdb_api` in a larger script or application
### Exceptions
There are several exceptions you may catch, these can be imported from `tvdb_api`:
- `tvdb_error` - this is raised when there is an error communicating with [thetvdb.com][tvdb] (a network error most commonly)
- `tvdb_userabort` - raised when a user aborts the Select Series dialog (by `ctrl+c`, or entering `q`)
- `tvdb_shownotfound` - raised when `t['show name']` cannot find anything
- `tvdb_seasonnotfound` - raised when the requested series (`t['show name][99]`) does not exist
- `tvdb_episodenotfound` - raised when the requested episode (`t['show name][1][99]`) does not exist.
- `tvdb_attributenotfound` - raised when the requested attribute is not found (`t['show name']['an attribute']`, `t['show name'][1]['an attribute']`, or ``t['show name'][1][1]['an attribute']``)
### Series data
All data exposed by [thetvdb.com][tvdb] is accessible via the `Show` class. A Show is retrieved by doing..
>>> import tvdb_api
>>> t = tvdb_api.Tvdb()
>>> show = t['scrubs']
>>> type(show)
<class 'tvdb_api.Show'>
For example, to find out what network Scrubs is aired:
>>> t['scrubs']['network']
u'ABC'
The data is stored in an attribute named `data`, within the Show instance:
>>> t['scrubs'].data.keys()
['networkid', 'rating', 'airs_dayofweek', 'contentrating', 'seriesname', 'id', 'airs_time', 'network', 'fanart', 'lastupdated', 'actors', 'ratingcount', 'status', 'added', 'poster', 'imdb_id', 'genre', 'banner', 'seriesid', 'language', 'zap2it_id', 'addedby', 'firstaired', 'runtime', 'overview']
Although each element is also accessible via `t['scrubs']` for ease-of-use:
>>> t['scrubs']['rating']
u'9.0'
This is the recommended way of retrieving "one-off" data (for example, if you are only interested in "seriesname"). If you wish to iterate over all data, or check if a particular show has a specific piece of data, use the `data` attribute,
>>> 'rating' in t['scrubs'].data
True
### Banners and actors
Since banners and actors are separate XML files, retrieving them by default is undesirable. If you wish to retrieve banners (and other fanart), use the `banners` Tvdb initialisation argument:
>>> from tvdb_api import Tvdb
>>> t = Tvdb(banners = True)
Then access the data using a `Show`'s `_banner` key:
>>> t['scrubs']['_banners'].keys()
['fanart', 'poster', 'series', 'season']
The banner data structure will be improved in future versions.
Extended actor data is accessible similarly:
>>> t = Tvdb(actors = True)
>>> actors = t['scrubs']['_actors']
>>> actors[0]
<Actor "Zach Braff">
>>> actors[0].keys()
['sortorder', 'image', 'role', 'id', 'name']
>>> actors[0]['role']
u'Dr. John Michael "J.D." Dorian'
Remember a simple list of actors is accessible via the default Show data:
>>> t['scrubs']['actors']
u'|Zach Braff|Donald Faison|Sarah Chalke|Christa Miller|Aloma Wright|Robert Maschio|Sam Lloyd|Neil Flynn|Ken Jenkins|Judy Reyes|John C. McGinley|Travis Schuldt|Johnny Kastl|Heather Graham|Michael Mosley|Kerry Bish\xe9|Dave Franco|Eliza Coupe|'
[tvdb]: http://thetvdb.com
[tvnamer]: http://github.com/dbr/tvnamer
libs/tvdb_api/setup.py
+35
View File
@@ -0,0 +1,35 @@
from setuptools import setup
setup(
name = 'tvdb_api',
version='1.8.2',
author='dbr/Ben',
description='Interface to thetvdb.com',
url='http://github.com/dbr/tvdb_api/tree/master',
license='unlicense',
long_description="""\
An easy to use API interface to TheTVDB.com
Basic usage is:
>>> import tvdb_api
>>> t = tvdb_api.Tvdb()
>>> ep = t['My Name Is Earl'][1][22]
>>> ep
<Episode 01x22 - Stole a Badge>
>>> ep['episodename']
u'Stole a Badge'
""",
py_modules = ['tvdb_api', 'tvdb_ui', 'tvdb_exceptions', 'tvdb_cache'],
classifiers=[
"Intended Audience :: Developers",
"Natural Language :: English",
"Operating System :: OS Independent",
"Programming Language :: Python",
"Topic :: Multimedia",
"Topic :: Utilities",
"Topic :: Software Development :: Libraries :: Python Modules",
]
)
libs/tvdb_api/tests/gprof2dot.py
+1638
View File
File diff suppressed because it is too large Load Diff
libs/tvdb_api/tests/runtests.py
Executable
+28
View File
@@ -0,0 +1,28 @@
#!/usr/bin/env python
#encoding:utf-8
#author:dbr/Ben
#project:tvdb_api
#repository:http://github.com/dbr/tvdb_api
#license:unlicense (http://unlicense.org/)
import sys
import unittest
import test_tvdb_api
def main():
suite = unittest.TestSuite([
unittest.TestLoader().loadTestsFromModule(test_tvdb_api)
])
runner = unittest.TextTestRunner(verbosity=2)
result = runner.run(suite)
if result.wasSuccessful():
return 0
else:
return 1
if __name__ == '__main__':
sys.exit(
int(main())
)
libs/tvdb_api/tests/test_tvdb_api.py
+526
View File
@@ -0,0 +1,526 @@
#!/usr/bin/env python
#encoding:utf-8
#author:dbr/Ben
#project:tvdb_api
#repository:http://github.com/dbr/tvdb_api
#license:unlicense (http://unlicense.org/)
"""Unittests for tvdb_api
"""
import os
import sys
import datetime
import unittest
# Force parent directory onto path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import tvdb_api
import tvdb_ui
from tvdb_api import (tvdb_shownotfound, tvdb_seasonnotfound,
tvdb_episodenotfound, tvdb_attributenotfound)
class test_tvdb_basic(unittest.TestCase):
# Used to store the cached instance of Tvdb()
t = None
def setUp(self):
if self.t is None:
self.__class__.t = tvdb_api.Tvdb(cache = True, banners = False)
def test_different_case(self):
"""Checks the auto-correction of show names is working.
It should correct the weirdly capitalised 'sCruBs' to 'Scrubs'
"""
self.assertEquals(self.t['scrubs'][1][4]['episodename'], 'My Old Lady')
self.assertEquals(self.t['sCruBs']['seriesname'], 'Scrubs')
def test_spaces(self):
"""Checks shownames with spaces
"""
self.assertEquals(self.t['My Name Is Earl']['seriesname'], 'My Name Is Earl')
self.assertEquals(self.t['My Name Is Earl'][1][4]['episodename'], 'Faked His Own Death')
def test_numeric(self):
"""Checks numeric show names
"""
self.assertEquals(self.t['24'][2][20]['episodename'], 'Day 2: 3:00 A.M.-4:00 A.M.')
self.assertEquals(self.t['24']['seriesname'], '24')
def test_show_iter(self):
"""Iterating over a show returns each seasons
"""
self.assertEquals(
len(
[season for season in self.t['Life on Mars']]
),
2
)
def test_season_iter(self):
"""Iterating over a show returns episodes
"""
self.assertEquals(
len(
[episode for episode in self.t['Life on Mars'][1]]
),
8
)
def test_get_episode_overview(self):
"""Checks episode overview is retrieved correctly.
"""
self.assertEquals(
self.t['Battlestar Galactica (2003)'][1][6]['overview'].startswith(
'When a new copy of Doral, a Cylon who had been previously'),
True
)
def test_get_parent(self):
"""Check accessing series from episode instance
"""
show = self.t['Battlestar Galactica (2003)']
season = show[1]
episode = show[1][1]
self.assertEquals(
season.show,
show
)
self.assertEquals(
episode.season,
season
)
self.assertEquals(
episode.season.show,
show
)
class test_tvdb_errors(unittest.TestCase):
# Used to store the cached instance of Tvdb()
t = None
def setUp(self):
if self.t is None:
self.__class__.t = tvdb_api.Tvdb(cache = True, banners = False)
def test_seasonnotfound(self):
"""Checks exception is thrown when season doesn't exist.
"""
self.assertRaises(tvdb_seasonnotfound, lambda:self.t['CNNNN'][10][1])
def test_shownotfound(self):
"""Checks exception is thrown when episode doesn't exist.
"""
self.assertRaises(tvdb_shownotfound, lambda:self.t['the fake show thingy'])
def test_episodenotfound(self):
"""Checks exception is raised for non-existent episode
"""
self.assertRaises(tvdb_episodenotfound, lambda:self.t['Scrubs'][1][30])
def test_attributenamenotfound(self):
"""Checks exception is thrown for if an attribute isn't found.
"""
self.assertRaises(tvdb_attributenotfound, lambda:self.t['CNNNN'][1][6]['afakeattributething'])
self.assertRaises(tvdb_attributenotfound, lambda:self.t['CNNNN']['afakeattributething'])
class test_tvdb_search(unittest.TestCase):
# Used to store the cached instance of Tvdb()
t = None
def setUp(self):
if self.t is None:
self.__class__.t = tvdb_api.Tvdb(cache = True, banners = False)
def test_search_len(self):
"""There should be only one result matching
"""
self.assertEquals(len(self.t['My Name Is Earl'].search('Faked His Own Death')), 1)
def test_search_checkname(self):
"""Checks you can get the episode name of a search result
"""
self.assertEquals(self.t['Scrubs'].search('my first')[0]['episodename'], 'My First Day')
self.assertEquals(self.t['My Name Is Earl'].search('Faked His Own Death')[0]['episodename'], 'Faked His Own Death')
def test_search_multiresults(self):
"""Checks search can return multiple results
"""
self.assertEquals(len(self.t['Scrubs'].search('my first')) >= 3, True)
def test_search_no_params_error(self):
"""Checks not supplying search info raises TypeError"""
self.assertRaises(
TypeError,
lambda: self.t['Scrubs'].search()
)
def test_search_season(self):
"""Checks the searching of a single season"""
self.assertEquals(
len(self.t['Scrubs'][1].search("First")),
3
)
def test_search_show(self):
"""Checks the searching of an entire show"""
self.assertEquals(
len(self.t['CNNNN'].search('CNNNN', key='episodename')),
3
)
def test_aired_on(self):
"""Tests airedOn show method"""
sr = self.t['Scrubs'].airedOn(datetime.date(2001, 10, 2))
self.assertEquals(len(sr), 1)
self.assertEquals(sr[0]['episodename'], u'My First Day')
class test_tvdb_data(unittest.TestCase):
# Used to store the cached instance of Tvdb()
t = None
def setUp(self):
if self.t is None:
self.__class__.t = tvdb_api.Tvdb(cache = True, banners = False)
def test_episode_data(self):
"""Check the firstaired value is retrieved
"""
self.assertEquals(
self.t['lost']['firstaired'],
'2004-09-22'
)
class test_tvdb_misc(unittest.TestCase):
# Used to store the cached instance of Tvdb()
t = None
def setUp(self):
if self.t is None:
self.__class__.t = tvdb_api.Tvdb(cache = True, banners = False)
def test_repr_show(self):
"""Check repr() of Season
"""
self.assertEquals(
repr(self.t['CNNNN']),
"<Show Chaser Non-Stop News Network (CNNNN) (containing 3 seasons)>"
)
def test_repr_season(self):
"""Check repr() of Season
"""
self.assertEquals(
repr(self.t['CNNNN'][1]),
"<Season instance (containing 9 episodes)>"
)
def test_repr_episode(self):
"""Check repr() of Episode
"""
self.assertEquals(
repr(self.t['CNNNN'][1][1]),
"<Episode 01x01 - Terror Alert>"
)
def test_have_all_languages(self):
"""Check valid_languages is up-to-date (compared to languages.xml)
"""
et = self.t._getetsrc(
"http://thetvdb.com/api/%s/languages.xml" % (
self.t.config['apikey']
)
)
languages = [x.find("abbreviation").text for x in et.findall("Language")]
self.assertEquals(
sorted(languages),
sorted(self.t.config['valid_languages'])
)
class test_tvdb_languages(unittest.TestCase):
def test_episode_name_french(self):
"""Check episode data is in French (language="fr")
"""
t = tvdb_api.Tvdb(cache = True, language = "fr")
self.assertEquals(
t['scrubs'][1][1]['episodename'],
"Mon premier jour"
)
self.assertTrue(
t['scrubs']['overview'].startswith(
u"J.D. est un jeune m\xe9decin qui d\xe9bute"
)
)
def test_episode_name_spanish(self):
"""Check episode data is in Spanish (language="es")
"""
t = tvdb_api.Tvdb(cache = True, language = "es")
self.assertEquals(
t['scrubs'][1][1]['episodename'],
"Mi Primer Dia"
)
self.assertTrue(
t['scrubs']['overview'].startswith(
u'Scrubs es una divertida comedia'
)
)
def test_multilanguage_selection(self):
"""Check selected language is used
"""
class SelectEnglishUI(tvdb_ui.BaseUI):
def selectSeries(self, allSeries):
return [x for x in allSeries if x['language'] == "en"][0]
class SelectItalianUI(tvdb_ui.BaseUI):
def selectSeries(self, allSeries):
return [x for x in allSeries if x['language'] == "it"][0]
t_en = tvdb_api.Tvdb(
cache=True,
custom_ui = SelectEnglishUI,
language = "en")
t_it = tvdb_api.Tvdb(
cache=True,
custom_ui = SelectItalianUI,
language = "it")
self.assertEquals(
t_en['dexter'][1][2]['episodename'], "Crocodile"
)
self.assertEquals(
t_it['dexter'][1][2]['episodename'], "Lacrime di coccodrillo"
)
class test_tvdb_unicode(unittest.TestCase):
def test_search_in_chinese(self):
"""Check searching for show with language=zh returns Chinese seriesname
"""
t = tvdb_api.Tvdb(cache = True, language = "zh")
show = t[u'T\xecnh Ng\u01b0\u1eddi Hi\u1ec7n \u0110\u1ea1i']
self.assertEquals(
type(show),
tvdb_api.Show
)
self.assertEquals(
show['seriesname'],
u'T\xecnh Ng\u01b0\u1eddi Hi\u1ec7n \u0110\u1ea1i'
)
def test_search_in_all_languages(self):
"""Check search_all_languages returns Chinese show, with language=en
"""
t = tvdb_api.Tvdb(cache = True, search_all_languages = True, language="en")
show = t[u'T\xecnh Ng\u01b0\u1eddi Hi\u1ec7n \u0110\u1ea1i']
self.assertEquals(
type(show),
tvdb_api.Show
)
self.assertEquals(
show['seriesname'],
u'Virtues Of Harmony II'
)
class test_tvdb_banners(unittest.TestCase):
# Used to store the cached instance of Tvdb()
t = None
def setUp(self):
if self.t is None:
self.__class__.t = tvdb_api.Tvdb(cache = True, banners = True)
def test_have_banners(self):
"""Check banners at least one banner is found
"""
self.assertEquals(
len(self.t['scrubs']['_banners']) > 0,
True
)
def test_banner_url(self):
"""Checks banner URLs start with http://
"""
for banner_type, banner_data in self.t['scrubs']['_banners'].items():
for res, res_data in banner_data.items():
for bid, banner_info in res_data.items():
self.assertEquals(
banner_info['_bannerpath'].startswith("http://"),
True
)
def test_episode_image(self):
"""Checks episode 'filename' image is fully qualified URL
"""
self.assertEquals(
self.t['scrubs'][1][1]['filename'].startswith("http://"),
True
)
def test_show_artwork(self):
"""Checks various image URLs within season data are fully qualified
"""
for key in ['banner', 'fanart', 'poster']:
self.assertEquals(
self.t['scrubs'][key].startswith("http://"),
True
)
class test_tvdb_actors(unittest.TestCase):
t = None
def setUp(self):
if self.t is None:
self.__class__.t = tvdb_api.Tvdb(cache = True, actors = True)
def test_actors_is_correct_datatype(self):
"""Check show/_actors key exists and is correct type"""
self.assertTrue(
isinstance(
self.t['scrubs']['_actors'],
tvdb_api.Actors
)
)
def test_actors_has_actor(self):
"""Check show has at least one Actor
"""
self.assertTrue(
isinstance(
self.t['scrubs']['_actors'][0],
tvdb_api.Actor
)
)
def test_actor_has_name(self):
"""Check first actor has a name"""
self.assertEquals(
self.t['scrubs']['_actors'][0]['name'],
"Zach Braff"
)
def test_actor_image_corrected(self):
"""Check image URL is fully qualified
"""
for actor in self.t['scrubs']['_actors']:
if actor['image'] is not None:
# Actor's image can be None, it displays as the placeholder
# image on thetvdb.com
self.assertTrue(
actor['image'].startswith("http://")
)
class test_tvdb_doctest(unittest.TestCase):
# Used to store the cached instance of Tvdb()
t = None
def setUp(self):
if self.t is None:
self.__class__.t = tvdb_api.Tvdb(cache = True, banners = False)
def test_doctest(self):
"""Check docstring examples works"""
import doctest
doctest.testmod(tvdb_api)
class test_tvdb_custom_caching(unittest.TestCase):
def test_true_false_string(self):
"""Tests setting cache to True/False/string
Basic tests, only checking for errors
"""
tvdb_api.Tvdb(cache = True)
tvdb_api.Tvdb(cache = False)
tvdb_api.Tvdb(cache = "/tmp")
def test_invalid_cache_option(self):
"""Tests setting cache to invalid value
"""
try:
tvdb_api.Tvdb(cache = 2.3)
except ValueError:
pass
else:
self.fail("Expected ValueError from setting cache to float")
def test_custom_urlopener(self):
class UsedCustomOpener(Exception):
pass
import urllib2
class TestOpener(urllib2.BaseHandler):
def default_open(self, request):
print request.get_method()
raise UsedCustomOpener("Something")
custom_opener = urllib2.build_opener(TestOpener())
t = tvdb_api.Tvdb(cache = custom_opener)
try:
t['scrubs']
except UsedCustomOpener:
pass
else:
self.fail("Did not use custom opener")
class test_tvdb_by_id(unittest.TestCase):
t = None
def setUp(self):
if self.t is None:
self.__class__.t = tvdb_api.Tvdb(cache = True, actors = True)
def test_actors_is_correct_datatype(self):
"""Check show/_actors key exists and is correct type"""
self.assertEquals(
self.t[76156]['seriesname'],
'Scrubs'
)
class test_tvdb_zip(unittest.TestCase):
# Used to store the cached instance of Tvdb()
t = None
def setUp(self):
if self.t is None:
self.__class__.t = tvdb_api.Tvdb(cache = True, useZip = True)
def test_get_series_from_zip(self):
"""
"""
self.assertEquals(self.t['scrubs'][1][4]['episodename'], 'My Old Lady')
self.assertEquals(self.t['sCruBs']['seriesname'], 'Scrubs')
def test_spaces_from_zip(self):
"""Checks shownames with spaces
"""
self.assertEquals(self.t['My Name Is Earl']['seriesname'], 'My Name Is Earl')
self.assertEquals(self.t['My Name Is Earl'][1][4]['episodename'], 'Faked His Own Death')
class test_tvdb_show_search(unittest.TestCase):
# Used to store the cached instance of Tvdb()
t = None
def setUp(self):
if self.t is None:
self.__class__.t = tvdb_api.Tvdb(cache = True, useZip = True)
def test_search(self):
"""Test Tvdb.search method
"""
results = self.t.search("my name is earl")
all_ids = [x['seriesid'] for x in results]
self.assertTrue('75397' in all_ids)
if __name__ == '__main__':
runner = unittest.TextTestRunner(verbosity = 2)
unittest.main(testRunner = runner)
libs/tvdb_api/tvdb_api.py
+874
View File
@@ -0,0 +1,874 @@
#!/usr/bin/env python
#encoding:utf-8
#author:dbr/Ben
#project:tvdb_api
#repository:http://github.com/dbr/tvdb_api
#license:unlicense (http://unlicense.org/)
"""Simple-to-use Python interface to The TVDB's API (thetvdb.com)
Example usage:
>>> from tvdb_api import Tvdb
>>> t = Tvdb()
>>> t['Lost'][4][11]['episodename']
u'Cabin Fever'
"""
__author__ = "dbr/Ben"
__version__ = "1.8.2"
import os
import time
import urllib
import urllib2
import getpass
import StringIO
import tempfile
import warnings
import logging
import datetime
import zipfile
try:
import xml.etree.cElementTree as ElementTree
except ImportError:
import xml.etree.ElementTree as ElementTree
try:
import gzip
except ImportError:
gzip = None
from tvdb_cache import CacheHandler
from tvdb_ui import BaseUI, ConsoleUI
from tvdb_exceptions import (tvdb_error, tvdb_userabort, tvdb_shownotfound,
tvdb_seasonnotfound, tvdb_episodenotfound, tvdb_attributenotfound)
lastTimeout = None
def log():
return logging.getLogger("tvdb_api")
class ShowContainer(dict):
"""Simple dict that holds a series of Show instances
"""
def __init__(self):
self._stack = []
self._lastgc = time.time()
def __setitem__(self, key, value):
self._stack.append(key)
#keep only the 100th latest results
if time.time() - self._lastgc > 20:
tbd = self._stack[:-100]
i = 0
for o in tbd:
del self[o]
del self._stack[i]
i += 1
_lastgc = time.time()
del tbd
super(ShowContainer, self).__setitem__(key, value)
class Show(dict):
"""Holds a dict of seasons, and show data.
"""
def __init__(self):
dict.__init__(self)
self.data = {}
def __repr__(self):
return "<Show %s (containing %s seasons)>" % (
self.data.get(u'seriesname', 'instance'),
len(self)
)
def __getitem__(self, key):
if key in self:
# Key is an episode, return it
return dict.__getitem__(self, key)
if key in self.data:
# Non-numeric request is for show-data
return dict.__getitem__(self.data, key)
# Data wasn't found, raise appropriate error
if isinstance(key, int) or key.isdigit():
# Episode number x was not found
raise tvdb_seasonnotfound("Could not find season %s" % (repr(key)))
else:
# If it's not numeric, it must be an attribute name, which
# doesn't exist, so attribute error.
raise tvdb_attributenotfound("Cannot find attribute %s" % (repr(key)))
def airedOn(self, date):
ret = self.search(str(date), 'firstaired')
if len(ret) == 0:
raise tvdb_episodenotfound("Could not find any episodes that aired on %s" % date)
return ret
def search(self, term = None, key = None):
"""
Search all episodes in show. Can search all data, or a specific key (for
example, episodename)
Always returns an array (can be empty). First index contains the first
match, and so on.
Each array index is an Episode() instance, so doing
search_results[0]['episodename'] will retrieve the episode name of the
first match.
Search terms are converted to lower case (unicode) strings.
# Examples
These examples assume t is an instance of Tvdb():
>>> t = Tvdb()
>>>
To search for all episodes of Scrubs with a bit of data
containing "my first day":
>>> t['Scrubs'].search("my first day")
[<Episode 01x01 - My First Day>]
>>>
Search for "My Name Is Earl" episode named "Faked His Own Death":
>>> t['My Name Is Earl'].search('Faked His Own Death', key = 'episodename')
[<Episode 01x04 - Faked His Own Death>]
>>>
To search Scrubs for all episodes with "mentor" in the episode name:
>>> t['scrubs'].search('mentor', key = 'episodename')
[<Episode 01x02 - My Mentor>, <Episode 03x15 - My Tormented Mentor>]
>>>
# Using search results
>>> results = t['Scrubs'].search("my first")
>>> print results[0]['episodename']
My First Day
>>> for x in results: print x['episodename']
My First Day
My First Step
My First Kill
>>>
"""
results = []
for cur_season in self.values():
searchresult = cur_season.search(term = term, key = key)
if len(searchresult) != 0:
results.extend(searchresult)
return results
class Season(dict):
def __init__(self, show = None):
"""The show attribute points to the parent show
"""
self.show = show
def __repr__(self):
return "<Season instance (containing %s episodes)>" % (
len(self.keys())
)
def __getitem__(self, episode_number):
if episode_number not in self:
raise tvdb_episodenotfound("Could not find episode %s" % (repr(episode_number)))
else:
return dict.__getitem__(self, episode_number)
def search(self, term = None, key = None):
"""Search all episodes in season, returns a list of matching Episode
instances.
>>> t = Tvdb()
>>> t['scrubs'][1].search('first day')
[<Episode 01x01 - My First Day>]
>>>
See Show.search documentation for further information on search
"""
results = []
for ep in self.values():
searchresult = ep.search(term = term, key = key)
if searchresult is not None:
results.append(
searchresult
)
return results
class Episode(dict):
def __init__(self, season = None):
"""The season attribute points to the parent season
"""
self.season = season
def __repr__(self):
seasno = int(self.get(u'seasonnumber', 0))
epno = int(self.get(u'episodenumber', 0))
epname = self.get(u'episodename')
if epname is not None:
return "<Episode %02dx%02d - %s>" % (seasno, epno, epname)
else:
return "<Episode %02dx%02d>" % (seasno, epno)
def __getitem__(self, key):
try:
return dict.__getitem__(self, key)
except KeyError:
raise tvdb_attributenotfound("Cannot find attribute %s" % (repr(key)))
def search(self, term = None, key = None):
"""Search episode data for term, if it matches, return the Episode (self).
The key parameter can be used to limit the search to a specific element,
for example, episodename.
This primarily for use use by Show.search and Season.search. See
Show.search for further information on search
Simple example:
>>> e = Episode()
>>> e['episodename'] = "An Example"
>>> e.search("examp")
<Episode 00x00 - An Example>
>>>
Limiting by key:
>>> e.search("examp", key = "episodename")
<Episode 00x00 - An Example>
>>>
"""
if term == None:
raise TypeError("must supply string to search for (contents)")
term = unicode(term).lower()
for cur_key, cur_value in self.items():
cur_key, cur_value = unicode(cur_key).lower(), unicode(cur_value).lower()
if key is not None and cur_key != key:
# Do not search this key
continue
if cur_value.find( unicode(term).lower() ) > -1:
return self
class Actors(list):
"""Holds all Actor instances for a show
"""
pass
class Actor(dict):
"""Represents a single actor. Should contain..
id,
image,
name,
role,
sortorder
"""
def __repr__(self):
return "<Actor \"%s\">" % (self.get("name"))
class Tvdb:
"""Create easy-to-use interface to name of season/episode name
>>> t = Tvdb()
>>> t['Scrubs'][1][24]['episodename']
u'My Last Day'
"""
def __init__(self,
interactive = False,
select_first = False,
debug = False,
cache = True,
banners = False,
actors = False,
custom_ui = None,
language = None,
search_all_languages = False,
apikey = None,
forceConnect=False,
useZip=False):
"""interactive (True/False):
When True, uses built-in console UI is used to select the correct show.
When False, the first search result is used.
select_first (True/False):
Automatically selects the first series search result (rather
than showing the user a list of more than one series).
Is overridden by interactive = False, or specifying a custom_ui
debug (True/False) DEPRECATED:
Replaced with proper use of logging module. To show debug messages:
>>> import logging
>>> logging.basicConfig(level = logging.DEBUG)
cache (True/False/str/unicode/urllib2 opener):
Retrieved XML are persisted to to disc. If true, stores in
tvdb_api folder under your systems TEMP_DIR, if set to
str/unicode instance it will use this as the cache
location. If False, disables caching. Can also be passed
an arbitrary Python object, which is used as a urllib2
opener, which should be created by urllib2.build_opener
banners (True/False):
Retrieves the banners for a show. These are accessed
via the _banners key of a Show(), for example:
>>> Tvdb(banners=True)['scrubs']['_banners'].keys()
['fanart', 'poster', 'series', 'season']
actors (True/False):
Retrieves a list of the actors for a show. These are accessed
via the _actors key of a Show(), for example:
>>> t = Tvdb(actors=True)
>>> t['scrubs']['_actors'][0]['name']
u'Zach Braff'
custom_ui (tvdb_ui.BaseUI subclass):
A callable subclass of tvdb_ui.BaseUI (overrides interactive option)
language (2 character language abbreviation):
The language of the returned data. Is also the language search
uses. Default is "en" (English). For full list, run..
>>> Tvdb().config['valid_languages'] #doctest: +ELLIPSIS
['da', 'fi', 'nl', ...]
search_all_languages (True/False):
By default, Tvdb will only search in the language specified using
the language option. When this is True, it will search for the
show in and language
apikey (str/unicode):
Override the default thetvdb.com API key. By default it will use
tvdb_api's own key (fine for small scripts), but you can use your
own key if desired - this is recommended if you are embedding
tvdb_api in a larger application)
See http://thetvdb.com/?tab=apiregister to get your own key
forceConnect (bool):
If true it will always try to connect to theTVDB.com even if we
recently timed out. By default it will wait one minute before
trying again, and any requests within that one minute window will
return an exception immediately.
useZip (bool):
Download the zip archive where possibale, instead of the xml.
This is only used when all episodes are pulled.
And only the main language xml is used, the actor and banner xml are lost.
"""
global lastTimeout
# if we're given a lastTimeout that is less than 1 min just give up
if not forceConnect and lastTimeout != None and datetime.datetime.now() - lastTimeout < datetime.timedelta(minutes=1):
raise tvdb_error("We recently timed out, so giving up early this time")
self.shows = ShowContainer() # Holds all Show classes
self.corrections = {} # Holds show-name to show_id mapping
self.config = {}
if apikey is not None:
self.config['apikey'] = apikey
else:
self.config['apikey'] = "0629B785CE550C8D" # tvdb_api's API key
self.config['debug_enabled'] = debug # show debugging messages
self.config['custom_ui'] = custom_ui
self.config['interactive'] = interactive # prompt for correct series?
self.config['select_first'] = select_first
self.config['search_all_languages'] = search_all_languages
self.config['useZip'] = useZip
if cache is True:
self.config['cache_enabled'] = True
self.config['cache_location'] = self._getTempDir()
self.urlopener = urllib2.build_opener(
CacheHandler(self.config['cache_location'])
)
elif cache is False:
self.config['cache_enabled'] = False
self.urlopener = urllib2.build_opener() # default opener with no caching
elif isinstance(cache, basestring):
self.config['cache_enabled'] = True
self.config['cache_location'] = cache
self.urlopener = urllib2.build_opener(
CacheHandler(self.config['cache_location'])
)
elif isinstance(cache, urllib2.OpenerDirector):
# If passed something from urllib2.build_opener, use that
log().debug("Using %r as urlopener" % cache)
self.config['cache_enabled'] = True
self.urlopener = cache
else:
raise ValueError("Invalid value for Cache %r (type was %s)" % (cache, type(cache)))
self.config['banners_enabled'] = banners
self.config['actors_enabled'] = actors
if self.config['debug_enabled']:
warnings.warn("The debug argument to tvdb_api.__init__ will be removed in the next version. "
"To enable debug messages, use the following code before importing: "
"import logging; logging.basicConfig(level=logging.DEBUG)")
logging.basicConfig(level=logging.DEBUG)
# List of language from http://thetvdb.com/api/0629B785CE550C8D/languages.xml
# Hard-coded here as it is realtively static, and saves another HTTP request, as
# recommended on http://thetvdb.com/wiki/index.php/API:languages.xml
self.config['valid_languages'] = [
"da", "fi", "nl", "de", "it", "es", "fr","pl", "hu","el","tr",
"ru","he","ja","pt","zh","cs","sl", "hr","ko","en","sv","no"
]
# thetvdb.com should be based around numeric language codes,
# but to link to a series like http://thetvdb.com/?tab=series&id=79349&lid=16
# requires the language ID, thus this mapping is required (mainly
# for usage in tvdb_ui - internally tvdb_api will use the language abbreviations)
self.config['langabbv_to_id'] = {'el': 20, 'en': 7, 'zh': 27,
'it': 15, 'cs': 28, 'es': 16, 'ru': 22, 'nl': 13, 'pt': 26, 'no': 9,
'tr': 21, 'pl': 18, 'fr': 17, 'hr': 31, 'de': 14, 'da': 10, 'fi': 11,
'hu': 19, 'ja': 25, 'he': 24, 'ko': 32, 'sv': 8, 'sl': 30}
if language is None:
self.config['language'] = 'en'
else:
if language not in self.config['valid_languages']:
raise ValueError("Invalid language %s, options are: %s" % (
language, self.config['valid_languages']
))
else:
self.config['language'] = language
# The following url_ configs are based of the
# http://thetvdb.com/wiki/index.php/Programmers_API
self.config['base_url'] = "http://thetvdb.com"
if self.config['search_all_languages']:
self.config['url_getSeries'] = u"%(base_url)s/api/GetSeries.php?seriesname=%%s&language=all" % self.config
else:
self.config['url_getSeries'] = u"%(base_url)s/api/GetSeries.php?seriesname=%%s&language=%(language)s" % self.config
self.config['url_epInfo'] = u"%(base_url)s/api/%(apikey)s/series/%%s/all/%%s.xml" % self.config
self.config['url_epInfo_zip'] = u"%(base_url)s/api/%(apikey)s/series/%%s/all/%%s.zip" % self.config
self.config['url_seriesInfo'] = u"%(base_url)s/api/%(apikey)s/series/%%s/%%s.xml" % self.config
self.config['url_actorsInfo'] = u"%(base_url)s/api/%(apikey)s/series/%%s/actors.xml" % self.config
self.config['url_seriesBanner'] = u"%(base_url)s/api/%(apikey)s/series/%%s/banners.xml" % self.config
self.config['url_artworkPrefix'] = u"%(base_url)s/banners/%%s" % self.config
def _getTempDir(self):
"""Returns the [system temp dir]/tvdb_api-u501 (or
tvdb_api-myuser)
"""
if hasattr(os, 'getuid'):
uid = "u%d" % (os.getuid())
else:
# For Windows
try:
uid = getpass.getuser()
except ImportError:
return os.path.join(tempfile.gettempdir(), "tvdb_api")
return os.path.join(tempfile.gettempdir(), "tvdb_api-%s" % (uid))
def _loadUrl(self, url, recache = False, language=None):
global lastTimeout
try:
log().debug("Retrieving URL %s" % url)
resp = self.urlopener.open(url)
if 'x-local-cache' in resp.headers:
log().debug("URL %s was cached in %s" % (
url,
resp.headers['x-local-cache'])
)
if recache:
log().debug("Attempting to recache %s" % url)
resp.recache()
except (IOError, urllib2.URLError), errormsg:
if not str(errormsg).startswith('HTTP Error'):
lastTimeout = datetime.datetime.now()
raise tvdb_error("Could not connect to server: %s" % (errormsg))
# handle gzipped content,
# http://dbr.lighthouseapp.com/projects/13342/tickets/72-gzipped-data-patch
if 'gzip' in resp.headers.get("Content-Encoding", ''):
if gzip:
stream = StringIO.StringIO(resp.read())
gz = gzip.GzipFile(fileobj=stream)
return gz.read()
raise tvdb_error("Received gzip data from thetvdb.com, but could not correctly handle it")
if 'application/zip' in resp.headers.get("Content-Type", ''):
try:
# TODO: The zip contains actors.xml and banners.xml, which are currently ignored [GH-20]
log().debug("We recived a zip file unpacking now ...")
zipdata = StringIO.StringIO()
zipdata.write(resp.read())
myzipfile = zipfile.ZipFile(zipdata)
return myzipfile.read('%s.xml' % language)
except zipfile.BadZipfile:
if 'x-local-cache' in resp.headers:
resp.delete_cache()
raise tvdb_error("Bad zip file received from thetvdb.com, could not read it")
return resp.read()
def _getetsrc(self, url, language=None):
"""Loads a URL using caching, returns an ElementTree of the source
"""
src = self._loadUrl(url, language=language)
try:
# TVDB doesn't sanitize \r (CR) from user input in some fields,
# remove it to avoid errors. Change from SickBeard, from will14m
return ElementTree.fromstring(src.rstrip("\r"))
except SyntaxError:
src = self._loadUrl(url, recache=True, language=language)
try:
return ElementTree.fromstring(src.rstrip("\r"))
except SyntaxError, exceptionmsg:
errormsg = "There was an error with the XML retrieved from thetvdb.com:\n%s" % (
exceptionmsg
)
if self.config['cache_enabled']:
errormsg += "\nFirst try emptying the cache folder at..\n%s" % (
self.config['cache_location']
)
errormsg += "\nIf this does not resolve the issue, please try again later. If the error persists, report a bug on"
errormsg += "\nhttp://dbr.lighthouseapp.com/projects/13342-tvdb_api/overview\n"
raise tvdb_error(errormsg)
def _setItem(self, sid, seas, ep, attrib, value):
"""Creates a new episode, creating Show(), Season() and
Episode()s as required. Called by _getShowData to populate show
Since the nice-to-use tvdb[1][24]['name] interface
makes it impossible to do tvdb[1][24]['name] = "name"
and still be capable of checking if an episode exists
so we can raise tvdb_shownotfound, we have a slightly
less pretty method of setting items.. but since the API
is supposed to be read-only, this is the best way to
do it!
The problem is that calling tvdb[1][24]['episodename'] = "name"
calls __getitem__ on tvdb[1], there is no way to check if
tvdb.__dict__ should have a key "1" before we auto-create it
"""
if sid not in self.shows:
self.shows[sid] = Show()
if seas not in self.shows[sid]:
self.shows[sid][seas] = Season(show = self.shows[sid])
if ep not in self.shows[sid][seas]:
self.shows[sid][seas][ep] = Episode(season = self.shows[sid][seas])
self.shows[sid][seas][ep][attrib] = value
def _setShowData(self, sid, key, value):
"""Sets self.shows[sid] to a new Show instance, or sets the data
"""
if sid not in self.shows:
self.shows[sid] = Show()
self.shows[sid].data[key] = value
def _cleanData(self, data):
"""Cleans up strings returned by TheTVDB.com
Issues corrected:
- Replaces &amp; with &
- Trailing whitespace
"""
data = data.replace(u"&amp;", u"&")
data = data.strip()
return data
def search(self, series):
"""This searches TheTVDB.com for the series name
and returns the result list
"""
series = urllib.quote(series.encode("utf-8"))
log().debug("Searching for show %s" % series)
seriesEt = self._getetsrc(self.config['url_getSeries'] % (series))
allSeries = []
for series in seriesEt:
result = dict((k.tag.lower(), k.text) for k in series.getchildren())
result['id'] = int(result['id'])
result['lid'] = self.config['langabbv_to_id'][result['language']]
log().debug('Found series %(seriesname)s' % result)
allSeries.append(result)
return allSeries
def _getSeries(self, series):
"""This searches TheTVDB.com for the series name,
If a custom_ui UI is configured, it uses this to select the correct
series. If not, and interactive == True, ConsoleUI is used, if not
BaseUI is used to select the first result.
"""
allSeries = self.search(series)
if len(allSeries) == 0:
log().debug('Series result returned zero')
raise tvdb_shownotfound("Show-name search returned zero results (cannot find show on TVDB)")
if self.config['custom_ui'] is not None:
log().debug("Using custom UI %s" % (repr(self.config['custom_ui'])))
ui = self.config['custom_ui'](config = self.config)
else:
if not self.config['interactive']:
log().debug('Auto-selecting first search result using BaseUI')
ui = BaseUI(config = self.config)
else:
log().debug('Interactively selecting show using ConsoleUI')
ui = ConsoleUI(config = self.config)
return ui.selectSeries(allSeries)
def _parseBanners(self, sid):
"""Parses banners XML, from
http://thetvdb.com/api/[APIKEY]/series/[SERIES ID]/banners.xml
Banners are retrieved using t['show name]['_banners'], for example:
>>> t = Tvdb(banners = True)
>>> t['scrubs']['_banners'].keys()
['fanart', 'poster', 'series', 'season']
>>> t['scrubs']['_banners']['poster']['680x1000']['35308']['_bannerpath']
u'http://thetvdb.com/banners/posters/76156-2.jpg'
>>>
Any key starting with an underscore has been processed (not the raw
data from the XML)
This interface will be improved in future versions.
"""
log().debug('Getting season banners for %s' % (sid))
bannersEt = self._getetsrc( self.config['url_seriesBanner'] % (sid) )
banners = {}
for cur_banner in bannersEt.findall('Banner'):
bid = cur_banner.find('id').text
btype = cur_banner.find('BannerType')
btype2 = cur_banner.find('BannerType2')
if btype is None or btype2 is None:
continue
btype, btype2 = btype.text, btype2.text
if not btype in banners:
banners[btype] = {}
if not btype2 in banners[btype]:
banners[btype][btype2] = {}
if not bid in banners[btype][btype2]:
banners[btype][btype2][bid] = {}
for cur_element in cur_banner.getchildren():
tag = cur_element.tag.lower()
value = cur_element.text
if tag is None or value is None:
continue
tag, value = tag.lower(), value.lower()
banners[btype][btype2][bid][tag] = value
for k, v in banners[btype][btype2][bid].items():
if k.endswith("path"):
new_key = "_%s" % (k)
log().debug("Transforming %s to %s" % (k, new_key))
new_url = self.config['url_artworkPrefix'] % (v)
banners[btype][btype2][bid][new_key] = new_url
self._setShowData(sid, "_banners", banners)
def _parseActors(self, sid):
"""Parsers actors XML, from
http://thetvdb.com/api/[APIKEY]/series/[SERIES ID]/actors.xml
Actors are retrieved using t['show name]['_actors'], for example:
>>> t = Tvdb(actors = True)
>>> actors = t['scrubs']['_actors']
>>> type(actors)
<class 'tvdb_api.Actors'>
>>> type(actors[0])
<class 'tvdb_api.Actor'>
>>> actors[0]
<Actor "Zach Braff">
>>> sorted(actors[0].keys())
['id', 'image', 'name', 'role', 'sortorder']
>>> actors[0]['name']
u'Zach Braff'
>>> actors[0]['image']
u'http://thetvdb.com/banners/actors/43640.jpg'
Any key starting with an underscore has been processed (not the raw
data from the XML)
"""
log().debug("Getting actors for %s" % (sid))
actorsEt = self._getetsrc(self.config['url_actorsInfo'] % (sid))
cur_actors = Actors()
for curActorItem in actorsEt.findall("Actor"):
curActor = Actor()
for curInfo in curActorItem:
tag = curInfo.tag.lower()
value = curInfo.text
if value is not None:
if tag == "image":
value = self.config['url_artworkPrefix'] % (value)
else:
value = self._cleanData(value)
curActor[tag] = value
cur_actors.append(curActor)
self._setShowData(sid, '_actors', cur_actors)
def _getShowData(self, sid, language):
"""Takes a series ID, gets the epInfo URL and parses the TVDB
XML file into the shows dict in layout:
shows[series_id][season_number][episode_number]
"""
if self.config['language'] is None:
log().debug('Config language is none, using show language')
if language is None:
raise tvdb_error("config['language'] was None, this should not happen")
getShowInLanguage = language
else:
log().debug(
'Configured language %s override show language of %s' % (
self.config['language'],
language
)
)
getShowInLanguage = self.config['language']
# Parse show information
log().debug('Getting all series data for %s' % (sid))
seriesInfoEt = self._getetsrc(
self.config['url_seriesInfo'] % (sid, getShowInLanguage)
)
for curInfo in seriesInfoEt.findall("Series")[0]:
tag = curInfo.tag.lower()
value = curInfo.text
if value is not None:
if tag in ['banner', 'fanart', 'poster']:
value = self.config['url_artworkPrefix'] % (value)
else:
value = self._cleanData(value)
self._setShowData(sid, tag, value)
# Parse banners
if self.config['banners_enabled']:
self._parseBanners(sid)
# Parse actors
if self.config['actors_enabled']:
self._parseActors(sid)
# Parse episode data
log().debug('Getting all episodes of %s' % (sid))
if self.config['useZip']:
url = self.config['url_epInfo_zip'] % (sid, language)
else:
url = self.config['url_epInfo'] % (sid, language)
epsEt = self._getetsrc( url, language=language)
for cur_ep in epsEt.findall("Episode"):
seas_no = int(cur_ep.find('SeasonNumber').text)
ep_no = int(cur_ep.find('EpisodeNumber').text)
for cur_item in cur_ep.getchildren():
tag = cur_item.tag.lower()
value = cur_item.text
if value is not None:
if tag == 'filename':
value = self.config['url_artworkPrefix'] % (value)
else:
value = self._cleanData(value)
self._setItem(sid, seas_no, ep_no, tag, value)
def _nameToSid(self, name):
"""Takes show name, returns the correct series ID (if the show has
already been grabbed), or grabs all episodes and returns
the correct SID.
"""
if name in self.corrections:
log().debug('Correcting %s to %s' % (name, self.corrections[name]) )
sid = self.corrections[name]
else:
log().debug('Getting show %s' % (name))
selected_series = self._getSeries( name )
sname, sid = selected_series['seriesname'], selected_series['id']
log().debug('Got %(seriesname)s, id %(id)s' % selected_series)
self.corrections[name] = sid
self._getShowData(selected_series['id'], selected_series['language'])
return sid
def __getitem__(self, key):
"""Handles tvdb_instance['seriesname'] calls.
The dict index should be the show id
"""
if isinstance(key, (int, long)):
# Item is integer, treat as show id
if key not in self.shows:
self._getShowData(key, self.config['language'])
return self.shows[key]
key = key.lower() # make key lower case
sid = self._nameToSid(key)
log().debug('Got series id %s' % (sid))
return self.shows[sid]
def __repr__(self):
return str(self.shows)
def main():
"""Simple example of using tvdb_api - it just
grabs an episode name interactively.
"""
import logging
logging.basicConfig(level=logging.DEBUG)
tvdb_instance = Tvdb(interactive=True, cache=False)
print tvdb_instance['Lost']['seriesname']
print tvdb_instance['Lost'][1][4]['episodename']
if __name__ == '__main__':
main()
libs/tvdb_api/tvdb_cache.py
+251
View File
@@ -0,0 +1,251 @@
#!/usr/bin/env python
#encoding:utf-8
#author:dbr/Ben
#project:tvdb_api
#repository:http://github.com/dbr/tvdb_api
#license:unlicense (http://unlicense.org/)
"""
urllib2 caching handler
Modified from http://code.activestate.com/recipes/491261/
"""
from __future__ import with_statement
__author__ = "dbr/Ben"
__version__ = "1.8.2"
import os
import time
import errno
import httplib
import urllib2
import StringIO
from hashlib import md5
from threading import RLock
cache_lock = RLock()
def locked_function(origfunc):
"""Decorator to execute function under lock"""
def wrapped(*args, **kwargs):
cache_lock.acquire()
try:
return origfunc(*args, **kwargs)
finally:
cache_lock.release()
return wrapped
def calculate_cache_path(cache_location, url):
"""Checks if [cache_location]/[hash_of_url].headers and .body exist
"""
thumb = md5(url).hexdigest()
header = os.path.join(cache_location, thumb + ".headers")
body = os.path.join(cache_location, thumb + ".body")
return header, body
def check_cache_time(path, max_age):
"""Checks if a file has been created/modified in the [last max_age] seconds.
False means the file is too old (or doesn't exist), True means it is
up-to-date and valid"""
if not os.path.isfile(path):
return False
cache_modified_time = os.stat(path).st_mtime
time_now = time.time()
if cache_modified_time < time_now - max_age:
# Cache is old
return False
else:
return True
@locked_function
def exists_in_cache(cache_location, url, max_age):
"""Returns if header AND body cache file exist (and are up-to-date)"""
hpath, bpath = calculate_cache_path(cache_location, url)
if os.path.exists(hpath) and os.path.exists(bpath):
return(
check_cache_time(hpath, max_age)
and check_cache_time(bpath, max_age)
)
else:
# File does not exist
return False
@locked_function
def store_in_cache(cache_location, url, response):
"""Tries to store response in cache."""
hpath, bpath = calculate_cache_path(cache_location, url)
try:
outf = open(hpath, "wb")
headers = str(response.info())
outf.write(headers)
outf.close()
outf = open(bpath, "wb")
outf.write(response.read())
outf.close()
except IOError:
return True
else:
return False
@locked_function
def delete_from_cache(cache_location, url):
"""Deletes a response in cache."""
hpath, bpath = calculate_cache_path(cache_location, url)
try:
if os.path.exists(hpath):
os.remove(hpath)
if os.path.exists(bpath):
os.remove(bpath)
except IOError:
return True
else:
return False
class CacheHandler(urllib2.BaseHandler):
"""Stores responses in a persistant on-disk cache.
If a subsequent GET request is made for the same URL, the stored
response is returned, saving time, resources and bandwidth
"""
@locked_function
def __init__(self, cache_location, max_age = 21600):
"""The location of the cache directory"""
self.max_age = max_age
self.cache_location = cache_location
if not os.path.exists(self.cache_location):
try:
os.mkdir(self.cache_location)
except OSError, e:
if e.errno == errno.EEXIST and os.path.isdir(self.cache_location):
# File exists, and it's a directory,
# another process beat us to creating this dir, that's OK.
pass
else:
# Our target dir is already a file, or different error,
# relay the error!
raise
def default_open(self, request):
"""Handles GET requests, if the response is cached it returns it
"""
if request.get_method() is not "GET":
return None # let the next handler try to handle the request
if exists_in_cache(
self.cache_location, request.get_full_url(), self.max_age
):
return CachedResponse(
self.cache_location,
request.get_full_url(),
set_cache_header = True
)
else:
return None
def http_response(self, request, response):
"""Gets a HTTP response, if it was a GET request and the status code
starts with 2 (200 OK etc) it caches it and returns a CachedResponse
"""
if (request.get_method() == "GET"
and str(response.code).startswith("2")
):
if 'x-local-cache' not in response.info():
# Response is not cached
set_cache_header = store_in_cache(
self.cache_location,
request.get_full_url(),
response
)
else:
set_cache_header = True
return CachedResponse(
self.cache_location,
request.get_full_url(),
set_cache_header = set_cache_header
)
else:
return response
class CachedResponse(StringIO.StringIO):
"""An urllib2.response-like object for cached responses.
To determine if a response is cached or coming directly from
the network, check the x-local-cache header rather than the object type.
"""
@locked_function
def __init__(self, cache_location, url, set_cache_header=True):
self.cache_location = cache_location
hpath, bpath = calculate_cache_path(cache_location, url)
StringIO.StringIO.__init__(self, file(bpath, "rb").read())
self.url = url
self.code = 200
self.msg = "OK"
headerbuf = file(hpath, "rb").read()
if set_cache_header:
headerbuf += "x-local-cache: %s\r\n" % (bpath)
self.headers = httplib.HTTPMessage(StringIO.StringIO(headerbuf))
def info(self):
"""Returns headers
"""
return self.headers
def geturl(self):
"""Returns original URL
"""
return self.url
@locked_function
def recache(self):
new_request = urllib2.urlopen(self.url)
set_cache_header = store_in_cache(
self.cache_location,
new_request.url,
new_request
)
CachedResponse.__init__(self, self.cache_location, self.url, True)
@locked_function
def delete_cache(self):
delete_from_cache(
self.cache_location,
self.url
)
if __name__ == "__main__":
def main():
"""Quick test/example of CacheHandler"""
opener = urllib2.build_opener(CacheHandler("/tmp/"))
response = opener.open("http://google.com")
print response.headers
print "Response:", response.read()
response.recache()
print response.headers
print "After recache:", response.read()
# Test usage in threads
from threading import Thread
class CacheThreadTest(Thread):
lastdata = None
def run(self):
req = opener.open("http://google.com")
newdata = req.read()
if self.lastdata is None:
self.lastdata = newdata
assert self.lastdata == newdata, "Data was not consistent, uhoh"
req.recache()
threads = [CacheThreadTest() for x in range(50)]
print "Starting threads"
[t.start() for t in threads]
print "..done"
print "Joining threads"
[t.join() for t in threads]
print "..done"
main()
libs/tvdb_api/tvdb_exceptions.py
+52
View File
@@ -0,0 +1,52 @@
#!/usr/bin/env python
#encoding:utf-8
#author:dbr/Ben
#project:tvdb_api
#repository:http://github.com/dbr/tvdb_api
#license:unlicense (http://unlicense.org/)
"""Custom exceptions used or raised by tvdb_api
"""
__author__ = "dbr/Ben"
__version__ = "1.8.2"
__all__ = ["tvdb_error", "tvdb_userabort", "tvdb_shownotfound",
"tvdb_seasonnotfound", "tvdb_episodenotfound", "tvdb_attributenotfound"]
class tvdb_exception(Exception):
"""Any exception generated by tvdb_api
"""
pass
class tvdb_error(tvdb_exception):
"""An error with thetvdb.com (Cannot connect, for example)
"""
pass
class tvdb_userabort(tvdb_exception):
"""User aborted the interactive selection (via
the q command, ^c etc)
"""
pass
class tvdb_shownotfound(tvdb_exception):
"""Show cannot be found on thetvdb.com (non-existant show)
"""
pass
class tvdb_seasonnotfound(tvdb_exception):
"""Season cannot be found on thetvdb.com
"""
pass
class tvdb_episodenotfound(tvdb_exception):
"""Episode cannot be found on thetvdb.com
"""
pass
class tvdb_attributenotfound(tvdb_exception):
"""Raised if an episode does not have the requested
attribute (such as a episode name)
"""
pass
libs/tvdb_api/tvdb_ui.py
+153
View File
@@ -0,0 +1,153 @@
#!/usr/bin/env python
#encoding:utf-8
#author:dbr/Ben
#project:tvdb_api
#repository:http://github.com/dbr/tvdb_api
#license:unlicense (http://unlicense.org/)
"""Contains included user interfaces for Tvdb show selection.
A UI is a callback. A class, it's __init__ function takes two arguments:
- config, which is the Tvdb config dict, setup in tvdb_api.py
- log, which is Tvdb's logger instance (which uses the logging module). You can
call log.info() log.warning() etc
It must have a method "selectSeries", this is passed a list of dicts, each dict
contains the the keys "name" (human readable show name), and "sid" (the shows
ID as on thetvdb.com). For example:
[{'name': u'Lost', 'sid': u'73739'},
{'name': u'Lost Universe', 'sid': u'73181'}]
The "selectSeries" method must return the appropriate dict, or it can raise
tvdb_userabort (if the selection is aborted), tvdb_shownotfound (if the show
cannot be found).
A simple example callback, which returns a random series:
>>> import random
>>> from tvdb_ui import BaseUI
>>> class RandomUI(BaseUI):
... def selectSeries(self, allSeries):
... import random
... return random.choice(allSeries)
Then to use it..
>>> from tvdb_api import Tvdb
>>> t = Tvdb(custom_ui = RandomUI)
>>> random_matching_series = t['Lost']
>>> type(random_matching_series)
<class 'tvdb_api.Show'>
"""
__author__ = "dbr/Ben"
__version__ = "1.8.2"
import logging
import warnings
from tvdb_exceptions import tvdb_userabort
def log():
return logging.getLogger(__name__)
class BaseUI:
"""Default non-interactive UI, which auto-selects first results
"""
def __init__(self, config, log = None):
self.config = config
if log is not None:
warnings.warn("the UI's log parameter is deprecated, instead use\n"
"use import logging; logging.getLogger('ui').info('blah')\n"
"The self.log attribute will be removed in the next version")
self.log = logging.getLogger(__name__)
def selectSeries(self, allSeries):
return allSeries[0]
class ConsoleUI(BaseUI):
"""Interactively allows the user to select a show from a console based UI
"""
def _displaySeries(self, allSeries, limit = 6):
"""Helper function, lists series with corresponding ID
"""
if limit is not None:
toshow = allSeries[:limit]
else:
toshow = allSeries
print "TVDB Search Results:"
for i, cshow in enumerate(toshow):
i_show = i + 1 # Start at more human readable number 1 (not 0)
log().debug('Showing allSeries[%s], series %s)' % (i_show, allSeries[i]['seriesname']))
if i == 0:
extra = " (default)"
else:
extra = ""
print "%s -> %s [%s] # http://thetvdb.com/?tab=series&id=%s&lid=%s%s" % (
i_show,
cshow['seriesname'].encode("UTF-8", "ignore"),
cshow['language'].encode("UTF-8", "ignore"),
str(cshow['id']),
cshow['lid'],
extra
)
def selectSeries(self, allSeries):
self._displaySeries(allSeries)
if len(allSeries) == 1:
# Single result, return it!
print "Automatically selecting only result"
return allSeries[0]
if self.config['select_first'] is True:
print "Automatically returning first search result"
return allSeries[0]
while True: # return breaks this loop
try:
print "Enter choice (first number, return for default, 'all', ? for help):"
ans = raw_input()
except KeyboardInterrupt:
raise tvdb_userabort("User aborted (^c keyboard interupt)")
except EOFError:
raise tvdb_userabort("User aborted (EOF received)")
log().debug('Got choice of: %s' % (ans))
try:
selected_id = int(ans) - 1 # The human entered 1 as first result, not zero
except ValueError: # Input was not number
if len(ans.strip()) == 0:
# Default option
log().debug('Default option, returning first series')
return allSeries[0]
if ans == "q":
log().debug('Got quit command (q)')
raise tvdb_userabort("User aborted ('q' quit command)")
elif ans == "?":
print "## Help"
print "# Enter the number that corresponds to the correct show."
print "# a - display all results"
print "# all - display all results"
print "# ? - this help"
print "# q - abort tvnamer"
print "# Press return with no input to select first result"
elif ans.lower() in ["a", "all"]:
self._displaySeries(allSeries, limit = None)
else:
log().debug('Unknown keypress %s' % (ans))
else:
log().debug('Trying to return ID: %d' % (selected_id))
try:
return allSeries[selected_id]
except IndexError:
log().debug('Invalid show number entered!')
print "Invalid number (%s) selected!"
self._displaySeries(allSeries)