manghata / phase_2_extract_data.py
sameerbanchhor-work's picture
Upload folder using huggingface_hub
810ef2c verified
Raw History Blame Contribute Delete
3.16 kB
import os
import json
from bs4 import BeautifulSoup
import glob
INPUT_DIR = "pages"
OUTPUT_FILE = "videos_dataset.json"
def extract_data_from_html(file_path):
"""Parses an HTML file and extracts video data."""
extracted_data = []
with open(file_path, "r", encoding="utf-8") as f:
soup = BeautifulSoup(f, "html.parser")
# Find all video links
video_elements = soup.find_all("a", class_="video")
for video in video_elements:
try:
url = video.get("href", "").strip()
thumbnail = video.get("data-bg", "").strip()
# Title can be in the title attribute or the h2 tag
title = video.get("title", "")
if not title:
title_elem = video.find("h2", class_="vtitle")
if title_elem:
title = title_elem.text.strip()
# Time and duration
duration_elem = video.find("span", class_="time clock")
duration = duration_elem.text.strip() if duration_elem else ""
ago_elem = video.find("span", class_="ago")
uploaded_ago = ago_elem.text.strip() if ago_elem else ""
# Only add if we have at least a URL and title
if url and title:
extracted_data.append({
"url": url,
"thumbnail": thumbnail,
"title": title,
"duration": duration,
"uploaded_ago": uploaded_ago
})
except Exception as e:
print(f"Error extracting an item in {file_path}: {e}")
return extracted_data
def main():
print(f"Starting Phase 2: Data Extraction from '{INPUT_DIR}' directory...")
# Check if input directory exists
if not os.path.exists(INPUT_DIR):
print(f"Error: Directory '{INPUT_DIR}' not found. Please run phase 1 first.")
return
all_videos = []
# Get all HTML files in the pages directory
html_files = glob.glob(os.path.join(INPUT_DIR, "*.html"))
if not html_files:
print(f"No HTML files found in '{INPUT_DIR}'.")
return
print(f"Found {len(html_files)} HTML files. Processing...")
# Process each file
for file_path in html_files:
videos_in_file = extract_data_from_html(file_path)
all_videos.extend(videos_in_file)
# Optional progress output
# print(f"Extracted {len(videos_in_file)} videos from {os.path.basename(file_path)}")
print(f"Extraction complete! Extracted {len(all_videos)} total videos.")
# Remove duplicates based on URL just in case
unique_videos = {v['url']: v for v in all_videos}.values()
final_dataset = list(unique_videos)
print(f"Dataset contains {len(final_dataset)} unique videos.")
# Save to JSON
with open(OUTPUT_FILE, "w", encoding="utf-8") as f:
json.dump(final_dataset, f, indent=4, ensure_ascii=False)
print(f"Successfully saved dataset to {OUTPUT_FILE}")
if __name__ == "__main__":
main()