scraper419.py

from __future__ import annotations

import asyncio
import html
import re
from pathlib import Path

from playwright.async_api import Error as PlaywrightError
from playwright.async_api import TimeoutError as PlaywrightTimeoutError
from playwright.async_api import async_playwright


TARGET_URL = (
	"https://programs.sigchi.org/chi/2026/program/all"
	"?CONTENT_TYPE%5B%5D=14694"
	"&itemsType=SESSION"
	"&sortDirection=asc"
	"&sortType=TIME"
	"&viewType=LIST"
)
OUTPUT_FILE = "chi2026_papers.txt"
OUTPUT_PATH = Path(__file__).with_name(OUTPUT_FILE)


async def scroll_to_load_all_content(
	page,
	pause_ms: int = 1200,
	max_no_growth: int = 3,
	max_scrolls: int = 100,
) -> None:
	"""Scroll down until lazy-loaded content stops increasing page height."""
	no_growth_count = 0
	previous_height = await page.evaluate("() => document.body.scrollHeight")

	for _ in range(max_scrolls):
		await page.evaluate("() => window.scrollTo(0, document.body.scrollHeight)")
		await page.wait_for_timeout(pause_ms)

		current_height = await page.evaluate("() => document.body.scrollHeight")
		if current_height <= previous_height:
			no_growth_count += 1
			if no_growth_count >= max_no_growth:
				break
		else:
			no_growth_count = 0
			previous_height = current_height


def _clean_text(value: str) -> str:
	# Keep output plain text even if the source contains escaped entities or stray markup.
	text = html.unescape(value)
	text = re.sub(r"<[^>]+>", "", text)
	return " ".join(text.split())


async def wait_for_session_cards(page, retries: int = 2) -> bool:
	"""Wait for session cards to appear, with reload retries for flaky dynamic loads."""
	for _ in range(retries + 1):
		try:
			await page.wait_for_selector("a.link-block.session-card-header", timeout=15000)
		except PlaywrightTimeoutError:
			await page.reload(wait_until="domcontentloaded", timeout=120000)
			await page.wait_for_load_state("networkidle")
			continue

		if await page.locator("a.link-block.session-card-header").count() > 0:
			return True

	return False


async def ensure_program_loaded(page, attempts: int = 5) -> bool:
	"""Ensure we are not stuck on the lightweight shell page with no program cards."""
	for attempt in range(1, attempts + 1):
		if await wait_for_session_cards(page, retries=1):
			return True

		title = await page.title()
		header_count = await page.locator("a.link-block.session-card-header").count()
		print(
			f"Program not ready (attempt {attempt}/{attempts}): "
			f"title='{title}', session headers={header_count}. Retrying..."
		)

		if attempt < attempts:
			await page.wait_for_timeout(1200 * attempt)
			await page.reload(wait_until="domcontentloaded", timeout=120000)
			await page.wait_for_load_state("networkidle")

	return False


async def goto_with_retries(page, url: str, attempts: int = 3) -> None:
	"""Navigate with retries to tolerate transient network errors."""
	last_error: Exception | None = None
	for attempt in range(1, attempts + 1):
		try:
			await page.goto(url, wait_until="domcontentloaded", timeout=120000)
			return
		except PlaywrightError as exc:
			last_error = exc
			if attempt == attempts:
				raise
			await page.wait_for_timeout(1500 * attempt)

	if last_error:
		raise last_error


async def extract_session_paper_lines(page) -> list[str]:
	"""Iterate sessions and collect lines in the form: Session Name: Paper Title."""
	results: list[str] = []
	seen_lines: set[str] = set()

	session_cards = page.locator("session-card")
	session_count = await session_cards.count()

	for i in range(session_count):
		session_card = session_cards.nth(i)
		await session_card.scroll_into_view_if_needed()
		await page.wait_for_timeout(150)

		header = session_card.locator("a.link-block.session-card-header").first
		if await header.count() == 0:
			continue

		session_title_locator = session_card.locator(
			"a.link-block.session-card-header session-card-data program-card-name h4 span"
		).first
		if await session_title_locator.count() == 0:
			session_title_locator = session_card.locator(
				"a.link-block.session-card-header program-card-name .name"
			).first

		session_title = _clean_text(await session_title_locator.inner_text())

		# Expand session cards so nested papers become visible.
		names_before_count = await session_card.locator("program-card-name h4 span").count()
		aria_label = (await header.get_attribute("aria-label") or "").lower()
		if "collapsed" in aria_label or names_before_count <= 1:
			await header.click()
			await page.wait_for_timeout(700)

		# Prefer selectors that match expanded session body content cards.
		nested_selectors = [
			".session-card-body .content-card .name",
			".session-card-body .content-card h5 .name",
			".session-card-body .content-card h5 span.name",
		]

		session_paper_titles: list[str] = []
		try:
			await session_card.locator(".session-card-body .content-card").first.wait_for(timeout=2500)
		except PlaywrightTimeoutError:
			pass

		for selector in nested_selectors:
			items = session_card.locator(selector)
			if await items.count() == 0:
				continue

			for raw_text in await items.all_inner_texts():
				title = _clean_text(raw_text)
				if title and title != session_title:
					session_paper_titles.append(title)

			if session_paper_titles:
				break

		# Fallback: if nested selectors fail, parse h5 spans in expanded content cards.
		if not session_paper_titles:
			all_name_texts = await session_card.locator(".session-card-body .content-card h5 span").all_inner_texts()
			if all_name_texts:
				for raw_text in all_name_texts:
					title = _clean_text(raw_text)
					# Filter out schedule labels such as time ranges and the word "Papers".
					if (
						title
						and title != session_title
						and " - " not in title
						and title.lower() != "papers"
					):
						session_paper_titles.append(title)

		for title in session_paper_titles:
			line = f"{session_title}: {title}"
			if line not in seen_lines:
				seen_lines.add(line)
				results.append(line)

	# Global fallback for cases where content appears outside individual card subtrees.
	if not results:
		global_titles = await page.locator(".session-card-body .content-card .name").all_inner_texts()
		if not global_titles:
			global_titles = await page.locator(".content-card .name").all_inner_texts()
		for raw_text in global_titles:
			title = _clean_text(raw_text)
			if title:
				line = f"Unknown Session: {title}"
				if line not in seen_lines:
					seen_lines.add(line)
					results.append(line)

	return results


async def main() -> None:
	async with async_playwright() as p:
		browser = await p.chromium.launch(headless=True)
		page = await browser.new_page()

		print(f"Navigating to: {TARGET_URL}")
		await goto_with_retries(page, TARGET_URL)

		# Dynamic sites often need extra time for JavaScript-rendered content.
		await page.wait_for_load_state("networkidle")
		loaded = await ensure_program_loaded(page)
		if not loaded:
			raise RuntimeError(
				"Program content did not load after retries (session cards missing)."
			)

		print("Scrolling to load all lazy-loaded sessions/papers...")
		await scroll_to_load_all_content(page)
		await page.wait_for_load_state("networkidle")

		session_paper_lines = await extract_session_paper_lines(page)

		with OUTPUT_PATH.open("w", encoding="utf-8") as f:
			for line in session_paper_lines:
				f.write(f"{line}\n")

		title = await page.title()
		url = page.url
		html = await page.content()

		print(f"Loaded page title: {title}")
		print(f"Final URL: {url}")
		print(f"HTML size: {len(html)} chars")
		print(f"Extracted papers: {len(session_paper_lines)}")
		print(f"Saved to: {OUTPUT_PATH}")
		print("First 10 lines:")
		for line in session_paper_lines[:10]:
			print(f"- {line}")

		await browser.close()


if __name__ == "__main__":
	asyncio.run(main())