From 2643f4812f39d495730454cab31c478de6ffa25e Mon Sep 17 00:00:00 2001 From: Tyler Date: Tue, 6 Jun 2023 18:51:27 -0700 Subject: [PATCH] Bugfixing for slightly invalid queries and handling of malformed ones --- app.py | 241 ++++++++++++++++++++++++++++++------------ tools/sanity_check.py | 1 + 2 files changed, 173 insertions(+), 69 deletions(-) diff --git a/app.py b/app.py index c6ab54a..fc81a5e 100644 --- a/app.py +++ b/app.py @@ -1,5 +1,13 @@ # app.py -from flask import Flask, request, jsonify, send_file, render_template, redirect, Response +from flask import ( + Flask, + request, + jsonify, + send_file, + render_template, + redirect, + Response, +) import random from datetime import date, datetime, timedelta # used for daily verse import sqlite3 @@ -15,7 +23,7 @@ timestamp_runtime: datetime = datetime.now().strftime("%d%b%y-%H:%M") logging.basicConfig( filename=f"logs/kjv-api_{datetime.now().strftime('%d%b%y')}.log", encoding="utf-8", - level=logging.INFO, + level=logging.DEBUG, ) logging.info(f"Started: {timestamp_runtime}") @@ -32,7 +40,7 @@ except Exception as e: finally: logging.info(f"Sanity: {sanity}") -app = Flask(__name__, template_folder='html') +app = Flask(__name__, template_folder="html") """ Current work prefixed with * @@ -63,6 +71,7 @@ cached: dict = { i[0] for i in kjv_cur.execute("SELECT n FROM key_english").fetchall() ], "chapter_list": [], + "books_chapter_lengths": {}, "json_dictionary": json.load(open("data/1828_Webster_KJV.json", "r")), "html_response": open("html/response.html", "r").read(), "html_navbar": open("html/navbar.html", "r").read(), @@ -79,38 +88,47 @@ cached: dict = { if int(str(chap)[:-3]) not in cached["chapter_list"] ] cached["chapter_list"].sort() +for book in range(1, 66): + cached["books_chapter_lengths"][book] = kjv_cur.execute( + "SELECT c FROM fts_kjv where b = ? ORDER BY id DESC LIMIT 1;", (book,) + ).fetchone()[0] + def generate_short_id(): - remain = random.randrange(614_656,17_210_367) # resolves to length of 5 - chars="0123456789ACEFHJKLMNPRTUVWXY" - length=len(chars) - result="" - while remain>0: - pos = remain % length - remain = remain // length - result = chars[pos] + result - return result + remain = random.randrange(614_656, 17_210_367) # resolves to length of 5 + chars = "0123456789ACEFHJKLMNPRTUVWXY" + length = len(chars) + result = "" + while remain > 0: + pos = remain % length + remain = remain // length + result = chars[pos] + result + return result + + def lookup_bookname(book_id): - logging.debug(f"lookup_bookname({book_id})") result = kjv_cur.execute( f"SELECT n FROM key_english WHERE b = {book_id};" ).fetchone() if result: result = result[0] - logging.debug(f"lookup_bookname.result = '{result}'") return result -def lookup_book_id(bookname): - logging.debug(f"lookup_book_id({bookname})") + + +def lookup_book_id(bookname) -> int: result = kjv_cur.execute( f"SELECT b FROM key_english WHERE n = '{bookname}';" ).fetchone() if result: result = result[0] - logging.debug(f"lookup_book_id.result = '{result}'") return result + + def lookup_fts(kwds): logging.debug(f"lookup_fts.kwds = '{kwds}'") return kjv_cur.execute(f"SELECT * FROM fts_kjv('{kwds}')").fetchall() + + def lookup_by_verse_id(start_verse, end_verse=None): if not end_verse: results = kjv_cur.execute( @@ -121,21 +139,31 @@ def lookup_by_verse_id(start_verse, end_verse=None): f"SELECT * FROM fts_kjv WHERE id BETWEEN {start_verse} AND {end_verse};" ).fetchall() return results -def get_verse_id(book_id, chapter, verse): + + +def get_verse_id(book_id, chapter: str, verse: str) -> str: # verse_id is 00000000 (00 000 000) (book + chapter + verse) chapterid = chapter.zfill(3) verseid = verse.zfill(3) if verseid == "000": verseid = "001" # Avoid verse_ids list IndexErrors on :- searches return str(book_id) + chapterid + verseid -def get_next_prev_chapter(verseid): + + +def get_next_prev_chapter(verseid) -> dict: current_chapter = verseid[:-3] current_chapter_index = cached["chapter_list"].index(int(current_chapter)) - - #next_chapter_id = str(cached["chapter_list"][current_chapter_index + 1]) - next_chapter_id = str(cached["chapter_list"][(current_chapter_index + 1) % len(cached["chapter_list"])]) - #prev_chapter_id = str(cached["chapter_list"][current_chapter_index - 1]) - prev_chapter_id = str(cached["chapter_list"][(current_chapter_index - 1) % len(cached["chapter_list"])]) + + next_chapter_id = str( + cached["chapter_list"][ + (current_chapter_index + 1) % len(cached["chapter_list"]) + ] + ) + prev_chapter_id = str( + cached["chapter_list"][ + (current_chapter_index - 1) % len(cached["chapter_list"]) + ] + ) current_chapter_name = " ".join( (lookup_bookname(current_chapter[:-3]), str(current_chapter[-3:]).lstrip("0")) ) @@ -155,15 +183,20 @@ def get_next_prev_chapter(verseid): "next_chapter_link": next_chapter_link, } return result + + def ref_input_cleaning( ref, valid_chars=(string.ascii_letters + string.digits + ":-,"), ref_search=True ): logging.debug(f"ref_input_cleaning({ref})") ref = ref.replace(" ", "") if ref.count(":") != 1 and ref_search: - logging.debug("ref_input_cleaning.if ref.count(:): TRUE") - logging.debug(" -> Assuming user wanted entire chapter") - ref += ":-" + logging.debug("ref_input_cleaning: Assuming user wanted entire chapter") + if ref[-1] in string.digits: + ref += ":-" + else: + # TODO: This should return something like /browse and list the chapters + ref += "1:-" if ref.endswith(":"): ref += "-" if ref.count("-") > 1 and ref_search: @@ -191,8 +224,10 @@ def ref_input_cleaning( return prefix + "".join(cleaned_ref).capitalize() else: return "".join(cleaned_ref) + + def search_input_cleaning( - query, valid_chars=string.ascii_letters + string.digits + '+^* ' + query, valid_chars=string.ascii_letters + string.digits + "+^* " ): # should protect against SQL injection, maybe. Should have another person review. cleaned_query = [] @@ -200,13 +235,17 @@ def search_input_cleaning( if character in valid_chars: cleaned_query.append(character) return "".join(cleaned_query) + + def ref_parse_book(ref): for book in reversed(cached["book_names"]): if book.startswith(ref): return book return False + + @app.get("/find") -def parse_query(query_override:str = None): +def parse_query(query_override: str = None): query = request.args.get("kw", None) # /find should always return human readable (rich) but some users may try # to use it with other view arguments, we can try to support that. @@ -219,7 +258,7 @@ def parse_query(query_override:str = None): if query: query = query.strip() if request.args.get("view", "rich") != "rich": - return jsonify({"Error":"/find endpoint only supports rich responses"}) + return jsonify({"Error": "/find endpoint only supports rich responses"}) logging.info(f"Query: {query}") if not query: return send_file("html/find.html") @@ -235,7 +274,9 @@ def parse_query(query_override:str = None): result = "" result += search() result += "" - result += "

1828 Webster Definitions

Click on the word to get the definition." + result += ( + "

1828 Webster Definitions

Click on the word to get the definition." + ) split_query = query.replace(",", " ").split(" ") if len(split_query) > 20: return "Your Query is too long. 20 Words or less." @@ -252,9 +293,11 @@ def parse_query(query_override:str = None): logging.debug(f"parse_query.define = '{word}'") result += "" result += "
" - #result += search() + # result += search() logging.debug(f"parse_query.search = '{result}'") return result + + @app.get("/define") def define(word=None): ref = request.args.get("kw", None) @@ -285,11 +328,15 @@ def define(word=None): else: response = jsonify({ref: dictionary[ref]}) return response + + def return_dict_plain(ref, defs): response = "" for e in defs: response += e + "

" return ref.capitalize() + "

" + response + "\n" + + def return_dict_rich(ref, defs, simple): if simple: css = "" @@ -304,9 +351,11 @@ def return_dict_rich(ref, defs, simple): ) + css ) + + @app.get("/show") def show(kw=None, get_start_verse_id=False, human_readable=False): - ref = request.args.get("kw", None) + ref: str = request.args.get("kw", None) get_chapter: bool = request.args.get("all", False) if ref.endswith(","): ref = ref[:-1] @@ -315,23 +364,19 @@ def show(kw=None, get_start_verse_id=False, human_readable=False): if ref is None: return send_file("html/show.html") logging.debug(f"show({ref})") - refs = [] + refs: list = [] for r in ref.split(","): r_ = ref_input_cleaning(r.strip()) if r_: refs.append(r_) if not refs: return "Invalid Query" - verse_results = [] + verse_results: list = [] # check if the search is for multiple refs for ref in refs: logging.debug(f"show.reference({ref})") - maybe_chapter = ref.split(":")[0][3:] - #FIXME: Not including : in a book reference should take you to the first book - # not just crash like a retard - if not ":" in maybe_chapter: - get_chapter = True - ref_chapter = "" + maybe_chapter: str = ref.split(":")[0][3:] + ref_chapter: str = "" for maybedigit in maybe_chapter: if maybedigit.isdigit(): ref_chapter += maybedigit @@ -349,11 +394,14 @@ def show(kw=None, get_start_verse_id=False, human_readable=False): start_verse = ref_verse end_verse = None end_verse_id = None - book_str = ref_input_cleaning(ref, string.ascii_letters) - ref_bookname = ref_parse_book(book_str) - ref_book_id = lookup_book_id(ref_bookname) + book_str: str = ref_input_cleaning(ref, string.ascii_letters) + ref_bookname: str = ref_parse_book(book_str) + ref_book_id: int = lookup_book_id(ref_bookname) if not ref_book_id: return "Invalid Query" + maximum_chapters: int = cached.get("books_chapter_lengths").get(ref_book_id) + if int(ref_chapter) > maximum_chapters: + ref_chapter = str(maximum_chapters) start_verse_id = get_verse_id(ref_book_id, ref_chapter, start_verse) if get_start_verse_id: return start_verse_id @@ -362,7 +410,7 @@ def show(kw=None, get_start_verse_id=False, human_readable=False): logging.debug(f"show.start_verse_id({start_verse_id})") logging.debug(f"show.end_verse_id({end_verse_id})") results = lookup_by_verse_id(start_verse_id, end_verse_id) - logging.debug(f"show.results({results})") + logging.debug(f"show.results(len({results}))") for result in results: verse_results.append(result) response_list = scripture_response(verse_results, human_readable) @@ -374,6 +422,8 @@ def show(kw=None, get_start_verse_id=False, human_readable=False): else: response = jsonify(response_list) return response + + def response_dict(scripture): # used for JSON responses verse = { @@ -383,12 +433,16 @@ def response_dict(scripture): "text": scripture[4], } return verse + + def response_text(scripture): bookname = lookup_bookname(scripture[1]) chapter = str(scripture[2]) verse = str(scripture[3]) text = str(scripture[4]) return bookname + " " + chapter + ":" + verse + " " + text + "\n" + + def response_rich(scripture): # (passage, text) is the html formatting arguments bookname = lookup_bookname(scripture[1]) @@ -399,6 +453,8 @@ def response_rich(scripture): return cached["html_response"].format( chapter=passage.split(":")[0], passage=passage, text=text ) + + def scripture_response(scriptures, human_readable=False): response = [] arg_view = request.args.get("view", "json") @@ -407,10 +463,15 @@ def scripture_response(scriptures, human_readable=False): for scripture in scriptures: response.append(response_text(scripture)) elif arg_view == "rich" or human_readable: + try: + scriptures[0][0] + except IndexError: + logging.error(f"Malformed query: {request.args.get('kw',None)}") + return(("Malformed Query - if you believe this to be in error (your query should resolve), email tyler@dinsmoor.us",)) l_r_nav = get_next_prev_chapter( str(scriptures[0][0]) ) # get the first scripture, the id thereof - #response.append(cached["css_response"]) + # response.append(cached["css_response"]) response.append( cached["html_navbar"].format( query=request.args.get("kw", ""), @@ -419,7 +480,6 @@ def scripture_response(scriptures, human_readable=False): next_chapter_link=l_r_nav["next_chapter_link"], prev_chapter_name=l_r_nav["prev_chapter_name"], prev_chapter_link=l_r_nav["prev_chapter_link"], - ) ) @@ -433,25 +493,33 @@ def scripture_response(scriptures, human_readable=False): for scripture in scriptures: response.append(response_dict(scripture)) return response + + @app.errorhandler(404) def page_not_found(e): message = { "error": "You're lost! Isaiah 41:10 - Fear thou not; for I am with thee: be not dismayed; for I am thy God: I will strengthen thee; yea, I will help thee; yea, I will uphold thee with the right hand of my righteousness." } return jsonify(message) + + @app.get("/") @app.get("/") -def idx(shortcode=''): +def idx(shortcode=""): if not sanity: - return("Sanity check failed. Check log, or please notify tyler@dinsmoor.us") + return "Sanity check failed. Check log, or please notify tyler@dinsmoor.us" if shortcode: return resolve_shortcode(shortcode) logging.info("Index Served") return send_file("html/index.html") + + @app.get("/salvation") def salvation(): return send_file("html/heaven.html") -@app.get("/t/random", defaults={"view": "plain"}) # backwards-compatibility + + +@app.get("/t/random", defaults={"view": "plain"}) # backwards-compatibility @app.get("/random") def random_verse(): rand_id = random.SystemRandom().choice(cached["verse_ids"]) @@ -467,7 +535,9 @@ def random_verse(): response = jsonify(response_list) logging.debug(f"Search Response: {response}") return response -@app.get("/t/votd", defaults={"view": "plain"}) # backwards-compatibility + + +@app.get("/t/votd", defaults={"view": "plain"}) # backwards-compatibility @app.get("/votd") def verse_of_the_day(): # plain takes 38ms -TODO can we cache this? @@ -486,8 +556,10 @@ def verse_of_the_day(): response = jsonify(response_list) logging.debug(f"Search Response: {response}") return response + + def make_view(response_list=[]): - #TODO: Maybe use flask session to make this easier to deal with + # TODO: Maybe use flask session to make this easier to deal with # https://stackoverflow.com/a/53152394 arg_view = request.args.get("view", None) if arg_view == "plain": @@ -497,6 +569,7 @@ def make_view(response_list=[]): else: response = jsonify(response_list) + @app.get("/search") def search(): keywords = request.args.get("kw", None) @@ -524,6 +597,8 @@ def search(): response = jsonify(response_list) logging.debug(f"Search Response: {response}") return response + + @app.get("/browse") def bible_reader(): return render_template("reader.html", books=cached.get("book_names")) @@ -534,6 +609,8 @@ def bible_reader(): # header should have a navigation bar and "next/previous chapter" # Footer should have next/previous as well. # Maybe preliminary theming support + + @app.get("/plan") def reading_plan(): # needs 2 sets of data, books and time frame. @@ -542,26 +619,31 @@ def reading_plan(): # Should return reading_plan template. return render_template("reading_plan.html") + @app.get("/status") def status(): import resource + return f"""Sanity: {sanity}
UserMode: {usermode}
Memory: {resource.getrusage(resource.RUSAGE_SELF).ru_maxrss} """ + + @app.get("/seqreport") def sequential_report(uid=None): if usermode == "Single": - seq_conn = sqlite3.connect("data/seq.db", check_same_thread=False) + seq_conn = sqlite3.connect("data/seq.db", check_same_thread=False) else: seq_conn = sqlite3.connect("data/seq.db") seq_cur = seq_conn.cursor() uids = seq_cur.execute("select * from reading_tracker").fetchall() return render_template("seq_report.html", uids=uids) + + def sequential_purge(): - #TODO: Test this :^) if usermode == "Single": - seq_conn = sqlite3.connect("data/seq.db", check_same_thread=False) + seq_conn = sqlite3.connect("data/seq.db", check_same_thread=False) else: seq_conn = sqlite3.connect("data/seq.db") seq_cur = seq_conn.cursor() @@ -577,6 +659,8 @@ def sequential_purge(): seq_conn.commit() seq_cur.close() + + @app.get("/seq") def sequential_read(start=0, num=1, uid=None, getuid=False): # /seq?uid=1 @@ -590,11 +674,13 @@ def sequential_read(start=0, num=1, uid=None, getuid=False): if not uid and not getuid: return send_file("html/seq.html") if usermode == "Single": - seq_conn = sqlite3.connect("data/seq.db", check_same_thread=False) + seq_conn = sqlite3.connect("data/seq.db", check_same_thread=False) else: seq_conn = sqlite3.connect("data/seq.db") seq_cur = seq_conn.cursor() - seq_cur.execute("CREATE TABLE IF NOT EXISTS reading_tracker (uid TEXT PRIMARY KEY, verse INTEGER, last_used TEXT)") + seq_cur.execute( + "CREATE TABLE IF NOT EXISTS reading_tracker (uid TEXT PRIMARY KEY, verse INTEGER, last_used TEXT)" + ) last_used = str(datetime.now()) num -= 1 # 0 will mean "next" if num < 0: @@ -604,13 +690,15 @@ def sequential_read(start=0, num=1, uid=None, getuid=False): start = cached["verse_ids"].index( int(show(kw=start, get_start_verse_id=True)) ) - uid = generate_short_id() # range returns a 5 digit id + uid = generate_short_id() # range returns a 5 digit id try: seq_cur.execute( f"INSERT INTO reading_tracker VALUES('{uid}', {start}, '{last_used}')" ) except sqlite3.OperationalError as e: - logging.error(f"UID Database failed to insert {uid} into reading_tracker. {e}") + logging.error( + f"UID Database failed to insert {uid} into reading_tracker. {e}" + ) return jsonify( { "error": "Something went wrong with the UID database. Please notify tyler@dinsmoor.us" @@ -643,7 +731,7 @@ def sequential_read(start=0, num=1, uid=None, getuid=False): f"REPLACE INTO reading_tracker (uid, verse, last_used) VALUES('{uid}', {nextverse+num+1}, '{last_used}')" ) # Doing REPLACE INTO instead of UPDATE will make the updated row at the end of the table, making the purging - # only look at a small set of the oldest unused entries. + # only look at a small set of the oldest unused entries. seq_conn.commit() arg_view = request.args.get("view", None) if arg_view == "plain" or arg_view == "rich": @@ -652,17 +740,21 @@ def sequential_read(start=0, num=1, uid=None, getuid=False): response = jsonify(response_list) return response + + @app.get("/mksc/") def generate_shortcode(query: str = ""): query = search_input_cleaning(query) if not query: - return("Empty query.") + return "Empty query." if usermode == "Single": - short_conn = sqlite3.connect("data/shortcodes.db", check_same_thread=False) + short_conn = sqlite3.connect("data/shortcodes.db", check_same_thread=False) else: short_conn = sqlite3.connect("data/shortcodes.db") short_cur = short_conn.cursor() - short_cur.execute("CREATE TABLE IF NOT EXISTS shortcodes (shortcode TEXT PRIMARY KEY, query TEXT UNIQUE, creation_date TEXT)") + short_cur.execute( + "CREATE TABLE IF NOT EXISTS shortcodes (shortcode TEXT PRIMARY KEY, query TEXT UNIQUE, creation_date TEXT)" + ) short_conn.commit() creation_date = str(datetime.now()) shortcode = generate_short_id() @@ -673,23 +765,34 @@ def generate_shortcode(query: str = ""): short_cur.close() logging.info(f"Generated Shortcode {shortcode} for query '{query}'") return shortcode + + def resolve_shortcode(shortcode: str = ""): - shortcode = search_input_cleaning(shortcode, valid_chars="0123456789ACEFHJKLMNPRTUVWXY") + shortcode = search_input_cleaning( + shortcode, valid_chars="0123456789ACEFHJKLMNPRTUVWXY" + ) if not shortcode: - return("Empty shortcode.") + return "Empty shortcode." if usermode == "Single": - short_conn = sqlite3.connect("data/shortcodes.db", check_same_thread=False) + short_conn = sqlite3.connect("data/shortcodes.db", check_same_thread=False) else: short_conn = sqlite3.connect("data/shortcodes.db") short_cur = short_conn.cursor() - query = short_cur.execute(f"SELECT query FROM shortcodes WHERE shortcode = '{shortcode}'").fetchone()[0] + query = short_cur.execute( + f"SELECT query FROM shortcodes WHERE shortcode = '{shortcode}'" + ).fetchone()[0] short_cur.close() logging.info(f"Resolved Shortcode {shortcode} for query '{query}'") return redirect(f"/find?kw={query}&view=rich") + if __name__ == "__main__": - print("You must run this program within a uwsgi environment if you are running this for multiple users.") - print("Otherwise, you may end up with database corruption if multiple writes happen.") + print( + "You must run this program within a uwsgi environment if you are running this for multiple users." + ) + print( + "Otherwise, you may end up with database corruption if multiple writes happen." + ) print("Operating in single user mode.") app.run(host="127.0.0.1", port=1611) diff --git a/tools/sanity_check.py b/tools/sanity_check.py index 7ab1caa..f93a29d 100644 --- a/tools/sanity_check.py +++ b/tools/sanity_check.py @@ -43,6 +43,7 @@ def check_queries(url=None): for logged in log: if logged[1] != 200: logfile.write(f"{logged[1]} - {logged[0]}\n") + print(f"{logged[1]} - {logged[0]}\n") if __name__ == "__main__": import argparse