Upload folder using huggingface_hub

app.py

@@ -20,11 +20,8 @@ from chat_utils import (
     limited_chat_wrapper,
 )
 from config import SanatanConfig
-from db import SanatanDatabase
-from drive_downloader import ZipDownloader
 from graph_helper import generate_graph
 from modules.youtube_metadata.app import initialize_youtube_metadata_and_poll
-from nalayiram_helper import delete_taniyan
 import pycountry
 
 # Logging

db.py

@@ -783,6 +783,15 @@ class SanatanDatabase:
 
         return sorted(list(values))
 
+    def is_global_index_available_for_scripture(self, scripture: dict):
+        scripture_name = scripture["name"]
+        collection_name = scripture["collection_name"]
+        collection = self.chroma_client.get_or_create_collection(name=collection_name)
+        results = collection.get(include=["metadatas"], limit=1)
+        metadatas = results["metadatas"]
+        return True if (metadatas and "_global_index" in metadatas[0]) else False
+
+
     def build_global_index_for_scripture(self, scripture: dict, force: bool = False):
         scripture_name = scripture["name"]
         chapter_order = scripture.get("chapter_order", None)
@@ -823,6 +832,13 @@ class SanatanDatabase:
             chapter_order_mapping,
         )
 
+        if(not force and self.is_global_index_available_for_scripture(scripture)):
+            logger.warning(
+                "build_global_index_for_all_scriptures:%s: global index already available. skipping collection",
+                scripture_name,
+            )
+            return
+
         # Fetch all records (keep embeddings for upsert)
         MAX_RETRIES = 3
         RETRY_DELAY = 5  # seconds
@@ -832,15 +848,14 @@ class SanatanDatabase:
                 results = collection.get(include=["metadatas", "documents", "embeddings"])
                 break  # success → exit loop
             except Exception as e:
-                logger.error(
-                    "build_global_index_for_all_scriptures:%s Error getting data from chromadb (attempt %s/%s)",
-                    scripture_name,
-                    attempt,
-                    MAX_RETRIES,
-                    exc_info=True,
-                )
-
                 if attempt == MAX_RETRIES:
+                    logger.error(
+                        "build_global_index_for_all_scriptures:%s Error getting data from chromadb (attempt %s/%s)",
+                        scripture_name,
+                        attempt,
+                        MAX_RETRIES,
+                        exc_info=True,
+                    )
                     # still failing after 3 attempts
                     return
                 time.sleep(RETRY_DELAY)  # wait before retry
@@ -850,13 +865,6 @@ class SanatanDatabase:
         documents = results["documents"]
         embeddings = results.get("embeddings", [None] * len(ids))
 
-        if not force and metadatas and "_global_index" in metadatas[0]:
-            logger.warning(
-                "build_global_index_for_all_scriptures:%s: global index already available. skipping collection",
-                scripture_name,
-            )
-            return
-
         # Create a DataFrame for metadata sorting
         df = pd.DataFrame(metadatas)
         df["_id"] = ids

main.py

@@ -32,6 +32,7 @@ logger.setLevel(logging.INFO)
 
 
 def init():
+    logger.info("Application Initializing ...")
     load_dotenv(override=True)
     try:
         SanatanDatabase().test_sanity()
@@ -49,24 +50,30 @@ def init():
 
     # add global index
     # delete taniyan records
-    SanatanDatabase().delete_taniyans_in_divya_prabandham()
-    SanatanDatabase().build_global_index_for_all_scriptures()
+    logger.info("STARTED: Deleting Taniyans ...")
+    sanatanDatabase = SanatanDatabase()
+    sanatanDatabase.delete_taniyans_in_divya_prabandham()
+    logger.info("STARED: Building Global Index for all scriptures ...")
+    sanatanDatabase.build_global_index_for_all_scriptures()
+    logger.info("FINISHED: Building Global Index for all scriptures ...")
     # Launch the whole thing in a background thread
+    logger.info("STARTED: Initializing youtube metadata poller...")
     yt_init_thread = threading.Thread(target=initialize_youtube_metadata_and_poll, daemon=True)
     yt_init_thread.start()
+    logger.info("FINISHED: Initializing youtube metadata poller...")
 
 
 @asynccontextmanager
 async def fn_lifespan(app: FastAPI):
+    logging.info("🚀 Starting Lifespan ...")
+    ######  Initialize the database ######
+    init()
+    ######  Initialize the database ######
     logging.info("🚀 Starting Firestore listener...")
     start_firestore_listener()
     yield
     logging.info("🛑 Firestore listener shutdown (no explicit cleanup needed).")
     
-######  Initialize the database ######
-init()
-######  Initialize the database ######
-
 app = FastAPI(title="Sanatan AI Unified Server",lifespan=fn_lifespan)
 limiter = Limiter(key_func=get_remote_address)
 app.state.limiter = limiter

modules/firebase/listener.py

@@ -14,7 +14,7 @@ db = admin_firestore.client()
 _last_known_status = {}
 
 def handle_changes(doc_snapshot, changes, read_time):
-    print("handle_changes fired!", changes)
+    print("handle_changes fired!")
 
     for change in changes:
         doc = change.document

modules/firebase/messaging.py

@@ -1,15 +1,17 @@
 import json
 import os
+import time
+import uuid
 from dotenv import load_dotenv
 import firebase_admin
 from firebase_admin import messaging, credentials
+from firebase_admin import firestore as admin_firestore
 from google.oauth2.service_account import Credentials
 from pydantic import BaseModel
 from fastapi import HTTPException
 
 load_dotenv()
 
-
 class FcmRequest(BaseModel):
     token: str
     title: str
@@ -137,22 +139,41 @@ class FcmService:
             return {"status": "success", "message_id": response}
 
         except Exception as e:
+            print(f"Error Sending Message to topic {topic}", e)
             raise HTTPException(status_code=500, detail=str(e))
 
 
-    async def send_broadcast(self, title: str, body: str, data: dict | None = None):
-        """Send a broadcast notification to all users subscribed to 'global_broadcast'."""
-        print(f"send_broadcast: title={title} ; body = {body}; data={data}")
+    async def send_broadcast(self, id:str, title: str, body: str, data: dict | None = None):
+        """Send a broadcast notification and persist it to Firestore."""
+        broadcast_doc = {
+            "id" : id,
+            "title": title,
+            "body": body,
+            "data": data or {"type": "broadcast"},
+            "sent_at": int(time.time()),
+            "created_at": admin_firestore.SERVER_TIMESTAMP,
+        }
+
+        # 1️⃣ Persist to Firestore
+        try:
+            db = admin_firestore.client()
+            doc_ref = db.collection("broadcasts").document()
+            doc_ref.set(broadcast_doc)   # Admin SDK is sync, but non-blocking + safe inside async
+            print(f"[Firestore] Broadcast persisted: {doc_ref.id}")
+        except Exception as e:
+            print(f"[Firestore ERROR] Failed to store broadcast: {e}")
+
+        # 2️⃣ Send FCM broadcast
+        print(f"send_broadcast: title={title} ; body={body}; data={data}")
+
         return await self.send_to_topic(
             title=title,
             body=body,
-            topic=os.getenv("BROADCAST_TOPIC_NAME","global_broadcast"), # use local broadcast if available for dev env.
+            topic=os.getenv("BROADCAST_TOPIC_NAME", "global_broadcast"),
             data=data or {"type": "broadcast"},
             channel_id="broadcast_channel",
-            is_data_only_message=False
+            is_data_only_message=False,
         )
-        
-
 
 # ✅ Singleton instance
 fcm_service = FcmService()

server.py

@@ -621,6 +621,7 @@ async def summarize_scripture_verse(req: ScriptureRequest):
     return response
 
 class BroadcastRequest(BaseModel):
+    id: str
     title: str
     body: str
     data: dict | None = None
@@ -628,6 +629,7 @@ class BroadcastRequest(BaseModel):
 @router.post("/send_broadcast_message")
 async def send_broadcast_message(req: BroadcastRequest):
     response = await fcm_service.send_broadcast(
+        id=req.id,
         title=req.title,
         body=req.body,
         data=req.data