# Project export: Revi This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small. ## Project metadata - Hackathon: TreeHacks 2025 - Tagline: Aggregated reviews to help you make good purchases in seconds. - Devpost: https://devpost.com/software/revi - GitHub: https://github.com/Bloomh/revi - Video: https://www.youtube.com/embed/B5oGauHfLZA?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent - Team: 2 GitHub contributor(s) — Henry Bloom (15 commits), alexissfry (4 commits) ## Devpost submission (written by the team) ### Overview The Problem 🙇🏻‍♀️ While around 98% of consumers read reviews before making a purchase, the process remains time consuming. Consumers jump from site to site, reading numerous written reviews and ingesting video reviews on social media platforms. Revi is here to make that process much more informative and efficient, ensuring that consumers put their money towards quality purchases. ### What it does 🧐 Revi analyzes ratings from major retailers across the internet, such as Amazon, Walmart, and more niche retailers like Sephora, and provides a weighted average across tens of thousands of reviews. Further, it summarizes social media review content from YouTube and TikTok, assigns a rating to it, and provides a link to the post. ### How we built it ⚙️ Revi is a Flask app that uses Python for the backend and JS/HTML/CSS for the frontend. We use Oxylab's Web Scraper API to retrieve ratings, review counts, and images for products. For social media review pulling, we use the YouTube Data API and the EnsembleData TikTok API to search for relevant review posts on YouTube and TikTok, respectively. We then use yt-dlp to download audio from these videos as mp3, transcribe the videos with OpenAI's Whisper API, and finally feed the video metadata (such as the title and description/caption) and transcribed content into OpenAI's gpt-4-turbo to assign a star rating to the review and concisely summarize the content into a review format. ### What's next 🔜 Looking to the future, we envision Revi's business model to be a subscription-based platform that prompts users to pay for more than three queries a month. Technically, this means that we would have to build out user management and ensure that the application is scalable for deployment. We also see pathways to monetization through running ads or collecting user data. We aim to deploy Revi to make it accessible to all, but the heavy API usage required for the platform to analyze reviews will be costly at scale. We also plan to integrate more social media platforms into our aggregation mechanism, along with embedded video viewing and support for non-english videos. Eventually, we will provide users with tailored recommendations based on their query and viewing history, using our advanced review analysis platform to suggest products that they will likely enjoy based on the opinions of thousands of others! ## README (from the GitHub repository) # Revi Aggregated, AI-generated reviews to help you make good purchases in seconds. ## Setup 1. Create a virtual environment (recommended): ```bash python -m venv venv source venv/bin/activate ``` 2. Install dependencies: ```bash pip install -r requirements.txt ``` 3. Set up environment variables: 1. Copy `.env.template` to a new file named `.env`: ```bash cp .env.template .env ``` 2. Fill in your API keys in the `.env` file: - `YOUTUBE_API_KEY`: Get from [Google Cloud Console](https://console.cloud.google.com) - `OPENAI_API_KEY`: Get from [OpenAI](https://platform.openai.com/api-keys) - `OXYLABS_USER`, `OXYLABS_PASS`: Get from [Oxylabs](https://oxylabs.io) - `ENSEMBLEDDATA_API_KEY`: Get from [EnsembleData](https://ensembledata.com) - `PERPLEXITY_API_KEY`: Get from [Perplexity](https://perplexity.ai) All these APIs are required for full functionality. 4. Run the application: ```bash python app.py ``` 5. Open your browser and visit: `http://localhost:5000` ## Detected evidence (automated analysis) Indexed codebase: 13 recognized source files, 90 KB. - CSS (language) — detected in the code - Flask (technology) — detected in the code - HTML (language) — detected in the code - JavaScript (language) — detected in the code - OpenAI (technology) — detected in the code - Python (language) — detected in the code ## Codebase structure (from repository index) ### Files (16 of 16) ``` .env.template .gitignore app.py README.md requirements.txt review_generator.py reviews.py static/css/style.css static/js/results.js static/js/search.js templates/index.html templates/results.html tiktok_search.py transcribing_utils.py utils.py youtube_search.py ``` ### Dependencies - requirements.txt: Flask@==3.0.0, google-api-python-client@==2.116.0, google-auth@==2.27.0, google-auth-oauthlib@==1.2.0, httpx@==0.27.2, langdetect@==1.0.9, openai@==1.56.1, python-dotenv@==1.0.0, yt-dlp@==2025.1.26 ### Recent commits (newest first) - Update README.md - hotfix - setup instructions - fixed re-integrating the summarizing - tiktok reviews fixed! - tiktok summarizing works! - tiktok draft 1 - review generation! - review summary - reworked results page - youtube and transcribing fixed - ui draft 1 - first draft loading from yt and transcribing - env stuff - google review scraping - Merge remote-tracking branch 'origin/main' - Initial commit - Initial commit: Basic Flask application setup ## Key source files (fetched from GitHub, selected and truncated for size) ### requirements.txt ``` Flask==3.0.0 google-api-python-client==2.116.0 google-auth-oauthlib==1.2.0 google-auth==2.27.0 python-dotenv==1.0.0 yt-dlp==2025.1.26 openai==1.56.1 httpx==0.27.2 langdetect==1.0.9 ``` ### app.py ```python from flask import Flask, render_template, request, jsonify from youtube_search import search_videos as search_youtube_videos, download_audio as download_youtube_audio, get_video_dir as get_youtube_video_dir, save_video_data from tiktok_search import search_videos as search_tiktok_videos, download_audio as download_tiktok_audio, get_video_dir as get_tiktok_video_dir from transcribing_utils import transcribe_audio from review_generator import process_query_directory from reviews import get_product_reviews, get_review_summary import logging import json from utils import get_query_dir # Set up logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = Flask(__name__) @app.route('/') def home(): return render_template('index.html') @app.route('/search') def search(): query = request.args.get('product') if not query: logger.warning('No product query provided') error_response = {'error': 'No product query provided'} if request.headers.get('X-Requested-With') == 'XMLHttpRequest': return jsonify(error_response) return render_template('results.html', query='', results=error_response) try: # Get product reviews from existing sources logger.info(f'Searching for product: {query}') results = get_product_reviews(query) # Process image URLs if results.get('img_urls'): logger.info(f'Found {len(results["img_urls"])} images') valid_urls = [url for url in results['img_urls'] if url.startswith(('http://', 'https://'))] results['img_urls'] = valid_urls logger.info(f'Found {len(valid_urls)} valid images') summary_result = get_review_summary(query, results) if summary_result['error']: logger.warning(f'Error getting review summary: {summary_result["error"]}') else: logger.info('Successfully retrieved review summary') logger.info(f'Review summary: {summary_result["summary"]}') results['summary'] = summary_result["summary"] # Create directory for this search query query_dir = get_query_dir(query) # Start the YouTube search process youtube_videos = search_youtube_videos(query, max_results=4, query_dir=query_dir) youtube_reviews = [] # Start the TikTok search process tiktok_videos = search_tiktok_videos(query, max_results=8, query_dir=query_dir) tiktok_reviews = [] # Process YouTube videos if youtube_videos: for video in youtube_videos: try: logger.info(f'Processing YouTube video: {video["title"]} (ID: {video["video_id"]})') # Download audio and transcribe with Whisper audio_path = download_youtube_audio(video['video_url'], video['video_id'], video['title'], query_dir) logger.info(f'Audio download result: {"Success" if audio_path else "Failed"}') if audio_path: logger.info(f'Audio downloaded successfully: {audio_path}') whisper_result = transcribe_audio(audio_path) if whisper_result['available']: logger.info('Whisper transcription successful') # Save video data video_dir = get_youtube_video_dir(video['video_id'], video['title'], query_dir) save_video_data( video_dir=video_dir, video_info={ 'title': video['title'], 'description': video.get('description', ''), 'channel': video['channel'], 'publishedAt': video.get('published_at', ''), 'platform': 'youtube', 'statistics': { 'viewCount': str(video.get('view_count', 0)), 'likeCount': str(video.get('like_count', 0)), 'commentCount': str(video.get('comment_count', 0)) }, 'video_url': video.get('video_url', ''), }, transcript=whisper_result['transcript'] ) youtube_reviews.append({ 'title': video['title'], 'url': video['video_url'], 'transcript': whisper_result['transcript'], 'platform': 'youtube', 'channel': video['channel'] }) except Exception as e: logger.error(f'Error processing YouTube video: {str(e)}') # Process TikTok videos if tiktok_videos: for video in tiktok_videos: try: logger.info(f'Processing TikTok video: {video["title"]} (ID: {video["video_id"]})') # Download audio and transcribe with Whisper audio_path = download_tiktok_audio(video['video_url'], video['video_id'], video['title'], query_dir) logger.info(f'Audio download result: {"Success" if audio_path else "Failed"}') if audio_path: logger.info(f'Audio downloaded successfully: {audio_path}') whisper_result = transcribe_audio(audio_path) if whi [truncated — 3231 more characters] ``` ### utils.py ```python from pathlib import Path from datetime import datetime DOWNLOADS_DIR = Path('downloads') def get_query_dir(query): """ Get the directory for a specific search query's results. Args: query (str): Search query Returns: Path: Path to the query's directory """ # Clean the query to make it filesystem-friendly clean_query = ''.join(c for c in query if c.isalnum() or c in ' -_')[:50].strip() # Add timestamp to make each search unique timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') query_dir = DOWNLOADS_DIR / f"{clean_query}-{timestamp}" query_dir.mkdir(parents=True, exist_ok=True) return query_dir ``` ### transcribing_utils.py ```python """ Utility functions for transcribing audio using OpenAI's Whisper API. """ import os import logging from pathlib import Path from openai import OpenAI from langdetect import detect, DetectorFactory # Set seed for consistent language detection DetectorFactory.seed = 0 # Set up logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def is_english_text(text): """ Check if the given text is in English. Args: text (str): Text to check Returns: bool: True if text is in English, False otherwise """ try: detected_language = detect(text) logger.info(f"Detected language: {detected_language}") return detected_language == 'en' except: return False def transcribe_audio(audio_path): """ Transcribe audio using OpenAI's Whisper API. Args: audio_path (str): Path to the audio file Returns: dict: Dictionary containing transcription info { 'available': bool, 'transcript': str or None, 'transcript_path': str or None, 'error': str or None } """ try: api_key = os.getenv("OPENAI_API_KEY") if not api_key: return { 'available': False, 'transcript': None, 'transcript_path': None, 'error': "OPENAI_API_KEY not found in environment variables" } client = OpenAI(api_key=api_key) logger.info(f"Transcribing audio: {audio_path}") # Transcribe the audio with open(audio_path, "rb") as audio_file: response = client.audio.transcriptions.create( model="whisper-1", file=audio_file, response_format="text" ) transcript = str(response) if response else '' logger.info(f"Raw transcription response: {transcript[:200]}...") # Check if transcript is in English if not is_english_text(transcript): return { 'available': False, 'transcript': None, 'transcript_path': None, 'error': 'Transcript is not in English' } # Save transcript to file transcript_path = Path(audio_path).with_suffix('.txt') with open(transcript_path, 'w', encoding='utf-8') as f: f.write(transcript) return { 'available': True, 'transcript': transcript, 'transcript_path': str(transcript_path), 'error': None } except Exception as e: logger.error(f"Whisper transcription error: {str(e)}", exc_info=True) return { 'available': False, 'transcript': None, 'transcript_path': None, 'error': f"Whisper transcription failed: {str(e)}" } def save_video_data(video_dir, video_info, transcript): """ Save video information and transcript to a JSON file. Args: video_dir (Path): Directory to save the video data video_info (dict): Dictionary containing video information transcript (str): Video transcript """ import json try: data = { 'video_info': video_info, 'transcript': transcript } output_path = video_dir / 'video_data.json' with open(output_path, 'w', encoding='utf-8') as f: json.dump(data, f, indent=4, ensure_ascii=False) except Exception as e: logger.error(f"Error saving video data: {str(e)}", exc_info=True) ``` ### reviews.py ```python import requests from typing import Dict, Any from dotenv import load_dotenv import os import logging from openai import OpenAI # Set up logger logger = logging.getLogger(__name__) load_dotenv(override=True) def get_product_reviews(query: str, pages: int = 2) -> Dict[str, Any]: """ Fetch and analyze product reviews from Google Shopping. Args: query (str): The product search query pages (int): Number of pages to scrape (default: 2) Returns: Dict containing: - total_reviews: Total number of reviews - weighted_avg_rating: Average rating weighted by review count - img_urls: List of URLs of product images - error: Error message if any """ try: payload = { 'source': 'google_shopping_search', 'domain': 'com', 'query': query, 'pages': pages, 'parse': True, 'context': [ {'key': 'sort_by', 'value': 'r'}, ], } response = requests.request( 'POST', 'https://realtime.oxylabs.io/v1/queries', auth=(os.getenv('OXYLABS_USER'), os.getenv('OXYLABS_PASS')), json=payload, timeout=40 ) response.raise_for_status() data = response.json() # Log the full response structure print("\nFull API Response:") import json print(json.dumps(data, indent=2)) print("\n") total_reviews = 0 weighted_rating_sum = 0 img_urls = [] for result in data.get("results", []): print(f"\nProcessing result:") print(json.dumps(result, indent=2)) content = result.get("content", {}) print(f"\nContent section:") print(json.dumps(content, indent=2)) organic_results = content.get("results", {}).get("organic", []) print(f"\nFound {len(organic_results)} organic results") for idx, product in enumerate(organic_results): print(f"\nProduct {idx + 1}:") print(json.dumps(product, indent=2)) # Try multiple possible image fields img_url = None if product.get("thumbnail"): img_url = product.get("thumbnail") print(f"Found thumbnail URL: {img_url}") elif product.get("image"): img_url = product.get("image") print(f"Found image URL: {img_url}") elif product.get("images"): images = product.get("images") if isinstance(images, list) and images: img_url = images[0] print(f"Found URL in images array: {img_url}") if img_url: # Analyze URL structure print(f"URL Analysis:") print(f"- Full URL: {img_url}") print(f"- Starts with http/https: {img_url.startswith(('http://', 'https://'))}") print(f"- URL length: {len(img_url)}") print(f"- URL parts: {img_url.split('/')}") if img_url.startswith(('http://', 'https://')): img_urls.append(img_url) print("✓ URL added to valid images list") else: print("✗ URL rejected - invalid protocol") else: print(f"No image found in any field for product {idx + 1}") rating = product.get("rating") reviews_count = product.get("reviews_count") if rating is not None and reviews_count is not None: print(f"Rating: {rating}, Reviews: {reviews_count}") total_reviews += reviews_count weighted_rating_sum += rating * reviews_count else: print(f"Missing rating or reviews. Rating: {rating}, Reviews: {reviews_count}") weighted_avg_rating = round(weighted_rating_sum / total_reviews, 2) if total_reviews > 0 else None return { "total_reviews": total_reviews, "weighted_avg_rating": weighted_avg_rating, "img_urls": img_urls, "error": None } except requests.RequestException as e: return { "total_reviews": 0, "weighted_avg_rating": None, "img_urls": [], "error": f"Error fetching reviews: {str(e)}" } except Exception as e: return { "total_reviews": 0, "weighted_avg_rating": None, "img_urls": [], "error": f"Unexpected error: {str(e)}" } def get_review_summary(query: str, results: Dict[str, Any]) -> Dict[str, Any]: """ Get a summary of reviews for a product using Perplexity AI. Args: query (str): The product to get review summary for results (Dict[str, Any]): The results of the product reviews Returns: Dict containing: summary (str): Summary of reviews error (str): Error message if any, None otherwise """ try: api_key = os.getenv('OPENAI_API_KEY') if not api_key: return None, "OPENAI_API_KEY not found in environment variables" client = OpenAI(api_key=api_key) messages = [ { "role": "system", "content": ( "You are a review aggregator artificial intelligence assistant and you need to " "help the user summarize reviews for a product they queried from across the internet. " "Do not provide citations for any website in your response." ), }, { [truncated — 862 more characters] ``` ### review_generator.py ```python import os import json from pathlib import Path import openai from dotenv import load_dotenv load_dotenv(override=True) # Initialize OpenAI client openai.api_key = os.getenv('OPENAI_API_KEY') def generate_review(video_data, transcript): """ Generate a review from video data and transcript using OpenAI. Args: video_data (dict): Video metadata including title, description, etc. transcript (str): Video transcript text Returns: dict: Generated review with rating """ # Get platform from video info or default to YouTube platform = video_data.get('platform', 'YouTube') # Create a prompt that includes key video information prompt = f"""Based on this {platform} review: Title: {video_data['title']} Channel: {video_data['channel']} Description: {video_data.get('description', 'Not available')} Transcript: {transcript} Write a customer review as if you personally used the product. The review should: 1. Be 1-4 sentences long 2. Include specific details about the product's features and performance 3. Give a rating out of 5 stars 4. Focus on your direct experience with the product 5. NOT mention that this is based on a video or reference any reviewers 6. Be written in first person about your hands-on experience 7. Include both pros and cons Respond with a JSON object in this exact format, with no deviations: {{ "review_text": "Your 1-4 sentence review here", "rating": "Your rating out of 5 stars here" }} Make sure to: - Use proper JSON formatting with double quotes - Make the rating a number between 1 and 5 - Write as a customer who bought and used the product - Never mention YouTube, videos, or reviewers - Focus on personal experience with the product""" try: response = openai.chat.completions.create( model="gpt-4-turbo-preview", messages=[ {"role": "system", "content": "You are an expert at distilling product reviews into concise, authentic summaries."}, {"role": "user", "content": prompt} ], response_format={"type": "json_object"}, temperature=0.6 ) content = response.choices[0].message.content print("Review generated: ", content) # Try to extract JSON even if it's embedded in other text try: # First try direct JSON parsing review_data = json.loads(content) except json.JSONDecodeError: # If that fails, try to find JSON-like structure import re json_pattern = r'\{[^{}]*\}' # Simple pattern to match JSON object matches = re.findall(json_pattern, content) if matches: try: review_data = json.loads(matches[0]) except json.JSONDecodeError: raise Exception("Could not parse embedded JSON") else: raise Exception("No JSON-like structure found in response") # Validate the structure if not isinstance(review_data, dict): raise Exception("Response is not a dictionary") if 'review_text' not in review_data or 'rating' not in review_data: raise Exception("Missing required fields in response") # Convert rating to float/int if it's a string if isinstance(review_data['rating'], str): try: review_data['rating'] = float(review_data['rating']) except ValueError: raise Exception("Rating must be a number") if not isinstance(review_data['rating'], (int, float)) or not 1 <= float(review_data['rating']) <= 5: raise Exception("Invalid rating value") if not isinstance(review_data['review_text'], str) or len(review_data['review_text']) < 10: raise Exception("Invalid review text") return review_data except Exception as e: print(f"Error generating review: {str(e)}") return None def process_query_directory(query_dir): """ Process all videos in a query directory and generate reviews. Args: query_dir (str): Path to query directory Returns: list: List of generated reviews """ print(f"\nProcessing directory: {query_dir}") query_path = Path(query_dir) reviews = [] # Process each video directory for video_dir in query_path.iterdir(): print(f"\nChecking directory: {video_dir}") if not video_dir.is_dir(): print("Not a directory, skipping...") continue video_data_path = video_dir / 'video_data.json' print(f"Looking for video data at: {video_data_path}") if not video_data_path.exists(): print("No video data found, skipping...") continue # Load video data try: print("Loading video data...") with open(video_data_path, 'r') as f: video_data = json.load(f) print(f"Video title: {video_data['video_info']['title']}") print(f"Transcript length: {len(video_data.get('transcript', ''))} chars") # Generate review print("Generating review...") review = generate_review( video_data=video_data['video_info'], transcript=video_data.get('transcript', '') ) print("Review generated successfully!") if review and isinstance(review, dict) and 'review_text' in review and 'rating' in review: try: reviews.append({ 'video_title': video_data['video_info']['title'], 'channel': video_data['video_info']['channel'], 'review_text': review['review_text'], 'rating': review['rating'], [truncated — 1664 more characters] ``` ### tiktok_search.py ```python import logging import os import requests from datetime import datetime from pathlib import Path from dotenv import load_dotenv from transcribing_utils import transcribe_audio, is_english_text, save_video_data import yt_dlp # Set up logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Create downloads directory if it doesn't exist from utils import get_query_dir, DOWNLOADS_DIR DOWNLOADS_DIR.mkdir(exist_ok=True) def sanitize_filename(filename, max_length=50): """ Create a safe filename by removing invalid characters and limiting length. Args: filename (str): Original filename max_length (int): Maximum length for the filename Returns: str: Sanitized filename """ # Remove invalid characters safe_name = "".join(c if c.isalnum() or c in (' ', '-', '_') else '_' for c in filename) # Remove multiple consecutive underscores safe_name = '_'.join(filter(None, safe_name.split('_'))) # Limit length while preserving extension if present name_parts = safe_name.rsplit('.', 1) if len(name_parts) > 1: name, ext = name_parts return f"{name[:max_length-len(ext)-1]}.{ext}" return safe_name[:max_length] def get_video_dir(video_id, title, query_dir, create=True): """ Get the directory for a specific video's files. Args: video_id (str): TikTok video ID title (str): Video title (used for folder name) query_dir (Path): Directory for the search query results create (bool): Whether to create the directory if it doesn't exist Returns: Path: Path to the video's directory """ safe_title = sanitize_filename(title) video_dir = query_dir / f"{safe_title}_{video_id}" if create: video_dir.mkdir(exist_ok=True) return video_dir def download_audio(video_url, video_id, title, query_dir): """ Download audio from a TikTok video. Args: video_url (str): TikTok video URL video_id (str): TikTok video ID title (str): Video title query_dir (Path): Directory for the search query results Returns: str: Path to the downloaded audio file or None if file is too large """ video_dir = get_video_dir(video_id, title, query_dir) audio_path = video_dir / 'audio.mp3' if audio_path.exists(): logger.info(f"Audio already exists for video {video_id}") return str(audio_path) def try_api_download(): try: # Get video metadata using EnsembleData API api_key = os.getenv('ENSEMBLEDDATA_API_KEY') root = "https://ensembledata.com/apis" endpoint = "/tt/video/details" params = { "aweme_id": video_id, "token": api_key } response = requests.get(root + endpoint, params=params) response.raise_for_status() video_data = response.json() if 'data' in video_data and 'video' in video_data['data']: video_info = video_data['data']['video'] if 'play_addr' in video_info and 'url_list' in video_info['play_addr']: direct_url = video_info['play_addr']['url_list'][0] # Download video using requests video_response = requests.get(direct_url, stream=True) video_response.raise_for_status() temp_video = video_dir / 'temp.mp4' with open(temp_video, 'wb') as f: for chunk in video_response.iter_content(chunk_size=8192): if chunk: f.write(chunk) # Convert to audio using ffmpeg import subprocess subprocess.run([ 'ffmpeg', '-i', str(temp_video), '-vn', '-acodec', 'libmp3lame', '-q:a', '4', str(audio_path) ], check=True, capture_output=True) # Clean up temp file temp_video.unlink() return str(audio_path) return None except Exception as e: logger.error(f"API download failed: {str(e)}") return None def try_yt_dlp_download(): try: ydl_opts = { 'format': 'bestaudio/best', 'postprocessors': [{ 'key': 'FFmpegExtractAudio', 'preferredcodec': 'mp3', 'preferredquality': '192', }], 'outtmpl': str(video_dir / 'audio.%(ext)s'), 'max_filesize': 10000000, # 10MB limit 'quiet': True, 'no_warnings': True, 'nocheckcertificate': True, 'no_check_certificate': True, 'http_headers': { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8', 'Accept-Language': 'en-US,en;q=0.5', 'Sec-Fetch-Mode': 'navigate', 'Sec-Fetch-Site': 'none', 'Sec-Fetch-Dest': 'document', 'Cookie': 'tt_webid_v2=1234567890123456789' } } with yt_dlp.YoutubeDL(ydl_opts) as ydl: ydl.download([video_url]) return str(audio_path) except Exception as e: logger.error(f"yt-dlp download failed: {str(e)}") return None # Try API downlo [truncated — 5643 more characters] ``` ### youtube_search.py ```python from googleapiclient.discovery import build from google_auth_oauthlib.flow import InstalledAppFlow from google.auth.transport.requests import Request from google.oauth2.credentials import Credentials from dotenv import load_dotenv import os import logging import yt_dlp from pathlib import Path import html import pickle from transcribing_utils import transcribe_audio, is_english_text, save_video_data from datetime import datetime # Set up logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # OAuth2 configuration SCOPES = [ 'https://www.googleapis.com/auth/youtube.force-ssl', 'https://www.googleapis.com/auth/youtube.readonly' ] TOKEN_FILE = 'token.pickle' CREDENTIALS_FILE = 'youtube_client_secrets.json' # Create downloads directory if it doesn't exist from utils import get_query_dir, DOWNLOADS_DIR DOWNLOADS_DIR.mkdir(exist_ok=True) # Load environment variables load_dotenv(override=True) def get_video_details(youtube, video_id): """ Get detailed information about a specific video. Args: youtube: Authenticated YouTube service instance video_id (str): YouTube video ID Returns: dict: Video details including statistics and description """ try: request = youtube.videos().list( part="snippet,statistics,contentDetails", id=video_id ) response = request.execute() if response['items']: video = response['items'][0] return { 'title': video['snippet'].get('title', ''), 'description': video['snippet'].get('description', ''), 'channel': video['snippet'].get('channelTitle', ''), 'publishedAt': video['snippet'].get('publishedAt', ''), 'statistics': video['statistics'], } return None except Exception as e: logger.error(f"Error getting video details: {str(e)}") return None """ Search for YouTube videos using the YouTube Data API. Args: query (str): Search query max_results (int): Maximum number of results to return (default: 2) query_dir (Path): Directory to save video data (default: None) Returns: list: List of video information dictionaries """ def search_videos(query, max_results=2, query_dir=None): # If no query_dir provided, create a default one if query_dir is None: query_dir = get_query_dir(query) # Get both API keys api_key = os.getenv("YOUTUBE_API_KEY") api_key_2 = os.getenv("YOUTUBE_API_KEY_2") for current_key in [api_key, api_key_2]: if not current_key: continue logger.info(f"Trying API key: {current_key[-10:]}") try: # Create YouTube API client with cache disabled youtube = build('youtube', 'v3', developerKey=current_key, cache_discovery=False, static_discovery=False) review_query = f"{query} review" # First get video IDs search_response = youtube.search().list( q=review_query, part='id', # Only get IDs, not snippets maxResults=max_results, type='video', fields='items(id/videoId)' # Only get video IDs to minimize response size ).execute() videos = [] video_ids = [item['id']['videoId'] for item in search_response.get('items', [])] if not video_ids: return [] logger.info(f'Found {len(video_ids)} video IDs, fetching full details...') # Get full video details in a single request video_response = youtube.videos().list( part='snippet,statistics,contentDetails', id=','.join(video_ids), fields='items(id,snippet(title,description,channelTitle,publishedAt,thumbnails/high/url),statistics,contentDetails/duration)' ).execute() logger.info(f'Retrieved details for {len(video_response.get("items", []))} videos') for video_details in video_response.get('items', []): snippet = video_details['snippet'] stats = video_details['statistics'] duration = video_details['contentDetails']['duration'] title = snippet['title'] video_info = { 'title': title, 'description': snippet.get('description', ''), # Get full description from video details 'channel': snippet['channelTitle'], 'published_at': snippet['publishedAt'], 'thumbnail': snippet['thumbnails']['high']['url'], 'video_id': video_details['id'], 'video_url': f'https://www.youtube.com/watch?v={video_details["id"]}', 'view_count': int(stats.get('viewCount', 0)), 'like_count': int(stats.get('likeCount', 0)), 'comment_count': int(stats.get('commentCount', 0)), 'duration': duration, } logger.info(f'Processed video info:') logger.info(f' - Title: {video_info["title"]}') logger.info(f' - Channel: {video_info["channel"]}') logger.info(f' - Duration: {video_info["duration"]}') logger.info(f' - Views: {video_info["view_count"]}') logger.info(f' - URL: {video_info["video_url"]}') videos.append(video_info) # Print video details print(f"\nVideo found:") print(f"Title: {video_info['title']}") print(f"Channel: {video_info['channel']}") print(f"Views: {video_info['view_count']}") print(f"Likes: {video_info['like_count']}") [truncated — 11426 more characters] ``` ### static/js/results.js ```javascript document.addEventListener('DOMContentLoaded', function() { // Set up image loading handlers const productImage = document.querySelector('.product-image'); if (productImage) { // Handle successful image load productImage.addEventListener('load', function() { this.classList.add('loaded'); }); // Handle image load error productImage.addEventListener('error', function() { this.style.display = 'none'; const wrapper = document.querySelector('.single-image-wrapper'); if (wrapper) { wrapper.innerHTML = `
No images available
`; } }); // If image is already loaded (from cache) if (productImage.complete) { productImage.classList.add('loaded'); } } // Handle form submission const searchForm = document.querySelector('.search-form'); const loadingOverlay = document.getElementById('loading-overlay'); if (searchForm) { searchForm.addEventListener('submit', function() { loadingOverlay.classList.add('visible'); }); } }); ``` [4 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]