pip install scrapegraph-py

from scrapegraph_py import ScrapeGraphAI
 
sgai = ScrapeGraphAI(api_key="your-api-key-here")

from scrapegraph_py import ScrapeGraphAI
from pydantic import BaseModel, Field
from typing import List, Optional
import urllib.parse
 
class SearchResult(BaseModel):
    position: int = Field(description="Rank position in results (1, 2, 3...)")
    title: str = Field(description="Page title shown in result")
    url: str = Field(description="Destination URL")
    description: str = Field(description="Meta description/snippet shown")
    is_sponsored: bool = Field(description="Whether this is a paid/sponsored result")
    domain: str = Field(description="Root domain of the result")
 
class YahooSearchPage(BaseModel):
    query: str = Field(description="Search query used")
    total_results_estimate: Optional[str] = Field(description="Yahoo's estimated result count")
    results: List[SearchResult]
 
def scrape_yahoo_search(query: str, api_key: str) -> YahooSearchPage:
    """Scrape Yahoo search engine results for a given query."""
    sgai = ScrapeGraphAI(api_key=api_key)
 
    encoded_query = urllib.parse.quote_plus(query)
    url = f"https://search.yahoo.com/search?p={encoded_query}"
 
    response = sgai.extract(ExtractRequest(
        website_url=url,
        prompt=(
            f"Extract all search results for the query '{query}'. "
            "For each result, get the title, URL, description snippet, "
            "rank position, whether it's sponsored/paid, and the domain. "
            "Also capture the total estimated results count Yahoo shows."
        ),
        output_schema=YahooSearchPage
    ))
 
    sgai.close()
    return response['result']
 
# Usage
results = scrape_yahoo_search("best AI tools 2026", api_key="your-api-key")
print(f"Query: {results['query']}")
print(f"Total results: {results['total_results_estimate']}")
print()
for r in results['results']:
    label = "[AD]" if r['is_sponsored'] else f"[{r['position']}]"
    print(f"{label} {r['title']}")
    print(f"    {r['url']}")
    print(f"    {r['description'][:100]}...")

def scrape_yahoo_organic_results(query: str, api_key: str) -> list:
    """Scrape only organic (non-sponsored) Yahoo search results."""
    sgai = ScrapeGraphAI(api_key=api_key)
    encoded_query = urllib.parse.quote_plus(query)
 
    class OrganicResult(BaseModel):
        position: int
        title: str
        url: str
        description: str
        domain: str
 
    class OrganicResults(BaseModel):
        organic_results: List[OrganicResult] = Field(
            description="Only the non-sponsored, organic search listings"
        )
 
    response = sgai.extract(ExtractRequest(
        url=f"https://search.yahoo.com/search?p={encoded_query}",
        prompt=(
            "Extract only the organic (non-sponsored, non-paid) search results. "
            "Skip any results labeled as 'Ad', 'Sponsored', or marked with a yellow/green Ad badge. "
            "For each organic result, capture: position, title, URL, snippet description, and domain."
        ),
        output_schema=OrganicResults
    ))
 
    sgai.close()
    return response['result']['organic_results']
 
# Usage
organic = scrape_yahoo_organic_results("python web scraping", api_key="your-api-key")
for result in organic:
    print(f"#{result['position']} {result['title']} — {result['domain']}")

import time
 
def scrape_yahoo_search_pages(query: str, num_pages: int, api_key: str) -> list:
    """Scrape multiple pages of Yahoo search results."""
    sgai = ScrapeGraphAI(api_key=api_key)
    all_results = []
    encoded_query = urllib.parse.quote_plus(query)
 
    for page in range(num_pages):
        # Yahoo uses 'b' parameter for pagination (b=1, b=11, b=21, ...)
        start = page * 10 + 1
        url = f"https://search.yahoo.com/search?p={encoded_query}&b={start}"
 
        response = sgai.extract(ExtractRequest(
            website_url=url,
            prompt=f"Extract all search results from this Yahoo results page (page {page + 1}). Get title, URL, description, and whether it's sponsored."
        ))
 
        page_results = response['result'].get('results', [])
        # Add absolute position
        for r in page_results:
            if isinstance(r, dict):
                r['absolute_position'] = start + page_results.index(r)
        all_results.extend(page_results)
 
        print(f"Page {page + 1}: {len(page_results)} results")
        time.sleep(2)  # Be respectful with request timing
 
    sgai.close()
    return all_results
 
# Scrape top 3 pages
results = scrape_yahoo_search_pages("machine learning jobs", num_pages=3, api_key="your-api-key")
print(f"Total results scraped: {len(results)}")

from scrapegraph_py import ScrapeGraphAI
from pydantic import BaseModel, Field
from typing import List, Optional
 
class TrendingTopic(BaseModel):
    rank: int = Field(description="Trending rank (1 = most trending)")
    topic: str = Field(description="The trending search term or topic")
    category: Optional[str] = Field(description="Topic category (News, Entertainment, Sports, etc.)")
    trend_indicator: Optional[str] = Field(description="Search volume indicator if shown (e.g. '50K+ searches')")
    related_url: Optional[str] = Field(description="Yahoo URL for this trending topic")
 
class YahooTrending(BaseModel):
    trending_topics: List[TrendingTopic]
    captured_at: Optional[str] = Field(description="Timestamp indicator if shown on page")
 
def scrape_yahoo_trending_searches(api_key: str) -> YahooTrending:
    """Scrape Yahoo's trending searches."""
    sgai = ScrapeGraphAI(api_key=api_key)
 
    response = sgai.extract(ExtractRequest(
        url="https://www.yahoo.com/trending",
        prompt=(
            "Extract all trending search topics from Yahoo's trending page. "
            "For each trending item, get: rank position, topic name, category (News/Entertainment/Sports/etc.), "
            "search volume indicator if shown, and any Yahoo search URL associated with it."
        ),
        output_schema=YahooTrending
    ))
 
    sgai.close()
    return response['result']
 
# Usage
trending = scrape_yahoo_trending_searches(api_key="your-api-key")
print("Yahoo Trending Searches:\n")
for topic in trending['trending_topics']:
    category = f"[{topic['category']}]" if topic.get('category') else ""
    volume = topic.get('trend_indicator', '')
    print(f"#{topic['rank']} {topic['topic']} {category} {volume}")

import json
from datetime import datetime
 
def monitor_yahoo_trends(api_key: str, output_file: str = "yahoo_trends.jsonl"):
    """Append trending searches to a file for trend analysis over time."""
    trending = scrape_yahoo_trending_searches(api_key)
 
    entry = {
        "timestamp": datetime.now().isoformat(),
        "topics": trending['trending_topics']
    }
 
    with open(output_file, "a") as f:
        f.write(json.dumps(entry) + "\n")
 
    print(f"Saved {len(trending['trending_topics'])} trending topics at {entry['timestamp']}")
    return trending
 
# Run every hour via cron: 0 * * * * python monitor_trends.py
monitor_yahoo_trends(api_key="your-api-key")

from scrapegraph_py import ScrapeGraphAI
from pydantic import BaseModel, Field
from typing import List, Optional
import urllib.parse
 
class VideoResult(BaseModel):
    title: str = Field(description="Video title")
    url: str = Field(description="Video page URL on Yahoo or source platform")
    duration: Optional[str] = Field(description="Video duration (e.g. '5:32')")
    source: Optional[str] = Field(description="Source platform (YouTube, Dailymotion, etc.)")
    channel: Optional[str] = Field(description="Channel or creator name")
    views: Optional[str] = Field(description="View count if shown")
    published_date: Optional[str] = Field(description="Publication date if shown")
    thumbnail_url: Optional[str] = Field(description="Thumbnail image URL")
 
class YahooVideoResults(BaseModel):
    query: str
    videos: List[VideoResult]
 
def scrape_yahoo_videos(query: str, api_key: str) -> YahooVideoResults:
    """Build a Yahoo videos scraper for any search query."""
    sgai = ScrapeGraphAI(api_key=api_key)
    encoded_query = urllib.parse.quote_plus(query)
 
    response = sgai.extract(ExtractRequest(
        url=f"https://video.yahoo.com/search?q={encoded_query}",
        prompt=(
            f"Extract all video results for the query '{query}'. "
            "For each video, get: title, URL, duration, source platform (YouTube/Dailymotion/etc.), "
            "channel name, view count, publication date, and thumbnail URL."
        ),
        output_schema=YahooVideoResults
    ))
 
    sgai.close()
    return response['result']
 
# Usage
videos = scrape_yahoo_videos("machine learning tutorial", api_key="your-api-key")
print(f"Found {len(videos['videos'])} videos for '{videos['query']}':\n")
for video in videos['videos']:
    print(f"Title: {video['title']}")
    print(f"  Source: {video.get('source', 'Unknown')} | Duration: {video.get('duration', 'N/A')}")
    print(f"  Views: {video.get('views', 'N/A')} | Channel: {video.get('channel', 'N/A')}")
    print(f"  URL: {video['url']}")
    print()

def track_video_trends(topics: list, api_key: str) -> dict:
    """Track video content across multiple topics for trend analysis."""
    import time
 
    trend_data = {}
 
    for topic in topics:
        print(f"Scraping videos for: {topic}")
        results = scrape_yahoo_videos(topic, api_key)
        trend_data[topic] = {
            "count": len(results['videos']),
            "videos": results['videos'],
            "top_sources": {}
        }
 
        # Count videos by source platform
        for video in results['videos']:
            source = video.get('source', 'Unknown')
            trend_data[topic]["top_sources"][source] = \
                trend_data[topic]["top_sources"].get(source, 0) + 1
 
        time.sleep(2)
 
    return trend_data
 
topics = ["artificial intelligence", "machine learning", "large language models"]
data = track_video_trends(topics, api_key="your-api-key")
 
for topic, info in data.items():
    print(f"\n{topic}: {info['count']} videos")
    for source, count in sorted(info["top_sources"].items(), key=lambda x: -x[1]):
        print(f"  {source}: {count} videos")

from scrapegraph_py import ScrapeGraphAI
from pydantic import BaseModel, Field
from typing import List, Optional
import urllib.parse
 
class ImageResult(BaseModel):
    title: str = Field(description="Image title or alt text")
    image_url: str = Field(description="Direct URL to the image file")
    source_url: str = Field(description="URL of the page hosting this image")
    source_domain: str = Field(description="Domain of the source page")
    dimensions: Optional[str] = Field(description="Image dimensions if shown (e.g. '1920x1080')")
    format: Optional[str] = Field(description="Image format (JPG, PNG, GIF, etc.)")
 
class YahooImageResults(BaseModel):
    query: str
    images: List[ImageResult]
 
def scrape_yahoo_images(query: str, api_key: str) -> YahooImageResults:
    """Build a Yahoo images scraper for visual research."""
    sgai = ScrapeGraphAI(api_key=api_key)
    encoded_query = urllib.parse.quote_plus(query)
 
    response = sgai.extract(ExtractRequest(
        url=f"https://images.search.yahoo.com/search/images?p={encoded_query}",
        prompt=(
            f"Extract all image results for the query '{query}'. "
            "For each image, capture: title/alt text, direct image URL, "
            "source page URL, source domain, image dimensions if shown, "
            "and file format (JPG/PNG/GIF/WebP)."
        ),
        output_schema=YahooImageResults
    ))
 
    sgai.close()
    return response['result']
 
# Usage
images = scrape_yahoo_images("data visualization dashboard", api_key="your-api-key")
print(f"Found {len(images['images'])} images for '{images['query']}':\n")
for img in images['images'][:5]:
    print(f"Title: {img['title']}")
    print(f"  Format: {img.get('format', 'N/A')} | Dimensions: {img.get('dimensions', 'N/A')}")
    print(f"  Source: {img['source_domain']}")
    print()

from scrapegraph_py import ScrapeGraphAI
from pydantic import BaseModel, Field
from typing import List, Optional
import urllib.parse
 
class RelatedSearch(BaseModel):
    query: str = Field(description="The related search suggestion text")
    url: Optional[str] = Field(description="Yahoo search URL for this related query")
    position: int = Field(description="Position in the related searches list")
 
class RelatedSearches(BaseModel):
    original_query: str
    related_searches: List[RelatedSearch]
 
def scrape_yahoo_related_searches(query: str, api_key: str) -> RelatedSearches:
    """Scrape Yahoo's related search suggestions for keyword research."""
    sgai = ScrapeGraphAI(api_key=api_key)
    encoded_query = urllib.parse.quote_plus(query)
 
    response = sgai.extract(ExtractRequest(
        url=f"https://search.yahoo.com/search?p={encoded_query}",
        prompt=(
            "Extract the 'related searches' or 'people also search for' section "
            "from this Yahoo search results page. These are usually shown at the bottom "
            "of the page or in a sidebar. Get each related search query and its Yahoo URL."
        ),
        output_schema=RelatedSearches
    ))
 
    sgai.close()
    return response['result']
 
# Usage for keyword research
related = scrape_yahoo_related_searches("web scraping python", api_key="your-api-key")
print(f"Related searches for '{related['original_query']}':\n")
for search in related['related_searches']:
    print(f"  {search['position']}. {search['query']}")

import time
 
def build_keyword_cluster(seed_keyword: str, depth: int, api_key: str) -> dict:
    """
    Recursively expand a seed keyword using Yahoo related searches.
    depth=1: get related searches for seed
    depth=2: also get related searches for each of those
    """
    cluster = {seed_keyword: []}
 
    first_level = scrape_yahoo_related_searches(seed_keyword, api_key)
    first_queries = [r['query'] for r in first_level['related_searches']]
    cluster[seed_keyword] = first_queries
 
    if depth >= 2:
        for query in first_queries[:5]:  # Limit to 5 to avoid too many requests
            time.sleep(2)
            second_level = scrape_yahoo_related_searches(query, api_key)
            cluster[query] = [r['query'] for r in second_level['related_searches']]
 
    return cluster
 
# Example: expand "AI scraping" into a keyword cluster
cluster = build_keyword_cluster("AI scraping", depth=2, api_key="your-api-key")
for keyword, related in cluster.items():
    print(f"\n{keyword}:")
    for r in related:
        print(f"  - {r}")

import time
import json
from datetime import datetime
from scrapegraph_py import ScrapeGraphAI
 
class YahooMonitor:
    def __init__(self, api_key: str):
        self.sgai = ScrapeGraphAI(api_key=api_key)
        self.api_key = api_key
 
    def run_daily_report(self, keywords: list) -> dict:
        """Run a complete Yahoo intelligence report for a list of keywords."""
        report = {
            "generated_at": datetime.now().isoformat(),
            "keywords": {},
            "trending": None
        }
 
        # Get trending searches
        print("Fetching Yahoo trending searches...")
        try:
            trending = scrape_yahoo_trending_searches(self.api_key)
            report["trending"] = trending['trending_topics']
        except Exception as e:
            print(f"Trending scrape failed: {e}")
 
        # Process each keyword
        for keyword in keywords:
            print(f"\nProcessing: {keyword}")
            report["keywords"][keyword] = {}
 
            # Search results
            try:
                search = scrape_yahoo_organic_results(keyword, self.api_key)
                report["keywords"][keyword]["organic_results"] = search
                print(f"  Search results: {len(search)} organic listings")
            except Exception as e:
                print(f"  Search scrape failed: {e}")
 
            time.sleep(2)
 
            # Related searches
            try:
                related = scrape_yahoo_related_searches(keyword, self.api_key)
                report["keywords"][keyword]["related_searches"] = \
                    [r['query'] for r in related['related_searches']]
                print(f"  Related searches: {len(related['related_searches'])} queries")
            except Exception as e:
                print(f"  Related searches scrape failed: {e}")
 
            time.sleep(2)
 
        return report
 
    def save_report(self, report: dict, filename: str = None):
        """Save the report to a JSON file."""
        if not filename:
            filename = f"yahoo_report_{datetime.now().strftime('%Y%m%d_%H%M')}.json"
        with open(filename, "w") as f:
            json.dump(report, f, indent=2)
        print(f"\nReport saved to {filename}")
 
# Usage
monitor = YahooMonitor(api_key="your-api-key")
report = monitor.run_daily_report(keywords=["AI tools", "web scraping", "data extraction"])
monitor.save_report(report)

import time
 
def scrape_with_retry(url: str, prompt: str, api_key: str, max_retries: int = 3) -> dict:
    """Scrape with automatic retry on failure."""
    sgai = ScrapeGraphAI(api_key=api_key)
 
    for attempt in range(max_retries):
        try:
            response = sgai.extract(ExtractRequest(website_url=url, prompt=prompt))
            sgai.close()
            return response['result']
        except Exception as e:
            if attempt < max_retries - 1:
                wait = 2 ** attempt  # 1s, 2s, 4s
                print(f"Attempt {attempt + 1} failed, retrying in {wait}s: {e}")
                time.sleep(wait)
            else:
                sgai.close()
                raise
 
    return {}

response = sgai.extract(ExtractRequest(
    url="https://news.yahoo.com/",
    prompt="Extract the top 20 news headlines with title, URL, source, and publication time"
))