-
Notifications
You must be signed in to change notification settings - Fork 0
/
application.py
295 lines (252 loc) · 10.2 KB
/
application.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
import os
from flask import Flask, session, render_template, request, redirect, url_for, flash, jsonify
from flask_session import Session
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker
import requests
import json
app = Flask(__name__)
def page_not_found(e):
return render_template('404.html'), 404
app.register_error_handler(404, page_not_found)
# Check for environment variable
if not os.getenv("DATABASE_URL"):
raise RuntimeError("DATABASE_URL is not set")
# Configure session to use filesystem
app.config["SESSION_PERMANENT"] = False
app.config["SESSION_TYPE"] = "filesystem"
Session(app)
# Set up database
engine = create_engine(os.getenv("DATABASE_URL"))
db = scoped_session(sessionmaker(bind=engine))
GOODREADS_KEY = "lQPRX6ne1CNGAV8raRov1w"
class Book:
def __init__(self,isbn,title,year,author):
self.isbn = isbn
self.title = title
self.year = year
self.author = author
@app.route("/")
def index():
try:
session['user_id']
return render_template("search.html")
except KeyError:
return render_template("registration.html")
@app.route("/logout", methods=["POST","GET"])
def logout():
error = None
session.clear()
return render_template("login.html")
@app.route("/register", methods=["POST","GET"])
# def registration():
# return render_template("registration.html")
#@app.route("/registrationcheck", methods=["POST","GET"])
def register():
if request.method == 'POST':
username = request.form.get("username")
password = request.form.get("password")
password_confirmation = request.form.get("password_confirmation")
error = None
if not username:
error = 'Username is required.'
elif not password:
error = 'Password is required.'
elif password != password_confirmation:
error = 'Passwords did not match.'
else:
if user_exists(username):
error = 'Username is already registered.'
else:
#insert new user into database
insert_user(username,password)
user = login_success(username,password)
session["user_id"] = user['user_id']
return redirect(url_for('search'))
flash(error)
return render_template("registration.html")
def user_exists(username):
exists = db.execute("""SELECT user_id FROM user_accounts
WHERE username = :username""",
{"username": username}).fetchone()
if exists is not None:
#user already exists
return True
else:
#user does not exist
return False
def insert_user(username,password):
db.execute("""INSERT INTO user_accounts (username,password)
VALUES (:username, :password)""",
{"username": username, "password": password})
db.commit()
@app.route("/login", methods=["POST","GET"])
def login():
error = None
if request.method == 'POST':
username = request.form.get("username")
password = request.form.get("password")
error = None
user = login_success(username,password)
if user is not None:
#correct login
session.clear()
session["user_id"] = user['user_id']
return redirect(url_for('search'))
else:
#incorrect login
error = 'There has been a problem with your credentials. Please, try again.'
flash(error)
return render_template("login.html")
def login_success(username,password):
user = db.execute("""SELECT * FROM user_accounts
WHERE username = :username AND password = :password""",
{"username": username, "password":password}).fetchone()
return(user)
@app.route("/search", methods=["POST","GET"])
def search():
try:
session["user_id"]
results = []
if request.method == 'POST':
keyword = request.form.get("keyword")
results = book_search(keyword)
if len(results) == 0:
error = 'No results found.'
flash(error)
return render_template("search.html")
return render_template('search.html',results=results)
except KeyError:
error='You must be logged in to search for books.'
flash(error)
return redirect(url_for('login'))
def book_search(keyword):
book_list = []
#keyword = '%'+keyword+'%'
results = db.execute("""SELECT * FROM books
WHERE book_isbn like :keyword
OR UPPER(title) like :keyword
OR UPPER(author) like :keyword
ORDER BY title, author, book_isbn""",
{"keyword":'%'+keyword.upper()+'%'})
for result in results:
isbn=result['book_isbn']
title=result['title']
year=result['year']
author=result['author']
new_book=Book(isbn,title,year,author)
book_list.append(new_book)
return(book_list)
@app.route("/details/<book_isbn>", methods=["GET"])
def details(book_isbn):
error=None
try:
user_id = session["user_id"]
dict_goodreads = get_goodreads_data(book_isbn)
book = get_book_by_isbn(book_isbn)
reviews = get_reviews(book_isbn)
allow_review = single_book_review_success(user_id,book_isbn)
return render_template('book.html',book=book, avg_rating=dict_goodreads["avg_rating"],ratings_count=dict_goodreads["ratings_count"], reviews=reviews, allow_review=allow_review)
except KeyError:
error='You must be logged in to search for books.'
flash(error)
return redirect(url_for('login'))
@app.route("/details/<book_isbn>", methods=["GET","POST"])
def reviews(book_isbn):
error=None
try:
dict_goodreads = get_goodreads_data(book_isbn)
user_id = session["user_id"]
review = request.form.get("review")
rating = request.form['rating']
book = get_book_by_isbn(book_isbn)
error = add_review(book_isbn,user_id,review,rating)
reviews = get_reviews(book_isbn)
allow_review = single_book_review_success(user_id,book_isbn)
if error:
flash(error)
return render_template('book.html',book=book, avg_rating=dict_goodreads["avg_rating"],ratings_count=dict_goodreads["ratings_count"], reviews=reviews, allow_review=allow_review)
except KeyError:
error='You must be logged in to search for books.'
flash(error)
return redirect(url_for('login'))
@app.route("/api/<book_isbn>", methods=["GET"])
def api(book_isbn):
json = get_book_api_data(book_isbn)
return(json)
def get_book_cover(book_isbn):
res = ("https://covers.openlibrary.org/b/isbn/{book_isbn}-M.jpg",book_isbn)
return(res)
def get_goodreads_data(book_isbn):
try:
res = requests.get("https://www.goodreads.com/book/review_counts.json", params={"key": GOODREADS_KEY, "isbns": book_isbn})
response = res.json()
avg_rating = response['books'][0]['average_rating']
ratings_count = response['books'][0]['work_ratings_count']
except:
avg_rating='Not available'
ratings_count = 'Not available'
dict_goodreads_data = {"avg_rating": avg_rating,
"ratings_count": ratings_count}
return(dict_goodreads_data)
def get_book_by_isbn(book_isbn):
consult = db.execute("SELECT * FROM books WHERE book_isbn = :book_isbn",
{"book_isbn":book_isbn}).fetchone()
isbn=consult['book_isbn']
title=consult['title']
year=consult['year']
author=consult['author']
book=Book(isbn,title,year,author)
return(book)
def add_review(book_isbn,user_id,review,rating):
error = None
if single_book_review_success(user_id,book_isbn):
db.execute("""INSERT INTO reviews (book_id,user_id,review_text,review_rating)
VALUES (:book_id, :user_id, :review_text, :review_rating)""",
{"book_id": book_isbn, "user_id": user_id, "review_text": review, "review_rating": rating})
db.commit()
else:
error='The user has already written a review for the selected book'
return(error)
def single_book_review_success(user_id,book_isbn):
result = db.execute("""SELECT review_text, review_rating FROM reviews
WHERE book_id = :book_id AND user_id=:user_id""",
{"user_id":user_id,"book_id":book_isbn}).fetchone()
if result is None:
return(True)
else:
return(False)
def get_reviews(book_isbn):
results = db.execute("""SELECT u.username, r.review_text, r.review_rating
FROM reviews r
INNER JOIN user_accounts u on u.user_id = r.user_id
WHERE r.book_id = :book_isbn""",
{"book_isbn":book_isbn}).fetchall()
return(results)
def get_book_api_data(book_isbn):
try:
result = jsonify(dict(db.execute("""SELECT b.title, b.author, b.year,
b.book_isbn, count(r.review_id), avg(r.review_rating)
FROM books b
INNER JOIN reviews r on r.book_id = b.book_isbn
WHERE b.book_isbn = :book_isbn
GROUP BY b.title, b.author, b.year,
b.book_isbn""",
{"book_isbn":book_isbn}).fetchone()))
return(result)
except TypeError:
return render_template("404.html")
@app.route("/profile", methods=["GET"])
def profile():
try:
user_id = session['user_id']
reviews = db.execute("""SELECT r.review_text, r.review_rating, b.book_isbn, b.title, b.author
FROM reviews r
INNER JOIN books b on b.book_isbn = r.book_id
WHERE r.user_id = :user_id""",
{"user_id":user_id}).fetchall()
return render_template("profile.html", reviews=reviews)
except KeyError:
error='You must be logged in.'
flash(error)
return redirect(url_for('login'))