145 lines
5.7 KiB
Python
145 lines
5.7 KiB
Python
"""
|
|
Mark videos as watched using a headless browser.
|
|
Visiting the page with authenticated cookies triggers the watch tracking JavaScript.
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
import logging
|
|
import asyncio
|
|
from urllib.parse import urlparse, parse_qs
|
|
from typing import Optional, Dict
|
|
|
|
log = logging.getLogger("headless_watcher")
|
|
|
|
# Try to import playwright
|
|
try:
|
|
from playwright.async_api import async_playwright
|
|
HAS_PLAYWRIGHT = True
|
|
except ImportError:
|
|
HAS_PLAYWRIGHT = False
|
|
log.warning("Playwright not installed, headless watching will not work")
|
|
|
|
|
|
class HeadlessWatcher:
|
|
"""Uses headless browser to visit pages and trigger watch tracking."""
|
|
def __init__(self, cookie_file: str):
|
|
self.cookie_file = cookie_file
|
|
self.domain_cookies = self._parse_cookie_file()
|
|
|
|
def _parse_cookie_file(self) -> Dict[str, list]:
|
|
"""Parse Netscape cookie file and group by domain."""
|
|
cookies_by_domain = {}
|
|
|
|
if not os.path.exists(self.cookie_file):
|
|
log.warning(f"Cookie file not found: {self.cookie_file}")
|
|
return cookies_by_domain
|
|
|
|
try:
|
|
with open(self.cookie_file, "r") as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
parts = line.split("\t")
|
|
if len(parts) >= 7:
|
|
domain = parts[0].lstrip(".")
|
|
path = parts[2]
|
|
secure = parts[3] == "TRUE"
|
|
name = parts[5]
|
|
value = parts[6]
|
|
if domain not in cookies_by_domain:
|
|
cookies_by_domain[domain] = []
|
|
cookies_by_domain[domain].append({
|
|
"name": name,
|
|
"value": value,
|
|
"domain": parts[0],
|
|
"path": path,
|
|
"secure": secure,
|
|
"httpOnly": False,
|
|
})
|
|
log.debug(f"Parsed cookies for {len(cookies_by_domain)} domains")
|
|
except Exception as e:
|
|
log.error(f"Error parsing cookie file: {e}")
|
|
return cookies_by_domain
|
|
|
|
async def visit_page(self, url: str, wait_seconds: int = 5) -> bool:
|
|
"""Visit a page with cookies to trigger watch tracking."""
|
|
if not HAS_PLAYWRIGHT:
|
|
log.error("Playwright not installed")
|
|
return False
|
|
|
|
parsed_url = urlparse(url)
|
|
domain = parsed_url.netloc.lower()
|
|
|
|
cookies = []
|
|
for cookie_domain, domain_cookies in self.domain_cookies.items():
|
|
if domain in cookie_domain or cookie_domain in domain:
|
|
cookies.extend(domain_cookies)
|
|
|
|
if not cookies:
|
|
log.warning(f"No cookies found for domain: {domain}")
|
|
return False
|
|
|
|
log.info(f"Visiting {url} with {len(cookies)} cookies")
|
|
|
|
try:
|
|
async with async_playwright() as p:
|
|
# Use system Chromium if available (for Alpine/Docker)
|
|
chromium_path = os.environ.get("PLAYWRIGHT_CHROMIUM_EXECUTABLE_PATH")
|
|
if chromium_path and os.path.exists(chromium_path):
|
|
log.debug(f"Using system Chromium: {chromium_path}")
|
|
browser = await p.chromium.launch(headless=True, executable_path=chromium_path)
|
|
else:
|
|
browser = await p.chromium.launch(headless=True)
|
|
try:
|
|
context = await browser.new_context(
|
|
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
|
|
)
|
|
await context.add_cookies(cookies)
|
|
page = await context.new_page()
|
|
response = await page.goto(url, wait_until="networkidle", timeout=30000)
|
|
if response and response.status == 200:
|
|
log.debug(f"Page loaded, waiting {wait_seconds}s")
|
|
await asyncio.sleep(wait_seconds)
|
|
log.info("Successfully triggered watch tracking")
|
|
return True
|
|
else:
|
|
status = response.status if response else "no response"
|
|
log.warning(f"Failed to load page, status: {status}")
|
|
return False
|
|
finally:
|
|
await browser.close()
|
|
except Exception as e:
|
|
log.error(f"Error in headless browser: {e}")
|
|
return False
|
|
|
|
class PHEadlessWatcher(HeadlessWatcher):
|
|
"""PornHub specific headless watcher."""
|
|
|
|
DOMAINS = ["pornhub.com", "www.pornhub.com", "de.pornhub.com", "fr.pornhub.com", "es.pornhub.com", "it.pornhub.com", "rt.pornhub.com"]
|
|
|
|
def can_handle(self, url: str) -> bool:
|
|
parsed = urlparse(url)
|
|
domain = parsed.netloc.lower()
|
|
return any(d in domain for d in self.DOMAINS)
|
|
|
|
async def mark_watched(self, url: str, wait_seconds: int = 5) -> bool:
|
|
return await self.visit_page(url, wait_seconds=wait_seconds)
|
|
|
|
|
|
async def headless_mark_watched(url: str, cookie_file: str, wait_seconds: int = 5) -> bool:
|
|
"""Mark a video as watched using headless browser."""
|
|
if not HAS_PLAYWRIGHT:
|
|
log.error("Playwright is not installed")
|
|
return False
|
|
|
|
# Try PH handler first
|
|
ph_handler = PHHeadlessWatcher(cookie_file)
|
|
if ph_handler.can_handle(url):
|
|
return await ph_handler.mark_watched(url, wait_seconds)
|
|
|
|
# Fallback to generic handler
|
|
generic_handler = HeadlessWatcher(cookie_file)
|
|
return await generic_handler.visit_page(url, wait_seconds)
|