""" ScrapeGraphAI integration — semantic extraction of the 4 PDP sections (Scientific Studies / Results Timeline / Comparison Table / FAQs) from each of the 7 reference competitor brand PDPs that Umar called out on 21 May 2026. This is the ONE job where ScrapeGraphAI beats Apify and Playwright direct: each of the 7 sites has a different DOM, but the *semantic* structure of the 4 sections is the same. LLM-driven extraction handles all 7 with a single schema. Output: data/sources/competitor-section-extracts/.json — one file per brand, containing the 4 section blocks in our normalised schema. Usage: pip install -U scrapegraphai playwright pydantic playwright install chromium export OPENAI_API_KEY=sk-... # or use another supported provider python scripts/scrape-competitor-sections-sga.py Cost: ~$0.05-0.20 per page on gpt-4o-mini. Full 7-brand sweep ≈ $1-2. """ import json import os import sys from pathlib import Path from typing import List, Optional try: from pydantic import BaseModel, Field from scrapegraphai.graphs import SmartScraperGraph except ImportError: print("Missing deps. Run: pip install -U scrapegraphai playwright pydantic && playwright install chromium") sys.exit(1) ROOT = Path(__file__).resolve().parent.parent OUT_DIR = ROOT / "data" / "sources" / "competitor-section-extracts" OUT_DIR.mkdir(parents=True, exist_ok=True) # ---------- Target PDPs (from Umar's 21 May email + competitor list) ---------- COMPETITOR_PDPS: List[dict] = [ {"brand": "Wild Nutrition", "handle": "wild-nutrition", "url": "https://www.wildnutrition.com/products/food-grown-magnesium", "category": "mineral"}, {"brand": "Heights", "handle": "heights", "url": "https://www.heights.com/products/biotic", "category": "category"}, {"brand": "Spacegoods", "handle": "spacegoods", "url": "https://spacegoods.com/products/rainbow-dust-starter-kit", "category": "category"}, {"brand": "Dirtea", "handle": "dirtea", "url": "https://www.dirteaworld.com/products/pure-essentials-dirtea-creatine", "category": "category"}, {"brand": "Ancient & Brave", "handle": "ancient-and-brave", "url": "https://ancientandbrave.earth/", "category": "collagen"}, {"brand": "Elavate", "handle": "elavate", "url": "https://elavate.com/products/multi-collagen-main", "category": "collagen"}, {"brand": "Ritual", "handle": "ritual", "url": "https://ritual.com/products/hyacera-for-skin-hydration", "category": "collagen"}, {"brand": "Grüns", "handle": "gruns", "url": "https://gruns.co/", "category": "multivitamin"}, {"brand": "Club EarlyBird", "handle": "club-earlybird", "url": "https://clubearlybird.com/products/brain-drops", "category": "energy"}, {"brand": "Lumity Life", "handle": "lumity-life", "url": "https://lumitylife.co.uk/products/mens-morning-night-supplement", "category": "vitamin"}, {"brand": "MoonBrew", "handle": "moonbrew", "url": "https://moonbrew.co/", "category": "category"}, {"brand": "Nothing Fishy", "handle": "nothing-fishy", "url": "https://www.nothingfishy.co/products/complete-collagen", "category": "collagen"}, {"brand": "Hunter & Gather", "handle": "hunter-and-gather", "url": "https://hunterandgatherfoods.com/products/grass-fed-collagen-peptides-protein-powder", "category": "collagen"}, {"brand": "Absolute Collagen", "handle": "absolute-collagen", "url": "https://www.absolutecollagen.com/", "category": "collagen"}, {"brand": "Sunna Supplements", "handle": "sunna", "url": "https://sunnasupplements.com/", "category": "collagen"}, {"brand": "8Hours", "handle": "8hours", "url": "https://www.8-hours.com/products/8hours-daily-sleep-recovery-supplement", "category": "sleep"}, {"brand": "LIT", "handle": "lit", "url": "https://litsalt.com/products/lit-daily-hydration", "category": "hydration"}, ] # ---------- Normalised schema for the 4 sections ---------- class ScientificStudy(BaseModel): present: bool = Field(..., description="True if a scientific-studies / clinically-shown / science section is present on the page") headline: Optional[str] = None body_copy: Optional[str] = None citations: List[str] = Field(default_factory=list, description="URLs or study citations on the page") stat_tiles: List[dict] = Field(default_factory=list, description="[{value, label}] pairs shown as stats") notes: Optional[str] = None class TimelinePhase(BaseModel): phase_label: str = Field(..., description="e.g. 'Day 1', 'Week 1-2', 'Month 1'") body_copy: str class ResultsTimeline(BaseModel): present: bool headline: Optional[str] = None subtitle: Optional[str] = None intro_copy: Optional[str] = None phases: List[TimelinePhase] = Field(default_factory=list) notes: Optional[str] = None class ComparisonRow(BaseModel): feature_label: str us_value: Optional[str] = None competitor_values: List[str] = Field(default_factory=list) class ComparisonTable(BaseModel): present: bool section_title: Optional[str] = None subtitle: Optional[str] = None column_headers: List[str] = Field(default_factory=list, description="Including this brand and competitor columns") rows: List[ComparisonRow] = Field(default_factory=list) notes: Optional[str] = None class FaqPair(BaseModel): question: str answer: str class FaqBlock(BaseModel): present: bool section_title: Optional[str] = None cta_label: Optional[str] = None pairs: List[FaqPair] = Field(default_factory=list) notes: Optional[str] = None class PdpSections(BaseModel): brand: str url: str scientific_study: ScientificStudy results_timeline: ResultsTimeline comparison_table: ComparisonTable faq_block: FaqBlock PROMPT = """ You are extracting four specific PDP sections from this supplement / vitamin product page. Be honest about absence — if a section is not present, set `present: false` and leave fields empty. Do NOT invent content. Sections to extract: 1. **scientific_study** — any block titled "Scientific Studies", "The Science", "Clinically Shown", "Backed by Research", or similar. Capture the headline, any body paragraph, any explicit study citations (URLs/journal names), and any stat tiles (e.g. "32% better sleep"). 2. **results_timeline** — any structured "Results" / "Timeline" / "What to expect" block with phases like "Day 1 / Day 30 / Day 60" or "Month 1 / Month 2 / Month 3". Capture the headline, any subtitle, any intro paragraph, and the phase pairs. 3. **comparison_table** — any "Us vs Them", "How we compare", "Brand X vs Y", or feature-matrix block. Capture column headers (including this brand's own column), and rows as feature_label + this brand's value + each competitor's value in column order. 4. **faq_block** — the Frequently Asked Questions section. Capture each Q&A pair verbatim, plus any section title and CTA label (e.g. "Still unsure? Get in touch"). Return strict JSON matching the schema. Do not fabricate or paraphrase. """ def run_for_brand(brand: dict, model: str = "gpt-4o-mini") -> dict: config = { "llm": { "model": f"openai/{model}", "api_key": os.environ.get("OPENAI_API_KEY"), "temperature": 0, }, "verbose": False, "headless": True, } print(f"\n[{brand['handle']}] scraping {brand['url']}") graph = SmartScraperGraph( prompt=PROMPT, source=brand["url"], schema=PdpSections, config=config, ) result = graph.run() return { "brand": brand["brand"], "handle": brand["handle"], "url": brand["url"], "category": brand["category"], "extractedAt": __import__("datetime").datetime.utcnow().isoformat() + "Z", "model": model, "sections": result, "source": "ScrapeGraphAI SmartScraperGraph", } def main(): if not os.environ.get("OPENAI_API_KEY"): print("Set OPENAI_API_KEY before running. Example:") print(" $env:OPENAI_API_KEY = 'sk-...' # PowerShell") print(" export OPENAI_API_KEY=sk-... # bash") sys.exit(2) only = sys.argv[1:] if len(sys.argv) > 1 else None failures = [] for brand in COMPETITOR_PDPS: if only and brand["handle"] not in only: continue out_file = OUT_DIR / f"{brand['handle']}.json" try: data = run_for_brand(brand) out_file.write_text(json.dumps(data, indent=2, default=str), encoding="utf-8") print(f" -> {out_file.relative_to(ROOT)}") except Exception as e: failures.append((brand["handle"], str(e))) print(f" FAIL [{brand['handle']}]: {e}") if failures: print(f"\n{len(failures)} failures:") for h, e in failures: print(f" {h}: {e}") print(f"\nDone. Output: {OUT_DIR}") if __name__ == "__main__": main()