ScrapeGraphAI

pip install cognee-community-tasks-scrapegraph cognee

uv pip install cognee-community-tasks-scrapegraph cognee

export LLM_API_KEY="sk-..."
export SGAI_API_KEY="sgai-..."

import asyncio
from cognee_community_tasks_scrapegraph import scrape_urls

async def main():
    results = await scrape_urls(
        urls=[
            "https://cognee.ai",
            "https://docs.cognee.ai",
        ],
        user_prompt="Extract the product description, key features, and target use cases",
    )

    for item in results:
        if item.get("error"):
            print(f"[!] {item['url']}: {item['error']}")
        else:
            print(f"\n=== {item['url']} ===")
            print(str(item["content"])[:500])

asyncio.run(main())

import asyncio
import cognee
from cognee_community_tasks_scrapegraph import scrape_and_add

async def main():
    await cognee.prune.prune_data()
    await cognee.prune.prune_system(metadata=True)

    await scrape_and_add(
        urls=[
            "https://cognee.ai",
            "https://docs.cognee.ai",
            "https://github.com/topoteretes/cognee",
        ],
        user_prompt="Extract the product description, key features, and target use cases",
        dataset_name="cognee_research",
    )

    results = await cognee.recall(
        query_text="What is Cognee and what problems does it solve?",
        datasets=["cognee_research"],
    )
    for r in results:
        print(r)

asyncio.run(main())

import asyncio
import os
import cognee
from pydantic import BaseModel, Field
from scrapegraph_py import Client

class ProductPage(BaseModel):
    name: str = Field(description="Product or company name")
    tagline: str = Field(description="One-line value proposition")
    features: list[str] = Field(description="List of key features or capabilities")
    pricing_model: str = Field(description="How the product is priced")
    target_audience: str = Field(description="Who the product is primarily aimed at")

async def scrape_structured(urls: list[str]) -> list[str]:
    client = Client(api_key=os.environ["SGAI_API_KEY"])
    texts = []
    try:
        for url in urls:
            response = client.smartscraper(
                website_url=url,
                user_prompt="Extract product information",
                output_schema=ProductPage,
            )
            result = response.get("result", {})
            text = f"""Source: {url}
Product: {result.get('name', 'N/A')}
Tagline: {result.get('tagline', 'N/A')}
Features: {', '.join(result.get('features', []))}
Pricing: {result.get('pricing_model', 'N/A')}
Audience: {result.get('target_audience', 'N/A')}"""
            texts.append(text)
    finally:
        client.close()
    return texts

async def main():
    await cognee.prune.prune_data()
    await cognee.prune.prune_system(metadata=True)

    urls = ["https://cognee.ai", "https://scrapegraphai.com", "https://firecrawl.dev"]
    texts = await scrape_structured(urls)

    combined = "\n\n".join(texts)
    await cognee.remember(combined, dataset_name="product_landscape")

    results = await cognee.recall(
        query_text="How do these products differ in their approach?",
        datasets=["product_landscape"],
    )
    for r in results:
        print(r)

asyncio.run(main())