Spaces:
Sleeping
Sleeping
| import os | |
| import sys | |
| import json | |
| import sqlite3 | |
| import math | |
| import logging | |
| from datetime import datetime | |
| logger = logging.getLogger("app.oracle") | |
| # Paths | |
| BASE_DIR = os.path.dirname(os.path.abspath(__file__)) | |
| PSALM_DIR = os.path.join(BASE_DIR, "daily_psalms_api") | |
| def get_oracle_data(name: str = "Seeker", topic: str = "humanity", date_str: str = ""): | |
| """ | |
| Consolidated Oracle Tool with Legacy Parity. | |
| date_str: YYYY-MM-DD | |
| """ | |
| # 1. Prepare Environment | |
| original_cwd = os.getcwd() | |
| if PSALM_DIR not in sys.path: sys.path.append(PSALM_DIR) | |
| os.chdir(PSALM_DIR) | |
| try: | |
| from gematria import calculate_gematria | |
| from utils import date_to_words as ps_date_to_words | |
| import torah | |
| import quran | |
| import bible | |
| # 2. Date Handling | |
| if not date_str or date_str.lower() == "today": | |
| date_str = datetime.now().strftime("%Y-%m-%d") | |
| clean_date = date_str.replace("-00-00", "").replace("-00", "") | |
| date_words = ps_date_to_words(clean_date) | |
| # 3. Gematria Key Calculation (Legacy: Name + Date) | |
| search_input = f"{name} {topic}".strip() | |
| combined = f"{search_input} {date_words}".strip() | |
| gematria_sum = calculate_gematria(combined) | |
| # ELS Constant Adjustment | |
| ADJUSTMENT_CONSTANT = 137.035999177 | |
| adjusted_sum = math.ceil(gematria_sum + (gematria_sum / ADJUSTMENT_CONSTANT)) | |
| results_payload = { | |
| "query_context": {"name": name, "topic": topic, "date": date_str, "signal": adjusted_sum}, | |
| "wisdom_nodes": [] | |
| } | |
| # --- Helper for formatted nodes --- | |
| def format_node(match, category, els_text="", els_signal=0): | |
| if not match: return None | |
| try: | |
| trans = "" | |
| # Handle different tuple structures | |
| if len(match) == 5: | |
| words, trans, book, ch, vs = match | |
| elif len(match) == 4: | |
| words, book, ch, vs = match | |
| else: | |
| return None | |
| # Dynamic Translation Fallback if 'trans' is missing/empty | |
| if not trans or trans == words: | |
| try: | |
| from deep_translator import GoogleTranslator | |
| trans = GoogleTranslator(source='auto', target='en').translate(words) | |
| except: trans = "..." | |
| return { | |
| "category": category, | |
| "reference": f"{book} {ch}:{vs}", | |
| "original": words, | |
| "english": trans, | |
| "els_signal": els_signal | |
| } | |
| except Exception as e: | |
| logger.error(f"Error formatting node: {e}") | |
| return None | |
| # --- A. PSALM (Torah ELS) --- | |
| # Legacy: Torah ELS -> Psalm | |
| els_results_torah = torah.process_json_files(1, 1, adjusted_sum, "1,-1", 0, "en", True, True, True, True) | |
| if els_results_torah: | |
| t_res = els_results_torah[0] | |
| els_text_heb = t_res.get("result_text", "") | |
| els_text_eng = t_res.get("translated_text", "") | |
| # Fallback Translation | |
| if not els_text_eng or els_text_eng == els_text_heb: | |
| try: | |
| from deep_translator import GoogleTranslator | |
| els_text_eng = GoogleTranslator(source='auto', target='en').translate(els_text_heb) | |
| except: pass | |
| torah_signal = calculate_gematria(els_text_heb) | |
| # Find Psalm | |
| db_path = os.path.join(PSALM_DIR, "gematria.db") | |
| if os.path.exists(db_path): | |
| try: | |
| with sqlite3.connect(db_path) as conn: | |
| c = conn.cursor() | |
| c.execute("SELECT words, translation, book, chapter, verse FROM results WHERE gematria_sum = ? AND book = 'Psalms' ORDER BY LENGTH(words) ASC LIMIT 1", (torah_signal,)) | |
| p_row = c.fetchone() | |
| node = format_node(p_row, "Biblical Psalm", els_text_heb, torah_signal) | |
| if node: results_payload["wisdom_nodes"].append(node) | |
| except: pass | |
| # --- B. REVELATION (Matthew ELS) --- | |
| # Legacy: Matthew (Book 40) ELS -> Revelation | |
| matthew_results = bible.process_json_files(40, 40, adjusted_sum, "1,-1", 0, "en", True, True, True) | |
| # Handle list structure and potential error dicts | |
| valid_m_res = [r for r in matthew_results if isinstance(r, dict) and "error" not in r] | |
| if valid_m_res: | |
| m_res = None | |
| # Prefer round 1 | |
| r1 = [r for r in valid_m_res if r.get('round') == 1] | |
| if r1: m_res = r1[0] | |
| else: m_res = valid_m_res[0] | |
| if m_res and "result_text" in m_res: | |
| matthew_text = m_res["result_text"] | |
| matthew_signal = calculate_gematria(matthew_text) | |
| # Find Revelation | |
| db_path = os.path.join(PSALM_DIR, "bible.db") | |
| if os.path.exists(db_path): | |
| with sqlite3.connect(db_path) as conn: | |
| c = conn.cursor() | |
| c.execute("SELECT words, book, chapter, verse FROM results WHERE gematria_sum = ? AND book = 'Revelation' ORDER BY LENGTH(words) ASC LIMIT 1", (matthew_signal,)) | |
| b_row = c.fetchone() | |
| node = format_node(b_row, "NT Prophecy", matthew_text, matthew_signal) | |
| if node: results_payload["wisdom_nodes"].append(node) | |
| # --- C. SURA (Quran ELS) --- | |
| # Legacy: Quran ELS -> Sura Match | |
| def dummy_cache(func, *args, **kwargs): return func(*args, **kwargs) | |
| q_res = quran.get_first_els_result_quran(adjusted_sum, "en", cached_process_func=dummy_cache) | |
| if q_res and "result_text" in q_res: | |
| q_text = q_res["result_text"] | |
| quran_signal = calculate_gematria(q_text) | |
| # Find Sura | |
| db_path = os.path.join(PSALM_DIR, "abjad.db") | |
| if os.path.exists(db_path): | |
| with sqlite3.connect(db_path) as conn: | |
| c = conn.cursor() | |
| c.execute("SELECT words, book, chapter, verse FROM results WHERE gematria_sum = ? ORDER BY LENGTH(words) ASC LIMIT 1", (quran_signal,)) | |
| s_row = c.fetchone() | |
| node = format_node(s_row, "Quranic Sura", q_text, quran_signal) | |
| if node: results_payload["wisdom_nodes"].append(node) | |
| return results_payload | |
| except Exception as e: | |
| logger.error(f"Oracle Error: {e}") | |
| return {"error": str(e)} | |
| finally: | |
| os.chdir(original_cwd) | |