from __future__ import annotations import json import re import os from typing import Any, Optional import httpx from bs4 import BeautifulSoup def can_handle(host: str) -> bool: return "youporn.com" in host.lower() def _best_image_url(img: Any) -> Optional[str]: """Extract the best image URL from an img element, checking multiple lazy-load attributes.""" if img is None: return None # Check common lazy-loading attributes in order of preference # Youporn uses: data-poster (primary image thumbnail) video_fallback = None # Store first video URL as fallback for attr in ("data-poster", "data-src", "data-original", "data-lazy", "data-image", "src", "data-mediabook"): value = img.get(attr) if not value: continue url = str(value).strip() if not url or "data:image" in url: continue # Check if it's a video file url_lower = url.lower() is_video = any(url_lower.endswith(ext) or f'{ext}/' in url_lower or f'{ext}?' in url_lower for ext in ('.mp4', '.webm', '.m3u8', '.ts')) if is_video: # Save as fallback but keep looking for image if not video_fallback: video_fallback = url continue # Found a valid image URL! return url # If no image URL found, use video as fallback (better than null) return video_fallback def get_categories() -> list[dict]: try: current_dir = os.path.dirname(os.path.abspath(__file__)) json_path = os.path.join(current_dir, "categories.json") with open(json_path, 'r', encoding='utf-8') as f: return json.load(f) except Exception: return [] async def fetch_html(url: str) -> str: headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8", # Mobile cookie might be needed? Usually desktop is safer for parsing. # YouPorn might not strict check 'platform=pc' but good practice if structure varies } async with httpx.AsyncClient( follow_redirects=True, timeout=httpx.Timeout(20.0, connect=20.0), headers=headers, ) as client: resp = await client.get(url) resp.raise_for_status() return resp.text async def _resolve_proxy_url(proxy_url: str) -> list[dict]: """ Resolve a YouPorn proxy URL (e.g., /media/mp4/?s=...) to actual CDN streams. Returns a list of stream objects with quality, url, format. """ try: headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Accept": "application/json", } async with httpx.AsyncClient(headers=headers, timeout=10.0, follow_redirects=True) as client: resp = await client.get(proxy_url) if resp.status_code != 200: return [] data = resp.json() if isinstance(data, list): streams = [] for item in data: quality = item.get("quality") video_url = item.get("videoUrl") fmt = item.get("format", "mp4") if video_url: # Convert quality to string if isinstance(quality, int): quality = str(quality) streams.append({ "quality": quality if quality else "unknown", "url": video_url, "format": fmt }) return streams except Exception: pass return [] def _extract_video_streams(html: str) -> dict[str, Any]: streams = [] hls_url = None # Find mediaDefinitions in HTML and extract the array efficiently media_defs = [] idx = html.find('mediaDefinitions') if idx != -1: # Find the opening bracket after mediaDefinitions chunk_start = idx chunk = html[chunk_start:chunk_start + 10000] # Limit search space array_start_idx = chunk.find('[') if array_start_idx != -1: # Count brackets to find the matching closing bracket depth = 0 array_content = None for i, char in enumerate(chunk[array_start_idx:], array_start_idx): if char == '[': depth += 1 elif char == ']': depth -= 1 if depth == 0: array_content = chunk[array_start_idx:i + 1] break if array_content: try: media_defs = json.loads(array_content) except Exception: pass # Process mediaDefinitions if media_defs: for md in media_defs: video_url = md.get("videoUrl") if not video_url: continue # Skip poster/thumbnail images if video_url.endswith('.jpg') or video_url.endswith('.jpeg') or video_url.endswith('.png'): continue fmt = md.get("format") # mp4, hls quality = md.get("quality") # 720p, 1080p, etc if isinstance(quality, list): quality = str(quality[0]) # Check format if fmt == "hls" or ".m3u8" in video_url: # Extract quality from URL or metadata parsed_quality = "adaptive" if quality: if isinstance(quality, str) and quality.isdigit(): parsed_quality = f"{quality}p" else: parsed_quality = str(quality) # Try regex on URL if quality is not specific if not parsed_quality or parsed_quality == "adaptive": # Try finding /1080P/ or similar patterns m_q = re.search(r'/(\d{3,4})[pP]?/', video_url) if not m_q: m_q = re.search(r'(\d{3,4})[pP]_', video_url) if m_q: parsed_quality = f"{m_q.group(1)}p" hls_url = video_url streams.append({ "quality": parsed_quality, "url": video_url, "format": "hls" }) elif fmt == "mp4" or ".mp4" in video_url: streams.append({ "quality": str(quality) if quality else "unknown", "url": video_url, "format": "mp4" }) # Generic fallback if no JSON found: check for