File size: 5,006 Bytes
c35855b | 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 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 | import os
import re
import pdfplumber
from flask import Blueprint, request, jsonify
from werkzeug.utils import secure_filename
from assistant.storage import (
generate_doc_id,
save_document,
load_document,
save_metadata,
get_all_docs,
)
assistant_bp = Blueprint('assistant_bp', __name__)
UPLOAD_FOLDER = './uploads'
os.makedirs(UPLOAD_FOLDER, exist_ok=True)
FALLBACK = "This information is not available in your uploaded notes."
STOP_WORDS = {
"what", "is", "a", "an", "the", "of", "to", "in", "and", "why",
"how", "who", "when", "where", "are", "do", "does", "did", "for",
"on", "it", "this", "that", "tell", "me", "about", "define",
"explain", "give", "meaning", "can", "you", "please", "with",
"was", "will", "has", "have", "had", "be", "been", "by", "or",
"at", "from", "as", "its", "my", "your"
}
# ---------------------------------------------------------------------------
# POST /upload_notes
# ---------------------------------------------------------------------------
@assistant_bp.route('/upload_notes', methods=['POST'])
def upload_notes():
if 'file' not in request.files:
return jsonify({"error": "No file part in request"}), 400
file = request.files['file']
if file.filename == '':
return jsonify({"error": "No selected file"}), 400
filename = secure_filename(file.filename)
filepath = os.path.join(UPLOAD_FOLDER, filename)
file.save(filepath)
try:
text = ""
with pdfplumber.open(filepath) as pdf:
for page in pdf.pages:
extracted = page.extract_text()
if extracted:
text += extracted + "\n"
if not text.strip():
return jsonify({"error": "Could not extract any text from PDF"}), 422
doc_id = generate_doc_id()
save_document(doc_id, text)
save_metadata(doc_id, filename)
return jsonify({"doc_id": doc_id, "message": "Saved successfully"}), 200
except Exception as e:
return jsonify({"error": str(e)}), 500
finally:
if os.path.exists(filepath):
os.remove(filepath)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _extract_keywords(question: str) -> list:
words = re.findall(r'\b\w+\b', question.lower())
keywords = [w for w in words if w not in STOP_WORDS and len(w) > 1]
return keywords if keywords else words
# ---------------------------------------------------------------------------
# POST /ask
# ---------------------------------------------------------------------------
@assistant_bp.route('/ask', methods=['POST'])
def ask_question():
data = request.get_json()
if not data:
return jsonify({"error": "Request body must be JSON"}), 400
if 'doc_id' not in data or 'question' not in data:
return jsonify({"error": "Missing 'doc_id' or 'question'"}), 400
doc_id = data['doc_id']
question = data['question'].strip()
if not question:
return jsonify({"answer": "Please provide a non-empty question."}), 200
# STEP 1 — Load document
text = load_document(doc_id)
if text is None:
return jsonify({"error": f"Document '{doc_id}' not found"}), 404
# STEP 2 — Extract Keywords
keywords = _extract_keywords(question)
# STEP 3 — Match context with line tracking (Pure RAG)
lines = text.split("\n")
matched_data = []
for i, line in enumerate(lines):
line_lower = line.lower()
for kw in keywords:
if kw in line_lower:
matched_data.append({
"line": i + 1,
"text": line.strip()
})
break
if len(matched_data) >= 3: # Keep it concise for direct RAG
break
# Build context string
context = " ".join([item["text"] for item in matched_data])
# STEP 4 — Guard: check context length
if len(context.strip()) < 30:
return jsonify({
"answer": FALLBACK,
"source": None
}), 200
# STEP 5 — Determine source metadata (first match)
first_match = matched_data[0]
# Find the column index of the first keyword in the first matched line
column_index = first_match["text"].lower().find(keywords[0]) if keywords else 0
# STEP 6 — Return the raw extracted context natively as the answer
answer = f"According to your notes: {context}"
return jsonify({
"answer": answer,
"source": {
"line": first_match["line"],
"column": column_index
}
}), 200
# ---------------------------------------------------------------------------
# GET /documents
# ---------------------------------------------------------------------------
@assistant_bp.route('/documents', methods=['GET'])
def list_documents():
return jsonify(get_all_docs()), 200
|