Error %s - This page does not exist.
" % status tmpl = lookup.get_template("error.html") return tmpl.render(message=message) def handle_error(): """ Handle different type of errors. """ message = "Sorry, an error occured.
" cherrypy.response.status = 500 cherrypy.response.body = [message] class RestrictedArea(object): """ All methods in this controller (and subcontrollers) is open only to members of the admin group """ _cp_config = { 'auth.auth.require': [auth.member_of('admin')] } @cherrypy.expose def index(self): message = "This is the admin only area.
" tmpl = lookup.get_template("error.html") return tmpl.render(message=message) class pyAggr3g470r(object): """ Main class. All pages of pyAggr3g470r are described in this class. """ _cp_config = {'request.error_response': handle_error, \ 'tools.sessions.on': True, \ 'tools.auth.on': True} def __init__(self): """ """ self.auth = auth.AuthController() restricted = RestrictedArea() self.mongo = mongodb.Articles(conf.MONGODB_ADDRESS, conf.MONGODB_PORT, \ conf.MONGODB_DBNAME, conf.MONGODB_USER, conf.MONGODB_PASSWORD) @auth.require() def index(self): """ Main page containing the list of feeds and articles. """ feeds = self.mongo.get_all_feeds() nb_unread_articles = self.mongo.nb_unread_articles() nb_favorites = self.mongo.nb_favorites() nb_mail_notifications = self.mongo.nb_mail_notifications() tmpl = lookup.get_template("index.html") return tmpl.render(feeds=feeds, nb_feeds=len(feeds), mongo=self.mongo, \ nb_favorites=nb_favorites, nb_unread_articles=nb_unread_articles, \ nb_mail_notifications=nb_mail_notifications, header_text=nb_unread_articles) index.exposed = True @auth.require() def management(self): """ Management page. Allows adding and deleting feeds. Export functions of the MongoDB data base and display some statistics. """ feeds = self.mongo.get_all_feeds() nb_mail_notifications = self.mongo.nb_mail_notifications() nb_favorites = self.mongo.nb_favorites() nb_articles = format(self.mongo.nb_articles(), ",d") nb_unread_articles = format(self.mongo.nb_unread_articles(), ",d") tmpl = lookup.get_template("management.html") return tmpl.render(feeds=feeds, nb_mail_notifications=nb_mail_notifications, \ nb_favorites=nb_favorites, nb_articles=nb_articles, \ nb_unread_articles=nb_unread_articles, \ mail_notification_enabled=conf.MAIL_ENABLED) management.exposed = True @auth.require() def statistics(self, word_size=6): """ More advanced statistics. """ articles = self.mongo.get_articles() top_words = utils.top_words(articles, n=50, size=int(word_size)) tag_cloud = utils.tag_cloud(top_words) tmpl = lookup.get_template("statistics.html") return tmpl.render(articles=articles, word_size=word_size, tag_cloud=tag_cloud) statistics.exposed = True @auth.require() def search(self, query=None): """ Simply search for the string 'query' in the description of the article. """ param, _, value = query.partition(':') feed_id = None if param == "Feed": feed_id, _, query = value.partition(':') search_result = self.mongo.full_search(param) tmpl = lookup.get_template("search.html") return tmpl.render(search_result=search_result, query=query, feed_id=feed_id, mongo=self.mongo) search.exposed = True @auth.require() def fetch(self, param=None): """ Fetch all feeds. """ feed_link = None if None != param: # Fetch only the feed specified in parameter feed_link = self.mongo.get_feed(param)["feed_link"] feed_getter = feedgetter.FeedGetter() feed_getter.retrieve_feed(feed_url=feed_link) return self.index() fetch.exposed = True @auth.require() def article(self, param): """ Display the article in parameter in a new Web page. """ try: feed_id, article_id = param.split(':') feed = self.mongo.get_feed(feed_id) articles = self.mongo.get_articles(feed_id) article = self.mongo.get_articles(feed_id, article_id) except: return self.error("Bad URL. This article do not exists.
") if article["article_readed"] == False: # if the current article is not yet readed, update the database self.mark_as_read("Article:"+article["article_id"]+":"+feed["feed_id"]) # Description (full content) of the article description = article["article_content"] if description: p = re.compile(r'<')
q = re.compile(r'>')
description = p.sub('<', description)
description = q.sub('>', description)
description = description + "\nThis feed do not exists.
") if articles != []: today = datetime.datetime.now() last_article = articles[0]["article_date"] first_article = articles[self.mongo.nb_articles(feed_id)-2]["article_date"] delta = last_article - first_article elapsed = today - last_article average = round(nb_articles_feed / abs(delta.days), 2) top_words = utils.top_words(articles = self.mongo.get_articles(feed_id), n=50, size=int(word_size)) tag_cloud = utils.tag_cloud(top_words) tmpl = lookup.get_template("feed.html") return tmpl.render(feed=feed, articles=articles, favorites=favorites, \ nb_articles_feed=nb_articles_feed, nb_articles_total=nb_articles_total, nb_unread_articles_feed=nb_unread_articles_feed, \ nb_favorites = nb_favorites, first_post_date=first_article, end_post_date=last_article, \ average=average, delta=delta, elapsed=elapsed, \ tag_cloud=tag_cloud, word_size=word_size, \ mail_to=conf.mail_to, mail_notification_enabled=conf.MAIL_ENABLED) feed.exposed = True @auth.require() def articles(self, feed_id): """ This page displays all articles of a feed. """ try: feed = self.mongo.get_feed(feed_id) articles = self.mongo.get_articles(feed_id) except KeyError: return self.error("This feed do not exists.
") tmpl = lookup.get_template("articles.html") return tmpl.render(articles=articles, feed=feed) articles.exposed = True @auth.require() def unread(self, feed_id=""): """ This page displays all unread articles of a feed. """ feeds = self.mongo.get_all_feeds() tmpl = lookup.get_template("unread.html") return tmpl.render(feeds=feeds, feed_id=feed_id, mongo=self.mongo) unread.exposed = True @auth.require() def history(self, query="all", m=""): """ This page enables to browse articles chronologically. """ feeds = self.mongo.get_all_feeds() tmpl = lookup.get_template("history.html") return tmpl.render(feeds=feeds, mongo=self.mongo, query=query, m=m) history.exposed = True @auth.require() def plain_text(self, target): """ Display an article in plain text (without HTML tags). """ try: feed_id, article_id = target.split(':') feed = self.mongo.get_feed(feed_id) article = self.mongo.get_articles(feed_id, article_id) except: return self.error("Bad URL. This article do not exists.
") description = utils.clear_string(article["article_content"]) if not description: description = "Unvailable" tmpl = lookup.get_template("plain_text.html") return tmpl.render(feed_title=feed["feed_title"], \ article_title=article["article_title"], description = description) plain_text.exposed = True @auth.require() def error(self, message): """ Display a message (bad feed id, bad article id, etc.) """ tmpl = lookup.get_template("error.html") return tmpl.render(message=message) error.exposed = True @auth.require() def mark_as_read(self, target=""): """ Mark one (or more) article(s) as read by setting the value of the field 'article_readed' of the MongoDB database to 'True'. """ param, _, identifiant = target.partition(':') # Mark all articles as read. if param == "": self.mongo.mark_as_read(True, None, None) # Mark all articles from a feed as read. elif param == "Feed" or param == "Feed_FromMainPage": self.mongo.mark_as_read(True, identifiant, None) # Mark an article as read. elif param == "Article": self.mongo.mark_as_read(True, identifiant.split(':')[1], identifiant.split(':')[0]) return self.index() mark_as_read.exposed = True @auth.require() def notifications(self): """ List all active e-mail notifications. """ feeds = self.mongo.get_all_feeds(condition=("mail",True)) tmpl = lookup.get_template("notifications.html") return tmpl.render(feeds=feeds, mail_to=conf.mail_to, mail_notification_enabled=conf.MAIL_ENABLED) notifications.exposed = True @auth.require() def mail_notification(self, param): """ Enable or disable to notifications of news for a feed. """ try: action, feed_id = param.split(':') except: return self.error("Bad URL. This feed do not exists.
") return self.index() mail_notification.exposed = True @auth.require() def like(self, param): """ Mark or unmark an article as favorites. """ try: like, feed_id, article_id = param.split(':') articles = self.mongo.get_articles(feed_id, article_id) except: return self.error("Bad URL. This article do not exists.
") self.mongo.like_article("1"==like, feed_id, article_id) return self.article(feed_id+":"+article_id) like.exposed = True @auth.require() def favorites(self): """ List of favorites articles """ feeds = self.mongo.get_all_feeds() articles = {} for feed in feeds: articles[feed["feed_id"]] = self.mongo.get_favorites(feed["feed_id"]) tmpl = lookup.get_template("favorites.html") return tmpl.render(feeds=feeds, \ articles=articles) favorites.exposed = True @auth.require() def inactives(self, nb_days=365): """ List of favorites articles """ feeds = self.mongo.get_all_feeds() today = datetime.datetime.now() inactives = [] for feed in feeds: more_recent_article = self.mongo.get_articles(feed["feed_id"], limit=1) last_post = next(more_recent_article)["article_date"] elapsed = today - last_post if elapsed > datetime.timedelta(days=int(nb_days)): inactives.append((feed, elapsed)) tmpl = lookup.get_template("inactives.html") return tmpl.render(inactives=inactives, nb_days=int(nb_days)) inactives.exposed = True @auth.require() def languages(self): """ Filter by languages. """ try: from guess_language import guess_language_name except: tmpl = lookup.get_template("error.html") return tmpl.render(message='Module guess_language not installed.
') result = {} feeds = self.mongo.get_all_feeds() for feed in feeds: for article in self.mongo.get_articles(feed["feed_id"]): language = guess_language_name(utils.clear_string(article["article_content"])) result.setdefault(language, defaultdict(list)) result[language][feed["feed_id"]].append(article) tmpl = lookup.get_template("languages.html") return tmpl.render(articles_sorted_by_languages=result, mongo=self.mongo) languages.exposed = True @auth.require() def add_feed(self, url): """ Add a new feed with the URL of a page. """ # search the feed in the HTML page with BeautifulSoup feed_url = utils.search_feed(url) if feed_url is None: return self.error("Impossible to find a feed at this URL.
") # if a feed exists else: result = utils.add_feed(feed_url) # if the feed is not in the file feed.lst if result is False: message = "You are already following this feed!
" else: message = """Feed added. You can now fetch your feeds.
""" tmpl = lookup.get_template("confirmation.html") return tmpl.render(message=message) add_feed.exposed = True @auth.require() def remove_feed(self, feed_id): """ Remove a feed from the file feed.lst and from the MongoDB database. """ feed = self.mongo.get_feed(feed_id) self.mongo.delete_feed(feed_id) utils.remove_feed(feed["feed_link"]) message = """All articles from the feed %s are now removed from the base.
""" % (feed["feed_title"],) tmpl = lookup.get_template("confirmation.html") return tmpl.render(message=message) remove_feed.exposed = True @auth.require() def change_feed_url(self, feed_id, old_feed_url, new_feed_url): """ Enables to change the URL of a feed already present in the database. """ self.mongo.update_feed(feed_id, {"feed_link":new_feed_url}) result = utils.change_feed_url(old_feed_url, new_feed_url) if result: tmpl = lookup.get_template("confirmation.html") message = "The URL of the feed has been changed.
" else: tmpl = lookup.get_template("error.html") message = "Error when changing the URL of the feed.
" return tmpl.render(message=message) change_feed_url.exposed = True @auth.require() def change_feed_name(self, feed_id, new_feed_name): """ Enables to change the name of a feed. """ self.mongo.update_feed(feed_id, {"feed_title":new_feed_name}) tmpl = lookup.get_template("confirmation.html") return tmpl.render(message="The name of the feed has been changed.
") change_feed_name.exposed = True @auth.require() def change_feed_logo(self, feed_id, new_feed_logo): """ Enables to change the name of a feed. """ self.mongo.update_feed(feed_id, {"feed_image":new_feed_logo}) tmpl = lookup.get_template("confirmation.html") return tmpl.render(message="The logo of the feed has been changed.
") change_feed_logo.exposed = True @auth.require() def change_username(self, new_username): """ Enables to change the username of a user. """ result = auth.change_username(self.auth.username, new_username) if result: self.auth.username = new_username message = "Your username has been changed.
" tmpl = lookup.get_template("confirmation.html") return tmpl.render(message=message) else: return self.error("Impossible to change the username.
") change_username.exposed = True @auth.require() def change_password(self, new_password): """ Enables to change the password of a user. """ result = auth.change_password(self.auth.username, new_password) if result: message = "Your password has been changed.
" tmpl = lookup.get_template("confirmation.html") return tmpl.render(message=message) else: return self.error("Impossible to change the password.
") change_password.exposed = True @auth.require() def delete_article(self, param): """ Delete an article. """ try: feed_id, article_id = param.split(':') self.mongo.delete_article(feed_id, article_id) except: return self.error("Bad URL. This article do not exists.
") return self.index() delete_article.exposed = True @auth.require() def logout(self): """ Close the session. """ return self.auth.logout() logout.exposed = True @auth.require() def drop_base(self): """ Delete all articles. """ self.mongo.drop_database() return self.index() drop_base.exposed = True @auth.require() def export(self, export_method): """ Export articles currently loaded from the MongoDB database with the appropriate function of the 'export' module. """ getattr(export, export_method)(self.mongo) try: getattr(export, export_method)(self.mongo) except Exception as e: return self.error(e) tmpl = lookup.get_template("confirmation.html") return tmpl.render(message="Export successfully terminated.
Check the folder: " + conf.path + "/var/export/.
This article do not exists.
") try: folder = conf.path + "/var/export/epub/" os.makedirs(folder) except OSError: # directories already exists (not a problem) pass section = ez_epub.Section() section.title = article["article_title"].decode('utf-8') section.paragraphs = [utils.clear_string(article["article_content"])] ez_epub.makeBook(article["article_title"], [feed["feed_title"]], [section], \ os.path.normpath(folder) + "article.epub", lang='en-US', cover=None) return self.article(param) epub.exposed = True if __name__ == '__main__': # Point of entry in execution mode root = pyAggr3g470r() root.favicon_ico = cherrypy.tools.staticfile.handler(filename=os.path.join(conf.path + "/img/favicon.png")) cherrypy.config.update({'error_page.404': error_404}) cherrypy.quickstart(root, "/" ,config=conf.path + "/cfg/cherrypy.cfg")