stubbfel/feedcrawler
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

add load and setup methods

Browse Source
This commit is contained in:
stubbfel
2014-07-30 22:59:03 +02:00
parent d1e64e66de
commit 67e74675ed
10 changed files with 113 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/config/config.py
View File

@@ -1,7 +1,7 @@
__author__ = 'dev'
Config = {
"FeedLinkTypes" : {"application/x-bittorrent","mp3"},
"FeedLinkTypes" : {"audio/mpeg","application/x-bittorrent"},
"FeedTypes" : {"podcast","rss"},
"FeedStorageFolder" : "testfeeds",
"FeedStoragePrefix" : { "0" : "FeedType", "1" : "FeedName"},

71
src/feed/feed.py
View File

@@ -1,40 +1,67 @@
__author__ = 'dev'
from utilities.jsonfile import JsonFile
from utilities.feedfile import FeedFile
from config.config import Config
from utilities.podcastserializer import PodcastSerializer
import os
import feedparser
class Feed:
def __init__(self, feed_name=None, feed_type=None, link_type=None, feed_url=None):
self.valid_file_types = {"application/x-bittorrent","mp3"}
self.config_path = None
self.feed_folder_path = None
self.feed_config = None
self.setup_feed(feed_name, feed_type, link_type, feed_url)
def load_feed(self):
assert self.feed_config is not None
return feedparser.parse(self.feed_config["FeedUrl"])
def setup_feed(self, feed_name, feed_type, link_type, feed_url):
assert feed_name is not None
assert feed_type is not None
assert link_type is not None
assert feed_url is not None
self.feed_config = self.create_feed_config_file(feed_name, feed_type, link_type, feed_url)
self.config_path = self.create_feed_config_path()
if os.path.exists(self.config_path) is False:
self.write_feed_config_file()
def new_feed_folder(self):
feed_folder = Config["FeedStorageFolder"]
feed_folder_prefix = Config["FeedStoragePrefix"]
for i in range(len(feed_folder_prefix)):
value = feed_folder_prefix.get(str(i))
if value is not None:
tmpValue = self.feed_config.get(value)
if tmpValue is not None:
value = tmpValue
if self.feed_folder_path is None:
assert self.feed_config is not None
feed_folder = Config.get("FeedStorageFolder")
assert feed_folder is not None
feed_folder_prefix = Config.get("FeedStoragePrefix")
assert feed_folder_prefix is not None
feed_folder += "/" + value
else:
break
for i in range(len(feed_folder_prefix)):
value = feed_folder_prefix.get(str(i))
if value is not None:
tmp_value = self.feed_config.get(value)
if tmp_value is not None:
value = tmp_value
feed_folder += "/" + value
else:
break
if not os.path.exists(feed_folder):
self.feed_folder_path = feed_folder
if not os.path.exists(self.feed_folder_path):
os.mkdir(feed_folder)
return os.path.abspath(feed_folder)
return self.feed_folder_path
def create_feed_config_file(self, feed_name, feed_type, link_type, feed_url):
feed_config =\
assert feed_name is not None
assert feed_type is not None
assert link_type is not None
assert feed_url is not None
feed_config = \
{
"FeedName": feed_name,
"FeedType": feed_type,
@@ -43,7 +70,11 @@ class Feed:
}
return feed_config
def create_feed_config_path(self):
if self.config_path is None:
self.config_path = self.new_feed_folder() + "/" + self.feed_config["FeedName"] + Config["ConfigFileSuffix"] + Config["ConfigFileExtension"]
return self.config_path
def write_feed_config_file(self, feed_folder):
file = JsonFile(feed_folder + self.feed_config["FeedName"] + Config["ConfigFileSuffix"] + Config["ConfigFileExtension"])
def write_feed_config_file(self):
file = JsonFile(self.create_feed_config_path())
file.write_json_file(self.feed_config)

2
src/utilities/feedfile.py
View File

@@ -21,7 +21,7 @@ class FeedFile(JsonFile):
feed_str = self.read_text_file()
return feedparser.parse(feed_str)
def write_feedjson_file(self, feed, serializer=None):
def write_feed_json_file(self, feed, serializer=None):
"""
method write a feedparpser object to a text file. It use a certain serializer or the default serializer
if type of the serializer is none

6
src/utilities/podcastserializer.py
View File

@@ -1,9 +1,13 @@
__author__ = 'dev'
import json
from utilities.serializer import Serializer
from config.config import Config
class PodcastSerializer(Serializer):
def __init__(self, episode_file_type):
self.episode_file_type = episode_file_type
def serialize(self, dict_object):
"""
method convert i dict to a string.
@@ -18,7 +22,7 @@ class PodcastSerializer(Serializer):
for episode in dict_object["entries"]:
filelink = ""
for links in episode["links"]:
if links["type"] == 'application/x-bittorrent':
if links["type"] == self.episode_file_type:
filelink = links["href"]
tempepisode = {

5
testenv/testfeeds/podcast/sbk-config.json
View File

@@ -1,5 +0,0 @@
{
"FeedName": "sbk",
"FeedUrl": "http://www.staatsbuergerkunde-podcast.de/feed/mp3-rss/",
"LinkType": "application/x-bittorrent"
}

6
testenv/testfeeds/podcast/sbk/sbk-config.json Normal file
View File

@@ -0,0 +1,6 @@
{
"LinkType": "application/x-bittorrent",
"FeedUrl": "http://www.staatsbuergerkunde-podcast.de/feed/mp3-rss/",
"FeedType": "podcast",
"FeedName": "sbk"
}

0
testenv/testfeeds/podcast/sbk.xml → testenv/testfeeds/podcast/sbk/sbk.xml
View File

50
testenv/tests/feed/testfeed.py
View File

@@ -5,14 +5,13 @@ from feed.feed import Feed
from utilities.jsonfile import JsonFile
from config.config import Config
import os.path
import feedparser
class TestFeed(unittest.TestCase):
def setUp(self):
super().setUp()
self.testFeed = Feed("sbk", "podcast", "application/x-bittorrent",
"http://www.staatsbuergerkunde-podcast.de/feed/mp3-rss/")
self.testFeed = Feed("sbk", "podcast", "audio/mpeg", "testfeeds/podcast/sbk/sbk.xml")
def test_load_new_feed(self):
self.assertTrue(False)
@@ -27,7 +26,19 @@ class TestFeed(unittest.TestCase):
self.assertTrue(False)
def test_load_feed(self):
self.assertTrue(False)
feed = self.testFeed.load_feed()
self.assertIsInstance(self.testFeed.load_feed(), feedparser.FeedParserDict)
self.assertEqual(0, feed["bozo"])
def test_load_feed_error_none_config(self):
self.testFeed.feed_config = None
self.assertRaises(AssertionError, self.testFeed.load_feed)
def test_load_feed_error_wrong_url(self):
self.testFeed = Feed("sbk", "podcast", "audio/mpeg", "testfeeds/podcast/sbk/sbk.xm")
feed = self.testFeed.load_feed()
self.assertIsInstance(self.testFeed.load_feed(), feedparser.FeedParserDict)
self.assertEqual(1, feed["bozo"])
def test_load_feed_error_loading_failed(self):
self.assertTrue(False)
@@ -35,18 +46,45 @@ class TestFeed(unittest.TestCase):
def test_setup_feed(self):
self.assertTrue(False)
def test_setup_feed_error_none_parameter(self):
self.assertRaises(AssertionError, self.testFeed.setup_feed, None, "podcast", "audio/mpeg",
"http://www.staatsbuergerkunde-podcast.de/feed/mp3-rss/")
self.assertRaises(AssertionError, self.testFeed.setup_feed, "sbk", None, "audio/mpeg",
"http://www.staatsbuergerkunde-podcast.de/feed/mp3-rss/")
self.assertRaises(AssertionError, self.testFeed.setup_feed, "sbk", "podcast", None,
"http://www.staatsbuergerkunde-podcast.de/feed/mp3-rss/")
self.assertRaises(AssertionError, self.testFeed.setup_feed, "sbk", "podcast","audio/mpeg", None)
def test_create_feed_config_path(self):
self.testFeed.config_path = "foo"
self.assertEqual(self.testFeed.create_feed_config_path(), "foo")
def test_new_feed_folder(self):
self.assertTrue(os.path.exists(self.testFeed.new_feed_folder()))
def test_new_feed_folder_no_feed_config(self):
self.testFeed.feed_folder_path = None
self.testFeed.feed_config = None
self.assertRaises(AssertionError, self.testFeed.new_feed_folder)
def test_create_feed_config_file(self):
feed_config = self.testFeed.create_feed_config_file("sbk", "podcast", "application/x-bittorrent",
feed_config = self.testFeed.create_feed_config_file("sbk", "podcast", "audio/mpeg",
"http://www.staatsbuergerkunde-podcast.de/feed/mp3-rss/")
self.assertIsNotNone(feed_config)
self.assertEqual("sbk", feed_config["FeedName"])
self.assertEqual("podcast", feed_config["FeedType"])
self.assertEqual("application/x-bittorrent", feed_config["LinkType"])
self.assertEqual("audio/mpeg", feed_config["LinkType"])
self.assertEqual("http://www.staatsbuergerkunde-podcast.de/feed/mp3-rss/", feed_config["FeedUrl"])
def test_create_feed_config_file_error_none_parameter(self):
self.assertRaises(AssertionError, self.testFeed.create_feed_config_file, None, "podcast",
"audio/mpeg", "http://www.staatsbuergerkunde-podcast.de/feed/mp3-rss/")
self.assertRaises(AssertionError, self.testFeed.create_feed_config_file, "sbk", None,
"audio/mpeg", "http://www.staatsbuergerkunde-podcast.de/feed/mp3-rss/")
self.assertRaises(AssertionError, self.testFeed.create_feed_config_file, "sbk", "podcast", None,
"http://www.staatsbuergerkunde-podcast.de/feed/mp3-rss/")
self.assertRaises(AssertionError, self.testFeed.create_feed_config_file, "sbk", "podcast",
"audio/mpeg", None)
if __name__ == '__main__':
unittest.main()

4
testenv/tests/utilities/testfeedfile.py
View File

@@ -17,7 +17,7 @@ class FeedUtilitiesTest(unittest.TestCase):
feed = self.__ff.read_feed_file()
ff2 = FeedFile("testfeed.xml")
serializer = None
ff2.write_feedjson_file(feed, serializer)
ff2.write_feed_json_file(feed, serializer)
feed2 = ff2.read_feed_file()
self.assertTrue(isinstance(feed2, feedparser.FeedParserDict))
ff2.check_and_remove_file()
@@ -27,7 +27,7 @@ class FeedUtilitiesTest(unittest.TestCase):
feed = "Fail"
ff2 = FeedFile("testfeed.xml")
serializer = None
self.assertRaises(AssertionError, ff2.write_feedjson_file, feed, serializer)
self.assertRaises(AssertionError, ff2.write_feed_json_file, feed, serializer)
ff2.check_and_remove_file()
if __name__ == '__main__':

6
testenv/tests/utilities/testpodcastserializer.py
View File

@@ -7,12 +7,12 @@ from utilities.feedfile import FeedFile
class PodCastSerializerUtilitiesTest(unittest.TestCase):
def setUp(self):
self.__podcastfeed= FeedFile("testfeeds/podcast/sbk.xml").read_feed_file()
self.__serializer = PodcastSerializer()
self.__podcastfeed= FeedFile("testfeeds/podcast/sbk/sbk.xml").read_feed_file()
self.__serializer = PodcastSerializer("application/x-bittorrent")
def test_serialize(self):
ff2 = FeedFile("testfeed.json")
ff2.write_feedjson_file(self.__podcastfeed, self.__serializer)
ff2.write_feed_json_file(self.__podcastfeed, self.__serializer)
feed = ff2.read_json_file()
self.assertIsNotNone(feed)