from modules.youtube_metadata.db import get_youtube_metadata_collection
import pandas as pd
page_size = 10 # change if you like
# -------------------------------
# Fetch channel videos as HTML table with pagination
# -------------------------------
def fetch_channel_html(channel_id: str, page: int = 1, page_size: int = 10):
collection = get_youtube_metadata_collection()
offset = (page - 1) * page_size
all_results = collection.get(
where={"channel_id": channel_id}, include=["metadatas"]
)
total_count = (
len(all_results["metadatas"])
if all_results and "metadatas" in all_results
else 0
)
results = collection.get(
where={"channel_id": channel_id},
include=["documents", "metadatas"],
limit=page_size,
offset=offset,
)
# handle empty
if not results or not results.get("metadatas"):
return f"""
⚠️ No videos found for this channel (page {page}).
"""
videos = results["metadatas"]
# build table
html = (
f"Total: {total_count} videos
"
+ """
| # |
Title |
Video URL |
Description |
"""
)
for idx, v in enumerate(videos, start=offset + 1):
html += f"""
| {idx} |
{v.get('video_title','')} |
Watch Video |
{v.get('description','')} |
"""
html += "
"
return html
# -------------------------------
# Fetch channel videos as HTML table with pagination
# -------------------------------
def fetch_channel_dataframe(channel_id: str):
collection = get_youtube_metadata_collection()
results = collection.get(
where={"channel_id": channel_id}, include=["documents", "metadatas"]
)
total_count = len(results["metadatas"]) if results and "metadatas" in results else 0
# handle empty
if not results or not results.get("metadatas"):
return pd.DataFrame(data=[])
videos = results["metadatas"]
items = []
for idx, v in enumerate(videos, start=1):
item = {
"#": idx,
"title": v.get("video_title", "-"),
"description": v.get("description", ""),
"url": f"""▶️Watch Video""",
}
items.append(item)
return pd.DataFrame(data=items)
def update_table(channel_id, page):
return fetch_channel_html(channel_id, page, page_size), f"Page {page}"
def prev_page(channel_id, page):
new_page = max(1, page - 1)
return (
fetch_channel_html(channel_id, new_page, page_size),
f"Page {new_page}",
new_page,
)
def next_page(channel_id, page):
new_page = page + 1
return (
fetch_channel_html(channel_id, new_page, page_size),
f"Page {new_page}",
new_page,
)