goon/scripts/site_icons.py

"""Loga stron do zakladki Sites: favicon kazdego tube'a jako PNG 64x64 w app/static.

Po co przez backend, a nie z telefonu: apka nie ma uderzac do 44 domen NSFW ani do
zewnetrznego serwisu ikon (Google/DDG) z adresem IP uzytkownika. VPS pobiera raz,
normalizuje (PIL: ico/png/svg-bez, przezroczystosc zachowana) i serwuje statycznie,
a API `/sources` dokleja `icon_url`, gdy plik istnieje.

Zrodlo ikony, w kolejnosci: `<link rel="apple-touch-icon">` (zwykle najwiekszy PNG),
`<link rel="icon"|"shortcut icon">`, na koncu `/favicon.ico`. Strony za Cloudflare,
ktore blokuja VPS, idą przez proxy Bright Data, jak w ingest HTML.

Odswiezanie: plik mlodszy niz `_TTL_DNI` zostaje, brak/porazka nie zostawia sladu
(nastepny przebieg sprobuje znowu). Job w schedulerze raz na dobe, recznie:
  docker compose run --rm worker python3 -u /srv/scripts/site_icons.py [--force] [--only sitetag]
"""
from __future__ import annotations

import argparse
import html as html_mod
import io
import logging
import re
import sys
import time
from pathlib import Path
from urllib.parse import urljoin

from PIL import Image

from app.api.sources import _sitetag_to_display
from app.config import get_settings
from app.extractors._fetch import browser_get

log = logging.getLogger(__name__)

ICON_DIR = Path(__file__).resolve().parent.parent / "app" / "static" / "site-icons"
ICON_PX = 64
_TTL_DNI = 14

_LINK_RE = re.compile(r"<link\b[^>]*>", re.IGNORECASE)
_ATTR_RE = re.compile(r"""([a-zA-Z-]+)\s*=\s*(?:"([^"]*)"|'([^']*)'|([^\s>]+))""")


def _attrs(tag: str) -> dict[str, str]:
    out: dict[str, str] = {}
    for m in _ATTR_RE.finditer(tag):
        out[m.group(1).lower()] = html_mod.unescape(m.group(2) or m.group(3) or m.group(4) or "")
    return out


def _icon_candidates(page_url: str, page_html: str) -> list[str]:
    """Adresy ikon z <link rel=...icon...>, najpierw apple-touch-icon, potem reszta."""
    apple: list[str] = []
    plain: list[str] = []
    for tag in _LINK_RE.findall(page_html):
        a = _attrs(tag)
        rel = a.get("rel", "").lower()
        href = a.get("href", "").strip()
        if not href or "icon" not in rel:
            continue
        if href.lower().endswith(".svg"):
            continue  # PIL nie czyta SVG, a to i tak zwykle logo, nie favicon
        full = urljoin(page_url, href)
        (apple if "apple-touch" in rel else plain).append(full)
    seen: set[str] = set()
    out: list[str] = []
    for u in apple + plain + [urljoin(page_url, "/favicon.ico")]:
        if u not in seen:
            seen.add(u)
            out.append(u)
    return out


def _fetch(url: str, *, proxy: str | None, timeout: float = 20.0):
    """HTML strony glownej. Najpierw bez proxy, przy blokadzie przez Bright Data."""
    try:
        r = browser_get(url, timeout=timeout)
        if r.status_code == 200:
            return r
    except Exception as e:  # noqa: BLE001
        log.debug("site-icons: %s bez proxy: %s", url, e)
    if not proxy:
        return None
    try:
        r = browser_get(url, timeout=timeout, proxy=proxy)
        return r if r.status_code == 200 else None
    except Exception as e:  # noqa: BLE001
        log.debug("site-icons: %s przez proxy: %s", url, e)
        return None


def _fetch_bytes(url: str, *, proxy: str | None, timeout: float = 15.0) -> bytes | None:
    """Binarka ikony. `browser_get` oddaje tylko tekst, wiec ten sam klient (curl_cffi z
    odciskiem Chrome) wolany wprost, z tym samym proxy w odwodzie."""
    from curl_cffi import requests as cf

    for px in ([None, proxy] if proxy else [None]):
        try:
            r = cf.get(
                url,
                timeout=timeout,
                impersonate="chrome120",
                allow_redirects=True,
                proxies={"http": px, "https": px} if px else None,
                verify=not px,
            )
            if r.status_code == 200 and r.content:
                return r.content
        except Exception as e:  # noqa: BLE001
            log.debug("site-icons: %s (%s): %s", url, "proxy" if px else "direct", e)
    return None


def _to_png(data: bytes) -> bytes | None:
    try:
        im = Image.open(io.BytesIO(data))
        # ICO trzyma kilka rozmiarow, PIL otwiera najwiekszy dopiero po .size sprawdzeniu
        if getattr(im, "format", None) == "ICO":
            sizes = sorted(getattr(im, "ico", None).sizes()) if hasattr(im, "ico") else []
            if sizes:
                im.size = sizes[-1]
        im = im.convert("RGBA")
    except Exception:  # noqa: BLE001
        return None
    if im.width < 8 or im.height < 8:
        return None
    im.thumbnail((ICON_PX, ICON_PX), Image.LANCZOS)
    canvas = Image.new("RGBA", (ICON_PX, ICON_PX), (0, 0, 0, 0))
    canvas.paste(im, ((ICON_PX - im.width) // 2, (ICON_PX - im.height) // 2))
    buf = io.BytesIO()
    canvas.save(buf, format="PNG", optimize=True)
    return buf.getvalue()


# Sitetagi bez TLD w nazwie (display name = sama nazwa) albo z inna domena niz sugeruje
# nazwa (galaxyporn.net, nie .com). Zrodlo: _BASE w scraperach.
_DOMAIN_OVERRIDES: dict[str, str] = {
    "fullporner": "fullporner.com",
    "fullvideosporn": "fullvideosporn.com",
    "galaxyporncom": "galaxyporn.net",
    "hentaigasm": "hentaigasm.com",
    "hentaimama": "hentaimama.io",
    "hentaiplay": "hentaiplay.net",
    "javflix": "javflix.cc",
    "naughtymachinima": "www.naughtymachinima.com",
    "porndishcom": "www.porndish.com",
    "porndupe": "www.porndupe.com",
    "pornmike": "pornmike.com",
    "rule34video": "rule34video.com",
    "supjav": "supjav.com",
    "tubepornclassic": "tubepornclassic.com",
    "vjav": "vjav.com",
    "xanimeporn": "xanimeporn.com",
    "yespornvip": "yesporn.vip",
}


def fetch_icon(sitetag: str, *, proxy: str | None) -> bytes | None:
    domain = _DOMAIN_OVERRIDES.get(sitetag) or _sitetag_to_display(sitetag)
    if "." not in domain:
        return None
    home = f"https://{domain}/"
    page = _fetch(home, proxy=proxy)
    candidates = _icon_candidates(home, page.text) if page is not None else [home + "favicon.ico"]
    for url in candidates[:4]:
        raw = _fetch_bytes(url, proxy=proxy)
        if not raw:
            continue
        png = _to_png(raw)
        if png:
            return png
    return None


def refresh_site_icons(sitetags: list[str], *, force: bool = False) -> dict[str, int]:
    ICON_DIR.mkdir(parents=True, exist_ok=True)
    proxy = get_settings().brightdata_proxy_url
    stats = {"ok": 0, "fresh": 0, "miss": 0}
    now = time.time()
    for tag in sitetags:
        out = ICON_DIR / f"{tag}.png"
        if not force and out.exists() and now - out.stat().st_mtime < _TTL_DNI * 86400:
            stats["fresh"] += 1
            continue
        png = fetch_icon(tag, proxy=proxy)
        if png:
            out.write_bytes(png)
            stats["ok"] += 1
            log.info("site-icons: %s ok (%d B)", tag, len(png))
        else:
            stats["miss"] += 1
            log.info("site-icons: %s brak ikony", tag)
    return stats


def live_sitetags() -> list[str]:
    from sqlalchemy import text

    from app.db import session_scope

    with session_scope() as s:
        rows = s.execute(
            text(
                "SELECT DISTINCT origin FROM playback_sources "
                "WHERE dead_at IS NULL AND origin LIKE 'tube:%'"
            )
        ).all()
    return sorted(r[0].split(":", 1)[1] for r in rows)


def main() -> int:
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    ap = argparse.ArgumentParser()
    ap.add_argument("--force", action="store_true")
    ap.add_argument("--only", help="jeden sitetag")
    args = ap.parse_args()
    tags = [args.only] if args.only else live_sitetags()
    stats = refresh_site_icons(tags, force=args.force)
    print(f"site-icons: {stats} z {len(tags)} stron; katalog {ICON_DIR}")
    return 0


if __name__ == "__main__":
    sys.exit(main())