1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
|
"""Api endpoints for gnqa"""
import json
from datetime import datetime, timezone
from flask import Blueprint
from flask import current_app
from flask import jsonify
from flask import request
from gn3.llms.process import get_gnqa
from gn3.llms.errors import LLMError
from gn3.auth.authorisation.oauth2.resource_server import require_oauth
from gn3.auth import db
gnqa = Blueprint("gnqa", __name__)
@gnqa.route("/search", methods=["POST"])
def search():
"""Main gnqa endpoint"""
query = request.json.get("querygnqa", "")
if not query:
return jsonify({"error": "querygnqa is missing in the request"}), 400
fahamu_token = current_app.config.get("FAHAMU_AUTH_TOKEN")
if not fahamu_token:
raise LLMError(
"Request failed:an LLM authorisation token is required ", query)
task_id, answer, refs = get_gnqa(
query, fahamu_token, current_app.config.get("DATA_DIR"))
response = {
"task_id": task_id,
"query": query,
"answer": answer,
"references": refs
}
with (db.connection(current_app.config["LLM_DB_PATH"]) as conn,
require_oauth.acquire("profile user") as token):
cursor = conn.cursor()
cursor.execute("""CREATE TABLE IF NOT EXISTS
history(user_id TEXT NOT NULL,
task_id TEXT NOT NULL,
query TEXT NOT NULL,
results TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY(task_id)) WITHOUT ROWID""")
cursor.execute(
"""INSERT INTO history(user_id, task_id, query, results)
VALUES(?, ?, ?, ?)
""", (str(token.user.user_id), str(task_id["task_id"]),
query,
json.dumps(response))
)
return response
@gnqa.route("/rating/<task_id>", methods=["POST"])
@require_oauth("profile")
def rate_queries(task_id):
"""Endpoint for rating qnqa query and answer"""
with (require_oauth.acquire("profile") as token,
db.connection(current_app.config["LLM_DB_PATH"]) as conn):
results = request.json
user_id, query, answer, weight = (token.user.user_id,
results.get("query"),
results.get("answer"),
results.get("weight", 0))
cursor = conn.cursor()
create_table = """CREATE TABLE IF NOT EXISTS Rating(
user_id TEXT NOT NULL,
query TEXT NOT NULL,
answer TEXT NOT NULL,
weight INTEGER NOT NULL DEFAULT 0,
task_id TEXT NOT NULL UNIQUE,
created_at TIMESTAMP
)"""
cursor.execute(create_table)
cursor.execute("""INSERT INTO Rating(user_id, query,
answer, weight, task_id, created_at)
VALUES(?, ?, ?, ?, ?, ?)
ON CONFLICT(task_id) DO UPDATE SET
weight=excluded.weight
""", (str(user_id), query, answer, weight, task_id,
datetime.now(timezone.utc)))
return {
"message": "You have successfully rated this query.Thank you!"
}, 200
@gnqa.route("/history", methods=["GET"])
@require_oauth("profile user")
def fetch_prev_history():
""" api method to fetch search query records"""
with (require_oauth.acquire("profile user") as token,
db.connection(current_app.config["LLM_DB_PATH"]) as conn):
cursor = conn.cursor()
if request.args.get("search_term"):
cursor.execute(
"""SELECT results from history Where task_id=? and user_id=?""",
(request.args.get("search_term"),
str(token.user.user_id),))
return dict(cursor.fetchone())["results"]
cursor.execute(
"""SELECT task_id,query from history WHERE user_id=?""",
(str(token.user.user_id),))
return jsonify([dict(item) for item in cursor.fetchall()])
|
