Meal-Planner/scripts/spike_lucky_scrape.py

"""
R2-A live-scrape spike for Lucky California weekly ad.

Performs ONE live fetch of https://luckysupermarkets.com weekly-ad page,
saves the rendered HTML, a full-page screenshot, and a META.md describing
the fetch result. This is the deferred-risk spike demanded by review §2.4.

Run once. Do not loop. Polite citizen: identifying user-agent, single
request, no auth bypass attempts.

Usage:
    python scripts/spike_lucky_scrape.py
"""

from __future__ import annotations

import sys
from datetime import datetime, timezone
from pathlib import Path

from playwright.sync_api import sync_playwright

# Path to backend/app on import path so we can reuse the parser.
REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO_ROOT / "backend"))

from app.scraper.lucky_ca_scraper import LuckyCaliforniaScraper  # noqa: E402
from bs4 import BeautifulSoup  # noqa: E402

FIXTURE_DIR = REPO_ROOT / "backend" / "tests" / "fixtures" / "lucky_ca"
TARGET_URL = "https://luckysupermarkets.com/coupons/Coupon%2Flu-featured-in-ad"
USER_AGENT = (
    "Mozilla/5.0 (compatible; MealPlannerSpike/0.1; "
    "+https://github.com/MealPlanner; spike=R2-A)"
)


def main() -> int:
    FIXTURE_DIR.mkdir(parents=True, exist_ok=True)
    html_path = FIXTURE_DIR / "weekly_ad.html"
    png_path = FIXTURE_DIR / "weekly_ad.png"
    meta_path = FIXTURE_DIR / "META.md"

    started = datetime.now(timezone.utc)
    print(f"[spike] Fetching {TARGET_URL}")

    status_code: int | None = None
    final_url: str = TARGET_URL
    error: str | None = None
    captcha_or_block = False
    captcha_signal = ""

    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        try:
            ctx = browser.new_context(user_agent=USER_AGENT)
            page = ctx.new_page()

            response = None

            def _capture(resp):
                nonlocal response
                # Capture only the main document response.
                if response is None and resp.url.rstrip("/") == TARGET_URL.rstrip("/"):
                    response = resp

            page.on("response", _capture)

            try:
                nav_resp = page.goto(
                    TARGET_URL,
                    wait_until="networkidle",
                    timeout=45_000,
                )
                if nav_resp is not None:
                    status_code = nav_resp.status
                    final_url = nav_resp.url
                elif response is not None:
                    status_code = response.status
                    final_url = response.url
            except Exception as exc:  # noqa: BLE001 -- spike, capture and report
                error = f"{type(exc).__name__}: {exc}"

            try:
                html = page.content()
            except Exception as exc:  # noqa: BLE001
                html = ""
                error = (error or "") + f" content_err={exc}"

            html_path.write_text(html, encoding="utf-8")

            try:
                page.screenshot(path=str(png_path), full_page=True)
            except Exception as exc:  # noqa: BLE001
                error = (error or "") + f" screenshot_err={exc}"

            # Heuristic captcha / block detection.
            lower = html.lower()
            for needle in (
                "captcha",
                "are you a robot",
                "access denied",
                "akamai",
                "cloudflare",
                "px-captcha",
                "perimeterx",
                "incapsula",
            ):
                if needle in lower:
                    captcha_or_block = True
                    captcha_signal = needle
                    break

        finally:
            browser.close()

    # Run parser portion against the captured HTML (no network).
    scraper = LuckyCaliforniaScraper()
    items: list[dict] = scraper.parse_featured_coupons_html(html)
    _ = BeautifulSoup  # keep import for type-stability if reused later

    completed = datetime.now(timezone.utc)

    meta = f"""# Lucky California weekly-ad spike (R2-A)

| Field | Value |
| --- | --- |
| URL | {TARGET_URL} |
| Final URL | {final_url} |
| Started (UTC) | {started.isoformat()} |
| Completed (UTC) | {completed.isoformat()} |
| HTTP status | {status_code} |
| HTML bytes | {len(html)} |
| Items parsed | {len(items)} |
| Captcha/block signal | {"YES (" + captcha_signal + ")" if captcha_or_block else "no"} |
| Error | {error or "none"} |
| User-Agent | `{USER_AGENT}` |

## First parsed item (sample)

```json
{__import__("json").dumps(items[0], indent=2) if items else "null"}
```

## Notes

- Single live fetch performed. Do not rerun without reason.
- HTML and PNG saved alongside this file.
- Parser used: `LuckyCaliforniaScraper._parse_coupon_item` against
  elements matching `h2/h3/a` with `$N.NN` text.
"""
    meta_path.write_text(meta, encoding="utf-8")

    print(f"[spike] status={status_code} html_bytes={len(html)} items={len(items)} "
          f"block={captcha_or_block}")
    print(f"[spike] wrote: {html_path}")
    print(f"[spike] wrote: {png_path}")
    print(f"[spike] wrote: {meta_path}")

    if captcha_or_block:
        print("[spike] WARNING: captcha/anti-bot signal detected; review META.md")
    if not items:
        print("[spike] WARNING: zero items parsed; selectors may be stale")

    return 0


if __name__ == "__main__":
    raise SystemExit(main())