PyPI 包: olostep | 要求: Python 3.11+

安装

pip install olostep

认证

从 Olostep Dashboard 获取你的 API 密钥。

快速开始

根据你的使用场景，SDK 提供了两种客户端选项：

同步客户端 (`Olostep`)

最适合：脚本和简单用例，当你更喜欢阻塞操作时。

同步客户端提供了一个更简单的阻塞接口，如果你对 async/await 不熟悉，这会更容易上手。

异步客户端 (`AsyncOlostep`)

最适合：生产应用程序和处理大量并发请求。

异步客户端提供非阻塞操作，是需要高吞吐量的生产应用程序的推荐选择。

同步客户端 (Olostep)

同步客户端 (Olostep) 提供了一个阻塞接口，非常适合脚本和简单用例。

from olostep import Olostep

# 提供 API 密钥，可以通过传递 'api_key' 参数或设置 OLOSTEP_API_KEY 环境变量

# 同步客户端自动处理资源管理
# 不需要显式关闭 - 每次操作后资源会被清理
client = Olostep(api_key="YOUR_REAL_KEY")
scrape_result = client.scrapes.create(url_to_scrape="https://example.com")

基本网页抓取

from olostep import Olostep

client = Olostep(api_key="your-api-key")

# 简单抓取
result = client.scrapes.create(url_to_scrape="https://example.com")
print(f"抓取了 {len(result.html_content)} 个字符")

# 多种格式
result = client.scrapes.create(
    url_to_scrape="https://example.com",
    formats=["html", "markdown"]
)
print(f"HTML: {len(result.html_content)} 字符")
print(f"Markdown: {len(result.markdown_content)} 字符")

批量处理

from olostep import Olostep

client = Olostep(api_key="your-api-key")

# 高效处理多个 URL
batch = client.batches.create(
    urls=[
        "https://www.google.com/search?q=python",
        "https://www.google.com/search?q=javascript",
        "https://www.google.com/search?q=typescript"
    ]
)

# 等待完成并处理结果
for item in batch.items():
    content = item.retrieve(["html"])
    print(f"处理了 {item.url}: {len(content.html_content)} 字节")

智能网页爬取

from olostep import Olostep

client = Olostep(api_key="your-api-key")

# 使用智能过滤进行爬取
crawl = client.crawls.create(
    start_url="https://www.bbc.com",
    max_pages=100,
    include_urls=["/articles/**", "/blog/**"],
    exclude_urls=["/admin/**"]
)

for page in crawl.pages():
    content = page.retrieve(["html"])
    print(f"爬取了: {page.url}")

网站映射

from olostep import Olostep

client = Olostep(api_key="your-api-key")

# 从网站提取所有链接
maps = client.maps.create(url="https://example.com")

# 获取所有发现的 URL
urls = []
for url in maps.urls():
    urls.append(url)
    if len(urls) >= 10:  # 演示限制
        break

print(f"发现了 {len(urls)} 个 URL")

AI 驱动的答案

from olostep import Olostep

client = Olostep(api_key="your-api-key")

# 使用 AI 从网页获取答案
answer = client.answers.create(
    task="What is the main topic of https://example.com?"
)
print(f"答案: {answer.answer}")

异步客户端 (AsyncOlostep)

异步客户端 (AsyncOlostep) 是高性能应用程序、后端服务以及需要处理大量并发请求的推荐客户端。

from olostep import AsyncOlostep

# 提供 API 密钥，可以通过传递 'api_key' 参数或设置 OLOSTEP_API_KEY 环境变量

# 资源管理
# ===================
# SDK 支持两种资源管理使用模式：

# 1. 上下文管理器（推荐用于一次性使用）：
#    自动处理资源清理
async with AsyncOlostep(api_key="YOUR_REAL_KEY") as client:
    scrape_result = await client.scrapes.create(url_to_scrape="https://example.com")
# 传输在此处自动关闭

# 2. 显式关闭（用于长时间运行的服务）：
#    需要手动资源清理
client = AsyncOlostep(api_key="YOUR_REAL_KEY")
try:
    scrape_result = await client.scrapes.create(url_to_scrape="https://example.com")
finally:
    await client.close()  # 手动关闭传输

基本网页抓取

import asyncio
from olostep import AsyncOlostep

async def main():
    async with AsyncOlostep(api_key="your-api-key") as client:
        # 简单抓取
        result = await client.scrapes.create(url_to_scrape="https://example.com")
        print(f"抓取了 {len(result.html_content)} 个字符")

        # 多种格式
        result = await client.scrapes.create(
            url_to_scrape="https://example.com",
            formats=["html", "markdown"]
        )
        print(f"HTML: {len(result.html_content)} 字符")
        print(f"Markdown: {len(result.markdown_content)} 字符")

asyncio.run(main())

批量处理

import asyncio
from olostep import AsyncOlostep

async def main():
    async with AsyncOlostep(api_key="your-api-key") as client:
        # 高效处理多个 URL
        batch = await client.batches.create(
            urls=[
                "https://www.google.com/search?q=python",
                "https://www.google.com/search?q=javascript",
                "https://www.google.com/search?q=typescript"
            ]
        )

        # 等待完成并处理结果
        async for item in batch.items():
            content = await item.retrieve(["html"])
            print(f"处理了 {item.url}: {len(content.html_content)} 字节")

asyncio.run(main())

智能网页爬取

import asyncio
from olostep import AsyncOlostep

async def main():
    async with AsyncOlostep(api_key="your-api-key") as client:
        # 使用智能过滤进行爬取
        crawl = await client.crawls.create(
            start_url="https://www.bbc.com",
            max_pages=100,
            include_urls=["/articles/**", "/blog/**"],
            exclude_urls=["/admin/**"]
        )

        async for page in crawl.pages():
            content = await page.retrieve(["html"])
            print(f"爬取了: {page.url}")

asyncio.run(main())

网站映射

import asyncio
from olostep import AsyncOlostep

async def main():
    async with AsyncOlostep(api_key="your-api-key") as client:
        # 从网站提取所有链接
        maps = await client.maps.create(url="https://example.com")

        # 获取所有发现的 URL
        urls = []
        async for url in maps.urls():
            urls.append(url)
            if len(urls) >= 10:  # 演示限制
                break

        print(f"发现了 {len(urls)} 个 URL")

asyncio.run(main())

AI 驱动的答案

import asyncio
from olostep import AsyncOlostep

async def main():
    async with AsyncOlostep(api_key="your-api-key") as client:
        # 使用 AI 从网页获取答案
        answer = await client.answers.create(
            task="What is the main topic of https://example.com?"
        )
        print(f"答案: {answer.answer}")

asyncio.run(main())

SDK 参考

方法结构

两个 SDK 客户端都提供相同的清晰、Python 风格的接口，组织为逻辑命名空间：

命名空间	目的	关键方法
`scrapes`	单个 URL 提取	`create()`, `get()`
`batches`	多 URL 处理	`create()`, `info()`, `items()`
`crawls`	网站遍历	`create()`, `info()`, `pages()`
`maps`	链接提取	`create()`, `urls()`
`answers`	AI 驱动的提取	`create()`, `get()`
`retrieve`	内容检索	`get()`

每个操作都返回具有符合人体工程学方法的有状态对象，用于后续操作。

错误处理

使用基础异常类捕获所有 SDK 错误：

from olostep import Olostep, Olostep_BaseError

client = Olostep(api_key="your-api-key")

try:
    result = client.scrapes.create(url_to_scrape="https://example.com")
except Olostep_BaseError as e:
    print(f"发生错误: {type(e).__name__}")
    print(f"错误信息: {e}")

有关详细的错误处理信息，包括完整的异常层次结构和细粒度的错误处理选项，请参见详细错误处理。

自动重试

SDK 会根据 RetryStrategy 配置自动重试瞬态错误（网络问题、临时服务器问题）。你可以在创建客户端时通过传递 RetryStrategy 实例来自定义重试行为：

from olostep import Olostep, RetryStrategy

retry_strategy = RetryStrategy(
    max_retries=3,
    initial_delay=1.0,
    jitter_min=0.2,
    jitter_max=0.8
)

client = Olostep(api_key="your-api-key", retry_strategy=retry_strategy)
result = client.scrapes.create("https://example.com")

有关详细的重试配置选项和最佳实践，请参见重试策略。

高级功能

智能输入强制

SDK 智能处理各种输入格式，以最大方便性：

from olostep import Olostep, Country

client = Olostep(api_key="your-api-key")

# 格式：字符串、列表或枚举
client.scrapes.create(url_to_scrape="https://example.com", formats="html")
client.scrapes.create(url_to_scrape="https://example.com", formats=["html", "markdown"])

# 国家：不区分大小写的字符串或枚举
client.scrapes.create(url_to_scrape="https://example.com", country="us")
client.scrapes.create(url_to_scrape="https://example.com", country=Country.US)

# 列表：单个值或列表
client.batches.create(urls="https://example.com")    # 单个 URL
client.batches.create(urls=["https://a.com", "https://b.com"])  # 多个 URL

高级抓取选项

from olostep import Olostep, Format, Country, WaitAction, FillInputAction

client = Olostep(api_key="your-api-key")

# 完全控制抓取行为
result = client.scrapes.create(
    url_to_scrape="https://news.google.com/",
    wait_before_scraping=3000,
    formats=[Format.HTML, Format.MARKDOWN],
    remove_css_selectors=["script", ".popup"],
    actions=[
        WaitAction(milliseconds=1500),
        FillInputAction(selector="searchbox", value="olostep")
    ],
    parser="@olostep/google-news",
    country=Country.US,
    remove_images=True
)

带自定义 ID 的批量处理

from olostep import Olostep, Country

client = Olostep(api_key="your-api-key")

batch = client.batches.create([
    {"url": "https://www.google.com/search?q=python", "custom_id": "search_1"},
    {"url": "https://www.google.com/search?q=javascript", "custom_id": "search_2"},
    {"url": "https://www.google.com/search?q=typescript", "custom_id": "search_3"}
],
country=Country.US,
parser="@olostep/google-search"
)

# 按自定义 ID 处理结果
# 使用解析器时，检索 JSON 内容而不是 HTML
for item in batch.items():
    if item.custom_id == "search_2":
        content = item.retrieve(["json"])
        print(f"搜索结果: {content.json_content}")

智能爬取

from olostep import Olostep

client = Olostep(api_key="your-api-key")

# 使用智能过滤进行爬取
crawl = client.crawls.create(
    start_url="https://www.bbc.com",
    max_pages=1000,
    max_depth=3,
    include_urls=["/articles/**", "/news/**"],
    exclude_urls=["/ads/**", "/tracking/**"],
    include_external=False,
    include_subdomain=True,
)

for page in crawl.pages():
    content = page.retrieve(["html"])
    print(f"爬取了: {page.url}")

带过滤器的网站映射

from olostep import Olostep

client = Olostep(api_key="your-api-key")

# 提取所有链接并进行高级过滤
maps = client.maps.create(
    url="https://www.bbc.com",
    include_subdomain=True,
    include_urls=["/articles/**", "/news/**"],
    exclude_urls=["/ads/**", "/tracking/**"]
)

# 获取过滤后的 URL
urls = []
for url in maps.urls():
    urls.append(url)

print(f"发现了 {len(urls)} 个相关 URL")

答案检索

from olostep import Olostep

client = Olostep(api_key="your-api-key")

# 首先创建一个答案
created_answer = client.answers.create(
    task="What is the main topic of https://example.com?"
)

# 然后使用 ID 检索它
answer = client.answers.get(answer_id=created_answer.id)
print(f"答案: {answer.answer}")

内容检索

from olostep import Olostep

client = Olostep(api_key="your-api-key")

# 通过检索 ID 获取内容
result = client.retrieve.get(retrieve_id="ret_123")

# 获取多种格式
result = client.retrieve.get(retrieve_id="ret_123", formats=["html", "markdown", "text", "json"])

日志记录

启用日志记录以调试问题：

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("olostep")
logger.setLevel(logging.INFO)  # 使用 DEBUG 获取详细输出

日志级别: INFO（推荐），DEBUG（详细），WARNING，ERROR

重试策略配置

RetryStrategy 类通过自动重试、指数退避和抖动控制 Olostep SDK 如何处理瞬态 API 错误。这有助于确保在生产环境中可靠运行，其中临时网络问题、速率限制和服务器过载可能导致间歇性故障。

默认行为

默认情况下，SDK 使用以下重试配置：

最大重试次数: 5 次尝试
初始延迟: 2 秒
退避: 指数（2^attempt）
抖动: 延迟的 10-90%（随机）

这意味着：

尝试 1：立即
尝试 2：~2-3.6 秒延迟
尝试 3：~4-7.2 秒延迟
尝试 4：~8-14.4 秒延迟
尝试 5：~16-28.8 秒延迟

最大持续时间：~57 秒用于所有重试（最坏情况）

自定义配置

from olostep import AsyncOlostep, RetryStrategy

# 创建自定义重试策略
retry_strategy = RetryStrategy(
    max_retries=3,
    initial_delay=1.0,
    jitter_min=0.2,  # 最小抖动 20%
    jitter_max=0.8,  # 最大抖动 80%
)

# 与客户端一起使用
async with AsyncOlostep(
    api_key="your-api-key",
    retry_strategy=retry_strategy
) as client:
    result = await client.scrapes.create("https://example.com")

何时发生重试

SDK 会自动重试：

临时服务器问题 (OlostepServerError_TemporaryIssue)
超时响应 (OlostepServerError_NoResultInResponse)

其他错误（认证、验证、资源未找到等）立即失败，不重试。

传输与调用者重试

SDK 有两个重试层：

传输层：处理网络级连接失败（DNS、超时等）
调用者层：处理 API 级瞬态错误（由 RetryStrategy 控制）

两个层是独立的，并且有单独的配置。总的最大持续时间是两个层的总和。

计算最大持续时间

retry_strategy = RetryStrategy(max_retries=5, initial_delay=2.0)
max_duration = retry_strategy.max_duration()
print(f"最大调用持续时间: {max_duration:.2f}s")

配置示例

以下是如何为不同用例配置重试策略的一些示例。

保守策略

# 更少的重试，较短的延迟
retry_strategy = RetryStrategy(
    max_retries=3,
    initial_delay=1.0,
    jitter_min=0.2,
    jitter_max=0.8
)
# 最大持续时间：~12.6s

激进策略

# 关键操作的更多重试
retry_strategy = RetryStrategy(
    max_retries=10,
    initial_delay=0.5
)
# 最大持续时间：~969.75s

无重试（快速失败）

# 禁用重试以立即反馈失败
retry_strategy = RetryStrategy(max_retries=0)

client = AsyncOlostep(api_key="your-api-key", retry_strategy=retry_strategy)

高吞吐量策略

# 为高容量操作优化
retry_strategy = RetryStrategy(
    max_retries=2,
    initial_delay=0.5,
    jitter_min=0.1,
    jitter_max=0.3  # 较低的抖动以获得更可预测的时间
)
# 最大持续时间：~1.95s

理解抖动

抖动添加随机化以防止“雷鸣般的羊群”问题，当许多客户端同时重试时。抖动计算如下：

base_delay = initial_delay * (2 ** attempt)
jitter_range = base_delay * (jitter_max - jitter_min)
jitter = random.uniform(base_delay * jitter_min, base_delay * jitter_min + jitter_range)
final_delay = base_delay + jitter

例如，使用 initial_delay=2.0，jitter_min=0.1，jitter_max=0.9：

尝试 0：基数=2.0s，抖动=0.2-1.8s，最终=2.2-3.8s
尝试 1：基数=4.0s，抖动=0.4-3.6s，最终=4.4-7.6s
尝试 2：基数=8.0s，抖动=0.8-7.2s，最终=8.8-15.2s

最佳实践

对于生产应用程序

# 生产的平衡方法
retry_strategy = RetryStrategy(
    max_retries=5,
    initial_delay=2.0,
    jitter_min=0.1,
    jitter_max=0.9
)

对于开发/测试

# 开发的快速反馈
retry_strategy = RetryStrategy(
    max_retries=2,
    initial_delay=0.5,
    jitter_min=0.1,
    jitter_max=0.3
)

对于批量操作

# 大批量作业的保守策略
retry_strategy = RetryStrategy(
    max_retries=3,
    initial_delay=1.0,
    jitter_min=0.2,
    jitter_max=0.8
)

监控和调试

SDK 在 DEBUG 级别记录重试信息：

DEBUG: 临时问题，2.34s 后重试
DEBUG: 响应中无结果，4.67s 后重试

启用调试日志以监控重试行为：

import logging
logging.getLogger("olostep").setLevel(logging.DEBUG)

错误处理

当所有重试都耗尽时，抛出原始错误：

try:
    result = await client.scrapes.create("https://example.com")
except OlostepServerError_TemporaryIssue as e:
    print(f"所有重试后失败: {e}")
    # 处理永久失败

性能考虑

内存：每次重试尝试使用额外的内存用于请求/响应对象
时间：启用重试时，总操作时间可能会显著延长
API 限制：重试计入你的 API 使用限制
网络：由于重试尝试，更多的网络流量

根据你的应用程序对可靠性与性能的要求选择重试策略。

详细错误处理

异常层次结构

Olostep SDK 为不同的故障场景提供了一个全面的异常层次结构。所有异常都继承自 Olostep_BaseError。有三种主要错误类型直接继承自 Olostep_BaseError：

Olostep_APIConnectionError - 网络级连接失败
OlostepServerError_BaseError - API 服务器引发的错误
OlostepClientError_BaseError - 客户端 SDK 引发的错误

为什么连接错误是分开的

Olostep_APIConnectionError 与服务器错误分开，因为它表示在 API 处理请求之前发生的网络级失败。这些是传输层问题（DNS 或 HTTP 失败、超时、连接被拒绝等），而不是 API 级错误。HTTP 状态码（4xx, 5xx）被视为 API 响应，并被分类为服务器错误，尽管它们表明存在问题。

Olostep_BaseError
├── Olostep_APIConnectionError
├── OlostepServerError_BaseError
│   ├── OlostepServerError_TemporaryIssue
│   │   ├── OlostepServerError_NetworkBusy
│   │   └── OlostepServerError_InternalNetworkIssue
│   ├── OlostepServerError_RequestUnprocessable
│   │   ├── OlostepServerError_ParserNotFound
│   │   └── OlostepServerError_OutOfResources
│   ├── OlostepServerError_BlacklistedDomain
│   ├── OlostepServerError_FeatureApprovalRequired
│   ├── OlostepServerError_AuthFailed
│   ├── OlostepServerError_CreditsExhausted
│   ├── OlostepServerError_InvalidEndpointCalled
│   ├── OlostepServerError_ResourceNotFound
│   ├── OlostepServerError_NoResultInResponse
│   └── OlostepServerError_UnknownIssue
└── OlostepClientError_BaseError
    ├── OlostepClientError_RequestValidationFailed
    ├── OlostepClientError_ResponseValidationFailed
    ├── OlostepClientError_NoAPIKey
    ├── OlostepClientError_AsyncContext
    ├── OlostepClientError_BetaFeatureAccessRequired
    └── OlostepClientError_Timeout

细粒度错误处理

如果你需要更具体的错误处理，直接捕获特定的错误类型。避免使用 OlostepServerError_BaseError 或 OlostepClientError_BaseError - 这些基类仅指示谁引发了错误（服务器 vs 客户端），而不是谁负责修复它。这是一个实现细节，对错误处理逻辑没有帮助。相反，捕获指示实际问题的特定错误类型：

from olostep import (
    AsyncOlostep,
    Olostep_BaseError,
    Olostep_APIConnectionError,
    OlostepServerError_AuthFailed,
    OlostepServerError_CreditsExhausted,
    OlostepClientError_NoAPIKey,
)

try:
    result = await client.scrapes.create(url_to_scrape="https://example.com")
except Olostep_APIConnectionError as e:
    print(f"网络错误: {type(e).__name__}")
except OlostepServerError_AuthFailed:
    print("无效的 API 密钥")
except OlostepServerError_CreditsExhausted:
    print("信用已用尽")
except OlostepClientError_NoAPIKey:
    print("未提供 API 密钥")
except Olostep_BaseError as e:
    print(f"发生错误: {type(e).__name__}")

配置

环境变量

变量	描述	默认值
`OLOSTEP_API_KEY`	你的 API 密钥	必需
`OLOSTEP_BASE_API_URL`	API 基础 URL	`https://api.olostep.com/v1`
`OLOSTEP_API_TIMEOUT`	请求超时（秒）	`150`

​安装

​认证

​快速开始

同步客户端 (`Olostep`)

异步客户端 (`AsyncOlostep`)

​同步客户端 (Olostep)

​基本网页抓取

​批量处理

​智能网页爬取

​网站映射

​AI 驱动的答案

​异步客户端 (AsyncOlostep)

​基本网页抓取

​批量处理

​智能网页爬取

​网站映射

​AI 驱动的答案

​SDK 参考

​方法结构

​错误处理

​自动重试

​高级功能

​智能输入强制

​高级抓取选项

​带自定义 ID 的批量处理

​智能爬取

​带过滤器的网站映射

​答案检索

​内容检索

​日志记录

​重试策略配置

​默认行为

​自定义配置

​何时发生重试

​传输与调用者重试

​计算最大持续时间

​配置示例

​保守策略

​激进策略

​无重试（快速失败）

​高吞吐量策略

​理解抖动

​最佳实践

​对于生产应用程序

​对于开发/测试

​对于批量操作

​监控和调试

​错误处理

​性能考虑

​详细错误处理

​异常层次结构

​为什么连接错误是分开的

​推荐的错误处理

​细粒度错误处理

​配置

​环境变量

​获取帮助

​资源

PyPI 包

获取 API 密钥

安装

认证

快速开始

同步客户端 (Olostep)

基本网页抓取

批量处理

智能网页爬取

网站映射

AI 驱动的答案

异步客户端 (AsyncOlostep)

基本网页抓取

批量处理

智能网页爬取

网站映射

AI 驱动的答案

SDK 参考

方法结构

错误处理

自动重试

高级功能

智能输入强制

高级抓取选项

带自定义 ID 的批量处理

智能爬取

带过滤器的网站映射

答案检索

内容检索

日志记录

重试策略配置

默认行为

自定义配置

何时发生重试

传输与调用者重试

计算最大持续时间

配置示例

保守策略

激进策略

无重试（快速失败）

高吞吐量策略

理解抖动

最佳实践

对于生产应用程序

对于开发/测试

对于批量操作

监控和调试

错误处理

性能考虑

详细错误处理

异常层次结构

为什么连接错误是分开的

推荐的错误处理

细粒度错误处理

配置

环境变量

获取帮助

资源