676 lines
25 KiB
Python
676 lines
25 KiB
Python
# app.py
|
|
from flask import Flask, request, jsonify, send_file, render_template, redirect
|
|
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 sys
|
|
import logging
|
|
from tools import sanity_check
|
|
|
|
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}")
|
|
|
|
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.
|
|
"""
|
|
|
|
if __name__ == "__main__":
|
|
# single user mode, running without uwsgi
|
|
usermode: str = "Single"
|
|
kjv_cur = sqlite3.connect("data/kjv.db", check_same_thread=False).cursor()
|
|
else:
|
|
# you're running with uwsgi
|
|
usermode: str = "Multi"
|
|
kjv_cur = sqlite3.connect("data/kjv.db").cursor()
|
|
|
|
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": [],
|
|
"json_dictionary": json.load(open("data/1828_Webster_KJV.json", "r")),
|
|
"html_response": open("html/response.html", "r").read(),
|
|
"html_navbar": open("html/navbar.html", "r").read(),
|
|
"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(),
|
|
"js_collapsible": open("js/collapsible.js", "r").read(),
|
|
}
|
|
|
|
# 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()
|
|
|
|
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):
|
|
logging.debug(f"lookup_bookname({book_id})")
|
|
result = kjv_cur.execute(
|
|
f"SELECT n FROM key_english WHERE b = {book_id};"
|
|
).fetchone()
|
|
if result:
|
|
result = result[0]
|
|
logging.debug(f"lookup_bookname.result = '{result}'")
|
|
return result
|
|
def lookup_book_id(bookname):
|
|
logging.debug(f"lookup_book_id({bookname})")
|
|
result = kjv_cur.execute(
|
|
f"SELECT b FROM key_english WHERE n = '{bookname}';"
|
|
).fetchone()
|
|
if result:
|
|
result = result[0]
|
|
logging.debug(f"lookup_book_id.result = '{result}'")
|
|
return result
|
|
def lookup_fts(kwds):
|
|
logging.debug(f"lookup_fts.kwds = '{kwds}'")
|
|
return kjv_cur.execute(f"SELECT * FROM fts_kjv('{kwds}')").fetchall()
|
|
def lookup_by_verse_id(start_verse, end_verse=None):
|
|
if not end_verse:
|
|
results = kjv_cur.execute(
|
|
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_verse_id(book_id, chapter, verse):
|
|
# verse_id is 00000000 (00 000 000) (book + chapter + verse)
|
|
chapterid = chapter.zfill(3)
|
|
verseid = verse.zfill(3)
|
|
if verseid == "000":
|
|
verseid = "001" # Avoid verse_ids list IndexErrors on :- searches
|
|
return str(book_id) + chapterid + verseid
|
|
def get_next_prev_chapter(verseid):
|
|
current_chapter = verseid[:-3]
|
|
current_chapter_index = cached["chapter_list"].index(int(current_chapter))
|
|
|
|
#next_chapter_id = str(cached["chapter_list"][current_chapter_index + 1])
|
|
next_chapter_id = str(cached["chapter_list"][(current_chapter_index + 1) % len(cached["chapter_list"])])
|
|
#prev_chapter_id = str(cached["chapter_list"][current_chapter_index - 1])
|
|
prev_chapter_id = str(cached["chapter_list"][(current_chapter_index - 1) % len(cached["chapter_list"])])
|
|
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:
|
|
logging.debug("ref_input_cleaning.if ref.count(:): TRUE")
|
|
logging.debug(" -> Assuming user wanted entire chapter")
|
|
ref += ":-"
|
|
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 + '+^* '
|
|
):
|
|
# 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
|
|
@app.get("/find")
|
|
def parse_query(query_override:str = None):
|
|
query = request.args.get("kw", None)
|
|
if query_override:
|
|
query = query_override
|
|
if query:
|
|
query = query.strip()
|
|
logging.info(f"Query: {query}")
|
|
if not query:
|
|
return send_file("html/find.html")
|
|
if query.endswith(","):
|
|
query = query[:-1]
|
|
result = show(kw=query)
|
|
if not hasattr(result, "__len__"):
|
|
logging.debug(f"parse_query.show = '{result}'")
|
|
return result
|
|
|
|
if len(result) > 17:
|
|
return result
|
|
result = "<!DOCTYPE html>"
|
|
result += search()
|
|
result += "<style>" + cached["css_collapsible"] + "</style>"
|
|
result += "<h3>1828 Webster Definitions</h3> Click on the word to get the definition."
|
|
split_query = query.replace(",", " ").split(" ")
|
|
if len(split_query) > 20:
|
|
return "Your Query is too long. 20 Words or less."
|
|
for word in split_query:
|
|
definition = define(word)
|
|
if not definition:
|
|
continue
|
|
result += (
|
|
"<button type='button' class='collapsible'>"
|
|
+ word.capitalize()
|
|
+ "</button>"
|
|
)
|
|
result += "<div class='definition'>" + definition + "</div>"
|
|
logging.debug(f"parse_query.define = '{word}'")
|
|
result += "<script>" + cached["js_collapsible"] + "</script>"
|
|
result += "<hr>"
|
|
#result += search()
|
|
logging.debug(f"parse_query.search = '{result}'")
|
|
return result
|
|
@app.get("/define")
|
|
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")
|
|
def show(kw=None, get_start_verse_id=False):
|
|
ref = request.args.get("kw", None)
|
|
get_chapter: bool = request.args.get("all", False)
|
|
if ref.endswith(","):
|
|
ref = ref[:-1]
|
|
if kw:
|
|
ref = kw
|
|
if ref is None:
|
|
return send_file("html/show.html")
|
|
logging.debug(f"show({ref})")
|
|
refs = []
|
|
for r in ref.split(","):
|
|
r_ = ref_input_cleaning(r.strip())
|
|
if r_:
|
|
refs.append(r_)
|
|
if not refs:
|
|
return "Invalid Query"
|
|
verse_results = []
|
|
# check if the search is for multiple refs
|
|
for ref in refs:
|
|
logging.debug(f"show.reference({ref})")
|
|
maybe_chapter = ref.split(":")[0][3:]
|
|
#FIXME: Not including : in a book reference should take you to the first book
|
|
# not just crash like a retard
|
|
if not ":" in maybe_chapter:
|
|
get_chapter = True
|
|
ref_chapter = ""
|
|
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 = ref_input_cleaning(ref, string.ascii_letters)
|
|
ref_bookname = ref_parse_book(book_str)
|
|
ref_book_id = lookup_book_id(ref_bookname)
|
|
if not ref_book_id:
|
|
return "Invalid Query"
|
|
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({results})")
|
|
for result in results:
|
|
verse_results.append(result)
|
|
response_list = scripture_response(verse_results)
|
|
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)
|
|
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):
|
|
response = []
|
|
arg_view = request.args.get("view", "json")
|
|
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":
|
|
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"{len(scriptures)} Results")
|
|
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("/")
|
|
@app.get("/<shortcode>")
|
|
def idx(shortcode=''):
|
|
if not sanity:
|
|
return("There is an issue with permissions. Check log, or please notify tyler@dinsmoor.us")
|
|
if shortcode:
|
|
return resolve_shortcode(shortcode)
|
|
logging.info("Index Served")
|
|
return send_file("html/index.html")
|
|
@app.get("/salvation")
|
|
def salvation():
|
|
return send_file("html/heaven.html")
|
|
@app.get("/t/random", defaults={"view": "plain"}) # backwards-compatibility
|
|
@app.get("/random")
|
|
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 = "<br>".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
|
|
@app.get("/votd")
|
|
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 = "<br>".join(response_list)
|
|
else:
|
|
response = jsonify(response_list)
|
|
logging.debug(f"Search Response: {response}")
|
|
return response
|
|
@app.get("/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")
|
|
results = lookup_fts(cleaned_keywords)
|
|
if not results:
|
|
h = hunspell.HunSpell("data/kjv.dic", "data/kjv.aff")
|
|
suggestions = h.suggest(cleaned_keywords)
|
|
return "{} not found, try one of the following: {}".format(
|
|
cleaned_keywords, ", ".join(suggestions)
|
|
)
|
|
response_list = scripture_response(results)
|
|
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)
|
|
logging.debug(f"Search Response: {response}")
|
|
return response
|
|
@app.get("/reader")
|
|
def bible_reader():
|
|
return send_file("html/reader.html")
|
|
# 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")
|
|
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")
|
|
def status():
|
|
import resource
|
|
return f"""Sanity: {sanity} <br>
|
|
UserMode: {usermode} <br>
|
|
Memory: {resource.getrusage(resource.RUSAGE_SELF).ru_maxrss}
|
|
"""
|
|
@app.get("/seqreport")
|
|
def sequential_report(uid=None):
|
|
if usermode == "Single":
|
|
seq_conn = sqlite3.connect("data/seq.db", check_same_thread=False)
|
|
else:
|
|
seq_conn = sqlite3.connect("data/seq.db")
|
|
seq_cur = seq_conn.cursor()
|
|
uids = seq_cur.execute("select * from reading_tracker").fetchall()
|
|
return render_template("seq_report.html", uids=uids)
|
|
def sequential_purge():
|
|
#TODO: Test this :^)
|
|
if usermode == "Single":
|
|
seq_conn = sqlite3.connect("data/seq.db", check_same_thread=False)
|
|
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(f"DELETE FROM reading_tracker WHERE uid = '{uid[0]}'")
|
|
|
|
seq_conn.commit()
|
|
seq_cur.close()
|
|
@app.get("/seq")
|
|
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(
|
|
f"INSERT INTO reading_tracker VALUES('{uid}', {start}, '{last_used}')"
|
|
)
|
|
except sqlite3.OperationalError as e:
|
|
logging.error(f"UID Database failed to insert {uid} into reading_tracker. {e}")
|
|
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(
|
|
f"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(
|
|
f"REPLACE INTO reading_tracker (uid, verse, last_used) VALUES('{uid}', {nextverse+num+1}, '{last_used}')"
|
|
)
|
|
# Doing REPLACE INTO instead of UPDATE will make the updated row at the end of the table, making the purging
|
|
# only look at a small set of the oldest unused entries.
|
|
seq_conn.commit()
|
|
arg_view = request.args.get("view", None)
|
|
if arg_view == "plain" or arg_view == "rich":
|
|
response = "<br>".join(response_list)
|
|
else:
|
|
response = jsonify(response_list)
|
|
|
|
return response
|
|
@app.get("/mksc/<query>")
|
|
def generate_shortcode(query: str = ""):
|
|
query = search_input_cleaning(query)
|
|
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(
|
|
f"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(f"SELECT query FROM shortcodes WHERE shortcode = '{shortcode}'").fetchone()[0]
|
|
short_cur.close()
|
|
logging.info(f"Resolved Shortcode {shortcode} for query '{query}'")
|
|
|
|
return redirect(f"/find?kw={query}&view=rich")
|
|
|
|
if __name__ == "__main__":
|
|
print("You must run this program within a uwsgi environment if you are running this for multiple users.")
|
|
print("Otherwise, you may end up with database corruption if multiple writes happen.")
|
|
print("Operating in single user mode.")
|
|
app.run(host="127.0.0.1", port=1611)
|