""" Mark videos as watched on various websites after successful download. Uses the same cookies that were used for downloading. """ import os import re import logging from urllib.parse import urlparse, parse_qs from typing import Optional, Dict, Callable from headless_watcher import headless_mark_watched, HAS_PLAYWRIGHT log = logging.getLogger("mark_watched") # Try to import curl_cffi for better impersonation, fallback to requests try: from curl_cffi import requests as curl_requests HAS_CURL_CFFI = True except ImportError: HAS_CURL_CFFI = False import requests class MarkWatchedHandler: """Base class for mark-as-watched handlers.""" def __init__(self, cookie_file: str): self.cookie_file = cookie_file def can_handle(self, url: str) -> bool: """Check if this handler can handle the given URL.""" raise NotImplementedError async def mark_watched(self, url: str) -> bool: """Mark the video as watched. Returns True on success.""" raise NotImplementedError def _load_cookies(self) -> Dict[str, str]: """Load cookies from the Netscape cookie file.""" cookies = {} if not os.path.exists(self.cookie_file): log.warning(f"Cookie file not found: {self.cookie_file}") return cookies try: with open(self.cookie_file, "r") as f: for line in f: line = line.strip() if not line or line.startswith("#"): continue # Netscape format: domain flag path secure expiration name value parts = line.split("\t") if len(parts) >= 7: name = parts[5] value = parts[6] cookies[name] = value log.debug(f"Loaded {len(cookies)} cookies from {self.cookie_file}") except Exception as e: log.error(f"Error loading cookies: {e}") return cookies def _make_request(self, method: str, url: str, **kwargs) -> Optional[object]: """Make HTTP request using curl_cffi if available, else requests.""" try: if HAS_CURL_CFFI: # Use curl_cffi for better impersonation session = curl_requests.Session() # Set browser impersonation session.impersonate = "chrome124" response = session.request(method, url, **kwargs) return response else: return requests.request(method, url, **kwargs) except Exception as e: log.error(f"Request failed: {e}") return None class PHHandler(MarkWatchedHandler): """Handler for PornHub.""" 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): parsed = urlparse(url) domain = parsed.netloc.lower() return any(d in domain for d in self.DOMAINS) async def mark_watched(self, url): # Extract viewkey from URL parsed = urlparse(url) params = parse_qs(parsed.query) viewkey = params.get("viewkey", [None])[0] if not viewkey: # Try to extract from path match = re.search(r"viewkey=([^&]+)", url) if match: viewkey = match.group(1) if not viewkey: log.warning(f"Could not extract viewkey from URL: {url}") return False log.info(f"Marking video as watched, viewkey: {viewkey}") # Load cookies cookies = self._load_cookies() if not cookies: log.warning("No cookies available, cannot mark as watched") return False # Try multiple endpoints as PH may use different URLs # First try the AJAX endpoint which is most commonly used endpoints = [ # AJAX endpoint (most common) ("POST", f"https://www.pornhub.com/user/watched/add/video/{viewkey}"), # Alternative format ("GET", f"https://www.pornhub.com/user/watched/add/video/{viewkey}"), # Legacy endpoint ("POST", f"https://www.pornhub.com/user/watched/video/viewkey/{viewkey}"), ] headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", "Accept": "application/json, text/javascript, */*; q=0.01", "Accept-Language": "en-US,en;q=0.9", "X-Requested-With": "XMLHttpRequest", "Referer": url, } for method, api_url in endpoints: log.debug(f"Trying {method} {api_url}") response = self._make_request(method, api_url, headers=headers, cookies=cookies, allow_redirects=True) if response is None: continue log.debug(f"Response status: {response.status_code}") if response.status_code == 200: log.info(f"Successfully marked video as watched using {api_url}") return True elif response.status_code == 302: # Redirect often means success on PH log.info(f"Successfully marked video as watched (redirect from {api_url})") return True # API methods failed, try headless browser as fallback log.info("API methods failed, trying headless browser approach") if HAS_PLAYWRIGHT: return await headless_mark_watched(url, self.cookie_file, wait_seconds=5) else: log.warning("Playwright not available for headless browser fallback") return False class YouTubeHandler(MarkWatchedHandler): """Handler for YouTube - uses YouTube API or internal endpoints.""" DOMAINS = ["youtube.com", "www.youtube.com", "youtu.be", "m.youtube.com"] def can_handle(self, url): parsed = urlparse(url) domain = parsed.netloc.lower() return any(d in domain for d in self.DOMAINS) async def mark_watched(self, url): # YouTube marking as watched requires more complex handling # typically done through the browse endpoint with protobuf # This is a simplified implementation log.info("YouTube mark-as-watched not yet fully implemented") return False # Registry of all handlers HANDLERS = [ PHHandler, YouTubeHandler, ] def get_handler(url: str, cookie_file: str) -> Optional[MarkWatchedHandler]: """ Get the appropriate handler for a URL. Args: url: The video URL cookie_file: Path to the Netscape cookie file Returns: Handler instance if found, None otherwise """ if not cookie_file or not os.path.exists(cookie_file): return None for handler_class in HANDLERS: handler = handler_class(cookie_file) if handler.can_handle(url): return handler return None async def mark_watched(url: str, cookie_file: str) -> bool: """ Mark a video as watched on its respective site. Args: url: The video URL cookie_file: Path to the Netscape cookie file Returns: True if successfully marked as watched, False otherwise """ handler = get_handler(url, cookie_file) if not handler: log.debug(f"No mark-watched handler available for URL: {url}") return False try: return await handler.mark_watched(url) except Exception as e: log.error(f"Error marking video as watched: {e}") return False