Files
kjv_api/app.py
T

1355 lines
54 KiB
Python

# app.py
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
import string # for literals shortcut
import json # for dictionary
import hunspell # for spell checking (not really used)
import os
import sys
import logging
import socket
import json
from tools import sanity_check
try:
from dotenv import load_dotenv
load_dotenv()
SERMON_DB = os.getenv("KJV_SERMON_DB")
SECRET_API_KEY = os.getenv("KJV_SECRET_API_KEY")
except:
KJV_SERMON_DB="./data/sermons.db"
# this does not do anything unless sermon_analyzer.py is running in tcp socket mode as a separate process
KJV_SECRET_API_KEY="testing"
sequential_purge_delta: timedelta = timedelta(days=90)
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,
)
logging.info(f"Started: {timestamp_runtime}")
logging.info(f"Using Sermon Database: {SERMON_DB}")
try:
sanity_check.check_sanity()
sanity: bool = True
except PermissionError as e:
logging.critical(f"PERMISSIONS FAILURE: {e}")
sanity: bool = False
except Exception as e:
logging.critical(e)
sanity: bool = False
finally:
logging.info(f"Sanity: {sanity}")
app = Flask(__name__, template_folder="html")
"""
Current work prefixed with *
TODO:
Replace send_file with something that loads headers and nav templates
Search phrase alias dictionary
Search highlighting
Bible Reading Plan
/seq Going backwards (works but is buggy, disabled for now)
/seq Wrap back to Gen 1:1 when you reach the end of Rev
/seq Querying what your UID's next verse is going to be
https://docs.python.org/3/library/sqlite3.html#sqlite3-placeholders
URL Shortener for sharing groups of scripture.
"""
def dict_factory(cursor, row):
fields = [column[0] for column in cursor.description]
return {key: value for key, value in zip(fields, row)}
kjv_cur = None
sermon_cur = None
if __name__ == "__main__":
# single user mode, running without uwsgi
usermode: str = "Single"
kjv_cur = sqlite3.connect("data/kjv.db", check_same_thread=False).cursor()
if SERMON_DB:
sermon_con = sqlite3.connect(SERMON_DB, check_same_thread=False)
sermon_con.row_factory = dict_factory
sermon_cur = sermon_con.cursor()
else:
# you're running with uwsgi
usermode: str = "Multi"
kjv_cur = sqlite3.connect("data/kjv.db").cursor()
if SERMON_DB:
sermon_con = sqlite3.connect(SERMON_DB)
sermon_con.row_factory = dict_factory
sermon_cur = sermon_con.cursor()
if not kjv_cur:
exit("CRITICAL: Error connecting to KJV Database")
if not sermon_cur:
print("WARN: Error connecting to Sermon Database")
# Build grammatical variants lookup from word groups
variant_groups = json.load(open("data/grammatical_variants.json", "r"))
grammatical_variants_map = {}
for group in variant_groups:
for word in group:
# Map each word to all other words in its group
grammatical_variants_map[word] = [w for w in group if w != word]
# Build theological synonyms lookup from word groups
theological_synonym_groups = json.load(open("data/theological_synonyms.json", "r"))
theological_synonyms_map = {}
for group in theological_synonym_groups:
for word in group:
# Map each word to other words in its group
theological_synonyms_map[word] = [w for w in group if w != word]
# Load concept mappings for common theological terms not in KJV
concept_mappings = json.load(open("data/concept_mappings.json", "r"))
concept_mappings_dict = {mapping["query"].lower(): mapping for mapping in concept_mappings}
cached: dict = {
"verse_ids": [i[0] for i in kjv_cur.execute("SELECT id FROM fts_kjv;").fetchall()],
"book_names": [
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")),
"grammatical_variants": grammatical_variants_map,
"theological_synonyms": theological_synonyms_map,
"concept_mappings": concept_mappings_dict,
"html_response": open("html/response.html", "r").read(),
"html_navbar": open("html/navbar.html", "r").read(),
"css_navbar": open("styles/header.css", "r").read(),
"css_response": open("styles/human_readable.css", "r").read(),
"css_collapsible": open("styles/collapsible.css", "r").read(),
"css_page": open("styles/page.css", "r").read(),
"js_collapsible": open("js/collapsible.js", "r").read(),
"js_shortcode": open("js/shortcode.js", "r").read(),
"favicon_svg": open("static/favicon.svg", "r").read(),
"book_info": kjv_cur.execute(
'SELECT "order", title_short, category, otnt, chapters FROM book_info ORDER BY "order"'
).fetchall(),
}
# Fill the chapter list with unique chapters for seeking back and forward by index.
[
cached["chapter_list"].append(int(str(chap)[:-3]))
for chap in cached.get("verse_ids")
if int(str(chap)[:-3]) not in cached["chapter_list"]
]
cached["chapter_list"].sort()
for book in range(1, 67):
cached["books_chapter_lengths"][book] = kjv_cur.execute(
"SELECT c FROM fts_kjv where b = ? ORDER BY id DESC LIMIT 1;", (book,)
).fetchone()[0]
# Cache book_id to bookname mapping for fast lookups
cached["book_id_to_name"] = {}
for row in kjv_cur.execute("SELECT b, n FROM key_english").fetchall():
cached["book_id_to_name"][row[0]] = row[1]
# Organize books by testament and category for index page
from collections import defaultdict
testament_data = {"OT": defaultdict(list), "NT": defaultdict(list)}
for book in cached["book_info"]:
book_order, title, category, testament, num_chapters = book
testament_data[testament][category].append({
"title": title,
"chapters": num_chapters
})
cached["book_info_for_index"] = dict(testament_data)
# Render the index page once and cache it
with app.app_context():
from flask import render_template_string
template_content = open("html/index.html", "r").read()
cached["html_index"] = render_template_string(template_content, testament_data=cached["book_info_for_index"])
@app.get("/js/shortcode.js")
def serve_shortcode_js():
response = Response(cached.get("js_shortcode"), mimetype='application/javascript')
response.headers['Cache-Control'] = 'public, max-age=31536000'
return response
@app.get("/js/collapsible.js")
def serve_collapsible_js():
response = Response(cached.get("js_collapsible"), mimetype='application/javascript')
response.headers['Cache-Control'] = 'public, max-age=31536000'
return response
@app.get("/styles/page.css")
def serve_page_css():
response = Response(cached.get("css_page"), mimetype='text/css')
response.headers['Cache-Control'] = 'public, max-age=31536000'
return response
@app.get("/favicon.ico")
@app.get("/favicon.svg")
def serve_favicon():
response = Response(cached.get("favicon_svg"), mimetype='image/svg+xml')
response.headers['Cache-Control'] = 'public, max-age=31536000'
return response
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
def lookup_bookname(book_id):
# Ensure book_id is an integer for cache lookup
if isinstance(book_id, str):
book_id = int(book_id)
return cached["book_id_to_name"].get(book_id)
def lookup_book_id(bookname) -> int:
result = kjv_cur.execute(
"SELECT b FROM key_english WHERE n = ?;",(bookname,)).fetchone()
if result:
result = result[0]
return result
def lookup_fts(kwds):
logging.debug(f"lookup_fts.kwds = '{kwds}'")
return kjv_cur.execute("SELECT * FROM fts_kjv(?)", (kwds,)).fetchall()
def lookup_by_verse_id(start_verse, end_verse=None) -> list:
# This is an internal function - we don't use placeholders because SQL injection should be impossible here
if not end_verse:
results = kjv_cur.execute(
f"SELECT * FROM fts_kjv WHERE id = {start_verse};"
).fetchall()
else:
results = kjv_cur.execute(
f"SELECT * FROM fts_kjv WHERE id BETWEEN {start_verse} AND {end_verse};"
).fetchall()
return results
def get_sermons_by_verse_id(verse_id) -> list:
if not sermon_cur:
return
return sermon_cur.execute(f"select * from verse_refs where verse_id={verse_id}").fetchall()
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_ref_from_verse_id(verse_id):
# verse_id is 00000000 (00 000 000) (book + chapter + verse)
verse_id = str(verse_id).zfill(8)
book_id = verse_id[:2]
chapter = verse_id[2:5]
verse = verse_id[5:8]
bookname = kjv_cur.execute("SELECT n FROM key_english WHERE b = ?;",(book_id,)).fetchone()[0]
return bookname + " " + chapter + ":" + verse
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) % 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"))
)
next_chapter_name = " ".join(
(lookup_bookname(next_chapter_id[:-3]), str(next_chapter_id[-3:]).lstrip("0"))
)
prev_chapter_name = " ".join(
(lookup_bookname(prev_chapter_id[:-3]), str(prev_chapter_id[-3:]).lstrip("0"))
)
next_chapter_link = f"/show?kw={next_chapter_name}&view=rich"
prev_chapter_link = f"/show?kw={prev_chapter_name}&view=rich"
result = {
"current_chapter_name": current_chapter_name,
"prev_chapter_name": prev_chapter_name,
"prev_chapter_link": prev_chapter_link,
"next_chapter_name": next_chapter_name,
"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:
if len(ref) == 0:
return "Invalid Query"
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:
logging.debug("ref_input_cleaning.if ref.count(-): TRUE")
return False
if not ref[-1].isdigit() and ref[-1] != "-" and ref_search:
logging.debug("ref_input_cleaning.if ref[-1] TRUE")
return False
ref = list(ref)
# preserve book number if we can guess the first
# character is a number and 3nd is a letter
if ref[0].isdigit() and ref[1].isalpha() and ref_search:
prefix = ref.pop(0) + " "
else:
prefix = ""
# clean the rest of the reference
cleaned_ref = []
for character in ref:
if character in valid_chars:
cleaned_ref.append(character)
# Was there a problem?
if len(cleaned_ref) == 0:
return False
if ref_search:
return prefix + "".join(cleaned_ref).capitalize()
else:
return "".join(cleaned_ref)
def search_input_cleaning(
query, valid_chars=string.ascii_letters + string.digits + "+^* ", extra_chars=''
):
valid_chars += extra_chars
# should protect against SQL injection, maybe. Should have another person review.
cleaned_query = []
for character in query:
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
def request_semantic_search(query):
HOST = 'localhost'
PORT = 1613
data = {"query": query}
data_string = json.dumps(data).encode()
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.connect((HOST, PORT))
s.sendall(data_string)
data = b''
while True:
part = s.recv(1024)
if not part:
break
data += part
return json.loads(data.decode())
@app.get("/natural")
def explain_natural():
res = 'Submit a POST request to this endpoint with the following params in the body:\n'
res += '{"token": "mytokenstring", "query" : "myquerystring"}'
return res
@app.post("/natural") # needs valid token and query in request body
def do_semantic_search():
request_data = request.get_json()
if request_data is None:
# Handle error: No JSON data sent
return jsonify({'error': 'Invalid request format'}), 400
token_string = request_data.get('token')
query_string = request_data.get('query')
if token_string is None:
# Handle error: Missing token in request
return jsonify({'error': 'Missing token in request body'}), 401
if token_string != SECRET_API_KEY:
return jsonify({'error': "Invalid Authorization"})
result = request_semantic_search(query=query_string)
return jsonify(result)
@app.get("/ref/<ref>") # find database verse_id by english scripture reference (1John5:7)
def get_single_reference_verse_id(ref:str):
ref = ref.split(",")[0]
ref = ref.split("-")[0]
# only want the first single ref, not a range, and not a group.
ref = ref_input_cleaning(ref)
maybe_chapter: str = ref.split(":")[0][3:]
ref_chapter: str = ""
for maybedigit in maybe_chapter:
if maybedigit.isdigit():
ref_chapter += maybedigit
ref_verse = ref.split(":")[1]
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
return get_verse_id(ref_book_id, ref_chapter, ref_verse)
@app.get("/sermon/search/<ref>") # find a sermon by english scripture reference (1John5:7)
def find_sermon_by_ref(ref:str=None):
verse_id = get_single_reference_verse_id(ref=ref)
if not sermon_cur:
return jsonify({'error':"Sermon database not found"})
sermons = get_sermons_by_verse_id(verse_id=verse_id)
for sermon in sermons:
sermon.update({'kjv':find_by_verseid(sermon.get("verse_id"))[4]})
return jsonify(sermons)
@app.get("/sermons") # lists all the sermons in our database
def list_sermons():
return jsonify(sermon_cur.execute("select distinct title,url,denomination,preacher,church_name,filename from verse_refs").fetchall())
@app.get("/sermon/<filename>/verses") # show all matched info on a sermon by its filename (hash)
def list_sermon_references(filename):
return jsonify(sermon_cur.execute("select distinct verse_id,text from verse_refs where filename=?", (filename,)).fetchall())
@app.get("/preachers") # return a list of preachers stored in database, their denomination and church name
def list_preachers():
return jsonify(sermon_cur.execute("select distinct denomination,preacher,church_name from verse_refs").fetchall())
@app.get("/churches") # return a list of churches with sermons in the database
def list_churches():
return jsonify(sermon_cur.execute("select distinct denomination,church_name from verse_refs").fetchall())
@app.get("/denominations") # return a list of denominations included in the database
def list_denominations():
return jsonify(sermon_cur.execute("select distinct denomination from verse_refs").fetchall())
@app.get("/sermons/preacher/<preacher>") # show all sermons by a paticular preacher
def find_sermons_by_preacher(preacher):
return jsonify(sermon_cur.execute("select distinct title,url,denomination,preacher,church_name from verse_refs where preacher=?", (preacher,)).fetchall())
@app.get("/sermons/church/<church_name>") # show all sermons by a paticular church
def find_sermons_by_church(church_name):
return jsonify(sermon_cur.execute("select distinct title,url,denomination,preacher,church_name from verse_refs where church_name=?", (church_name,)).fetchall())
@app.get("/sermons/denomination/<denomination>") # show all sermons by a paticular denomination
def find_sermons_by_denomination(denomination):
return jsonify(sermon_cur.execute("select distinct title,url,denomination,preacher,church_name from verse_refs where denomination=?", (denomination,)).fetchall())
@app.get("/sermon/<filename>") # show general information about a sermon by filename (hash)
def find_sermons_by_filename(filename):
return jsonify(sermon_cur.execute("select distinct title,url,denomination,preacher,church_name from verse_refs where filename=?", (filename,)).fetchall())
@app.get("/sermon/url/<url>") # show general information about a sermon by url
def find_sermons_by_url(url):
return jsonify(sermon_cur.execute("select distinct title,url,denomination,preacher,church_name from verse_refs where url=?", (url,)).fetchall())
def clean_verseid(verse_id):
verse_id = str(verse_id)
clean_verseid = ''
for char in verse_id:
if char in string.digits:
clean_verseid += char
return clean_verseid
def find_by_verseid(verse_id):
return lookup_by_verse_id(verse_id)[0]
# wow lol I just wrote this and forgot how it works already lmao
@app.get("/verse_id/<verse_id>") # return the database entry of an internal verse_id
def web_verseid(verse_id):
clean_verse_id = clean_verseid(verse_id=verse_id)
return jsonify(lookup_by_verse_id(clean_verse_id)[0])
@app.get("/find") # user-facing general search parser. This will move to an independant frontend.
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.
if not request.args.get("view", False):
human_readable = True
else:
human_readable = False
if query_override:
query = query_override
if query:
query = query.strip()
if request.args.get("view", "rich") != "rich":
return jsonify({"Error": "/find endpoint only supports rich responses"})
logging.info(f"Query: {query}")
if not query:
return send_file("html/find.html")
if query.endswith(","):
query = query[:-1]
# Check if query contains both letters and digits (required for reference parsing)
# If only numbers or only letters, skip reference parsing and go to keyword search
has_letters = any(char.isalpha() for char in query)
has_digits = any(char.isdigit() for char in query)
if not (has_letters and has_digits):
result = search()
else:
result = show(kw=query, human_readable=human_readable)
if not hasattr(result, "__len__"):
logging.debug(f"parse_query.show = '{result}'")
return result
if len(result) > 17:
return result
# Reference parsing failed, fall back to keyword search
result = search()
# Remove the closing </div> from search() so we can add dictionary content inside .responses
if result.endswith("</div>"):
result = result[:-6] # Remove last </div>
# Check if we have any dictionary definitions before showing the section
split_query = query.replace(",", " ").split(" ")
if len(split_query) > 20:
return "Your Query is too long. 20 Words or less."
definitions_html = ""
has_definitions = False
for word in split_query:
definition = define(word)
if not definition:
continue
has_definitions = True
definitions_html += (
"<button type='button' class='collapsible'>"
+ word.capitalize()
+ "</button>"
)
definitions_html += "<div class='definition'>" + definition + "</div>"
logging.debug(f"parse_query.define = '{word}'")
# Only add the dictionary section if we found at least one definition
if has_definitions:
result += "<style>" + cached["css_collapsible"] + "</style>"
result += "<div class='dictionary-section'><h3>1828 Webster Definitions</h3><p>Click on the word to get the definition.</p></div>"
result += definitions_html
result += "<script src='/js/collapsible.js'></script>"
result += "</div>"
return result
@app.get("/define") # user-facing Webster Dictionary, this will move to an independent frontend.
def define(word=None):
ref = request.args.get("kw", None)
if word:
ref = word
if ref:
ref = ref.strip()
if not ref:
return send_file("html/dictionary.html")
dictionary = cached["json_dictionary"]
ref = ref.lower()
if not ref in dictionary.keys():
if word:
return False
logging.warning(f"WEBSTER MISSING: {ref}")
return f"{ref} not found."
# search suggestions kind of suck
# h = hunspell.HunSpell("data/kjv.dic", "data/kjv.aff")
# suggestions = h.suggest(ref)
# return "{} not found, try one of the following: {}".format(
# ref, ", ".join(suggestions)
# )
arg_view = request.args.get("view", None)
if arg_view == "plain":
response = return_dict_plain(ref, dictionary[ref])
elif arg_view == "rich":
response = return_dict_rich(ref, dictionary[ref], simple=word)
else:
response = jsonify({ref: dictionary[ref]})
return response
def return_dict_plain(ref, defs):
response = ""
for e in defs:
response += e + "<br><br>"
return ref.capitalize() + "<br><br>" + response + "\n"
def return_dict_rich(ref, defs, simple):
if simple:
css = ""
else:
css = cached["css_response"]
response = ""
for e in defs:
response += e + "<br><br>"
return (
cached["html_response"].format(
chapter=ref.split(":")[0], passage=ref.capitalize(), text=response
)
+ css
)
@app.get("/show") # user facing to show a group of scripture. This will be deprecated and moved to backend.
def show(kw=None, get_start_verse_id=False, human_readable=False):
ref: str = request.args.get("kw", None)
get_chapter: bool = request.args.get("all", False)
if kw:
ref = kw
if not ref:
return send_file("html/show.html")
logging.debug(f"show({ref})")
if ref.endswith(","):
ref = ref[:-1]
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: list = []
# check if the search is for multiple refs
for ref in refs:
if len(ref) < 2:
continue
logging.debug(f"show.reference({ref})")
maybe_chapter: str = ref.split(":")[0][3:]
ref_chapter: str = ""
for maybedigit in maybe_chapter:
if maybedigit.isdigit():
ref_chapter += maybedigit
ref_verse = ref.split(":")[1]
if ref.__contains__("-") or get_chapter:
if ref.endswith("-") or get_chapter:
start_verse = ref_verse.split("-")[0]
end_verse = "500" # just get all the verses if it ends with a dash
else:
start_verse, end_verse = (
ref_verse.split("-")[0],
ref_verse.split("-")[1],
)
else:
start_verse = ref_verse
end_verse = None
end_verse_id = None
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
if end_verse:
end_verse_id = get_verse_id(ref_book_id, ref_chapter, end_verse)
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 len({len(results)})")
for result in results:
verse_results.append(result)
response_list = scripture_response(verse_results, human_readable)
arg_view = request.args.get("view", None)
if arg_view == "plain":
response = "<br>".join(response_list)
elif arg_view == "rich" or human_readable:
response = "".join(response_list)
else:
response = jsonify(response_list)
return response
def response_dict(scripture):
# used for JSON responses
verse = {
"bookname": lookup_bookname(scripture[1]),
"chapter": scripture[2],
"verse": scripture[3],
"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])
chapter = str(scripture[2])
verse = str(scripture[3])
text = str(scripture[4])
passage = bookname + " " + chapter + ":" + verse
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", "rich")
logging.debug(f"search ->requested_format = {arg_view}")
if arg_view == "plain":
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["html_navbar"].format(
query=request.args.get("kw", ""),
current_chapter_name=l_r_nav["current_chapter_name"],
next_chapter_name=l_r_nav["next_chapter_name"],
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"],
)
)
response.append(cached["css_navbar"])
response.append("<div class='responses'>")
response.append(f"<div class='results-count'>{len(scriptures)} Results</div>")
for scripture in scriptures:
response.append(response_rich(scripture))
response.append("</div>")
else:
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("/") # index
@app.get("/<shortcode>") # resolves a generated shortcode
def idx(shortcode=""):
if not sanity:
return "Sanity check failed. Check database ownership or please notify tyler@dinsmoor.us"
if shortcode:
logging.info(f"/{shortcode}")
return resolve_shortcode(shortcode)
logging.info(f"/{shortcode}")
return cached["html_index"]
@app.get("/fe")
def frontend():
return send_file('kjv_fe.html')
@app.get("/salvation") # a web page about going to heaven
def salvation():
return send_file("html/heaven.html")
@app.get("/t/random", defaults={"view": "plain"}) # backwards-compatibility plaintext random Bible verse
@app.get("/random") # truely random Bible verse
def random_verse():
rand_id = random.SystemRandom().choice(cached["verse_ids"])
scripture = lookup_by_verse_id(rand_id)
logging.info(f"Random Served: {scripture[0]}")
response_list = scripture_response(scripture)
arg_view = request.args.get("view", None)
if arg_view == "plain":
response = "<br>".join(response_list)
elif arg_view == "rich":
response = "".join(response_list)
else:
response = jsonify(response_list)
#logging.debug(f"Search Response: {response}")
return response
@app.get("/t/votd", defaults={"view": "plain"}) # backwards-compatibility plaintext "Today's" Bible verse
@app.get("/votd") # "today's" Bible verse
def verse_of_the_day():
# plain takes 38ms -TODO can we cache this?
logging.info("VOTD Served")
today = int(str(date.today()).replace("-", ""))
# this is deterministic because of today's date being used as the seed
random.seed(today)
scripture = lookup_by_verse_id(random.choice(cached["verse_ids"]))
response_list = scripture_response(scripture)
arg_view = request.args.get("view", None)
if arg_view == "plain":
response = "<br>".join(response_list)
elif arg_view == "rich":
response = "".join(response_list)
else:
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
# https://stackoverflow.com/a/53152394
arg_view = request.args.get("view", None)
if arg_view == "plain":
response = "<br>".join(response_list)
elif arg_view == "rich":
response = "<br>".join(response_list)
else:
response = jsonify(response_list)
@app.get("/search") # user-facing keyword search. To be moved to backend, possibly replaced with semantic search.
def search():
keywords = request.args.get("kw", None)
logging.debug(f"search({keywords})")
if keywords:
cleaned_keywords = search_input_cleaning(keywords)
if not cleaned_keywords:
return send_file("html/search.html")
else:
return send_file("html/search.html")
# Check if this is a common theological concept not in KJV
concept_mapping = cached["concept_mappings"].get(cleaned_keywords.lower(), None)
concept_notice = None
if concept_mapping:
# Build hyperlinks for all search terms
original_query = cleaned_keywords
search_links = []
for term in concept_mapping["search_terms"]:
search_link = f"/search?kw={term.replace(' ', '+')}&view=rich"
search_links.append(f"<a href='{search_link}'><strong>{term}</strong></a>")
links_html = ", ".join(search_links)
concept_notice = f"<div class='correction-notice'><strong>{original_query.capitalize()}</strong>: {concept_mapping['explanation']}<br>You may find what you are looking for with these search terms: {links_html}</div>"
# Use the first search term for the actual search
cleaned_keywords = concept_mapping["search_terms"][0]
def score_result_by_proximity(verse_text, query_words):
"""
Score a verse based on how close together the query words appear.
Higher score = words appear closer together.
"""
verse_lower = verse_text.lower()
# Find all positions of each query word
word_positions = {}
for word in query_words:
word_lower = word.lower()
positions = []
start = 0
while True:
pos = verse_lower.find(word_lower, start)
if pos == -1:
break
positions.append(pos)
start = pos + 1
word_positions[word] = positions
# If any word is missing, score is 0
if any(len(positions) == 0 for positions in word_positions.values()):
return 0
# Calculate minimum distance between all query words
# For each combination of positions, find the span (max_pos - min_pos)
# Smaller span = higher score
import itertools
# Get all combinations of positions for all words
all_position_combos = list(itertools.product(*word_positions.values()))
if not all_position_combos:
return 0
# Find the combination with minimum span
min_span = float('inf')
for combo in all_position_combos:
span = max(combo) - min(combo)
if span < min_span:
min_span = span
# Convert span to score (smaller span = higher score)
# Use inverse: 1000 / (span + 1) so adjacent words score ~1000, far apart score low
score = 1000.0 / (min_span + 1)
return score
results = lookup_fts(cleaned_keywords)
# Sort results by proximity score if we have multiple query words
if results and len(cleaned_keywords.split()) > 1:
query_words = cleaned_keywords.split()
scored_results = []
for result in results:
verse_text = result[4] # The text is at index 4
score = score_result_by_proximity(verse_text, query_words)
scored_results.append((score, result))
# Sort by score descending, then keep just the results
scored_results.sort(key=lambda x: x[0], reverse=True)
results = [result for score, result in scored_results]
if not results:
# PRIORITY 1: Try grammatical variant expansion first
words = cleaned_keywords.split()
def sort_variants_by_similarity(original, variants):
"""Sort variants by morphological similarity: forwards (longer/more specific) then backwards (shorter/simpler)"""
def get_common_prefix_length(w1, w2):
"""Find how many characters match from the start"""
prefix_len = 0
for i in range(min(len(w1), len(w2))):
if w1[i].lower() == w2[i].lower():
prefix_len += 1
else:
break
return prefix_len
scored_variants = []
for variant in variants:
prefix_len = get_common_prefix_length(original, variant)
# Reject variants that don't share the stem (beginning doesn't match)
if prefix_len < min(2, len(original) // 2):
continue
# Score: prioritize common prefix length (stem preservation) heavily,
# then prefer longer variants (forward/more specific) over shorter (backward/simpler)
# Prefix is weighted 1000x to ensure stem-preservation is primary
score = (prefix_len * 1000) + len(variant)
scored_variants.append((score, variant))
# Sort by score descending: highest score = most similar with longest form first
return [v for _, v in sorted(scored_variants, key=lambda x: x[0], reverse=True)]
for word in words:
variants = cached["grammatical_variants"].get(word.lower(), [])
if variants:
# Sort variants: forwards in likeness (longer/specific) then backwards (shorter/simple)
sorted_variants = sort_variants_by_similarity(word.lower(), variants)
# Try each variant in order of similarity
for variant in sorted_variants:
variant_query = cleaned_keywords.replace(word, variant, 1)
results = lookup_fts(variant_query)
if results:
response_list = scripture_response(results)
arg_view = request.args.get("view", None)
if arg_view == "plain":
response = f"Showing results for '{variant_query}' (grammatical variant of '{cleaned_keywords}')\n<br>" + "<br>".join(response_list)
elif arg_view == "rich":
variant_notice = f"<div class='variant-notice'>Showing results for '<strong>{variant_query}</strong>' (grammatical variant of '{cleaned_keywords}')</div>"
response_str = "".join(response_list)
response = response_str.replace("<div class='responses'>", f"<div class='responses'>{variant_notice}", 1)
else:
response = jsonify({"original": cleaned_keywords, "variant": variant_query, "results": response_list})
return response
# PRIORITY 2: If grammatical variants didn't help, try spell-check
h = hunspell.HunSpell("data/kjv.dic", "data/kjv.aff")
corrected_words = []
had_corrections = False
for word in words:
if h.spell(word):
# Word is spelled correctly
corrected_words.append(word)
else:
# Word is misspelled, get suggestions
word_suggestions = h.suggest(word)
if word_suggestions:
corrected_words.append(word_suggestions[0])
had_corrections = True
else:
# No suggestions, keep original
corrected_words.append(word)
if had_corrections:
corrected_keyword = " ".join(corrected_words)
results = lookup_fts(corrected_keyword)
if results:
response_list = scripture_response(results)
arg_view = request.args.get("view", None)
if arg_view == "plain":
response = f"Showing results for '{corrected_keyword}' (corrected from '{cleaned_keywords}')\n<br>" + "<br>".join(response_list)
elif arg_view == "rich":
correction_notice = f"<div class='correction-notice'>Showing results for '<strong>{corrected_keyword}</strong>' (corrected from '{cleaned_keywords}')</div>"
response_str = "".join(response_list)
response = response_str.replace("<div class='responses'>", f"<div class='responses'>{correction_notice}", 1)
else:
response = jsonify({"corrected_from": cleaned_keywords, "corrected_to": corrected_keyword, "results": response_list})
return response
# PRIORITY 3: Try theological synonym expansion
for word in words:
synonyms = cached["theological_synonyms"].get(word.lower(), None)
if synonyms:
for synonym in synonyms:
synonym_query = cleaned_keywords.replace(word, synonym, 1)
results = lookup_fts(synonym_query)
if results:
response_list = scripture_response(results)
arg_view = request.args.get("view", None)
explanation = f"In the KJV, '{word}' and '{synonym}' often refer to similar theological concepts. Consider also searching for related verses."
if arg_view == "plain":
response = f"Showing results for '{synonym_query}' (theological equivalent of '{cleaned_keywords}')\n{explanation}\n<br>" + "<br>".join(response_list)
elif arg_view == "rich":
theological_notice = f"<div class='theological-notice'>Showing results for '<strong>{synonym_query}</strong>' (theological equivalent of '{cleaned_keywords}')<br><em>{explanation}</em></div>"
response_str = "".join(response_list)
response = response_str.replace("<div class='responses'>", f"<div class='responses'>{theological_notice}", 1)
else:
response = jsonify({"original": cleaned_keywords, "theological_equivalent": synonym_query, "explanation": explanation, "results": response_list})
return response
# PRIORITY 4: If we still have no results, show spelling suggestions
suggestions = h.suggest(cleaned_keywords)
arg_view = request.args.get("view", "rich")
if arg_view == "rich":
# Build a nice no-results page with header
no_results_html = cached["html_navbar"].format(
query=cleaned_keywords,
current_chapter_name="",
next_chapter_name="",
next_chapter_link="",
prev_chapter_name="",
prev_chapter_link=""
)
no_results_html += cached["css_navbar"]
no_results_html += "<div class='responses'>"
no_results_html += f"<div class='correction-notice'><strong>No results found</strong> for '{cleaned_keywords}'.</div>"
if suggestions:
no_results_html += "<div class='theological-suggestions'><h3>Did you mean?</h3>"
no_results_html += "<p>We couldn't find any verses matching your search. Here are some spelling suggestions:</p>"
for suggestion in suggestions[:5]: # Limit to 5 suggestions
search_link = f"/search?kw={suggestion}&view=rich"
no_results_html += f"<div class='suggestion-item'><a href='{search_link}'><strong>{suggestion}</strong></a></div>"
no_results_html += "</div>"
else:
no_results_html += "<div class='theological-suggestions'><h3>Search Tips</h3>"
no_results_html += "<p>We couldn't find any verses matching your search. Try:</p>"
no_results_html += "<ul style='font-family: var(--font-ui); color: var(--color-text-secondary);'>"
no_results_html += "<li>Checking your spelling</li>"
no_results_html += "<li>Using different keywords</li>"
no_results_html += "<li>Searching for <a href='/search?kw=love&view=rich'>love</a>, <a href='/search?kw=faith&view=rich'>faith</a>, or <a href='/search?kw=hope&view=rich'>hope</a></li>"
no_results_html += "</ul></div>"
no_results_html += "</div>"
return no_results_html
else:
# Plain text fallback
return "{} not found, try one of the following: {}".format(
cleaned_keywords, ", ".join(suggestions) if suggestions else "No suggestions available"
)
response_list = scripture_response(results)
# Check if any words in the search have theological synonyms to suggest
theological_suggestions = []
words = cleaned_keywords.split()
for word in words:
synonyms = cached["theological_synonyms"].get(word.lower(), None)
if synonyms:
for synonym in synonyms:
theological_suggestions.append(synonym)
arg_view = request.args.get("view", "rich")
if arg_view == "plain":
response = "<br>".join(response_list)
if concept_notice:
response = concept_notice + "<br>" + response
if theological_suggestions:
response += "<br><br>Also consider searching: "
for suggestion in theological_suggestions:
response += f"{suggestion}<br>"
elif arg_view == "rich":
response = "".join(response_list)
# Insert concept notice at the top if it exists
if concept_notice:
# Find the first </div> after the opening <div class='responses'> and insert after it
responses_start = response.find("<div class='responses'>")
if responses_start != -1:
# Find the results-count div closing
insert_pos = response.find("</div>", responses_start + len("<div class='responses'>"))
if insert_pos != -1:
insert_pos += len("</div>")
response = response[:insert_pos] + concept_notice + response[insert_pos:]
if theological_suggestions:
# Add suggestions before the LAST closing </div> which closes the responses container
suggestions_html = "<div class='theological-suggestions'><h3>Related Theological Concepts</h3>"
suggestions_html += "<p>In the KJV, these terms often refer to similar theological concepts. Consider also searching for related verses.</p>"
for suggestion in theological_suggestions:
search_link = f"/search?kw={suggestion}&view=rich"
suggestions_html += f"<div class='suggestion-item'>"
suggestions_html += f"<a href='{search_link}'><strong>{suggestion.capitalize()}</strong></a>"
suggestions_html += "</div>"
suggestions_html += "</div>"
# Find the last </div> and insert before it
last_div_pos = response.rfind("</div>")
if last_div_pos != -1:
response = response[:last_div_pos] + suggestions_html + response[last_div_pos:]
else:
response = jsonify(response_list)
#logging.debug(f"Search Response: {response}")
return response
@app.get("/browse") # not implemented. TO be moved to frontend.
def bible_reader():
return render_template("reader.html", books=cached.get("book_names"))
# We should return an index of all books with
# drop down menus of each chapter, and a search
# bar at the top for quick seeking, which can use
# show(). Each page should return a header and footer.
# header should have a navigation bar and "next/previous chapter"
# Footer should have next/previous as well.
# Maybe preliminary theming support
@app.get("/plan") # not implmeneted. To be moved to frontend.
def reading_plan():
# needs 2 sets of data, books and time frame.
# should default to the entire Bible in 1 year.
# should start today() and end today()+1year
# Should return reading_plan template.
return render_template("reading_plan.html")
@app.get("/status") # basic instance info
def status():
import resource
return f"""Sanity: {sanity} <br>
UserMode: {usermode} <br>
Memory: {resource.getrusage(resource.RUSAGE_SELF).ru_maxrss}
"""
@app.get("/seqreport") # shows current tracked scripture sequences
def sequential_report(uid=None):
if usermode == "Single":
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)
@app.get("/screport") # shows current remembered shortcodes
def shortcode_report(shortcode=None):
if usermode == "Single":
short_conn = sqlite3.connect("data/shortcodes.db", check_same_thread=False)
else:
short_conn = sqlite3.connect("data/shortcodes.db")
short_cur = short_conn.cursor()
shortcodes = short_cur.execute(
"SELECT * FROM shortcodes"
).fetchall()
return render_template("shortcode_report.html", shortcodes=shortcodes)
def sequential_purge():
if usermode == "Single":
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").fetchmany(10)
evaluate_datetime = datetime.now()
for uid in uids:
last_used = datetime.fromisoformat(uid[2])
since_use: timedelta = evaluate_datetime - last_used
if since_use > sequential_purge_delta:
logging.info(f"Purging UID: {uid[0]}")
seq_cur.execute("DELETE FROM reading_tracker WHERE uid = ?", (uid[0],))
seq_conn.commit()
seq_cur.close()
@app.get("/seq") # allows you to get groups of scripture sequentially
def sequential_read(start=0, num=1, uid=None, getuid=False):
# /seq?uid=1
uid = request.args.get("uid", uid)
getuid = request.args.get("getuid", getuid)
start = request.args.get("start", start)
try:
num = int(request.args.get("num", num))
except ValueError:
return "Invalid num argument"
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)
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)"
)
last_used = str(datetime.now())
num -= 1 # 0 will mean "next"
if num < 0:
return jsonify({"error": "You may only increment forward."})
if getuid:
if start:
start = cached["verse_ids"].index(
int(show(kw=start, get_start_verse_id=True))
)
uid = generate_short_id() # range returns a 5 digit id
try:
seq_cur.execute(
"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}"
)
return jsonify(
{
"error": "Something went wrong with the UID database. Please notify tyler@dinsmoor.us"
}
)
seq_conn.commit()
logging.info(f"New SEQ UID: {uid}")
return f"Your new UID is: {uid}"
uid = ref_input_cleaning(uid, ref_search=False)
try:
nextverse = seq_cur.execute(
"SELECT verse FROM reading_tracker WHERE uid = ?", (uid,)
).fetchone()[0]
logging.debug(f"Queried UID {uid} on verse {nextverse}")
sequential_purge()
except TypeError:
logging.info(f"Invalid UID {uid}")
return jsonify({"error": "UID Invalid or purged (90 days of inactivity)"})
if len(cached["verse_ids"]) < (nextverse + num):
# Correct our range to go to ceiling to prevent IndexError
diff = len(cached["verse_ids"]) - (nextverse + num)
num = diff - num
start_verse = cached["verse_ids"][nextverse]
end_verse = cached["verse_ids"][nextverse + num]
result_list = lookup_by_verse_id(start_verse, end_verse)
response_list = scripture_response(result_list)
logging.info(f"Served {num+1} verses to UID {uid}")
seq_cur.execute(
"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.
seq_conn.commit()
arg_view = request.args.get("view", None)
if arg_view == "plain":
response = "<br>".join(response_list)
elif arg_view == "rich":
response = "".join(response_list)
else:
response = jsonify(response_list)
return response
@app.get("/mksc/<query>") # create and remember a new shortcode
def generate_shortcode(query: str = ""):
query = search_input_cleaning(query, extra_chars=":-,")
if not query:
return "Empty query."
if usermode == "Single":
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_conn.commit()
creation_date = str(datetime.now())
shortcode = generate_short_id()
short_cur.execute(
"REPLACE INTO shortcodes(shortcode,query,creation_date) VALUES(?, ?, ?)", (shortcode, query, creation_date)
)
short_conn.commit()
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"
)
if not shortcode:
return "Empty shortcode."
if usermode == "Single":
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(
"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("Operating in single user mode.")
app.run(host="0.0.0.0", port=1612, debug=True, threaded=True)