Upload folder using huggingface_hub

app.py

@@ -13,6 +13,8 @@ from modules.indexer import index_videos
 from modules.answerer import answer_query, LLMAnswer, VideoItem, build_video_html
 from dotenv import load_dotenv
 
+from youtube_sync import sync_channels_from_youtube
+
 load_dotenv()
 
 
@@ -58,52 +60,45 @@ def enable_if_not_none(question):
         return enable_component()
 
 
-# -------------------------------
-# Fetch & index channels
-# -------------------------------
-def refresh_channel(api_key, channel_url: str):
-    all_videos = []
-    for status, videos_batch in fetch_all_channel_videos(api_key, channel_url):
-        # enrich each video in the batch with channel_url
-        for v in videos_batch:
-            v["channel_url"] = channel_url
-        all_videos.extend(videos_batch)
-
-        # (optional) log progress
-        print(channel_url, status, "- total so far:", len(all_videos))
-
-    # index all at once
-    if all_videos:
-        print(f"{channel_url} - Adding {len(all_videos)} videos to database")
-        index_videos(all_videos, get_collection(), channel_url=channel_url)
-        print(f"{channel_url} - Loaded {len(all_videos)} videos to database")
-
-    return len(all_videos)
-
-
 def index_channels(channel_urls: str):
     yield "saving ...", gr.update(), gr.update()
     yt_api_key = os.environ["YOUTUBE_API_KEY"]
+
     urls = [u.strip() for u in re.split(r"[\n,]+", channel_urls) if u.strip()]
-    total_videos = sum(refresh_channel(yt_api_key, url) for url in urls)
-    yield f"✅ Indexed {total_videos} videos from {len(urls)} channels.", gr.update(
-        choices=list_channels_radio()
-    ), list_channels_radio()
-    return
+    total_videos = 0
+
+    # sync all channels, streaming progress
+    for message, videos_count in sync_channels_from_youtube(yt_api_key, urls):
+        total_videos += videos_count  # accumulate actual number of videos indexed
+        yield message, gr.update(), gr.update()
+
+    # final UI update
+    yield (
+        f"✅ Indexed {total_videos} videos from {len(urls)} channels.",
+        gr.update(choices=list_channels_radio()),
+        list_channels_radio(),
+    )
 
 
 def refresh_all_channels():
     yt_api_key = os.environ["YOUTUBE_API_KEY"]
     channels = get_indexed_channels(get_collection())
+
     if not channels:
         return "⚠️ No channels available to refresh.", list_channels_radio()
-    total_videos = 0
+
+    # build list of URLs
+    urls = []
     for key, val in channels.items():
         url = val.get("channel_url") if isinstance(val, dict) else key
         if url:
-            total_videos += refresh_channel(yt_api_key, url)
+            urls.append(url)
+
+    # re-index all at once
+    total_videos = sync_channels_from_youtube(yt_api_key, urls)
+
     return (
-        f"🔄 Refreshed {len(channels)} channels, re-indexed {total_videos} videos.",
+        f"🔄 Refreshed {len(urls)} channels, re-indexed {total_videos} videos.",
         list_channels_radio(),
     )
 
@@ -134,8 +129,7 @@ def fetch_channel_html(channel_id: str):
     # query your collection/db instead of YouTube API
     collection = get_collection()
     results = collection.get(
-        where={"channel_id": channel_id},
-        include=["documents", "metadatas"]
+        where={"channel_id": channel_id}, include=["documents", "metadatas"]
     )
 
     if not results or not results.get("metadatas"):
@@ -177,7 +171,6 @@ def fetch_channel_html(channel_id: str):
     return html
 
 
-
 # Delete a channel
 # -------------------------------
 def delete_channel(channel_url: str):
@@ -252,16 +245,17 @@ with gr.Blocks() as demo:
         # Sidebar
         with gr.Sidebar() as my_sidebar:
             gr.Markdown("### 📺 Channels")
-            channel_list_state = gr.State([c for c in list_channels_radio()])
+            channel_list_values = list_channels_radio()
+            channel_list_state = gr.State(channel_list_values)
 
             no_channels_message = gr.Markdown(
                 "⚠️ **No channels available.**",
-                visible=False if channel_list_state.value else True,
+                visible=False if channel_list_values else True,
             )
             channel_radio = gr.Radio(
-                choices=channel_list_state.value,
+                choices=channel_list_values,
                 label="Select a Channel",
-                visible=True if channel_list_state.value else False,
+                visible=True if channel_list_values else False,
             )
 
             with gr.Row():
@@ -272,8 +266,19 @@ with gr.Blocks() as demo:
                     variant="secondary",
                     interactive=False,
                 )
+                refresh_btn = gr.Button(
+                    "🔄Refresh",
+                    size="sm",
+                    scale=0,
+                    variant="huggingface",
+                    visible=False,
+                )
                 refresh_all_btn = gr.Button(
-                    "🔄Refresh", size="sm", scale=0, variant="huggingface"
+                    "🔄Refresh",
+                    size="sm",
+                    scale=0,
+                    variant="huggingface",
+                    visible=False,
                 )
                 add_channels_btn = gr.Button(
                     "➕ Add", size="sm", scale=0, variant="primary"
@@ -291,6 +296,7 @@ with gr.Blocks() as demo:
                 outputs=[refresh_status, channel_radio],
             )
 
+            refresh_btn.click(fn=list_channels_radio, outputs=[channel_radio])
             add_channels_btn.click(close_component, outputs=[my_sidebar]).then(
                 show_component, outputs=[add_channel_modal]
             )

modules/collector.py

@@ -15,11 +15,11 @@ def fetch_all_channel_videos(api_key: str, channel_url: str, max_results_per_cal
 
     final_videos = []
     for videos in fetch_channel_videos_by_id(api_key, channel_id, max_results_per_call):
-        final_videos.extend(videos)  # extend instead of append
+        final_videos.extend(videos)
         print("Fetched", len(final_videos))
-        yield (f"Fetched {len(final_videos)}", final_videos)
+        yield (f"Fetched {len(final_videos)}", videos)  # <-- only yield the *new* batch
 
-    yield (f"Fetched {len(final_videos)}", final_videos)
+    yield (f"Fetched {len(final_videos)}", [])  # final "summary"
 
 
 def fetch_channel_videos_by_id(api_key: str, channel_id: str, max_results=50):

modules/indexer.py

@@ -2,7 +2,7 @@
 from typing import Dict, List
 from openai import OpenAI
 
-def index_videos(videos: List[Dict], collection, channel_url: str, batch_size: int = 20):
+def index_videos(videos: List[Dict], collection, channel_url: str, batch_size: int = 50):
     client = OpenAI()
 
     total = len(videos)
@@ -11,7 +11,10 @@ def index_videos(videos: List[Dict], collection, channel_url: str, batch_size: i
     # Split into batches
     for start in range(0, total, batch_size):
         batch = videos[start:start + batch_size]
-        print(f"[INDEX] Processing batch {start+1} → {start+len(batch)} of {total}")
+        end = start + len(batch)
+        percent = round((end / total) * 100, 1)
+
+        print(f"[INDEX] Processing batch {start+1} → {end} of {total} — {percent}%")
 
         # Prepare text inputs
         texts = [f"{vid.get('title', '')} - {vid.get('description', '')}" for vid in batch]
@@ -49,6 +52,8 @@ def index_videos(videos: List[Dict], collection, channel_url: str, batch_size: i
             ids=ids,
         )
 
-        print(f"[INDEX] ✅ Indexed {len(batch)} videos (total so far: {start+len(batch)}/{total})")
+        print(f"[INDEX] ✅ Indexed {len(batch)} videos (total so far: {end}/{total} — {percent}%)")
 
     print(f"[INDEX] 🎉 Finished indexing {total} videos for channel={channel_url}")
+    return total
+

tests/test_num_records_in_channel.py

@@ -0,0 +1,28 @@
+# modules/test_utils.py
+import asyncio
+import time
+from typing import Optional
+from modules.db import get_collection 
+
+
+def count_records_for_channel(collection, channel_url: str) -> int:
+    """
+    Returns how many records are loaded in a collection for the given channel_url.
+    """
+    if not channel_url:
+        raise ValueError("channel_url must be provided")
+
+    results = collection.get(
+        where={"channel_url": channel_url},
+        include=[]
+    )
+    count = len(results["ids"])
+    print(f"[TEST] Channel '{channel_url}' has {count} records in collection.")
+    return count
+
+
+if __name__ == "__main__":
+    collection = get_collection()
+    while True:
+        count_records_for_channel(collection, "https://www.youtube.com/@SriYadugiriYathirajaMutt")
+        time.sleep(1)

youtube_sync.py

@@ -0,0 +1,78 @@
+import threading
+import gradio as gr
+from concurrent.futures import ThreadPoolExecutor, as_completed
+
+from modules.collector import fetch_all_channel_videos
+from modules.db import get_collection
+from modules.indexer import index_videos
+
+# global stop signal
+stop_event = threading.Event()
+MAX_BATCHES = 200  # safety cutoff
+
+def stop_sync():
+    """External call to stop the sync process."""
+    stop_event.set()
+
+def sync_channels_from_youtube(api_key, channel_urls: list, progress: gr.Progress = None):
+    """
+    Sync multiple channels, yielding (progress_message, videos_indexed_in_batch)
+    """
+    global stop_event
+    stop_event.clear()
+
+    total_channels = len(channel_urls)
+    total_videos = 0
+
+    for idx, channel_url in enumerate(channel_urls, 1):
+        if stop_event.is_set():
+            yield f"🛑 Stopped before processing channel: {channel_url}", 0
+            break
+
+        yield f"🔄 Syncing {channel_url} ({idx}/{total_channels})", 0
+
+        # stream video-level progress from inner generator
+        for update_message, batch_count in _refresh_single_channel(api_key, channel_url, progress):
+            total_videos += batch_count
+            yield update_message, batch_count
+
+    yield f"✅ Finished syncing. Total channels: {total_channels}, total videos: {total_videos}", 0
+
+
+def _refresh_single_channel(api_key, channel_url, progress):
+    # fetch all batches first
+    fetched_batches = list(fetch_all_channel_videos(api_key, channel_url))
+    all_videos = [v | {"channel_url": channel_url} for _, batch in fetched_batches for v in batch]
+    total_videos = len(all_videos)
+
+    if total_videos == 0:
+        yield f"{channel_url}: No videos found", 0
+        return
+
+    with ThreadPoolExecutor(max_workers=4) as executor:
+        futures = [
+            executor.submit(index_videos, batch, get_collection(), channel_url=channel_url)
+            for _, batch in fetched_batches
+        ]
+
+        completed_videos = 0
+        for f in as_completed(futures):
+            if stop_event.is_set():
+                yield "🛑 Stop requested during indexing stage", completed_videos
+                break
+
+            try:
+                indexed_count = f.result()
+                if indexed_count is None:
+                    indexed_count = len(all_videos)  # fallback if index_videos doesn't return
+            except Exception as e:
+                indexed_count = 0
+                yield f"⚠️ Error indexing {channel_url}: {e}", completed_videos
+
+            completed_videos += indexed_count
+            pct = 100.0 * completed_videos / max(1, total_videos)
+
+            if progress:
+                progress(completed_videos / total_videos)
+
+            yield f"{channel_url}: Indexed {completed_videos}/{total_videos} videos — {pct:.1f}%", completed_videos