1915 lines
77 KiB
Python
1915 lines
77 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
|
||
import hashlib
|
||
import re
|
||
import difflib
|
||
import math
|
||
import urllib.parse
|
||
import html as html_lib
|
||
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
|
||
# ---------------------------------------------------------------------------
|
||
# Verse menu helpers (favorites, definitions and cross references are saved by
|
||
# the browser itself; the server only supplies the lookups)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
DICT_STOPWORDS = {
|
||
"the", "and", "that", "for", "him", "his", "her", "but", "not", "with", "are",
|
||
"was", "were", "unto", "thou", "thee", "thy", "ye", "you", "all", "from", "they",
|
||
"them", "this", "have", "had", "shall", "will", "which", "into", "upon", "then",
|
||
}
|
||
|
||
|
||
def format_verse_ref(verse_id) -> str:
|
||
"""12345678 style id -> 'John 3:16'"""
|
||
verse_id = int(verse_id)
|
||
return f"{lookup_bookname(verse_id // 1_000_000)} {verse_id // 1000 % 1000}:{verse_id % 1000}"
|
||
|
||
|
||
@app.get("/xref/<int:verse_id>") # cross references for one verse, best ranked first
|
||
def cross_references(verse_id: int):
|
||
rows = kjv_cur.execute(
|
||
"SELECT sv, ev FROM cross_reference WHERE vid = ? ORDER BY r DESC LIMIT 15",
|
||
(verse_id,),
|
||
).fetchall()
|
||
results = []
|
||
for start_id, end_id in rows:
|
||
ref = format_verse_ref(start_id)
|
||
if end_id and end_id > start_id:
|
||
end_ref = format_verse_ref(end_id)
|
||
# "John 3:16-18" when the range stays in one chapter
|
||
same_chapter = start_id // 1000 == end_id // 1000
|
||
ref += "-" + (str(end_id % 1000) if same_chapter else end_ref)
|
||
verses = lookup_by_verse_id(start_id, end_id)
|
||
else:
|
||
verses = lookup_by_verse_id(start_id)
|
||
results.append({
|
||
"ref": ref,
|
||
"verse_id": start_id,
|
||
"text": " ".join(v[4] for v in verses),
|
||
})
|
||
return jsonify(results)
|
||
|
||
|
||
@app.get("/defs") # 1828 Webster definitions for several words at once: /defs?words=grace,loved
|
||
def definitions():
|
||
words = request.args.get("words", "").lower().replace(",", " ").split()[:40]
|
||
dictionary = cached["json_dictionary"]
|
||
found = {}
|
||
for word in words:
|
||
word = "".join(c for c in word if c.isalpha())
|
||
if word and word not in DICT_STOPWORDS and word in dictionary:
|
||
found[word] = dictionary[word]
|
||
return jsonify(found)
|
||
|
||
|
||
@app.get("/saved") # favorites, definitions and cross references kept in this browser
|
||
def saved_page():
|
||
return static_page("html/saved.html")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Bible reading plan: a /seq style tracker that can go backwards. The plan is
|
||
# its UID; the page URL (/plan/<uid>?n=<verses per day>) is what gets bookmarked.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
PLAN_UID_CHARS = "0123456789ACEFHJKLMNPRTUVWXY"
|
||
PLAN_MAX_PER_DAY = 500
|
||
|
||
|
||
def plan_connection():
|
||
if usermode == "Single":
|
||
conn = sqlite3.connect("data/seq.db", check_same_thread=False)
|
||
else:
|
||
conn = sqlite3.connect("data/seq.db")
|
||
conn.execute(
|
||
"CREATE TABLE IF NOT EXISTS reading_tracker (uid TEXT PRIMARY KEY, verse INTEGER, last_used TEXT)"
|
||
)
|
||
return conn
|
||
|
||
|
||
def plan_per_day(raw, default=25) -> int:
|
||
try:
|
||
return max(1, min(PLAN_MAX_PER_DAY, int(raw)))
|
||
except (TypeError, ValueError):
|
||
return default
|
||
|
||
|
||
def plan_start_index(ref: str):
|
||
"""Index into verse_ids for a reference like 'Matthew 1:1'; None if it can't be parsed."""
|
||
ref = (ref or "").strip()
|
||
if not ref:
|
||
return 0
|
||
try:
|
||
verse_id = int(show(kw=ref, get_start_verse_id=True))
|
||
return cached["verse_ids"].index(verse_id)
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def plan_error(message: str):
|
||
return static_html_doc(
|
||
"Reading Plan",
|
||
f"<h2>Reading Plan</h2><p>{html_lib.escape(message)}</p><p><a href='/plan'>Back to plan setup</a></p>",
|
||
), 400
|
||
|
||
|
||
@app.get("/plan") # setup form for a new reading plan
|
||
def reading_plan():
|
||
return static_page("html/plan.html")
|
||
|
||
|
||
@app.post("/plan/new")
|
||
def plan_create():
|
||
total = len(cached["verse_ids"])
|
||
start = plan_start_index(request.form.get("start", ""))
|
||
if start is None:
|
||
return plan_error("We couldn't find that starting point. Try something like 'Matthew 1:1'.")
|
||
if request.form.get("pace") == "verses":
|
||
per_day = plan_per_day(request.form.get("verses"))
|
||
else:
|
||
try:
|
||
days = max(1, min(3650, int(request.form.get("days", 365))))
|
||
except ValueError:
|
||
days = 365
|
||
per_day = plan_per_day(math.ceil((total - start) / days))
|
||
conn = plan_connection()
|
||
for _ in range(5): # a clash on a random 5 character id is very unlikely
|
||
uid = generate_short_id()
|
||
try:
|
||
conn.execute(
|
||
"INSERT INTO reading_tracker VALUES(?, ?, ?)", (uid, start, str(datetime.now()))
|
||
)
|
||
break
|
||
except sqlite3.IntegrityError:
|
||
continue
|
||
else:
|
||
return plan_error("Could not create a plan, please try again.")
|
||
conn.commit()
|
||
conn.close()
|
||
logging.info(f"New reading plan {uid} start={start} per_day={per_day}")
|
||
return redirect(f"/plan/{uid}?n={per_day}", code=303)
|
||
|
||
|
||
def plan_uid_or_none(raw: str):
|
||
uid = search_input_cleaning(raw, valid_chars=PLAN_UID_CHARS)
|
||
return uid if uid and uid == raw else None
|
||
|
||
|
||
@app.post("/plan/<uid>/move") # back / next / jump; POST so a page prefetch can never move your place
|
||
def plan_move(uid: str):
|
||
uid = plan_uid_or_none(uid)
|
||
per_day = plan_per_day(request.form.get("n"))
|
||
if not uid:
|
||
return plan_error("That reading plan id is not valid.")
|
||
conn = plan_connection()
|
||
row = conn.execute("SELECT verse FROM reading_tracker WHERE uid = ?", (uid,)).fetchone()
|
||
if row is None:
|
||
conn.close()
|
||
return plan_error("That reading plan was not found (plans unused for 90 days are removed).")
|
||
total = len(cached["verse_ids"])
|
||
position = row[0]
|
||
action = request.form.get("action")
|
||
if action == "next":
|
||
position = min(total, position + per_day)
|
||
elif action == "back":
|
||
position = max(0, position - per_day)
|
||
elif action == "restart":
|
||
position = 0
|
||
elif action == "goto":
|
||
target = plan_start_index(request.form.get("kw", ""))
|
||
if target is None:
|
||
conn.close()
|
||
return plan_error("We couldn't find that reference. Try something like 'John 3:1'.")
|
||
position = target
|
||
conn.execute(
|
||
"REPLACE INTO reading_tracker (uid, verse, last_used) VALUES(?, ?, ?)",
|
||
(uid, position, str(datetime.now())),
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
return redirect(f"/plan/{uid}?n={per_day}", code=303)
|
||
|
||
|
||
@app.get("/plan/<uid>") # today's reading for a plan. Viewing never moves your place.
|
||
def plan_view(uid: str):
|
||
uid = plan_uid_or_none(uid)
|
||
if not uid:
|
||
return plan_error("That reading plan id is not valid.")
|
||
per_day = plan_per_day(request.args.get("n"))
|
||
conn = plan_connection()
|
||
row = conn.execute("SELECT verse FROM reading_tracker WHERE uid = ?", (uid,)).fetchone()
|
||
if row is None:
|
||
conn.close()
|
||
return plan_error("That reading plan was not found (plans unused for 90 days are removed).")
|
||
position = row[0]
|
||
# keeps an active plan from being purged as unused
|
||
conn.execute(
|
||
"REPLACE INTO reading_tracker (uid, verse, last_used) VALUES(?, ?, ?)",
|
||
(uid, position, str(datetime.now())),
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
verse_ids = cached["verse_ids"]
|
||
total = len(verse_ids)
|
||
finished = position >= total
|
||
last_index = min(total, position + per_day) - 1
|
||
percent = round(100 * min(position, total) / total, 1)
|
||
esc = html_lib.escape
|
||
|
||
controls = f"""<form method='post' action='/plan/{uid}/move' class='plan-form'>
|
||
<input type='hidden' name='n' value='{per_day}'>"""
|
||
back_disabled = "disabled" if position <= 0 else ""
|
||
next_html = (
|
||
"" if finished else
|
||
"<button type='submit' name='action' value='next' class='plan-primary'>Done — next reading →</button>"
|
||
)
|
||
controls += f"""{next_html}
|
||
<button type='submit' name='action' value='back' class='plan-secondary' {back_disabled}>← Previous reading</button>
|
||
</form>
|
||
<form method='post' action='/plan/{uid}/move' class='plan-form plan-goto'>
|
||
<input type='hidden' name='n' value='{per_day}'>
|
||
<input type='hidden' name='action' value='goto'>
|
||
<label for='goto-kw'>Jump to</label>
|
||
<div class='plan-goto-row'><input type='text' id='goto-kw' name='kw' data-book-suggest placeholder='e.g. Psalms 23:1' autocapitalize='off' autocomplete='off'>
|
||
<button type='submit' class='plan-secondary'>Go</button></div>
|
||
</form>"""
|
||
|
||
if finished:
|
||
summary = (
|
||
"<div class='plan-summary'><strong>You've finished the plan!</strong>"
|
||
f"<div>You read all {total:,} verses. Use Previous reading to look back, or start over.</div></div>"
|
||
)
|
||
verses_html = ""
|
||
controls += f"""<form method='post' action='/plan/{uid}/move' class='plan-form'>
|
||
<input type='hidden' name='n' value='{per_day}'>
|
||
<button type='submit' name='action' value='restart' class='plan-secondary'>Start over from Genesis 1:1</button></form>"""
|
||
else:
|
||
first_ref = format_verse_ref(verse_ids[position])
|
||
last_ref = format_verse_ref(verse_ids[last_index])
|
||
count = last_index - position + 1
|
||
summary = (
|
||
"<div class='plan-summary'><strong>Today's reading</strong>"
|
||
f"<div class='plan-range'>{esc(first_ref)} – {esc(last_ref)}</div>"
|
||
f"<div class='plan-meta'>{count} verses · {percent}% of the Bible complete</div>"
|
||
f"<div class='plan-progress' role='img' aria-label='{percent}% complete'><span style='width:{percent}%'></span></div></div>"
|
||
)
|
||
verses = lookup_by_verse_id(verse_ids[position], verse_ids[last_index])
|
||
verses_html = "".join(response_rich(v) for v in verses)
|
||
|
||
bookmark_note = (
|
||
"<div class='plan-note'>Bookmark this page — it is your reading plan. "
|
||
"Come back each day, then press <em>Done</em> to move on.</div>"
|
||
)
|
||
store_script = (
|
||
"<script>try{localStorage.setItem('kjv.plan',JSON.stringify("
|
||
f"{{uid:'{uid}',n:{per_day},t:Date.now()}}))}}catch(e){{}}</script>"
|
||
)
|
||
return (
|
||
render_navbar("", bar_html="")
|
||
+ "<div class='responses'>"
|
||
+ summary
|
||
+ verses_html
|
||
+ f"<div class='plan-controls'>{controls}</div>"
|
||
+ bookmark_note
|
||
+ store_script
|
||
+ "</div>"
|
||
)
|
||
|
||
|
||
|
||
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}
|
||
|
||
# Static assets: each file's URL carries a hash of its own contents ("?v=...").
|
||
# A URL whose hash matches is served as immutable for a year; the hash changes
|
||
# whenever the file does, so a deploy is picked up immediately and only the
|
||
# files that actually changed get downloaded again. Any other URL for the same
|
||
# file (no hash, or an old one) is served with "no-cache", so it is revalidated
|
||
# every time instead of lingering in a browser for a year.
|
||
ASSET_SOURCES = {
|
||
"/styles/page.css": ("styles/page.css", "text/css"),
|
||
"/js/collapsible.js": ("js/collapsible.js", "application/javascript"),
|
||
"/js/shortcode.js": ("js/shortcode.js", "application/javascript"),
|
||
"/js/verse_menu.js": ("js/verse_menu.js", "application/javascript"),
|
||
"/js/book_suggest.js": ("js/book_suggest.js", "application/javascript"),
|
||
}
|
||
ASSET_VERSIONS: dict = {
|
||
url: hashlib.md5(open(path, "rb").read()).hexdigest()[:8]
|
||
for url, (path, _mime) in ASSET_SOURCES.items()
|
||
}
|
||
|
||
|
||
def asset_url(url: str) -> str:
|
||
return f"{url}?v={ASSET_VERSIONS[url]}"
|
||
|
||
|
||
app.jinja_env.globals["asset_url"] = asset_url
|
||
|
||
|
||
def load_html(path: str) -> str:
|
||
"""Read an html file, filling each '/js/x.js?v=__V__' with that file's own version."""
|
||
return re.sub(
|
||
r"(/(?:js|styles)/[\w.\-]+)\?v=__V__",
|
||
lambda m: asset_url(m.group(1)),
|
||
open(path, "r").read(),
|
||
)
|
||
|
||
|
||
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": load_html("html/navbar.html"),
|
||
"book_descriptions": json.load(open("data/book_descriptions.json", "r")),
|
||
"chapter_summaries": json.load(open("data/chapter_summaries.json", "r")),
|
||
"html_chapter_nav": open("html/chapter_nav.html", "r").read(),
|
||
"pages": {},
|
||
"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(),
|
||
"js_verse_menu": open("js/verse_menu.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(),
|
||
}
|
||
|
||
# Book name suggestions: the list of names is baked into the script once
|
||
cached["js_book_suggest"] = open("js/book_suggest.js", "r").read().replace(
|
||
"__BOOKS__", json.dumps(cached["book_names"])
|
||
)
|
||
|
||
# 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,
|
||
"about": cached["book_descriptions"].get(title, ""),
|
||
"summaries": cached["chapter_summaries"].get(title, []),
|
||
})
|
||
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 = load_html("html/index.html")
|
||
cached["html_index"] = render_template_string(template_content, testament_data=cached["book_info_for_index"])
|
||
|
||
def _make_asset_view(url_path: str, mimetype: str):
|
||
def view():
|
||
response = Response(cached["assets"][url_path], mimetype=mimetype)
|
||
response.add_etag()
|
||
if request.args.get("v") == ASSET_VERSIONS[url_path]:
|
||
response.headers["Cache-Control"] = "public, max-age=31536000, immutable"
|
||
else:
|
||
response.headers["Cache-Control"] = "no-cache"
|
||
return response.make_conditional(request)
|
||
view.__name__ = "asset_" + url_path.strip("/").replace("/", "_").replace(".", "_")
|
||
return view
|
||
|
||
|
||
cached["assets"] = {
|
||
"/styles/page.css": cached["css_page"],
|
||
"/js/collapsible.js": cached["js_collapsible"],
|
||
"/js/shortcode.js": cached["js_shortcode"],
|
||
"/js/verse_menu.js": cached["js_verse_menu"],
|
||
"/js/book_suggest.js": cached["js_book_suggest"],
|
||
}
|
||
for _url, (_path, _mime) in ASSET_SOURCES.items():
|
||
app.add_url_rule(_url, view_func=_make_asset_view(_url, _mime), methods=["GET"])
|
||
|
||
|
||
@app.after_request
|
||
def html_pages_are_never_stale(response):
|
||
"""Pages reference versioned assets, so they must always be revalidated to see new versions."""
|
||
if response.mimetype == "text/html" and "Cache-Control" not in response.headers:
|
||
response.headers["Cache-Control"] = "no-cache"
|
||
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=86400'
|
||
return response
|
||
|
||
def static_page(path: str):
|
||
"""Serve one of the simple html/ pages with its asset URLs versioned."""
|
||
if path not in cached["pages"]:
|
||
cached["pages"][path] = load_html(path)
|
||
return Response(cached["pages"][path], mimetype="text/html")
|
||
|
||
|
||
def static_html_doc(title: str, body: str):
|
||
"""Wrap a small html snippet in the standard page shell."""
|
||
return (
|
||
"<!DOCTYPE html><html lang='en'><head><meta charset='UTF-8'>"
|
||
"<meta name='viewport' content='width=device-width, initial-scale=1.0'>"
|
||
f"<title>{title} - KJV API</title>"
|
||
f"<link rel='stylesheet' href='{asset_url('/styles/page.css')}'></head><body>"
|
||
"<header><a href='/'>Home</a> <a href='/find'>Quick Search</a></header>"
|
||
f"{body}</body></html>"
|
||
)
|
||
|
||
|
||
def chapter_find_link(name: str) -> str:
|
||
"""Same link the verse headers use: the whole chapter, in the rich view."""
|
||
return "/find?kw=" + urllib.parse.quote(name) + "&view=rich"
|
||
|
||
|
||
def render_navbar(query: str = "", chapter_nav: dict = None, bar_html: str = "") -> str:
|
||
"""Opens the rich html document: <head>, site header and the chapter bar.
|
||
|
||
The caller adds the body content after it. bar_html replaces the chapter
|
||
bar for pages (like the reading plan) that bring their own.
|
||
"""
|
||
esc = lambda value: html_lib.escape(str(value), quote=True)
|
||
query = query or ""
|
||
title = esc(query.strip() or (chapter_nav or {}).get("current_chapter_name") or "Scripture")
|
||
chapter_html = bar_html
|
||
if chapter_nav:
|
||
values = {key: esc(value) for key, value in chapter_nav.items()}
|
||
values["current_chapter_link"] = esc(chapter_find_link(chapter_nav["current_chapter_name"]))
|
||
chapter_html = cached["html_chapter_nav"].format(**values)
|
||
return cached["html_navbar"].format(
|
||
title=title, css=cached["css_navbar"], query=esc(query), chapter_nav=chapter_html
|
||
)
|
||
|
||
|
||
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 static_page("html/find.html")
|
||
if query.endswith(","):
|
||
query = query[:-1]
|
||
|
||
# A query that reads as scripture references ("Exodus", "1 jn 3.16") is shown
|
||
# as scripture; anything else is a keyword search.
|
||
resolved = resolve_references(query)
|
||
if resolved:
|
||
result = show(kw=query, human_readable=human_readable)
|
||
if isinstance(result, str) and result != "Invalid Query":
|
||
notices = ""
|
||
for original, name in resolved["notes"]:
|
||
notices += (
|
||
f"<div class='correction-notice'>Showing <strong>{html_lib.escape(name)}</strong> "
|
||
f"(interpreted from '{html_lib.escape(original)}')</div>"
|
||
)
|
||
if resolved["bare"]:
|
||
keyword = urllib.parse.quote_plus(search_input_cleaning(query))
|
||
notices += (
|
||
f"<div class='correction-notice'>Showing <strong>{html_lib.escape(resolved['label'])}</strong>, "
|
||
f"chapter 1. Looking for the word instead? "
|
||
f"<a href='/search?kw={keyword}&view=rich'>Search the text for '{html_lib.escape(query)}'</a></div>"
|
||
)
|
||
if notices:
|
||
result = result.replace("<div class='responses'>", "<div class='responses'>" + notices, 1)
|
||
return result
|
||
if not search_input_cleaning(query).strip():
|
||
# nothing searchable left (punctuation, accents): show the search form
|
||
return static_page("html/find.html")
|
||
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 += f"<script src='{asset_url('/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 static_page("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 = ""
|
||
head = ""
|
||
else:
|
||
# standalone page: needs its own document head so phones scale it properly
|
||
css = ""
|
||
head = (
|
||
"<!DOCTYPE html><html lang='en'><head><meta charset='UTF-8'>"
|
||
"<meta name='viewport' content='width=device-width, initial-scale=1'>"
|
||
"<meta name='color-scheme' content='dark light'>"
|
||
f"<title>{html_lib.escape(ref.capitalize())} - KJV API</title>"
|
||
+ cached["css_navbar"]
|
||
+ "</head><body><div class='responses'>"
|
||
)
|
||
response = ""
|
||
for e in defs:
|
||
response += e + "<br><br>"
|
||
return (
|
||
head
|
||
+ cached["html_response"].format(
|
||
chapter=ref.split(":")[0], passage=ref.capitalize(), text=response, verse_id=""
|
||
)
|
||
+ css
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Scripture reference resolver. Understands the many ways people type a
|
||
# reference ("1 jn 3.16", "II Kings 2:1-4", "song of songs 2", "Exodus ") and
|
||
# turns it into ranges of verse ids (book*1000000 + chapter*1000 + verse).
|
||
# ---------------------------------------------------------------------------
|
||
|
||
# Extra spellings people use that the abbreviations table does not know.
|
||
EXTRA_BOOK_ALIASES = {
|
||
"Song of Solomon": ["song of songs", "songs of solomon", "songs of songs", "song of sol",
|
||
"song of sol", "canticles", "canticle of canticles", "sos", "sng", "ss"],
|
||
"Psalms": ["psalm", "psalter", "pslm", "psa", "pss", "psm", "ps"],
|
||
"Revelation": ["revelations", "revelation of john", "revelation of st john", "the revelation", "apocalypse"],
|
||
"Ecclesiastes": ["ecclesiasts", "eccles", "ecc", "qoheleth"],
|
||
"Philemon": ["phlm", "phm", "philem"],
|
||
"Judges": ["jdg", "jdgs", "jg"],
|
||
"Mark": ["mrk", "mk", "mr"],
|
||
"John": ["jhn", "jn"],
|
||
"Acts": ["acts of the apostles", "act"],
|
||
"James": ["jas", "jm"],
|
||
}
|
||
|
||
# Words that are books but also ordinary search words: a bare "numbers" or
|
||
# "mark" still shows the book, but /find offers a keyword search as well.
|
||
BARE_BOOK_EXTRA = {
|
||
"psalm", "songofsongs", "songsofsolomon", "songofsol", "canticles", "canticleofcanticles",
|
||
"revelations", "revelationofjohn", "therevelation",
|
||
}
|
||
|
||
ROMAN_PREFIX = r"(?:iii|ii|i)(?=[\s.])"
|
||
ORDINAL_PREFIX = r"(?:1st|2nd|3rd|first|second|third)"
|
||
BOOK_PIECE_RE = re.compile(
|
||
r"^\s*(?P<book>(?:(?:[1-3]|" + ROMAN_PREFIX + r"|" + ORDINAL_PREFIX + r")\.?\s*)?[a-z][a-z.\s]*?)"
|
||
r"\s*(?P<rest>\d[\d\s:.\-–—]*)?\s*$",
|
||
re.I,
|
||
)
|
||
CONTINUATION_RE = re.compile(r"^\s*\d[\d\s:.\-–—]*$")
|
||
|
||
|
||
def normalize_book_text(text: str) -> str:
|
||
"""'I Cor.' / '1st Corinthians' / '1 corinthians' -> '1corinthians'"""
|
||
text = text.lower().strip()
|
||
text = re.sub(r"^(the\s+)?(book\s+of\s+)", "", text)
|
||
text = re.sub(r"^(1st|first)(?=[\s.a-z]|$)", "1", text)
|
||
text = re.sub(r"^(2nd|second)(?=[\s.a-z]|$)", "2", text)
|
||
text = re.sub(r"^(3rd|third)(?=[\s.a-z]|$)", "3", text)
|
||
text = re.sub(r"^iii(?=[\s.]|$)", "3", text)
|
||
text = re.sub(r"^ii(?=[\s.]|$)", "2", text)
|
||
text = re.sub(r"^i(?=[\s.])", "1", text)
|
||
return re.sub(r"[^a-z0-9]", "", text)
|
||
|
||
|
||
def build_book_aliases():
|
||
"""alias key -> book id, plus the canonical keys on their own"""
|
||
canonical = {}
|
||
aliases = {}
|
||
for book_id, name in cached["book_id_to_name"].items():
|
||
canonical[normalize_book_text(name)] = book_id
|
||
aliases.update(canonical)
|
||
for abbreviation, book_id in kjv_cur.execute("SELECT a, b FROM key_abbreviations_english").fetchall():
|
||
aliases.setdefault(normalize_book_text(abbreviation), book_id)
|
||
for name, extras in EXTRA_BOOK_ALIASES.items():
|
||
book_id = lookup_book_id(name)
|
||
for extra in extras:
|
||
aliases[normalize_book_text(extra)] = book_id
|
||
return canonical, aliases
|
||
|
||
|
||
BOOK_CANONICAL, BOOK_ALIASES = build_book_aliases()
|
||
BOOK_ALIAS_KEYS = sorted(BOOK_ALIASES)
|
||
MAX_CHAPTER_ID_SPAN = 999
|
||
|
||
|
||
def resolve_book(token: str, loose: bool = True):
|
||
"""Returns (book_id, how) with how in exact|prefix|fuzzy, or None.
|
||
|
||
loose=False only accepts full names and spellings that are unmistakably
|
||
a book (used for a bare word with no chapter number).
|
||
"""
|
||
key = normalize_book_text(token)
|
||
if not key:
|
||
return None
|
||
if not loose:
|
||
if key in BOOK_CANONICAL or key in BARE_BOOK_EXTRA:
|
||
return (BOOK_ALIASES[key], "exact")
|
||
# a clear misspelling of a long full name ("genisis")
|
||
if len(key) >= 6 and not key[0].isdigit():
|
||
close = difflib.get_close_matches(key, list(BOOK_CANONICAL), n=1, cutoff=0.85)
|
||
if close:
|
||
return (BOOK_CANONICAL[close[0]], "fuzzy")
|
||
return None
|
||
if key in BOOK_ALIASES:
|
||
return (BOOK_ALIASES[key], "exact")
|
||
# unique (or best) prefix: "deut", "1cor", "matth"
|
||
prefixed = [k for k in BOOK_ALIAS_KEYS if k.startswith(key)]
|
||
if prefixed:
|
||
books = {BOOK_ALIASES[k] for k in prefixed}
|
||
canonical_hits = {BOOK_CANONICAL[k] for k in BOOK_CANONICAL if k.startswith(key)}
|
||
if len(canonical_hits) == 1:
|
||
return (next(iter(canonical_hits)), "prefix")
|
||
if len(books) == 1:
|
||
return (next(iter(books)), "prefix")
|
||
return (max(canonical_hits or books), "prefix") # ambiguous: the later book, as before
|
||
if len(key) >= 4:
|
||
close = difflib.get_close_matches(key, BOOK_ALIAS_KEYS, n=1, cutoff=0.78)
|
||
if close:
|
||
return (BOOK_ALIASES[close[0]], "fuzzy")
|
||
return None
|
||
|
||
|
||
def chapter_bounds(book_id: int, chapter: int) -> int:
|
||
"""Clamp a chapter number into the book (0 -> 1, past the end -> last chapter)."""
|
||
return max(1, min(chapter, cached["books_chapter_lengths"][book_id]))
|
||
|
||
|
||
def parse_ref_rest(rest: str, book_id: int, previous=None):
|
||
"""'3:16-18', '3 16', '3.16', '3-4', '3:16-4:2', '3', '3:-' -> (start_id, end_id|None, chapter, verse)"""
|
||
rest = rest.replace("–", "-").replace("—", "-")
|
||
rest = re.sub(r"\s*([:\-])\s*", r"\1", rest.strip())
|
||
base = book_id * 1_000_000
|
||
m = re.fullmatch(r"(\d+)", rest)
|
||
if m:
|
||
n = int(m.group(1))
|
||
if previous and previous["verse"]: # "John 3:16, 18": a verse of the same chapter
|
||
chapter = previous["chapter"]
|
||
return (base + chapter * 1000 + n, None, chapter, n)
|
||
chapter = chapter_bounds(book_id, n)
|
||
return (base + chapter * 1000 + 1, base + chapter * 1000 + MAX_CHAPTER_ID_SPAN, chapter, 0)
|
||
m = re.fullmatch(r"(\d+)(?::-|-)", rest) # "3:-" or "3-": the whole chapter
|
||
if m:
|
||
chapter = chapter_bounds(book_id, int(m.group(1)))
|
||
return (base + chapter * 1000 + 1, base + chapter * 1000 + MAX_CHAPTER_ID_SPAN, chapter, 0)
|
||
m = re.fullmatch(r"(\d+)-(\d+)", rest) # chapters 3 to 4
|
||
if m:
|
||
first, last = chapter_bounds(book_id, int(m.group(1))), chapter_bounds(book_id, int(m.group(2)))
|
||
if last < first:
|
||
first, last = last, first
|
||
return (base + first * 1000 + 1, base + last * 1000 + MAX_CHAPTER_ID_SPAN, first, 0)
|
||
m = re.fullmatch(r"(\d+)[:.\s](\d+)", rest)
|
||
if m:
|
||
chapter = chapter_bounds(book_id, int(m.group(1)))
|
||
verse = max(1, int(m.group(2)))
|
||
return (base + chapter * 1000 + verse, None, chapter, verse)
|
||
m = re.fullmatch(r"(\d+)[:.\s](\d+)-", rest) # "3:16-": from verse 16 to the end
|
||
if m:
|
||
chapter = chapter_bounds(book_id, int(m.group(1)))
|
||
return (base + chapter * 1000 + max(1, int(m.group(2))), base + chapter * 1000 + MAX_CHAPTER_ID_SPAN, chapter, int(m.group(2)))
|
||
m = re.fullmatch(r"(\d+)[:.\s](\d+)-(\d+)", rest)
|
||
if m:
|
||
chapter = chapter_bounds(book_id, int(m.group(1)))
|
||
first, last = sorted((max(1, int(m.group(2))), max(1, int(m.group(3)))))
|
||
return (base + chapter * 1000 + first, base + chapter * 1000 + last, chapter, first)
|
||
m = re.fullmatch(r"(\d+)[:.\s](\d+)-(\d+)[:.](\d+)", rest) # 3:16-4:2 across chapters
|
||
if m:
|
||
c1, c2 = chapter_bounds(book_id, int(m.group(1))), chapter_bounds(book_id, int(m.group(3)))
|
||
start = base + c1 * 1000 + max(1, int(m.group(2)))
|
||
end = base + c2 * 1000 + max(1, int(m.group(4)))
|
||
if end < start:
|
||
return None
|
||
return (start, end, c1, int(m.group(2)))
|
||
return None
|
||
|
||
|
||
def resolve_references(text: str, loose_bare: bool = False):
|
||
"""Parse a whole query ('John 3:16, Romans 5:8') into verse id ranges.
|
||
|
||
Returns None when it does not read as scripture references, so the
|
||
caller can treat it as a keyword search. Otherwise a dict with:
|
||
refs - list of (start_id, end_id or None)
|
||
notes - short explanations of guesses that were made
|
||
bare - the query was just a book name (shows its first chapter)
|
||
label - the name of the first book, for messages
|
||
"""
|
||
if not text or not text.strip():
|
||
return None
|
||
pieces = [p for p in re.split(r"[,;\n]+", text) if p.strip()]
|
||
if not pieces or len(pieces) > 20:
|
||
return None
|
||
refs, notes = [], []
|
||
bare = False
|
||
previous = None
|
||
label = None
|
||
for piece in pieces:
|
||
m = BOOK_PIECE_RE.match(piece)
|
||
if m:
|
||
token, rest = m.group("book"), m.group("rest")
|
||
if rest:
|
||
found = resolve_book(token)
|
||
else:
|
||
found = resolve_book(token, loose=loose_bare)
|
||
if not found:
|
||
return None
|
||
book_id, how = found
|
||
name = lookup_bookname(book_id)
|
||
if how == "fuzzy":
|
||
notes.append((token.strip(), name))
|
||
if not rest:
|
||
if len(pieces) == 1:
|
||
bare = True
|
||
start, end, chapter, verse = (book_id * 1_000_000 + 1001, book_id * 1_000_000 + 1999, 1, 0)
|
||
else:
|
||
parsed = parse_ref_rest(rest, book_id, None)
|
||
if not parsed:
|
||
return None
|
||
start, end, chapter, verse = parsed
|
||
elif CONTINUATION_RE.match(piece) and previous:
|
||
book_id = previous["book"]
|
||
name = lookup_bookname(book_id)
|
||
parsed = parse_ref_rest(piece, book_id, previous)
|
||
if not parsed:
|
||
return None
|
||
start, end, chapter, verse = parsed
|
||
else:
|
||
return None
|
||
previous = {"book": book_id, "chapter": chapter, "verse": verse}
|
||
label = label or name
|
||
refs.append((start, end))
|
||
return {"refs": refs, "notes": notes, "bare": bare, "label": label}
|
||
|
||
|
||
@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 static_page("html/show.html")
|
||
logging.debug(f"show({ref})")
|
||
resolved = resolve_references(ref, loose_bare=True)
|
||
if not resolved:
|
||
return "Invalid Query"
|
||
if get_start_verse_id:
|
||
return str(resolved["refs"][0][0])
|
||
verse_results: list = []
|
||
for start_id, end_id in resolved["refs"]:
|
||
if get_chapter: # ?all= : from the given verse to the end of its chapter
|
||
end_id = start_id // 1000 * 1000 + MAX_CHAPTER_ID_SPAN
|
||
results = lookup_by_verse_id(start_id, end_id)
|
||
logging.debug(f"show.results len({len(results)})")
|
||
verse_results.extend(results)
|
||
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, verse_id=scripture[0]
|
||
)
|
||
|
||
|
||
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",))
|
||
# Prev/next chapter links only make sense when reading a single chapter,
|
||
# not for a keyword search that spans books.
|
||
l_r_nav = None
|
||
if len({str(scripture[0])[:-3] for scripture in scriptures}) == 1:
|
||
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(
|
||
render_navbar(request.args.get("kw", ""), l_r_nav)
|
||
)
|
||
|
||
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 static_page("kjv_fe.html")
|
||
|
||
@app.get("/salvation") # a web page about going to heaven
|
||
def salvation():
|
||
return static_page("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 static_page("html/search.html")
|
||
else:
|
||
return static_page("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 = render_navbar(cleaned_keywords)
|
||
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("/status") # basic instance info
|
||
def status():
|
||
import resource
|
||
|
||
return static_html_doc(
|
||
"Status",
|
||
f"""<h2>Status</h2>
|
||
<p>Sanity: {sanity}</p>
|
||
<p>UserMode: {usermode}</p>
|
||
<p>Memory: {resource.getrusage(resource.RUSAGE_SELF).ru_maxrss}</p>""",
|
||
)
|
||
|
||
|
||
@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 static_page("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)
|