"""Module containing main views for web client"""
import datetime as dt
from flask import Blueprint, Response, flash, render_template, request, session, url_for
from werkzeug.datastructures import CombinedMultiDict
from werkzeug.exceptions import abort
from werkzeug.security import check_password_hash, generate_password_hash
from werkzeug.utils import redirect
from nnrss.background_update import BACKGROUND_TASK
from nnrss.database import RssHandler
from nnrss.decorators import (
admin_required,
guest_required,
login_required,
registration_enabled,
)
from nnrss.forms import (
AddFeedForm,
ExportForm,
ImportForm,
LoginForm,
RegistrationForm,
SettingsForm,
)
from nnrss.models import User
from nnrss.opml import OPMLWriter
from nnrss.thread_import import ImportThread
DEFAULT_WEB = Blueprint("default_web", __name__)
@DEFAULT_WEB.errorhandler(404)
def page_not_found(error):
return render_template("errors/404.html", error=error), 404
@DEFAULT_WEB.route("/")
@login_required
def index():
category = request.args.get("category", None)
handler = RssHandler()
if category:
entries = handler.get_unread_entries(session["logged_in"], category)
categories = None
else:
entries = handler.get_unread_entries(session["logged_in"])
categories = handler.get_all_categories(session["logged_in"])
return render_template(
"unread_entries.html",
entries=entries,
category_name=category,
categories=categories,
)
@DEFAULT_WEB.route("/list")
@login_required
def feed_list():
category = request.args.get("category", None)
handler = RssHandler()
if category:
feeds = handler.get_feeds_from_category(session["logged_in"], category)
categories = None
else:
feeds = handler.get_feeds(session["logged_in"])
categories = handler.get_all_categories(session["logged_in"])
return render_template(
"feed_list.html",
feeds=feeds,
categories=categories,
category_name=category,
)
@DEFAULT_WEB.route("/add", methods=["GET", "POST"])
@login_required
def add_feed():
form = AddFeedForm(request.form)
if request.method == "POST" and form.validate():
handler = RssHandler()
feed_added = handler.add_feeds(
[form.url.data], session["logged_in"], [form.category.data]
)
if not feed_added or feed_added is None:
flash("Not a valid feed URL", "error")
else:
flash("Feed {} added!".format(form.url.data), "info")
return render_template("feed_add.html", form=form)
@DEFAULT_WEB.route("/feed/<int:feed_id>")
@login_required
def view_feed_entries(feed_id=None):
if feed_id is None:
return "Enter id after URL", 400
page = int(request.args.get("page", 1))
handler = RssHandler()
if not handler.feed_exists(feed_id):
return abort(404, description="Feed not found")
entries = handler.get_entries_by_feed_id(feed_id, session["logged_in"])
selected_feed = handler.get_feed_without_details(feed_id, session["logged_in"])
return render_template(
"entries_list.html",
entries=entries.paginate(page=page),
feed=selected_feed,
)
@DEFAULT_WEB.route("/feed/<int:feed_id>/unread")
@login_required
def view_feed_entries_unread(feed_id=None):
if feed_id is None:
return "Enter id after URL", 400
page = int(request.args.get("page", 1))
handler = RssHandler()
if not handler.feed_exists(feed_id):
return abort(404, description="Feed not found")
entries = handler.get_unread_entries_by_feed_id(feed_id, session["logged_in"])
selected_feed = handler.get_feed_without_details(feed_id, session["logged_in"])
return render_template(
"entries_list_unread.html",
entries=entries.paginate(page=page),
feed=selected_feed,
)
@DEFAULT_WEB.route("/feed/<int:feed_id>/update", methods=["POST"])
@login_required
def update_feed_entries(feed_id=None):
if feed_id is None:
return "Enter id after URL", 400
handler = RssHandler()
if handler.update_feed(feed_id, session["logged_in"]):
flash("Feed updated!", "info")
else:
return abort(404, description="Feed not found")
return redirect(url_for("default_web.view_feed_entries", feed_id=feed_id))
@DEFAULT_WEB.route("/feed/<int:feed_id>/set_read")
@login_required
def set_read_entry(feed_id=None, entry_id=None):
if feed_id is None:
return "feed_id URL part missing", 400
entry_id = request.args.get("entry_id")
if entry_id is None:
return "entry_id parameter missing", 400
handler = RssHandler()
ok, link = handler.set_read_entry(feed_id, entry_id, session["logged_in"])
if ok:
return redirect(link, code=302)
else:
return abort(404, description="Feed not found")
@DEFAULT_WEB.route("/feed/<int:feed_id>/remove", methods=["GET", "POST"])
@login_required
def remove_feed(feed_id=None):
handler = RssHandler()
if request.method == "POST":
if feed_id is None:
return "Enter id after URL"
if handler.remove_feed(feed_id, session["logged_in"]):
flash("Feed removed!", "info")
else:
return abort(404, description="Feed not found")
return redirect(url_for("default_web.index"))
selected_feed = handler.get_feed_without_details(feed_id, session["logged_in"])
return render_template("delete_confirm.html", the_feed=selected_feed)
@DEFAULT_WEB.route("/admin")
@admin_required
def admin_index(feed_id=None):
handler = RssHandler()
users = handler.get_users()
feed_count = [handler.get_feeds_count(user.username) for user in users]
return render_template(
"admin_panel.html", users=handler.get_users(), feed_count=feed_count
)
@DEFAULT_WEB.route("/login", methods=["GET", "POST"])
@guest_required
def login():
form = LoginForm(request.form)
if request.method == "POST" and form.validate():
handler = RssHandler()
password = handler.get_user_password(form.username.data)
if password is not None and check_password_hash(password, form.password.data):
session["logged_in"] = form.username.data
flash("Logged in", "info")
else:
flash("Wrong credentials", "error")
return redirect(url_for("default_web.index"))
return render_template("login.html", form=form)
@DEFAULT_WEB.route("/logout")
@login_required
def logout():
session.pop("logged_in", None)
return redirect(url_for("default_web.index"))
@DEFAULT_WEB.route("/register", methods=["GET", "POST"])
@registration_enabled
@guest_required
def register():
form = RegistrationForm(request.form)
if request.method == "POST" and form.validate():
handler = RssHandler()
new_user = User(
form.username.data,
generate_password_hash(form.password.data, method="pbkdf2:sha512"),
)
is_ok = handler.add_user(new_user)
if is_ok:
flash("Registered", "info")
BACKGROUND_TASK.add_new_user_job(form.username.data)
return redirect(url_for("default_web.login"))
flash("Username taken", "error")
return redirect(url_for("default_web.register"))
return render_template("register.html", form=form)
@DEFAULT_WEB.route("/settings", methods=["GET", "POST"])
@login_required
def settings():
handler = RssHandler()
user_prefs = handler.get_user(session["logged_in"])
form = SettingsForm(request.form, user_prefs)
if request.method == "POST" and form.validate():
if form.password.data.startswith("pbkdf2"):
user_prefs.password = form.password.data
else:
user_prefs.password = generate_password_hash(
form.password.data, method="pbkdf2:sha512"
)
user_prefs.api_key = form.api_key.data
user_prefs.update_interval = form.update_interval.data
handler.add_user(user_prefs)
flash("Settings updated", "info")
BACKGROUND_TASK.reschedule(session["logged_in"])
return redirect(url_for("default_web.settings"))
return render_template("settings.html", form=form)
import_thread = dict()
@DEFAULT_WEB.route("/import", methods=["GET", "POST"])
@login_required
def import_feeds():
global import_thread
form = ImportForm(CombinedMultiDict((request.form, request.files)))
if request.method == "POST" and form.validate():
if form.import_file.data.mimetype != "text/x-opml+xml":
flash("Selected file is not OPML filetype", "error")
return redirect(url_for("default_web.import_feeds"))
import_thread[session["logged_in"]] = ImportThread(
form.import_file.data, session["logged_in"]
)
import_thread[session["logged_in"]].start()
flash("Import started", "info")
return redirect(url_for("default_web.index"))
try:
is_thread_alive = import_thread[session["logged_in"]].isAlive()
except KeyError:
is_thread_alive = False
except AttributeError:
is_thread_alive = (
True if import_thread[session["logged_in"]] is not None else False
)
if is_thread_alive:
flash("Import in progress, please wait.", "error")
return redirect(url_for("default_web.index"))
else:
return render_template("import_feeds.html", form=form)
@DEFAULT_WEB.route("/export", methods=["GET", "POST"])
@login_required
def export_feeds():
form = ExportForm()
if request.method == "POST" and form.validate():
handler = RssHandler()
feeds_list = handler.get_feeds(session["logged_in"])
writer = OPMLWriter(feeds_list)
return Response(
writer.get_exported_feeds(),
mimetype="text/xml",
headers={
"Content-Disposition": "attachment;filename=feeds-"
+ str(dt.date.today())
+ ".opml"
},
)
return render_template("export_feeds.html", form=form)