"""pornbusy.com — browse scraper (WordPress + Rank Math). Dodany 2026-07-27. Ripy studyjne (~60%) + JAV z kodami (~30%). Ocena orphan-risk LOW z pomiarem: 80% scen z próbki 100 ma performera, który u nas MA już ref tpdb/stashdb, a 21% tytułów mocno matchuje sceny które już mamy (czyli się scalą, nie zorphanują). **Paginacja homepage jest ZEPSUTA** — `/page/2/` i `/page/3/` zwracają ten sam zestaw co strona 1 (zweryfikowane). Dlatego listing bierzemy z SITEMAPY: `sitemap_index.xml` → 10× `post-sitemapN.xml` po ~500 URL-i, każdy z ``, posortowane od najnowszych. `crawl_page` tnie ten katalog na strony po `_PAGE_SIZE` (wzorzec jak `_playtube.BasePlayTubeScraper`). Metadane — każde pole to osobny węzeł `itemprop`, nic nie trzeba wyłuskiwać z tytułu-slug: - title `

`, duration ISO, uploadDate, thumbnailUrl, description - obsada: `
` — CZYSTA, bez pollution (na stronie sceny jest dokładnie tyle linków `/actor/` ilu realnych aktorów; brak sekcji Related) - kategorie + tagi: `
` (`/category/` i `/tag/`) - studio: **brak** — pornbusy nie ma pola studia ani prefiksu w tytule Playback: `itemprop="embedURL"` → extractor `pornbusycom` (loadvid / seekplayer / zpi.cx ≈ 69% katalogu). """ from __future__ import annotations import html import logging import re from app.connectors.base import ( RawPerformer, RawPlaybackSource, RawScene, RawTag, ) from app.connectors.direct_scrapers._browse_base import BaseBrowseScraper from app.connectors.direct_scrapers._playtube import _parse_iso_date from app.extractors import browser_get from app.normalize.text import slugify log = logging.getLogger(__name__) _BASE = "https://pornbusy.com" _PAGE_SIZE = 20 _SITEMAP_INDEX = f"{_BASE}/sitemap_index.xml" _LOC_RE = re.compile(r"\s*([^<]+?)\s*") _URL_BLOCK_RE = re.compile(r"(.*?)", re.DOTALL | re.IGNORECASE) _LASTMOD_RE = re.compile(r"\s*([^<]+?)\s*") _TITLE_RE = re.compile(r'

]*>(.*?)

', re.IGNORECASE | re.DOTALL) _DUR_RE = re.compile(r'itemprop="duration"\s+content="([^"]+)"', re.IGNORECASE) _DATE_RE = re.compile(r'itemprop="uploadDate"\s+content="([^"]+)"', re.IGNORECASE) _THUMB_RE = re.compile(r'itemprop="thumbnailUrl"\s+content="([^"]+)"', re.IGNORECASE) _DESC_RE = re.compile(r'itemprop="description"\s+content="([^"]*)"', re.IGNORECASE) _ACTORS_BLOCK_RE = re.compile(r'
(.*?)
', re.IGNORECASE | re.DOTALL) _ACTOR_RE = re.compile(r'/actor/[^/"]+/"[^>]*title="([^"]+)"', re.IGNORECASE) _ACTOR_TEXT_RE = re.compile(r">([^<>]+)") _TAGS_BLOCK_RE = re.compile(r'
(.*?)
', re.IGNORECASE | re.DOTALL) # Nazwa z atrybutu `title`, NIE z tekstu anchora: tekst poprzedza zagnieżdżona ikona # (`Blonde`), więc `>([^<]+)` # nie matchuje. `title` jest czysty i zawsze obecny. _TAXO_RE = re.compile( r'/(?:category|tag)/([a-z0-9\-]+)/"[^>]*title="([^"]+)"', re.IGNORECASE ) # `P0DT0H42M52S` _ISO_DUR_RE = re.compile( r"P(?:(\d+)D)?T?(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?", re.IGNORECASE ) # Junk-performer guard: pornbusy wrzuca do /actor/ także ogólniki. _JUNK_ACTORS = frozenset({"amateur", "unknown", "anonymous", "n-a", "na", "various"}) def _parse_iso_duration(value: str | None) -> int | None: if not value: return None m = _ISO_DUR_RE.match(value.strip()) if not m: return None d, h, mn, s = (int(g or 0) for g in m.groups()) total = d * 86400 + h * 3600 + mn * 60 + s return total or None def _clean(raw: str) -> str: return html.unescape(re.sub(r"\s+", " ", re.sub(r"<[^>]+>", " ", raw))).strip() class PornBusyScraper(BaseBrowseScraper): sitetag = "pornbusycom" def __init__(self) -> None: super().__init__() # Katalog z sitemapy, newest-first. Lazy raz per instancję (browse_latest i # deep_crawl tworzą instancję per run, więc 10 fetchy XML amortyzuje się). self._catalog: list[str] | None = None # Listing nie jest stronicowalny GET-em (homepage pagination zepsuta) — # paginację robi sitemap w `crawl_page`. Te dwie metody są abstrakcyjne w bazie. def _listing_url(self, page: int) -> str: # pragma: no cover - nieużywane return _SITEMAP_INDEX def _extract_scene_urls(self, listing_html: str) -> list[str]: # pragma: no cover return [] def _load_catalog(self) -> list[str] | None: if self._catalog is not None: return self._catalog try: idx = browser_get(_SITEMAP_INDEX, timeout=self._timeout) idx.raise_for_status() except Exception as e: log.warning("pornbusy: sitemap index fetch failed: %s", e) return None maps = [u for u in _LOC_RE.findall(idx.text) if "post-sitemap" in u] if not maps: log.warning("pornbusy: no post-sitemaps in index") return None entries: list[tuple[str, str]] = [] for sm_url in maps: try: sm = browser_get(sm_url, timeout=self._timeout) sm.raise_for_status() except Exception as e: log.warning("pornbusy: sitemap fetch failed %s: %s", sm_url, e) continue for block in _URL_BLOCK_RE.findall(sm.text): loc = _LOC_RE.search(block) if not loc: continue url = loc.group(1) # Pomijamy strony taksonomii/nav — sceny to `//` jednosegmentowe. if any(x in url for x in ("/actor/", "/category/", "/tag/", "/page/")): continue lm = _LASTMOD_RE.search(block) entries.append((lm.group(1) if lm else "", url)) if not entries: return None entries.sort(key=lambda e: e[0], reverse=True) seen: set[str] = set() catalog: list[str] = [] for _, url in entries: if url in seen: continue seen.add(url) catalog.append(url) log.info("pornbusy: catalog loaded — %d scenes from %d sitemaps", len(catalog), len(maps)) self._catalog = catalog return catalog def crawl_page(self, page: int) -> list[RawScene] | None: catalog = self._load_catalog() if catalog is None: return None chunk = catalog[(page - 1) * _PAGE_SIZE: page * _PAGE_SIZE] if not chunk: return [] out: list[RawScene] = [] for scene_url in chunk: try: res = browser_get(scene_url, timeout=self._timeout) res.raise_for_status() except Exception as e: log.info("pornbusy detail fetch failed %s: %s", scene_url, e) continue try: raw = self._parse_detail(scene_url, res.text) except Exception as e: log.warning("pornbusy detail parse failed %s: %s", scene_url, e) continue if raw is not None: out.append(raw) return out def _parse_detail(self, scene_url: str, detail_html: str) -> RawScene | None: tm = _TITLE_RE.search(detail_html) title = _clean(tm.group(1)) if tm else "" if not title: return None dm = _DUR_RE.search(detail_html) duration_sec = _parse_iso_duration(dm.group(1)) if dm else None um = _DATE_RE.search(detail_html) release_date = _parse_iso_date(um.group(1)) if um else None thm = _THUMB_RE.search(detail_html) thumbnail_url = thm.group(1) if thm else None dsm = _DESC_RE.search(detail_html) description = html.unescape(dsm.group(1)).strip() if dsm else None if description and description.strip().lower() == title.strip().lower(): description = None # opis bywa kopią tytułu — nie duplikujemy performers: list[RawPerformer] = [] seen_p: set[str] = set() ab = _ACTORS_BLOCK_RE.search(detail_html) if ab: names = _ACTOR_RE.findall(ab.group(1)) or _ACTOR_TEXT_RE.findall(ab.group(1)) for name in names: name = html.unescape(name).strip() sl = slugify(name) if not sl or sl in seen_p or sl in _JUNK_ACTORS: continue seen_p.add(sl) performers.append( RawPerformer(external_id=f"{self.sitetag}:performer:{sl}", name=name) ) tags: list[RawTag] = [] seen_t: set[str] = set() tb = _TAGS_BLOCK_RE.search(detail_html) if tb: for slug, name in _TAXO_RE.findall(tb.group(1)): name = html.unescape(name).strip() if not name or slug in seen_t or slug in seen_p: continue seen_t.add(slug) tags.append(RawTag(external_id=f"{self.sitetag}:tag:{slug}", name=name, slug=slug)) return RawScene( external_id=f"{self.sitetag}:{scene_url}", title=title, description=description, release_date=release_date, duration_sec=duration_sec, url=scene_url, # studio: pornbusy go nie ma (ani pole, ani prefiks tytułu) performers=performers, tags=tags, playback_sources=[ RawPlaybackSource( origin=f"tube:{self.sitetag}", page_url=scene_url, duration_sec=duration_sec, thumbnail_url=thumbnail_url, ) ], )