forked from zovos/bot_tg
479 lines
16 KiB
Python
479 lines
16 KiB
Python
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'<a[^>]+href="(?P<href>/view_video\.php\?viewkey=[^"]+)"(?P<attrs>[^>]*)>(?P<body>.*?)</a>',
|
||
re.IGNORECASE | re.DOTALL,
|
||
)
|
||
_IMG_URL_RE = re.compile(
|
||
r'(?:data-mediumthumb|data-path|data-thumb_url|src)="(?P<url>https?://[^"]+)"',
|
||
re.IGNORECASE,
|
||
)
|
||
_TITLE_ATTR_RE = re.compile(r'title="(?P<title>[^"]+)"', 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]
|