'+ nb_unread_articles +'pyAggr3g470r

Unread article(s): %s\n""" % \ (self.nb_unread_articles,) + + # The main page display all the feeds. for rss_feed_id in self.feeds.keys(): html += """

%s diff --git a/tuxdroid.py b/tuxdroid.py deleted file mode 100755 index c48fadf2..00000000 --- a/tuxdroid.py +++ /dev/null @@ -1,88 +0,0 @@ -#! /usr/local/bin/python -#-- coding: utf-8 -- - -"""Communication interface with Tux Droid. -""" - -author = "Cedric Bonhomme" -version = "$Revision: 0.1 $" -date = "$Date: 2010/02/26 $" -copyright = "Copyright (c) 2010 Cedric Bonhomme" -license = "GPL v3" - -import time -import sqlite3 - -try: - from tuxisalive.api import * -except: - raise NameError("The module tuxisalive is missing.") - -class Tux(object): - """Manage the connection with Tux Droid. - """ - def init(self, locutor = "Julie "): - """Connection to the server. - """ - self.tux = TuxAPI('127.0.0.1', 270) - self.tux.server.autoConnect(CLIENT_LEVEL_RESTRICTED, 'none', 'none') - print "Connection with Tux Droid..." - - if not self.tux.server.waitConnected(10.0): - raise Exception("The server of Tux Droid is not ready.") - print "Connected to the server of Tux Droid." - - if not self.tux.dongle.waitConnected(10.0): - raise Exception("Not connected to the dongle.") - print "Connected to the dongle." - - if not self.tux.radio.waitConnected(10.0): - raise Exception("Connection between Tux Droid and the dongle unpossible.") - print "Tux Droid connected to the dongle." - - if self.tux.access.waitAcquire(10.0, ACCESS_PRIORITY_NORMAL): - self.tux.tts.setLocutor(locutor) # set the locutor - self.tux.tts.setPitch(110) # set the Pitch - self.tux.tts.speakAsync("Hello") - - def disconnect(self): - """Disconnect. - """ - self.tux.server.disconnect() - self.tux.destroy() - - def onStart(self): - """ - """ - self.tux.flippers.onAsync(4, finalState = 'DOWN', speed = 3) - self.tux.tts.speakAsync("Hello !") - - def say_something(self, message): - """ - """ - self.tux.tts.speakAsync(message) - - def check_feeds(self): - """ - """ - unread_articles = [] - try: - conn = sqlite3.connect("./var/feed.db", isolation_level = None) - c = conn.cursor() - unread_articles = c.execute("SELECT * FROM articles WHERE article_readed='0'") - c.close() - except Exception, e: - pass - self.say_something("News to read !") - #self.say_something(str(len(unread_articles))) - #for unread_article in unread_articles: - #self.say_something() - - -if name == "main": - # Point of entry in execution mode - tux_reader = Tux() - while True: - time.sleep(10) - tux_reader.check_feeds() - tux_reader.disconnect() diff --git a/utils.py b/utils.py index 0b93fb1f..780ebf91 100755 --- a/utils.py +++ b/utils.py @@ -50,6 +50,7 @@ LOCKER = threading.Lock() import os import ConfigParser +# load the configuration config = ConfigParser.RawConfigParser() config.read("./cfg/pyAggr3g470r.cfg") path = os.path.abspath(".") @@ -60,6 +61,7 @@ smtp_server = config.get('mail','smtp') username = config.get('mail','username') password = config.get('mail','password') +# regular expression to chech URL url_finders = [ \ re.compile("([0-9]{1,3}\\.[0-9]{1,3}\\.[0-9]{1,3}\\.[0-9]{1,3}|(((news|telnet|nttp|file|http|ftp|https)://)|(www|ftp)[-A-Za-z0-9]\\.)[-A-Za-z0-9\\.]+)(:[0-9])?/[-A-Za-z0-9_\\$\\.\\+\\!\\\$\$,;:@&=\\?/~\\#\\%][^]'\\.}>\\),\\\"]"), \ re.compile("([0-9]{1,3}\\.[0-9]{1,3}\\.[0-9]{1,3}\\.[0-9]{1,3}|(((news|telnet|nttp|file|http|ftp|https)://)|(www|ftp)[-A-Za-z0-9]\\.)[-A-Za-z0-9\\.]+)(:[0-9])?"), \ @@ -70,6 +72,7 @@ url_finders = [ \ def detect_url_errors(list_of_urls): """ Detect URL errors. + Return a list of error(s). """ errors = [] for url in list_of_urls: @@ -108,8 +111,8 @@ def clear_string(data): Clear a string by removing HTML tags, HTML special caracters and consecutive white spaces (more that one). """ - p = re.compile(r'<[^<]?/?>') - q = re.compile(r'\s') + p = re.compile(r'<[^<]?/?>') # HTML tags + q = re.compile(r'\s') # consecutive white spaces return p.sub('', q.sub(' ', data)) def unescape(text): @@ -157,10 +160,12 @@ def tag_cloud(tags, query="word_count"): """ tags.sort(key=operator.itemgetter(0)) if query == "word_count": + # tags cloud from the management page return ' '.join([('%s\n' % \ (min(1 + count * 7 / max([tag[1] for tag in tags]), 7), word, count, word)) \ for (word, count) in tags]) if query == "year": + # tags cloud for the history return ' '.join([('%s\n' % \ (min(1 + count * 7 / max([tag[1] for tag in tags]), 7), query, word, count, word)) \ for (word, count) in tags]) @@ -213,6 +218,7 @@ def add_feed(feed_url): if os.path.exists("./var/feed.lst"): for line in open("./var/feed.lst", "r"): if feed_url in line: + # if the feed is already in the file return False with open("./var/feed.lst", "a") as f: f.write(feed_url + "\n") @@ -266,10 +272,12 @@ def create_base(): sqlite3.register_adapter(str, lambda s : s.decode('utf-8')) conn = sqlite3.connect(sqlite_base, isolation_level = None) c = conn.cursor() + # table of feeds c.execute('''create table if not exists feeds (feed_title text, feed_site_link text, \ feed_link text PRIMARY KEY, feed_image_link text, mail text)''') + # table of articles c.execute('''create table if not exists articles (article_date text, article_title text, \ article_link text PRIMARY KEY, article_description text, \ @@ -329,6 +337,7 @@ def load_feed(): sha1_hash.update(article[2].encode('utf-8')) article_id = sha1_hash.hexdigest() + # check the presence of the module for language detection if "oice" not in IMPORT_ERROR: if article[3] != "": language = detect_language(clear_string(article[3][:80]).encode('utf-8') + \ @@ -338,6 +347,7 @@ def load_feed(): else: language = "IMPORT_ERROR" + # informations about an article article_list = [article_id, article[0], unescape(article[1]), \ article[2], unescape(article[3]), \ article[4], language, article[6]] @@ -350,6 +360,7 @@ def load_feed(): else: articles[feed_id].append(article_list) + # informations about a feed feeds[feed_id] = (len(articles[feed_id]), \ len([article for article in articles[feed_id] \ if article[5]=="0"]), \ -- cgit