from __future__ import annotations import asyncio import html import json import logging import random import re from dataclasses import dataclass from urllib.parse import urljoin import aiohttp logger = logging.getLogger(__name__) PH_BASE_URL = "https://rt.pornhub.com" SHORTIES_LIST_URL = f"{PH_BASE_URL}/shorties" REQUEST_HEADERS = { "User-Agent": ( "Mozilla/5.0 (X11; Linux x86_64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/124.0.0.0 Safari/537.36" ), "Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8", "Referer": PH_BASE_URL, } _A_HREF_RE = re.compile( r']+href="(?P/view_video\.php\?viewkey=[^"]+)"(?P[^>]*)>(?P.*?)', re.IGNORECASE | re.DOTALL, ) _IMG_URL_RE = re.compile( r'(?:data-mediumthumb|data-path|data-thumb_url|src)="(?Phttps?://[^"]+)"', re.IGNORECASE, ) _TITLE_ATTR_RE = re.compile(r'title="(?P[^"]+)"', re.IGNORECASE) _SCRIPT_LD_JSON_RE = re.compile( r'<script[^>]+type="application/ld\+json"[^>]*>(?P<json>.*?)</script>', re.IGNORECASE | re.DOTALL, ) _MEDIA_DEFS_RE = re.compile(r'"mediaDefinitions"\s*:\s*(\[[^\]]+\])', re.IGNORECASE | re.DOTALL) _VIDEO_URL_RE = re.compile(r'"videoUrl"\s*:\s*"(?P<url>https?:\\?/\\?/[^"]+\.mp4[^"]*)"', re.IGNORECASE) _DURATION_RE = re.compile(r'"video_duration"\s*:\s*"?(?P<duration>\d+)"?', re.IGNORECASE) _VIEWS_RE = re.compile(r'"video_views"\s*:\s*"?(?P<views>[0-9,\.]+)"?', re.IGNORECASE) _JSON_SHORTIES_MARKER = "JSON_SHORTIES = insertAfterNthPosition(" _EMBED_SRC_RE = re.compile(r'<iframe[^>]+src="([^"]+)"', re.IGNORECASE) _CYRILLIC_RE = re.compile(r"[А-Яа-яЁё]") _RUSSIAN_MARKERS = { "russian", "russia", "russkiy", "russkaya", "russkoe", "русский", "русская", "русское", "россия", } @dataclass class ShortiesCandidate: page_url: str title: str thumb_url: str duration: str = "" views: str = "" def _clean_html_text(value: str) -> str: no_tags = re.sub(r"<[^>]+>", " ", value) return " ".join(html.unescape(no_tags).split()) def _absolute_url(value: str) -> str: return urljoin(PH_BASE_URL, value) def _extract_ldjson_candidates(page_html: str) -> list[ShortiesCandidate]: out: list[ShortiesCandidate] = [] for match in _SCRIPT_LD_JSON_RE.finditer(page_html): raw_json = html.unescape(match.group("json").strip()) if not raw_json: continue try: parsed = json.loads(raw_json) except json.JSONDecodeError: continue nodes: list[object] if isinstance(parsed, list): nodes = parsed else: nodes = [parsed] for node in nodes: if not isinstance(node, dict): continue if str(node.get("@type", "")).lower() != "videoobject": continue page_url = str(node.get("url") or "").strip() thumb = str(node.get("thumbnailUrl") or "").strip() title = str(node.get("name") or "").strip() or "Shorties" if not page_url: continue out.append( ShortiesCandidate( page_url=_absolute_url(page_url), title=title, thumb_url=_absolute_url(thumb) if thumb else "", ) ) return out def _extract_anchor_candidates(page_html: str) -> list[ShortiesCandidate]: out: list[ShortiesCandidate] = [] for match in _A_HREF_RE.finditer(page_html): href = match.group("href").strip() attrs = match.group("attrs") or "" body = match.group("body") or "" page_url = _absolute_url(href) title_match = _TITLE_ATTR_RE.search(attrs) or _TITLE_ATTR_RE.search(body) title = _clean_html_text(title_match.group("title")) if title_match else _clean_html_text(body) if not title: title = "Shorties" thumb_match = _IMG_URL_RE.search(attrs) or _IMG_URL_RE.search(body) thumb_url = _absolute_url(thumb_match.group("url")) if thumb_match else "" out.append(ShortiesCandidate(page_url=page_url, title=title, thumb_url=thumb_url)) return out def _dedupe_candidates(items: list[ShortiesCandidate]) -> list[ShortiesCandidate]: result: list[ShortiesCandidate] = [] seen: set[str] = set() for item in items: key = item.page_url.strip() if not key or key in seen: continue seen.add(key) result.append(item) return result async def _fetch_text(session: aiohttp.ClientSession, url: str) -> str: async with session.get(url, headers=REQUEST_HEADERS) as resp: if resp.status != 200: logger.warning("Shorties request failed: %s status=%s", url, resp.status) return "" return await resp.text() def _extract_best_mp4(page_html: str) -> str: media_match = _MEDIA_DEFS_RE.search(page_html) if media_match: raw = media_match.group(1) try: defs = json.loads(raw) except json.JSONDecodeError: defs = [] best_url = "" best_quality = -1 for item in defs: if not isinstance(item, dict): continue video_url = item.get("videoUrl") if not isinstance(video_url, str) or ".mp4" not in video_url: continue quality_raw = str(item.get("quality") or "").strip("p ") try: quality = int(quality_raw) except ValueError: quality = 0 if quality >= best_quality: best_quality = quality best_url = video_url if best_url: return best_url.replace("\\/", "/") for match in _VIDEO_URL_RE.finditer(page_html): candidate = match.group("url").replace("\\/", "/") if ".mp4" in candidate: return candidate return "" def _extract_balanced_json_array(page_html: str, marker: str) -> str: marker_idx = page_html.find(marker) if marker_idx < 0: return "" arr_start = page_html.find("[", marker_idx) if arr_start < 0: return "" depth = 0 in_string = False escaped = False for i in range(arr_start, len(page_html)): ch = page_html[i] if in_string: if escaped: escaped = False elif ch == "\\": escaped = True elif ch == '"': in_string = False continue if ch == '"': in_string = True continue if ch == "[": depth += 1 continue if ch == "]": depth -= 1 if depth == 0: return page_html[arr_start : i + 1] continue return "" def _extract_json_shorties(page_html: str) -> list[dict]: raw_array = _extract_balanced_json_array(page_html, _JSON_SHORTIES_MARKER) if not raw_array: return [] try: parsed = json.loads(raw_array) except json.JSONDecodeError: logger.warning("Shorties parser: failed to decode JSON_SHORTIES array") return [] if not isinstance(parsed, list): return [] return [item for item in parsed if isinstance(item, dict) and item.get("videoTitle")] def _pick_mp4_from_media_defs(media_defs: object) -> str: if not isinstance(media_defs, list): return "" best_url = "" best_quality = -1 for item in media_defs: if not isinstance(item, dict): continue if str(item.get("format", "")).lower() != "mp4": continue raw_url = item.get("videoUrl") if not isinstance(raw_url, str) or not raw_url: continue quality_raw = str(item.get("quality") or "").strip("p ") try: quality = int(quality_raw) except ValueError: quality = 0 if quality >= best_quality: best_quality = quality best_url = raw_url return best_url.replace("\\/", "/") def _pick_hls_from_media_defs(media_defs: object) -> str: if not isinstance(media_defs, list): return "" best_url = "" best_quality = -1 for item in media_defs: if not isinstance(item, dict): continue if str(item.get("format", "")).lower() != "hls": continue raw_url = item.get("videoUrl") if not isinstance(raw_url, str) or ".m3u8" not in raw_url: continue quality_raw = str(item.get("quality") or "").strip("p ") try: quality = int(quality_raw) except ValueError: quality = 0 if quality >= best_quality: best_quality = quality best_url = raw_url return best_url.replace("\\/", "/") def _is_russian_item(item: dict) -> bool: title = str(item.get("videoTitle") or item.get("metaTitle") or "") if _CYRILLIC_RE.search(title): return True pills = item.get("pillsData") if isinstance(pills, list): for pill in pills: if not isinstance(pill, dict): continue values = [str(pill.get("name") or ""), str(pill.get("slug") or "")] for raw in values: normalized = raw.strip().lower() if any(marker in normalized for marker in _RUSSIAN_MARKERS): return True return False def _is_russian_feed_item(item: dict) -> bool: title = str(item.get("title") or "") lowered = title.lower() return bool(_CYRILLIC_RE.search(title) or any(marker in lowered for marker in _RUSSIAN_MARKERS)) def _feed_from_json_shorties(items: list[dict]) -> list[dict]: feed: list[dict] = [] seen: set[str] = set() for item in items: if not _is_russian_item(item): continue hls_url = _pick_hls_from_media_defs(item.get("mediaDefinitions")) mp4_url = _pick_mp4_from_media_defs(item.get("mediaDefinitions")) if not hls_url and not mp4_url: continue source = str(item.get("linkUrl") or item.get("uniqueUrl") or item.get("shortieUrl") or "").strip() if source and source.startswith("/"): source = _absolute_url(source) key = source or hls_url or mp4_url if key in seen: continue seen.add(key) tracking = item.get("trackingTimeWatched") duration_raw = "" if isinstance(tracking, dict): duration_raw = str(tracking.get("video_duration") or "").strip() title = str(item.get("videoTitle") or item.get("metaTitle") or "Shorties").strip() thumb = str(item.get("imageUrl") or "").strip().replace("\\/", "/") views = str(item.get("likeInfo") or item.get("likeNumber") or "").strip() favorites = str(item.get("favoriteInfo") or item.get("favoriteNumber") or "").strip() embed_url = "" embed_raw = item.get("embedUrl") if isinstance(embed_raw, str) and embed_raw.strip(): unescaped = html.unescape(embed_raw.replace("\\/", "/")) match = _EMBED_SRC_RE.search(unescaped) if match: embed_url = match.group(1).strip() feed.append( { "type": "video", "url": hls_url or mp4_url, "sample": thumb or hls_url or mp4_url, "ext": "mp4", "score": 0, "fav_count": favorites or "0", "title": title, "duration": duration_raw, "views": views, "source": source or _absolute_url("/shorties"), "hls_url": hls_url, "mp4_url": mp4_url, "embed_url": embed_url, } ) return feed def _extract_meta_fields(page_html: str) -> tuple[str, str]: duration_match = _DURATION_RE.search(page_html) views_match = _VIEWS_RE.search(page_html) duration = duration_match.group("duration") if duration_match else "" views = views_match.group("views") if views_match else "" return duration, views async def _resolve_video_candidate( session: aiohttp.ClientSession, item: ShortiesCandidate, semaphore: asyncio.Semaphore, ) -> dict | None: async with semaphore: page_html = await _fetch_text(session, item.page_url) if not page_html: return None mp4_url = _extract_best_mp4(page_html) if not mp4_url: return None duration, views = _extract_meta_fields(page_html) if not item.duration: item.duration = duration if not item.views: item.views = views return { "type": "video", "url": mp4_url, "sample": item.thumb_url or mp4_url, "ext": "mp4", "score": 0, "fav_count": 0, "title": item.title, "duration": item.duration or "", "views": item.views or "", "source": item.page_url, } async def fetch_shorties_feed(count: int = 8, random_pages: int = 3) -> list[dict]: count = max(1, min(count, 20)) random_pages = max(1, min(random_pages, 2)) page_numbers = {1} while len(page_numbers) < random_pages: page_numbers.add(random.randint(1, 40)) list_urls = [f"{SHORTIES_LIST_URL}?page={page}" for page in sorted(page_numbers)] timeout = aiohttp.ClientTimeout(total=20) connector = aiohttp.TCPConnector(limit=16) async with aiohttp.ClientSession(timeout=timeout, connector=connector) as session: list_pages = await asyncio.gather(*[_fetch_text(session, url) for url in list_urls], return_exceptions=True) json_shorties_feed: list[dict] = [] for page in list_pages: if isinstance(page, Exception) or not page: continue shorties_items = _extract_json_shorties(page) if not shorties_items: continue json_shorties_feed.extend(_feed_from_json_shorties(shorties_items)) if json_shorties_feed: russian_only = [item for item in json_shorties_feed if _is_russian_feed_item(item)] # Строго пытаемся отдать russian first; если пусто — отдаём локализованную rt-ленту. selected_feed = russian_only if russian_only else json_shorties_feed random.shuffle(selected_feed) # dedupe by source/url after enrichment dedup: list[dict] = [] seen_keys: set[str] = set() for item in selected_feed: key = str(item.get("source") or item.get("url") or "").strip() if not key or key in seen_keys: continue seen_keys.add(key) dedup.append(item) return dedup[:count] candidates: list[ShortiesCandidate] = [] for page in list_pages: if isinstance(page, Exception) or not page: continue candidates.extend(_extract_ldjson_candidates(page)) candidates.extend(_extract_anchor_candidates(page)) candidates = _dedupe_candidates(candidates) if not candidates: logger.warning("Shorties parser: no candidates found") return [] random.shuffle(candidates) candidates = candidates[: max(count * 3, 12)] semaphore = asyncio.Semaphore(5) resolved = await asyncio.gather( *[_resolve_video_candidate(session, item, semaphore) for item in candidates], return_exceptions=True, ) feed: list[dict] = [] for item in resolved: if isinstance(item, dict) and item.get("url"): feed.append(item) russian_only_fallback = [item for item in feed if _is_russian_feed_item(item)] if russian_only_fallback: feed = russian_only_fallback random.shuffle(feed) return feed[:count]