322 lines
11 KiB
Python
322 lines
11 KiB
Python
"""Search-related API endpoints."""
|
|
|
|
import os
|
|
import re
|
|
import json
|
|
import yt_dlp
|
|
from pathlib import Path
|
|
from flask import Blueprint, request
|
|
from utils import make_response, make_error_response
|
|
|
|
search_bp = Blueprint('search', __name__, url_prefix='/api')
|
|
|
|
# Recent searches storage (use CONFIG_DIR env var for persistence in Docker)
|
|
config_dir = os.environ.get('CONFIG_DIR', str(Path.home() / '.config' / 'youtube_cli'))
|
|
recent_searches_file = Path(config_dir) / "recent_searches.json"
|
|
|
|
# Banned search terms (loaded from file)
|
|
_banned_terms_file = Path(__file__).parent.parent / "banned_terms.txt"
|
|
_banned_terms = []
|
|
|
|
|
|
def _load_banned_terms():
|
|
"""Load banned search terms from file."""
|
|
global _banned_terms
|
|
if _banned_terms_file.exists():
|
|
try:
|
|
with open(_banned_terms_file, 'r') as f:
|
|
terms = []
|
|
for line in f:
|
|
line = line.strip()
|
|
if line and not line.startswith('#'):
|
|
terms.append(line.lower())
|
|
_banned_terms = terms
|
|
except Exception:
|
|
_banned_terms = []
|
|
|
|
|
|
def _levenshtein(s1: str, s2: str) -> int:
|
|
"""Calculate Levenshtein distance between two strings."""
|
|
if len(s1) < len(s2):
|
|
return _levenshtein(s2, s1)
|
|
if len(s2) == 0:
|
|
return len(s1)
|
|
prev_row = range(len(s2) + 1)
|
|
for i, c1 in enumerate(s1):
|
|
curr_row = [i + 1]
|
|
for j, c2 in enumerate(s2):
|
|
insertions = prev_row[j + 1] + 1
|
|
deletions = curr_row[j] + 1
|
|
substitutions = prev_row[j] + (c1 != c2)
|
|
curr_row.append(min(insertions, deletions, substitutions))
|
|
prev_row = curr_row
|
|
return prev_row[-1]
|
|
|
|
|
|
def _is_banned(query: str) -> bool:
|
|
"""Check if a search query contains any banned terms (exact or fuzzy match)."""
|
|
if not _banned_terms:
|
|
_load_banned_terms()
|
|
query_lower = query.lower()
|
|
|
|
# Exact match check
|
|
for term in _banned_terms:
|
|
if term in query_lower:
|
|
return True
|
|
|
|
# Fuzzy match check for words in the query
|
|
query_words = query_lower.split()
|
|
for word in query_words:
|
|
# Skip very short words (3 chars or less)
|
|
if len(word) <= 3:
|
|
continue
|
|
for term in _banned_terms:
|
|
# Skip fuzzy matching for very short banned terms (too many false positives)
|
|
if len(term) <= 3:
|
|
continue
|
|
# Only fuzzy match for terms with similar length
|
|
if abs(len(word) - len(term)) > 2:
|
|
continue
|
|
# Allow up to 2 character differences for terms 4+ chars
|
|
threshold = 2
|
|
if _levenshtein(word, term) <= threshold:
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
# Load banned terms at startup
|
|
_load_banned_terms()
|
|
|
|
|
|
def _load_recent_searches():
|
|
"""Load recent searches from file."""
|
|
if recent_searches_file.exists():
|
|
try:
|
|
with open(recent_searches_file, 'r') as f:
|
|
return json.load(f)
|
|
except Exception:
|
|
return []
|
|
return []
|
|
|
|
|
|
def _save_recent_searches(searches):
|
|
"""Save recent searches to file."""
|
|
recent_searches_file.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(recent_searches_file, 'w') as f:
|
|
json.dump(searches, f, indent=2)
|
|
|
|
|
|
@search_bp.route('/search', methods=['GET'])
|
|
def search():
|
|
"""Search for YouTube videos using yt-dlp Python API."""
|
|
try:
|
|
query = request.args.get('q', '').strip()
|
|
page = int(request.args.get('page', 1))
|
|
limit = int(request.args.get('limit', 15))
|
|
|
|
if not query:
|
|
return make_error_response("Search query is required", 400)
|
|
|
|
if page < 1:
|
|
return make_error_response("Page must be greater than 0", 400)
|
|
|
|
if _is_banned(query):
|
|
return make_error_response("This search is not allowed. Please choose different keywords.", 400)
|
|
|
|
sanitized_query = re.sub(r'[^\w\s\-\'"\.]+', "", query)
|
|
search_query = f"ytsearch{limit * page}:{sanitized_query}"
|
|
|
|
ydl_opts = {
|
|
'extract_flat': True,
|
|
'no_warnings': True,
|
|
'quiet': True,
|
|
'no_progress': True,
|
|
'user_agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
|
|
}
|
|
|
|
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
|
info = ydl.extract_info(search_query, download=False)
|
|
|
|
if not info:
|
|
return make_response({
|
|
"query": query,
|
|
"page": page,
|
|
"results": [],
|
|
"total": 0,
|
|
"hasMore": False,
|
|
})
|
|
|
|
# Get all entries then slice to the correct page range
|
|
all_entries = info.get('entries', [info]) if isinstance(info, dict) else info
|
|
all_entries = all_entries or []
|
|
start_idx = limit * (page - 1)
|
|
end_idx = limit * page
|
|
entries = all_entries[start_idx:end_idx]
|
|
results = []
|
|
for entry in (entries or []):
|
|
if not entry:
|
|
continue
|
|
vid_id = entry.get('id', '')
|
|
url = entry.get('url', '') or entry.get('webpage_url', '') or f'https://www.youtube.com/watch?v={vid_id}'
|
|
duration = entry.get('duration', 0) or 0
|
|
duration_str = f"{int(duration // 60)}:{int(duration % 60):02d}" if duration else "0:00"
|
|
thumbnail = entry.get("thumbnail", "") or f"https://i.ytimg.com/vi/{vid_id}/hqdefault.jpg"
|
|
results.append({
|
|
"id": vid_id,
|
|
"videoId": vid_id,
|
|
"title": entry.get("title", "Unknown Title"),
|
|
"description": "",
|
|
"thumbnail": thumbnail,
|
|
"url": url,
|
|
"duration": duration_str,
|
|
"views": str(entry.get("view_count", 0) or 0),
|
|
"channel": entry.get("uploader", "Unknown"),
|
|
"isShort": "/shorts/" in url,
|
|
"published": "",
|
|
})
|
|
|
|
# Save to recent searches (only if not banned)
|
|
if not _is_banned(query):
|
|
searches = _load_recent_searches()
|
|
if query not in searches:
|
|
searches.insert(0, query)
|
|
searches = searches[:10]
|
|
_save_recent_searches(searches)
|
|
|
|
return make_response({
|
|
"query": query,
|
|
"page": page,
|
|
"results": results,
|
|
"total": len(results),
|
|
"hasMore": len(results) >= limit,
|
|
})
|
|
|
|
except Exception as e:
|
|
return make_error_response(f"Search failed: {str(e)}", 500)
|
|
|
|
|
|
@search_bp.route('/recent-searches', methods=['GET'])
|
|
def get_recent_searches():
|
|
"""Get list of recent search queries."""
|
|
try:
|
|
searches = _load_recent_searches()
|
|
return make_response(searches)
|
|
except Exception as e:
|
|
return make_error_response(f"Failed to load recent searches: {str(e)}", 500)
|
|
|
|
|
|
@search_bp.route('/recent-searches', methods=['POST'])
|
|
def save_recent_search():
|
|
"""Save a search query to recent searches."""
|
|
try:
|
|
data = request.get_json()
|
|
query = data.get('query', '').strip() if data else ''
|
|
if not query:
|
|
return make_error_response("Search query is required", 400)
|
|
|
|
searches = _load_recent_searches()
|
|
if query in searches:
|
|
searches.remove(query)
|
|
searches.insert(0, query)
|
|
searches = searches[:10]
|
|
_save_recent_searches(searches)
|
|
return make_response(searches)
|
|
except Exception as e:
|
|
return make_error_response(f"Failed to save recent search: {str(e)}", 500)
|
|
|
|
|
|
@search_bp.route('/recent-searches', methods=['DELETE'])
|
|
def clear_recent_searches():
|
|
"""Clear recent search history."""
|
|
try:
|
|
_save_recent_searches([])
|
|
return make_response({"message": "Recent searches cleared"})
|
|
except Exception as e:
|
|
return make_error_response(f"Failed to clear recent searches: {str(e)}", 500)
|
|
|
|
|
|
@search_bp.route('/recent-searches/<path:query>', methods=['DELETE'])
|
|
def remove_recent_search(query):
|
|
"""Remove a single recent search."""
|
|
try:
|
|
searches = _load_recent_searches()
|
|
if query in searches:
|
|
searches.remove(query)
|
|
_save_recent_searches(searches)
|
|
return make_response({"message": f"Search '{query}' removed", "searches": searches})
|
|
except Exception as e:
|
|
return make_error_response(f"Failed to remove recent search: {str(e)}", 500)
|
|
|
|
|
|
@search_bp.route('/video/<video_id>', methods=['GET'])
|
|
def get_video_details(video_id):
|
|
"""Get video details by YouTube video ID."""
|
|
try:
|
|
url = f"https://www.youtube.com/watch?v={video_id}"
|
|
ydl_opts = {
|
|
'dump_single_json': True,
|
|
'no_warnings': True,
|
|
'quiet': True,
|
|
'no_progress': True,
|
|
'write_thumbnail': False,
|
|
}
|
|
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
|
info = ydl.extract_info(url, download=False)
|
|
|
|
if not info:
|
|
return make_error_response("Video not found", 404)
|
|
|
|
return make_response({
|
|
"id": info.get("id", video_id),
|
|
"videoId": info.get("id", video_id),
|
|
"title": info.get("title", "Unknown Title"),
|
|
"description": info.get("description", ""),
|
|
"thumbnail": info.get("thumbnail", ""),
|
|
"url": info.get("webpage_url", url),
|
|
"duration": info.get("duration_string", "0:00"),
|
|
"views": str(info.get("view_count", 0) or 0),
|
|
"channel": info.get("uploader", "Unknown"),
|
|
"isShort": "/shorts/" in url,
|
|
"published": info.get("upload_date", ""),
|
|
})
|
|
except Exception as e:
|
|
return make_error_response(f"Failed to get video details: {str(e)}", 500)
|
|
|
|
|
|
@search_bp.route('/info', methods=['GET'])
|
|
def get_video_info():
|
|
"""Get video info by URL."""
|
|
try:
|
|
url = request.args.get('url', '').strip()
|
|
if not url:
|
|
return make_error_response("URL is required", 400)
|
|
|
|
ydl_opts = {
|
|
'dump_single_json': True,
|
|
'no_warnings': True,
|
|
'quiet': True,
|
|
'no_progress': True,
|
|
'write_thumbnail': False,
|
|
}
|
|
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
|
info = ydl.extract_info(url, download=False)
|
|
|
|
if not info:
|
|
return make_error_response("Video not found", 404)
|
|
|
|
return make_response({
|
|
"id": info.get("id", ""),
|
|
"videoId": info.get("id", ""),
|
|
"title": info.get("title", "Unknown Title"),
|
|
"description": info.get("description", ""),
|
|
"thumbnail": info.get("thumbnail", ""),
|
|
"url": info.get("webpage_url", url),
|
|
"duration": info.get("duration_string", "0:00"),
|
|
"views": str(info.get("view_count", 0) or 0),
|
|
"channel": info.get("uploader", "Unknown"),
|
|
"isShort": "/shorts/" in url,
|
|
"published": info.get("upload_date", ""),
|
|
})
|
|
except Exception as e:
|
|
return make_error_response(f"Failed to get video info: {str(e)}", 500) |