创建爬虫¶

# omnidata/data_sources/myplatform/spider.py
from pydantic import BaseModel, Field
from omnidata.core.base_web_spider import BaseWebSpider, SpiderResult

class MyParams(BaseModel):
    """请求参数模型"""
    url: str = Field(..., description="目标 URL")
    timeout: int = Field(default=30000, description="超时时间（毫秒）")

class MyPlatformActionSpider(BaseWebSpider):
    """爬虫类"""

    # 基本信息
    name = "myplatform_action"           # 爬虫名称（唯一）
    description = "我的爬虫描述"          # 描述
    version = "1.0.0"                     # 版本
    author = "your_name"                  # 作者
    platform = "我的平台"                 # 平台名称（中文）

    # 参数模型
    params_model = MyParams

    async def crawl(self, params: MyParams) -> SpiderResult:
        """爬虫核心逻辑"""
        async with self.new_page(namespace="my_namespace") as page:
            # 设置超时
            page.set_default_timeout(params.timeout)

            # 访问页面
            await page.goto(params.url)

            # 提取数据
            title = await page.title()

            return SpiderResult(
                success=True,
                data={
                    "title": title,
                    "url": page.url
                }
            )

from pydantic import BaseModel, Field

class StockQuoteParams(BaseModel):
    """股票行情参数"""
    secucode: str = Field(
        ...,
        description="股票代码",
        min_length=6,
        max_length=9
    )
    fields: list[str] = Field(
        default=["name", "price"],
        description="返回字段列表"
    )

# 自动验证
params = StockQuoteParams(
    secucode="000001",    # ✓ 正确
    secucode="123"        # ✗ 验证失败
)

# 方式1：自动管理（推荐）
async with self.new_page(namespace="my_namespace") as page:
    await page.goto("https://example.com")
    # page 会自动关闭

# 方式2：手动管理
context = await self.get_context(namespace="my_namespace")
page = await context.new_page()
try:
    await page.goto("https://example.com")
finally:
    await page.close()

# 导航
await page.goto(url)
await page.go_back()
await page.reload()

# 等待
await page.wait_for_selector(".content")
await page.wait_for_timeout(1000)

# 提取数据
title = await page.title()
text = await page.locator(".content").text_content()
html = await page.locator(".content").inner_html()

# 点击
await page.click(".button")

# 填写表单
await page.fill("#input", "text")

# 截图
screenshot = await page.screenshot()

from omnidata.core.base_web_spider import SpiderResult

return SpiderResult(
    success=True,                    # 是否成功
    data={...},                      # 返回数据
    metadata={                       # 元数据
        "url": page.url,
        "timestamp": int(time.time())
    }
)

try:
    data = await self._extract_data(page)
    return SpiderResult(success=True, data=data)
except Exception as e:
    return SpiderResult(
        success=False,
        error=str(e)
    )

from pydantic import BaseModel, Field
from omnidata.core.base_web_spider import BaseWebSpider, SpiderResult

class EastmoneyStockQuoteParams(BaseModel):
    """东方财富股票行情参数"""
    secucode: str = Field(..., description="股票代码，如 000001")

class EastmoneyStockQuoteSpider(BaseWebSpider):
    """东方财富股票行情爬虫"""
    name = "eastmoney_stock_quote"
    description = "获取股票实时行情"
    version = "1.0.0"
    author = "noimank"
    platform = "东方财富"
    params_model = EastmoneyStockQuoteParams

    async def crawl(self, params: EastmoneyStockQuoteParams) -> SpiderResult:
        """获取股票行情"""
        url = f"http://push2.eastmoney.com/api/qt/stock/get"
        async with self.new_page(namespace="eastmoney") as page:
            response = await page.goto(
                url,
                params={"secid": f"1.{params.secucode}"}
            )

            data = await response.json()

            if data and "data" in data:
                return SpiderResult(
                    success=True,
                    data={
                        "name": data["data"]["name"],
                        "price": data["data"]["f60"],
                        "change": data["data"]["f169"]
                    }
                )

            return SpiderResult(
                success=False,
                error="获取数据失败"
            )