from __future__ import annotations import json import re import os from typing import Any, Optional from bs4 import BeautifulSoup from app.core import pool def can_handle(host: str) -> bool: return "pornwex.tv" in host.lower() 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", "Accept-Language": "en-US,en;q=0.9", } session = await pool.get_session() async with session.get(url, headers=headers, allow_redirects=True) as resp: resp.raise_for_status() return await resp.text() def _text(el: Any) -> Optional[str]: if el is None: return None t = getattr(el, "get_text", None) if callable(t): return t(strip=True) or None return None def _best_image_url(img: Any) -> Optional[str]: if img is None: return None for k in ("data-src", "data-original", "data-lazy", "src"): v = img.get(k) if v and str(v).strip(): url = str(v).strip() if url.startswith("//"): return f"https:{url}" return url return None def _resolve_kt_url(raw: str) -> str: """ kt_player stores video URLs as 'function/0/' where the real_url is the /get_file/ endpoint. Strip the prefix and return the bare URL. If the value is already a plain URL, return as-is. """ # Strip 'function//' prefix that kt_player uses m = re.match(r'^function/\d+/(https?://.+)$', raw.strip()) if m: return m.group(1) if raw.startswith("//"): return f"https:{raw}" return raw async def _follow_to_direct_url(get_file_url: str) -> str: """ Follow the /get_file/ redirect chain to obtain the final CDN URL (typically remote_control.php on a CDN host). """ headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36", "Referer": "https://www.pornwex.tv/", } try: session = await pool.get_session() async with session.get( get_file_url, headers=headers, allow_redirects=True, ) as resp: # The final URL after redirects is the signed CDN link final = str(resp.url) # Only accept if it looks like a real media URL if any(x in final for x in ("remote_control", ".mp4", ".m3u8")): return final except Exception: pass # Fallback: return the get_file URL itself return get_file_url def parse_page(html: str, url: str) -> dict[str, Any]: soup = BeautifulSoup(html, "lxml") # Title title_node = soup.select_one("h1") title = _text(title_node) or _text(soup.find("title")) or "Unknown Video" # Clean title for suffix in (" | PornWex", " - PornWex"): if title.endswith(suffix): title = title[: -len(suffix)] # Description desc_node = soup.find("meta", property="og:description") description = desc_node.get("content") if desc_node else None # Thumbnail (og:image) og_img = soup.find("meta", property="og:image") thumbnail = og_img.get("content") if og_img else None if thumbnail and thumbnail.startswith("//"): thumbnail = f"https:{thumbnail}" # ------------------------------------------------------------------------- # Video Stream URL extraction # ------------------------------------------------------------------------- # PornWex uses kt_player. The player config object in the HTML contains: # # video_url: 'function/0/https://www.pornwex.tv/get_file//video.mp4/' # # The "function/0/" prefix is a kt_player directive — strip it to get the # /get_file/ URL, then follow its redirect chain to obtain the real signed # CDN URL (remote_control.php on a CDN host such as g02s01.pornwex.tv). # ------------------------------------------------------------------------- video_url = None streams = [] # Patterns for kt_player config keys that hold video source URLs. # They can be either plain https:// or prefixed with 'function//'. kt_url_keys = r'(?:video_url|video_url_text|video_alt_url|video_alt_url2)' kt_patterns = [ # single-quoted value rf"{kt_url_keys}\s*[:\=]\s*'([^']+)'", # double-quoted value rf'{kt_url_keys}\s*[:\=]\s*"([^"]+)"', ] raw_urls: list[tuple[str, str]] = [] # (key_name, raw_value) for pattern in kt_patterns: for m in re.finditer(pattern, html): raw = m.group(1).strip() key = re.match(r'(\w+)', m.group(0)).group(1) if raw and (raw, key) not in [(r, k) for r, k in raw_urls]: raw_urls.append((key, raw)) # Map key to quality hint: video_url = primary, alt variants = fallback quality_map = { "video_url": "default", "video_url_text": "default", "video_alt_url": "alt", "video_alt_url2": "alt2", } for key, raw in raw_urls: resolved = _resolve_kt_url(raw) if not resolved: continue # Must look like a /get_file/ or https:// media URL if "get_file" not in resolved and not re.search(r'\.mp4|\.m3u8', resolved): continue qual = quality_map.get(key, "default") if resolved not in [s["url"] for s in streams]: streams.append({ "url": resolved, "quality": qual, "format": "hls" if ".m3u8" in resolved else "mp4", "_needs_redirect": True, }) # Fallback 1: