diff options
Diffstat (limited to 'scraper/spiders/page.py')
| -rw-r--r-- | scraper/spiders/page.py | 450 |
1 files changed, 450 insertions, 0 deletions
diff --git a/scraper/spiders/page.py b/scraper/spiders/page.py new file mode 100644 index 0000000..4edcfa6 --- /dev/null +++ b/scraper/spiders/page.py @@ -0,0 +1,450 @@ +import asyncio +import datetime +import hashlib +import json +import logging +import os +import re +import urllib.parse +from pyquery import PyQuery as pq +from playwright.async_api import async_playwright + +logger = logging.getLogger(__name__) + +UNLIMITED = 2**31 + + +class PageScraper: + def __init__( + self, + url, + output_dir="data", + max_depth=5, + max_pages=None, + max_assets=None, + max_parallel=1, + timeout=60000, + page_timeout=120, + ): + self.url = url + self.output_dir = output_dir + self.max_depth = max_depth + self.max_pages = max_pages + self.max_assets = max_assets + self.max_parallel = max_parallel + self.timeout = timeout + self.page_timeout = page_timeout + self.page_count = 0 + self.asset_count = 0 + self.enqueued = set() + self._pages = {} + self._queue = asyncio.Queue() + self._busy = 0 + self._origin_base = ".".join( + urllib.parse.urlparse(url).netloc.rstrip(".").split(".")[-2:] + ) + + def _url_dir(self, url): + parsed = urllib.parse.urlparse(url) + path = parsed.path.strip("/") + return ( + os.path.join(self.output_dir, parsed.netloc, path) + if path + else os.path.join(self.output_dir, parsed.netloc) + ) + + def _read_meta(self, url): + meta_path = os.path.join(self._url_dir(url), "initial.json") + if not os.path.isfile(meta_path): + return None + try: + with open(meta_path) as f: + return json.load(f) + except (json.JSONDecodeError, FileNotFoundError): + return None + + def _content_changed(self, url, content): + new_hash = hashlib.sha256(content.encode()).hexdigest() + previous = self._read_meta(url) + if previous: + return previous.get("content", {}).get("initial_sha256") != new_hash + return True + + def _is_asset(self, resource_type): + return resource_type in ( + "stylesheet", + "image", + "media", + "font", + "script", + "fetch", + "xhr", + "other", + ) + + def _asset_path(self, url, content_type): + parsed = urllib.parse.urlparse(url) + netloc = parsed.netloc or parsed.path.split("/")[0] + path = parsed.path.strip("/") if parsed.path else netloc + if path.startswith(netloc): + path = "/" + path[len(netloc) :] + ext_map = { + "image/png": "png", + "image/jpeg": "jpg", + "image/webp": "webp", + "image/gif": "gif", + "image/svg+xml": "svg", + "image/x-icon": "ico", + "application/javascript": "js", + "text/javascript": "js", + "text/css": "css", + "font/woff2": "woff2", + "font/woff": "woff", + "application/json": "json", + "application/pdf": "pdf", + } + _, ext = os.path.splitext(path) + if ext: + return os.path.join(self.output_dir, netloc, path) + mime_ext = ext_map.get(content_type.split(";")[0].strip().lower()) + if mime_ext: + return os.path.join(self.output_dir, netloc, path, f"index.{mime_ext}") + return os.path.join(self.output_dir, netloc, path, "index.dat") + + async def _on_response(self, response, current_url_ref): + try: + url = response.url + if not self._is_asset(response.request.resource_type): + return + if self.max_assets is not None and self.asset_count >= self.max_assets: + return + content_type = response.headers.get("content-type", "") + asset_path = self._asset_path(url, content_type) + if os.path.isfile(asset_path): + return + self.asset_count += 1 + try: + body = await response.body() + except Exception: + return + try: + os.makedirs(os.path.dirname(asset_path), exist_ok=True) + except OSError: + return + with open(asset_path, "wb") as f: + f.write(body) + asset_hash = hashlib.sha256(body).hexdigest() + meta = { + "url": url, + "request": { + "method": response.request.method, + "headers": dict(response.request.headers), + }, + "response": { + "status": response.status, + "headers": dict(response.headers), + }, + "content": {"sha256": asset_hash}, + "scrape": { + "timestamp": datetime.datetime.now( + datetime.timezone.utc + ).isoformat(), + }, + } + with open(asset_path + ".json", "w", encoding="utf-8") as f: + json.dump(meta, f, indent=2) + page_url = current_url_ref.get() + if page_url and page_url in self._pages: + self._pages[page_url]["assets"].append( + { + "url": url, + } + ) + except Exception: + pass + + def _extract_article_links(self, doc, page_url): + links = set() + skip_patterns = [ + re.compile( + r"^/(about|contact|privacy|terms|ad|subscribe|newsletter|" + r"login|signup|careers|feedback|site|help|support|faq|" + r"sitemap|robots|feed|rss|api|static|assets|img|images|" + r"css|js|scripts|fonts|cdn|media|video|audio)$" + ), + re.compile( + r"\.(png|jpg|jpeg|gif|svg|ico|css|js|woff|woff2|ttf|pdf|xml|json|html)$" + ), + re.compile(r"^#"), + re.compile(r"^mailto:"), + re.compile(r"^javascript:"), + re.compile(r"^/$"), + ] + for a in doc("a").items(): + href = a.attr("href") or "" + resolved = urllib.parse.urljoin(page_url, href) + parsed = urllib.parse.urlparse(resolved) + parsed_base = ".".join(parsed.netloc.rstrip(".").split(".")[-2:]) + if parsed_base != self._origin_base: + continue + path = parsed.path.strip("/") + if not path: + continue + if any(s.search(parsed.path) for s in skip_patterns): + continue + segments = [s for s in path.split("/") if s] + if len(segments) >= 2: + links.add(resolved) + links.discard(self.url) + for seen in self.enqueued: + links.discard(seen) + return links + + async def _process_page(self, page, page_url, depth): + headers = {} + previous = self._read_meta(page_url) + if previous: + etag = previous.get("response", {}).get("etag") + if etag: + headers["if-none-match"] = etag + timestamp = previous.get("scrape", {}).get("timestamp") + if timestamp: + headers["if-modified-since"] = timestamp + try: + await page.set_extra_http_headers(headers) + resp = await page.goto(page_url, wait_until="domcontentloaded") + await page.set_extra_http_headers({}) + except Exception as e: + logger.error(f"Failed to navigate to {page_url}: {e}") + return + page_url = resp.url + if resp.request.redirected_from: + redirect_url = resp.request.redirected_from.url + page_dir = self._url_dir(redirect_url) + meta_path = os.path.join(page_dir, "initial.json") + if not os.path.exists(meta_path): + os.makedirs(page_dir, exist_ok=True) + prev_resp = resp.request.redirected_from.existing_response + meta = { + "request": { + "method": resp.request.method, + "url": redirect_url, + "final_url": page_url, + "redirect_chain": [], + "headers": dict(resp.request.headers), + }, + "response": { + "status": prev_resp.status if prev_resp else None, + "headers": dict(resp.headers), + }, + "content": {"sha256": ""}, + "scrape": { + "timestamp": datetime.datetime.now( + datetime.timezone.utc + ).isoformat(), + }, + } + with open(meta_path, "w", encoding="utf-8") as f: + json.dump(meta, f, indent=2) + if resp.status == 304: + page_dir = self._url_dir(page_url) + initial_path = os.path.join(page_dir, "initial.html") + if os.path.isfile(initial_path): + with open(initial_path, encoding="utf-8") as f: + local_html = f.read() + await self._enqueue_links(local_html, page_url, depth) + return + initial_html = await page.content() + content_hash = hashlib.sha256(initial_html.encode()).hexdigest() + changed = self._content_changed(page_url, initial_html) + + if changed: + at_limit = self.max_pages is not None and self.page_count >= self.max_pages + if at_limit: + return + self.page_count += 1 + logger.info(f"Scraping page {self.page_count}: {page_url}") + page_dir = self._url_dir(page_url) + os.makedirs(page_dir, exist_ok=True) + with open( + os.path.join(page_dir, "initial.html"), "w", encoding="utf-8" + ) as f: + f.write(initial_html) + + await self._wait_for_hydrated(page) + await self._scroll_page(page) + final_html = await page.content() + final_hash = hashlib.sha256(final_html.encode()).hexdigest() + with open(os.path.join(page_dir, "index.html"), "w", encoding="utf-8") as f: + f.write(final_html) + page_data = {"assets": []} + self._pages[page_url] = page_data + meta = { + "request": { + "method": resp.request.method, + "url": page_url, + "final_url": page_url, + "redirect_chain": [], + "headers": dict(resp.request.headers), + }, + "response": { + "status": resp.status, + "headers": dict(resp.headers), + "etag": resp.headers.get("etag"), + }, + "content": { + "sha256": final_hash, + "initial_sha256": content_hash, + }, + "scrape": { + "timestamp": datetime.datetime.now( + datetime.timezone.utc + ).isoformat(), + }, + } + with open( + os.path.join(page_dir, "initial.json"), "w", encoding="utf-8" + ) as f: + json.dump(meta, f, indent=2) + await self._enqueue_links(initial_html, page_url, depth) + await self._enqueue_links(final_html, page_url, depth) + else: + logger.info(f"Page unchanged, skipping: {page_url}") + page_dir = self._url_dir(page_url) + initial_path = os.path.join(page_dir, "initial.html") + if os.path.isfile(initial_path): + with open(initial_path, encoding="utf-8") as f: + local_html = f.read() + await self._enqueue_links(local_html, page_url, depth) + + async def _wait_for_hydrated(self, page): + prev_size = -1 + stable = 0 + for _ in range(10): + html = await page.content() + curr_size = len(html) + if abs(curr_size - prev_size) < 500: + stable += 1 + else: + stable = 0 + prev_size = curr_size + if stable >= 2: + break + await asyncio.sleep(1) + + async def _scroll_page(self, page): + """Scroll to bottom and back to trigger lazy loading.""" + await page.evaluate("window.scrollTo(0, document.body.scrollHeight)") + await asyncio.sleep(1) + await page.evaluate("window.scrollTo(0, 0)") + + async def _enqueue_links(self, html, page_url, depth): + if self.max_pages is not None and self.page_count >= self.max_pages: + return + if depth >= self.max_depth: + return + doc = pq(html) + remaining = ( + (self.max_pages - self.page_count) + if self.max_pages is not None + else UNLIMITED + ) + for link_url in self._extract_article_links(doc, page_url): + if remaining <= 0: + break + if link_url not in self.enqueued: + self.enqueued.add(link_url) + remaining -= 1 + await self._queue.put((link_url, depth + 1)) + + async def _worker(self, worker_id, pages): + page = pages[worker_id] + current_url_ref = {"url": None} + page.on("response", lambda resp, r=current_url_ref: self._on_response(resp, r)) + while True: + if self.max_pages is not None and self.page_count >= self.max_pages: + break + item = await self._queue.get() + if item is None: + break + url, depth = item + if depth >= self.max_depth: + self._queue.task_done() + continue + self._busy += 1 + try: + if current_url_ref["url"] and current_url_ref["url"] in self._pages: + self._pages[current_url_ref["url"]]["assets"] = self._pages[ + current_url_ref["url"] + ].get("assets", []) + current_url_ref["url"] = url + await asyncio.wait_for( + self._process_page(page, url, depth), + timeout=self.page_timeout, + ) + await asyncio.sleep(0.3) + except asyncio.TimeoutError: + logger.warning(f"Timeout processing {url}") + except Exception as e: + logger.error(f"Error processing {url}: {e}") + finally: + self._busy -= 1 + self._queue.task_done() + + async def run(self): + async with async_playwright() as p: + browser = await p.chromium.launch( + headless=True, + args=["--disable-dev-shm-usage"], + executable_path="/usr/bin/chromium-browser", + ) + context = await browser.new_context( + user_agent=( + "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" + ), + viewport={"width": 1920, "height": 1080}, + ) + await context.add_init_script( + "Object.defineProperty(navigator, 'webdriver', {get: () => undefined})" + ) + pages = [] + for i in range(self.max_parallel): + pg = await context.new_page() + pg.set_default_timeout(self.timeout) + pages.append(pg) + logger.info( + f"Scraping {self.url} (depth={self.max_depth}, parallel={self.max_parallel})" + ) + await self._queue.put((self.url, 0)) + workers = [ + asyncio.create_task(self._worker(i, pages)) + for i in range(self.max_parallel) + ] + watchdog = 0 + try: + while True: + await asyncio.sleep(0.5) + watchdog += 1 + if watchdog >= 300: + logger.warning("Watchdog: queue not draining") + watchdog = 0 + if self._queue.empty() and self._busy == 0: + for _ in range(self.max_parallel): + await self._queue.put(None) + break + if self.max_pages is not None and self.page_count >= self.max_pages: + for _ in range(self.max_parallel): + await self._queue.put(None) + break + await asyncio.gather(*workers) + except Exception: + logger.warning("Browser crashed, stopping scrape") + for _ in range(self.max_parallel): + await self._queue.put(None) + await asyncio.gather(*workers, return_exceptions=True) + logger.info(f"Done: {self.page_count} pages, {self.asset_count} assets") + try: + await browser.close() + except Exception: + pass |