scrape.py

"""Stage-0 web acquisition: Australian Government sites -> Markdown.

Pipeline position: scrape.py -> build/extracted/*.md -> extract.py -> Data Recipes.
Scrapy and html2text are optional (extras=scrape); imported lazily so users without
the scrape extras can still import this module for testing path-traversal guards.
"""
from __future__ import annotations

import argparse
import re
import sys
import tempfile
from pathlib import Path


# ── Optional dependency probe (module-level try/except; NO top-level scrapy import) ─────

try:
    import scrapy as _scrapy  # noqa: WPS433
    import html2text as _html2text  # noqa: WPS433
    _SCRAPY_AVAILABLE = True
    _SpiderBase: type = _scrapy.Spider
except ImportError:
    _SCRAPY_AVAILABLE = False
    _scrapy = None  # type: ignore[assignment]
    _html2text = None  # type: ignore[assignment]
    _SpiderBase = object


# ── Path traversal guard ──────────────────────────────────────────────────────

_ALLOWED_ABSOLUTE_PREFIXES = (tempfile.gettempdir(),)


def _validate_path(path: str) -> str:
    """Validate that the path is safe to use as an output destination.

    This addresses CR-04 (Path Traversal). Rejects:
    - Relative paths that use ``..`` components to escape the working directory.
    - Absolute paths that are not under cwd or the system temp directory.

    Absolute paths are allowed only under cwd or $TMPDIR (/tmp). All other
    absolute paths (including arbitrary user home directories) are rejected.
    """
    # Reject any path whose components include traversal sequences
    if ".." in Path(path).parts:
        raise ValueError(
            f"Path traversal detected: {path} is outside allowed directory."
        )
    root = Path.cwd().resolve()
    candidate = Path(path).resolve()
    try:
        candidate.relative_to(root)
        return str(candidate)
    except ValueError:
        pass
    # Absolute path outside cwd — only permit paths under the system temp directory
    resolved_str = str(candidate)
    if any(resolved_str.startswith(prefix) for prefix in _ALLOWED_ABSOLUTE_PREFIXES):
        return str(candidate)
    raise ValueError(
        f"Path traversal detected: {path} is outside allowed directory."
    )


# ── Scrapy item pipeline ──────────────────────────────────────────────────────

class MarkdownFilePipeline:
    """Scrapy item pipeline: collects markdown per spider, writes one .md file each."""

    def __init__(self, output_dir: str):
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)
        self._buffers: dict[str, list[str]] = {}

    @classmethod
    def from_crawler(cls, crawler):
        return cls(output_dir=crawler.settings.get("OUTPUT_DIR", "build/extracted"))

    def process_item(self, item, spider):
        key = spider.name
        self._buffers.setdefault(key, []).append(item["markdown"])
        return item

    def close_spider(self, spider):
        key = spider.name
        content = "\n\n---\n\n".join(self._buffers.get(key, []))
        out_path = self.output_dir / f"{key}-web.md"
        out_path.write_text(content, encoding="utf-8")
        print(f"[scrape] done  {key}  -> {out_path}  {len(content)} chars", flush=True)


# ── CrawlerProcess driver ─────────────────────────────────────────────────────

def _make_settings(output_dir: str, robotstxt: bool = True) -> dict:
    """Build shared CrawlerProcess settings dict."""
    return {
        "OUTPUT_DIR": output_dir,
        "DOWNLOAD_DELAY": 1.5,
        "ROBOTSTXT_OBEY": robotstxt,
        "DEPTH_LIMIT": 2,
        "ITEM_PIPELINES": {"scrape.MarkdownFilePipeline": 300},
        "LOG_LEVEL": "INFO",
        "USER_AGENT": "offlineaid-pack-builder/0.1 (+https://github.com/helenkwok/offlineaid-pack-builder)",
    }


def run_spiders(output_dir: str, spider_classes: list, robotstxt: bool | None = None) -> None:
    """Run the given spider classes via CrawlerProcess (D-06).

    `robotstxt`: if None (default), derive from each spider's custom_settings.
    When running as a subprocess (via CLI --no-robotstxt flag), pass False directly
    to avoid re-partitioning, since the Twisted reactor can only start once per process.
    """
    try:
        from scrapy.crawler import CrawlerProcess
    except ImportError:
        print("Install scrape dependencies: uv sync --extra scrape", flush=True)
        sys.exit(1)

    if robotstxt is not None:
        # Called from subprocess or explicit override: run all spiders with given setting.
        process = CrawlerProcess(_make_settings(output_dir, robotstxt=robotstxt))
        for spider_cls in spider_classes:
            process.crawl(spider_cls)
        process.start()
        return

    # Partition spiders by their requested ROBOTSTXT_OBEY setting.
    # Necessary because CrawlerProcess bleeds custom_settings across spiders,
    # and Twisted's reactor cannot restart — so each group runs in a subprocess.
    robots_true: list = []
    robots_false: list = []
    for spider_cls in spider_classes:
        cs = getattr(spider_cls, "custom_settings", {})
        if cs.get("ROBOTSTXT_OBEY", True) is False:
            robots_false.append(spider_cls)
        else:
            robots_true.append(spider_cls)

    # Run robots-respecting spiders in-process (this process starts the reactor).
    if robots_true:
        process = CrawlerProcess(_make_settings(output_dir, robotstxt=True))
        for spider_cls in robots_true:
            process.crawl(spider_cls)
        process.start()  # blocks; reactor started here

    # Run robots-exempt spiders in a subprocess (reactor already started above).
    if robots_false:
        import subprocess as _subprocess
        site_names = [cls.name for cls in robots_false]
        cmd = [sys.executable, __file__, "--output-dir", output_dir, "--no-robotstxt", "--sites", *site_names]
        result = _subprocess.run(cmd, check=False)
        if result.returncode not in (0, 1):  # exit 1 = partial scrape, not fatal
            print(
                f"[scrape] WARNING subprocess for sites={site_names} exited {result.returncode}",
                file=sys.stderr, flush=True,
            )


# ── CSS boilerplate stripping helper (D-08) ───────────────────────────────────

# D-08: CSS selector stripping list — removes nav/footer/aside boilerplate before conversion.
# Strips: "nav", "footer", "aside", "header" (structural chrome)
# Also strips: ".cookie-banner", "#cookie-notice", ".breadcrumb", site-specific wrappers.
_BOILERPLATE_SELECTORS = [
    "nav", "footer", "aside", "header",
    ".cookie-banner", "#cookie-notice", ".breadcrumb",
    ".skip-link", ".site-header", ".site-footer",
]


def _strip_and_convert(response) -> str:  # type: ignore[no-untyped-def]
    """Drop boilerplate then convert remaining body HTML to Markdown (D-08)."""
    for selector in _BOILERPLATE_SELECTORS:
        for el in response.css(selector):
            if el.root.getparent() is not None:
                el.root.getparent().remove(el.root)
    body_html = (
        response.css("main").get()
        or response.css("article").get()
        or response.css(".content").get()
        or response.css("body").get()
        or ""
    )
    converter = _html2text.HTML2Text()
    converter.ignore_links = False
    converter.body_width = 0  # disable wrapping; preserve paragraph structure
    return converter.handle(body_html)


# ── Base spider class ─────────────────────────────────────────────────────────

class _BaseGovSpider(_SpiderBase):  # type: ignore[misc, valid-type]
    """Base class for all Australian Government site spiders."""

    custom_settings = {
        "DOWNLOAD_DELAY": 1.5,
        "ROBOTSTXT_OBEY": True,
        "DEPTH_LIMIT": 2,
        "CONCURRENT_REQUESTS_PER_DOMAIN": 2,
    }

    def parse(self, response):  # type: ignore[no-untyped-def]
        markdown = _strip_and_convert(response)
        if markdown.strip():
            yield {"url": response.url, "markdown": f"## {response.url}\n\n{markdown}"}


# ── Spider classes at module level (D-05: three government sites) ─────────────

# D-05: Source #1 — ACCC Scamwatch website
class ScamwatchSpider(_BaseGovSpider):
    name = "scamwatch"
    allowed_domains = ["scamwatch.gov.au", "www.scamwatch.gov.au"]
    # URLs verified 2026-04-30; old /protect-yourself/* and sub-paths were 404 after site restructure.
    # New structure uses /stop-check-protect/ and updated /types-of-scams/ paths.
    start_urls = [
        "https://www.scamwatch.gov.au/types-of-scams",
        "https://www.scamwatch.gov.au/types-of-scams/buying-and-selling-scams",
        "https://www.scamwatch.gov.au/types-of-scams/investment-scams",
        "https://www.scamwatch.gov.au/types-of-scams/relationship-scams",
        "https://www.scamwatch.gov.au/types-of-scams/jobs-and-employment-scams",
        "https://www.scamwatch.gov.au/types-of-scams/threat-scams",
        "https://www.scamwatch.gov.au/types-of-scams/unexpected-money-scams",
        "https://www.scamwatch.gov.au/types-of-scams/phishing-scams",
        "https://www.scamwatch.gov.au/stop-check-protect",
        "https://www.scamwatch.gov.au/stop-check-protect/help-to-spot-and-avoid-scams",
        "https://www.scamwatch.gov.au/stop-check-protect/what-to-do-if-youve-been-scammed",
    ]


# D-05: Source #2 — Australian Government Disaster Assist
class DisasterAssistSpider(_BaseGovSpider):
    name = "disasterassist"
    allowed_domains = ["disasterassist.gov.au", "www.disasterassist.gov.au"]
    # URLs verified 2026-04-30; /staying-safe/* and /getting-help/* sub-paths returned 404
    # after a site restructure — replaced with confirmed-200 equivalents.
    start_urls = [
        "https://www.disasterassist.gov.au/find-a-disaster",
        "https://www.disasterassist.gov.au/getting-help",
        "https://www.disasterassist.gov.au/pages/disaster-recovery-allowance.aspx",
        "https://www.disasterassist.gov.au/disaster-arrangements",
        "https://www.disasterassist.gov.au/disaster-arrangements/disaster-recovery-funding-arrangements",
        "https://www.disasterassist.gov.au/find-a-disaster/australian-disasters",
        "https://www.disasterassist.gov.au/pages/about-us.aspx",
    ]


# D-05: Source #3 — Department of Foreign Affairs Smartraveller
class SmartravellerSpider(_BaseGovSpider):
    name = "smartraveller"
    allowed_domains = ["smartraveller.gov.au", "www.smartraveller.gov.au"]
    # Two mitigations needed for smartraveller.gov.au:
    # 1. ROBOTSTXT_OBEY=False: smartraveller.gov.au/robots.txt times out under Scrapy's
    #    twisted async handler (3× 180s retries = ~9min hang). curl confirms accessibility.
    # 2. USER_AGENT override: site blocks Scrapy's default bot UA (Scrapy/2.15.2) —
    #    drops the TCP connection silently. A browser UA (Chrome/122) passes through.
    _BROWSER_UA = (
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
        "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36"
    )
    custom_settings = {
        **_BaseGovSpider.custom_settings,
        "ROBOTSTXT_OBEY": False,
        "USER_AGENT": _BROWSER_UA,
    }
    # URLs verified 2026-04-30; /while-youre-away/staying-safe, /crisis/*, deprecated.
    # Current site structure uses /while-youre-away/crime/*, /while-youre-away/crisis-or-emergency.
    start_urls = [
        "https://www.smartraveller.gov.au/before-you-go",
        "https://www.smartraveller.gov.au/before-you-go/safety",
        "https://www.smartraveller.gov.au/before-you-go/health",
        "https://www.smartraveller.gov.au/while-youre-away",
        "https://www.smartraveller.gov.au/while-youre-away/crime",
        "https://www.smartraveller.gov.au/while-youre-away/crime/scammed",
        "https://www.smartraveller.gov.au/while-youre-away/crisis-or-emergency",
        "https://www.smartraveller.gov.au/while-youre-away/when-things-go-wrong",
    ]


# ── Spider registry ───────────────────────────────────────────────────────────

if _SCRAPY_AVAILABLE:
    SPIDER_REGISTRY: dict[str, type] = {
        "scamwatch": ScamwatchSpider,
        "disasterassist": DisasterAssistSpider,
        "smartraveller": SmartravellerSpider,
    }
else:
    SPIDER_REGISTRY = {}  # missing extras; main() will print install hint and exit 1


# ── CLI entrypoint ────────────────────────────────────────────────────────────

def main() -> None:
    """CLI entry point: parse args, validate path, run spiders."""
    parser = argparse.ArgumentParser(
        description="Stage-0 web acquisition: Australian Government sites -> Markdown"
    )
    parser.add_argument(
        "--output-dir",
        required=True,
        help="Directory to write per-site .md files into (e.g. build/extracted)",
    )
    parser.add_argument(
        "--sites",
        nargs="+",
        default=["scamwatch", "disasterassist", "smartraveller"],
        help="Site names to crawl (default: all three)",
    )
    parser.add_argument(
        "--no-robotstxt",
        action="store_true",
        default=False,
        help="Disable ROBOTSTXT_OBEY for all spiders in this invocation (used by subprocess group runner)",
    )
    args = parser.parse_args()

    # Path traversal guard (T-07-02-01)
    try:
        validated_output_dir = _validate_path(args.output_dir)
    except ValueError as exc:
        print(f"[scrape] ERROR {exc}", file=sys.stderr, flush=True)
        sys.exit(1)

    # If scrapy not available, SPIDER_REGISTRY will be empty
    if not SPIDER_REGISTRY:
        print("Install scrape dependencies: uv sync --extra scrape", flush=True)
        sys.exit(1)

    # Resolve --sites to spider classes; log unknown names
    selected_classes = []
    for name in args.sites:
        if name in SPIDER_REGISTRY:
            selected_classes.append(SPIDER_REGISTRY[name])
        else:
            print(f"[scrape] skipping unknown site: {name}", flush=True)

    if not selected_classes:
        print("[scrape] ERROR no valid sites to crawl", file=sys.stderr, flush=True)
        sys.exit(1)

    print(
        f"[scrape] starting  sites={args.sites}  output_dir={validated_output_dir}",
        flush=True,
    )

    # If --no-robotstxt is set, pass False explicitly (subprocess mode).
    robotstxt_override: bool | None = False if args.no_robotstxt else None

    try:
        run_spiders(validated_output_dir, selected_classes, robotstxt=robotstxt_override)
    except Exception as exc:
        print(f"[scrape] ERROR {exc}", file=sys.stderr, flush=True)
        sys.exit(1)


if __name__ == "__main__":
    main()