Spaces:
Sleeping
Sleeping
| # app.py (Updated with a toggle for the admin panel) | |
| import sys | |
| import subprocess | |
| import json | |
| from flask import Flask, render_template, request, flash, redirect, url_for, jsonify | |
| import torch | |
| from transformers import AutoTokenizer, AutoModel | |
| import os | |
| import chromadb | |
| from huggingface_hub import snapshot_download | |
| app = Flask(__name__) | |
| app.secret_key = os.urandom(24) | |
| # --- App Configuration --- | |
| CHROMA_PATH = "chroma_db" | |
| COLLECTION_NAME = "bible_verses" | |
| MODEL_NAME = "sentence-transformers/multi-qa-mpnet-base-dot-v1" | |
| DATASET_REPO = "broadfield-dev/bible-chromadb-multi-qa-mpnet" | |
| STATUS_FILE = "build_status.log" | |
| JSON_DIRECTORY = 'bible_json' | |
| SHOW_ADMIN_PANEL = os.environ.get('SHOW_ADMIN_PANEL', 'False').lower() in ('true', '1', 't', 'yes') | |
| # --- Globals and Helpers --- | |
| chroma_collection = None | |
| tokenizer = None | |
| embedding_model = None | |
| # (BOOK_ID_TO_NAME and BOOK_NAME_TO_ID dictionaries remain the same) | |
| BOOK_ID_TO_NAME = { | |
| 1: "Genesis", 2: "Exodus", 3: "Leviticus", 4: "Numbers", 5: "Deuteronomy", 6: "Joshua", 7: "Judges", 8: "Ruth", 9: "1 Samuel", 10: "2 Samuel", 11: "1 Kings", 12: "2 Kings", 13: "1 Chronicles", 14: "2 Chronicles", 15: "Ezra", 16: "Nehemiah", 17: "Esther", 18: "Job", 19: "Psalms", 20: "Proverbs", 21: "Ecclesiastes", 22: "Song of Solomon", 23: "Isaiah", 24: "Jeremiah", 25: "Lamentations", 26: "Ezekiel", 27: "Daniel", 28: "Hosea", 29: "Joel", 30: "Amos", 31: "Obadiah", 32: "Jonah", 33: "Micah", 34: "Nahum", 35: "Habakkuk", 36: "Zephaniah", 37: "Haggai", 38: "Zechariah", 39: "Malachi", 40: "Matthew", 41: "Mark", 42: "Luke", 43: "John", 44: "Acts", 45: "Romans", 46: "1 Corinthians", 47: "2 Corinthians", 48: "Galatians", 49: "Ephesians", 50: "Philippians", 51: "Colossians", 52: "1 Thessalonians", 53: "2 Thessalonians", 54: "1 Timothy", 55: "2 Timothy", 56: "Titus", 57: "Philemon", 58: "Hebrews", 59: "James", 60: "1 Peter", 61: "2 Peter", 62: "1 John", 63: "2 John", 64: "3 John", 65: "Jude", 66: "Revelation" | |
| } | |
| BOOK_NAME_TO_ID = {v: k for k, v in BOOK_ID_TO_NAME.items()} | |
| def get_verses_from_json(version, book_name, chapter=None): | |
| book_id = BOOK_NAME_TO_ID.get(book_name) | |
| if not book_id: return None | |
| file_path = os.path.join(JSON_DIRECTORY, f"t_{version.lower()}.json") | |
| if not os.path.exists(file_path): return None | |
| with open(file_path, 'r') as f: data = json.load(f) | |
| rows = data.get("resultset", {}).get("row", []) | |
| verses = [] | |
| for row in rows: | |
| field = row.get("field", []) | |
| if len(field) == 5: | |
| row_book_id, row_chapter, row_verse, text = field[1], field[2], field[3], field[4] | |
| if row_book_id == book_id and (chapter is None or row_chapter == chapter): | |
| verses.append({'chapter': row_chapter, 'verse': row_verse, 'text': text.strip()}) | |
| verses.sort(key=lambda v: (v['chapter'], v['verse'])) | |
| return verses | |
| def mean_pooling(model_output, attention_mask): | |
| token_embeddings = model_output[0] | |
| input_mask_expanded = attention_mask.unsqueeze(-1).expand(token_embeddings.size()).float() | |
| return torch.sum(token_embeddings * input_mask_expanded, 1) / torch.clamp(input_mask_expanded.sum(1), min=1e-9) | |
| def load_resources(): | |
| global chroma_collection, tokenizer, embedding_model | |
| if chroma_collection and embedding_model: return True | |
| print("Attempting to load resources...") | |
| try: | |
| if not os.path.exists(CHROMA_PATH) or not os.listdir(CHROMA_PATH): | |
| print(f"Local DB not found. Downloading from '{DATASET_REPO}'...") | |
| snapshot_download(repo_id=DATASET_REPO, repo_type="dataset", local_dir=CHROMA_PATH, local_dir_use_symlinks=False) | |
| client = chromadb.PersistentClient(path=CHROMA_PATH) | |
| chroma_collection = client.get_collection(name=COLLECTION_NAME) | |
| tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME) | |
| embedding_model = AutoModel.from_pretrained(MODEL_NAME) | |
| print(f"Resources loaded: DB has {chroma_collection.count()} items. Model is '{MODEL_NAME}'.") | |
| return True | |
| except Exception as e: | |
| print(f"Could not load resources. DB may not be built. Error: {e}") | |
| return False | |
| resources_loaded = load_resources() | |
| # --- Routes --- | |
| def home(): | |
| # *** CHANGE 2: PASS THE TOGGLE VARIABLE TO THE TEMPLATE *** | |
| return render_template('index.html', show_admin=SHOW_ADMIN_PANEL) | |
| # The /build-rag and /status routes are only used by the admin panel's javascript. | |
| # They will simply be unused if the panel is hidden, which is fine. No changes needed. | |
| def build_rag_route(): | |
| try: | |
| with open(STATUS_FILE, "w") as f: f.write("IN_PROGRESS: Starting build process...") | |
| subprocess.Popen([sys.executable, "build_rag.py"]) | |
| return jsonify({"status": "started"}) | |
| except Exception as e: | |
| with open(STATUS_FILE, "w") as f: f.write(f"FAILED: Could not start process - {e}") | |
| return jsonify({"status": "error", "message": str(e)}), 500 | |
| def status(): | |
| if not os.path.exists(STATUS_FILE): return jsonify({"status": "NOT_STARTED"}) | |
| with open(STATUS_FILE, "r") as f: status_line = f.read().strip() | |
| status_code, _, message = status_line.partition(': ') | |
| return jsonify({"status": status_code, "message": message}) | |
| def search(): | |
| global resources_loaded | |
| if not resources_loaded: | |
| resources_loaded = load_resources() | |
| if not resources_loaded: | |
| flash("Database not ready. Please build the database from the Admin Panel.", "error") | |
| return redirect(url_for('home')) | |
| user_query = request.form['query'] | |
| if not user_query: | |
| # *** CHANGE 3: PASS THE TOGGLE ON ALL RENDERS OF INDEX.HTML *** | |
| return render_template('index.html', results=[], show_admin=SHOW_ADMIN_PANEL) | |
| encoded_input = tokenizer([user_query], padding=True, truncation=True, return_tensors='pt') | |
| with torch.no_grad(): | |
| model_output = embedding_model(**encoded_input) | |
| query_embedding = mean_pooling(model_output, encoded_input['attention_mask']) | |
| search_results = chroma_collection.query(query_embeddings=query_embedding.cpu().tolist(), n_results=10) | |
| results_list = [] | |
| documents, metadatas, distances = search_results['documents'][0], search_results['metadatas'][0], search_results['distances'][0] | |
| for i in range(len(documents)): | |
| meta = metadatas[i] | |
| results_list.append({ | |
| 'score': distances[i], | |
| 'text': documents[i], | |
| 'reference': meta.get('reference', 'N/A'), | |
| 'version': meta.get('version', 'N/A'), | |
| 'book_name': meta.get('book_name', ''), | |
| 'chapter': meta.get('chapter', '') | |
| }) | |
| return render_template('index.html', results=results_list, query=user_query, show_admin=SHOW_ADMIN_PANEL) | |
| def view_chapter(version, book_name, chapter_num): | |
| verses = get_verses_from_json(version, book_name, chapter=chapter_num) | |
| if not verses: | |
| return "Chapter not found.", 404 | |
| return render_template('chapter.html', book_name=book_name, chapter_num=chapter_num, verses=verses, version=version) | |
| def view_book(version, book_name): | |
| all_verses = get_verses_from_json(version, book_name) | |
| if not all_verses: | |
| return "Book not found.", 404 | |
| chapters = {} | |
| for verse in all_verses: | |
| chap_num = verse['chapter'] | |
| if chap_num not in chapters: chapters[chap_num] = [] | |
| chapters[chap_num].append(verse) | |
| return render_template('book.html', book_name=book_name, chapters=chapters) | |
| if __name__ == '__main__': | |
| app.run(host='0.0.0.0', port=7860) |