[Python] How to scrape the full New York Times article content using Selenium and...

Discussão em 'Python' iniciado por Stack, Abril 18, 2025.

  1. Stack

    Stack Membro Participativo

    I'm building a scraper that fetches full article content from the New York Times using both the Article Search API and a hybrid static + Selenium-based HTML scraper. My goal is to extract complete article text for stock-related news (e.g., "NVDA Stock").

    The problem: I only get a few paragraphs, and the rest of the body is replaced with the following:


    We are having trouble retrieving the article content. Please enable JavaScript in your browser settings. Thank you for your patience while we verify access... Already a subscriber? Log in.
    ​

    I already use Selenium (headless mode) and wait explicitly for . As a fallback, I also parse the NEXT_DATA JSON to find the full text. Here's a simplified snippet of my scraping logic:

    Code

    # ───── Imports ─────
    import json, time, os, re, requests, json as js
    from datetime import datetime
    from typing import List, Dict, Any
    from bs4 import BeautifulSoup
    from requests.adapters import HTTPAdapter
    from urllib3.util.retry import Retry
    from selenium import webdriver
    from selenium.webdriver.chrome.options import Options
    from selenium.webdriver.common.by import By
    from selenium.webdriver.support.ui import WebDriverWait
    from selenium.webdriver.support import expected_conditions as EC

    # ───── CONFIG ─────
    API_URL = "https://api.nytimes.com/svc/search/v2/articlesearch.json"
    API_KEY = os.getenv("NYT_API_KEY", "Your_API_KEY")
    QUERY = "NVDA Stock"
    BEGIN_DATE, END_DATE = "20240801", "20240802" # YYYYMMDD
    SORT = "oldest"
    SECTIONS = ["Business", "Technology"] # [] → no filter
    PAGES_TO_FETCH, ARTICLES_PER_PAGE = 9, 10

    SCRAPE_BODY = True
    HEADLESS = True
    SEL_TIMEOUT = 15

    # ───── Requests session ─────
    session = requests.Session()
    session.headers.update({"User-Agent": "Mozilla/5.0"})
    session.mount(
    "https://",
    HTTPAdapter(
    max_retries=Retry(
    total=5, backoff_factor=2,
    status_forcelist=[429, 500, 502], allowed_methods=["GET"]
    )
    ),
    )

    # ───── Selenium driver ─────
    def make_driver(headless=True):
    opts = Options()
    if headless:
    opts.add_argument("--headless=new")
    opts.add_argument("--no-sandbox")
    opts.add_argument("--disable-dev-shm-usage")
    opts.add_argument("--disable-blink-features=AutomationControlled")
    opts.add_argument("--window-size=1920,1080")
    return webdriver.Chrome(options=opts)

    driver = make_driver(HEADLESS)

    # ───── Body‑scraping helpers ─────
    def _join_ps(ps):
    return " ".join(p.get_text(" ", strip=True) for p in ps)

    def _extract_paragraphs(container: BeautifulSoup) -> str:
    return _join_ps(container.find_all("p"))

    def _from_next_data(soup: BeautifulSoup) -> str:
    """Last‑ditch: NYT puts the article text in a big JSON blob."""
    node = soup.find("script", id="__NEXT_DATA__", type="application/json")
    if not node:
    return ""
    try:
    data = js.loads(node.string)
    # walk until we find an 'article' dict with 'body'
    stack: List[Any] = [data]
    while stack:
    cur = stack.pop()
    if isinstance(cur, dict):
    if "article" in cur and isinstance(cur["article"], dict):
    body = cur["article"].get("body") or cur["article"].get("articleBody")
    if body and isinstance(body, str):
    # strip HTML tags quickly
    return BeautifulSoup(body, "html.parser").get_text(" ", strip=True)
    stack.extend(cur.values())
    elif isinstance(cur, list):
    stack.extend(cur)
    except Exception:
    pass
    return ""

    def _scrape_static(url: str) -> str:
    try:
    r = session.get(url, timeout=10)
    r.raise_for_status()
    soup = BeautifulSoup(r.content, "html.parser")

    # 1) Most current template (metered & free)
    sec = soup.find("section", {"data-testid": "article-body"}) \
    or soup.find("section", {"id": "article-body"})
    if sec:
    return _extract_paragraphs(sec)

    # 2) Paywalled variant: <section class="meteredContent">
    sec = soup.find("section", class_=re.compile(r"meteredContent"))
    if sec:
    return _extract_paragraphs(sec)

    # 3) Newsletter / brief: <div class="css-1r7ky0e"> (hash class)
    sec = soup.find("div", class_=re.compile(r"css-[a-z0-9]+"))
    if sec and sec.find("p"):
    return _extract_paragraphs(sec)

    # 4) Legacy template
    sec = soup.find("section", {"name": "articleBody"})
    if sec:
    return _extract_paragraphs(sec)

    # 5) Individual paragraph tags
    paras = soup.find_all("p", {"data-testid": "paragraph"})
    if paras:
    return _join_ps(paras)

    # 6) __NEXT_DATA__ JSON blob
    body = _from_next_data(soup)
    if body:
    return body

    # 7) Fallback: any <p> inside article/main
    container = soup.find("article") or soup.find("main") or soup
    return _extract_paragraphs(container)
    except Exception:
    return ""

    def _scrape_with_selenium(url: str) -> str:
    try:
    driver.get(url)
    WebDriverWait(driver, SEL_TIMEOUT).until(
    EC.presence_of_element_located(
    (By.CSS_SELECTOR, "section[data-testid='article-body'], section#article-body, article p")
    )
    )
    soup = BeautifulSoup(driver.page_source, "html.parser")
    container = soup.find("section", {"data-testid": "article-body"}) \
    or soup.find("section", {"id": "article-body"}) \
    or soup.find("article") \
    or soup.find("main") \
    or soup
    return _extract_paragraphs(container)
    except Exception:
    return ""

    def get_body(url: str) -> str:
    if not SCRAPE_BODY:
    return ""
    text = _scrape_static(url)
    if text:
    return text
    return _scrape_with_selenium(url)

    # ───── API helpers ─────
    def build_params(page: int, with_filter: bool) -> Dict:
    p = {
    "q": QUERY, "begin_date": BEGIN_DATE, "end_date": END_DATE,
    "sort": SORT, "api-key": API_KEY, "page": page,
    }
    if with_filter and SECTIONS:
    quoted = " ".join(f'"{s}"' for s in SECTIONS)
    p["fq"] = f'(section_name:({quoted}) OR news_desk:({quoted}))'
    return p

    # ───── Main fetch routine ─────
    def fetch_articles() -> List[Dict]:
    collected, seen = [], set()
    for use_filter in (True, False if SECTIONS else False):
    probe = session.get(API_URL, params=build_params(0, use_filter), timeout=10)
    probe.raise_for_status()
    resp = probe.json()["response"]
    hits = (resp.get("meta") or resp.get("metadata", {})).get("hits", 0)
    docs0 = resp.get("docs") or []
    perpg = len(docs0)
    if hits == 0 or perpg == 0:
    if use_filter:
    print("0 hits with section filter → retrying without it …")
    continue
    print("No articles found.")
    return []
    pages = min(PAGES_TO_FETCH, -(-hits // perpg))
    print(f"Fetching {pages} page(s) (hits={hits}, server_filter={use_filter})")

    for pg in range(pages):
    r = session.get(API_URL, params=build_params(pg, use_filter), timeout=10)
    if r.status_code == 429:
    print("Rate‑limit 429 → sleeping 10 s"); time.sleep(10)
    r = session.get(API_URL, params=build_params(pg, use_filter), timeout=10)
    r.raise_for_status()
    for d in r.json()["response"]["docs"][:ARTICLES_PER_PAGE]:
    url = d.get("web_url", "")
    if not url or url in seen: continue
    seen.add(url)
    if not use_filter and SECTIONS:
    if d.get("section_name") not in SECTIONS and d.get("news_desk") not in SECTIONS:
    continue
    pub = d.get("pub_date") or ""
    date_str = datetime.strptime(pub, "%Y-%m-%dT%H:%M:%S%z").strftime("%Y-%m-%d") if pub else ""
    collected.append(
    {
    "url": url,
    "published_date": date_str,
    "section": d.get("section_name") or d.get("news_desk", ""),
    "headline": d.get("headline", {}).get("main", ""),
    "abstract": d.get("abstract", ""),
    "body": get_body(url),
    }
    )
    print(f" • Page {pg+1}/{pages} done; total {len(collected)}")
    time.sleep(2)
    break
    return collected

    # ───── Run & save ─────
    if __name__ == "__main__":
    try:
    arts = fetch_articles()
    out = datetime.now().strftime("nyt_%Y%m%d_%H%M%S.json")
    with open(out, "w", encoding="utf-8") as fp:
    json.dump(arts, fp, indent=2, ensure_ascii=False)
    print(f"Done — wrote {len(arts)} articles to {out}")
    finally:
    driver.quit()

    Continue reading...

Compartilhe esta Página