spider-console-sdk 0.1.0

Spider SDK：统一把爬虫运行状态/指标/错误上报到监控平台

spider_sdk 使用说明

一句话：spider_sdk 只做运行上报（start/heartbeat/end/error），且上报失败不影响主流程。

---

目录

1. 快速上手
2. Job 模式
3. Service 模式
4. with 上下文管理器模式
5. 错误上报机制
6. 动态身份（并发不串号）
7. 参数配置总规则
8. 关键参数速查
9. 平台判定说明
10. FAQ
11. 推荐公开接口

---

快速上手

第一步：放置 spider_sdk.ini

在项目根目录（或任意父目录）放置配置文件，SDK 自动向上查找：

[spider_sdk]
platform_url = https://spider-console.example.com
platform_key = replace-with-platform-key
timeout_sec = 2
max_retries = 2
heartbeat_min_sec = 10
error_detail_max = 8192
error_event_max = 200
enable_ip = true

spider_sdk.ini 只放固定项（连接/限速参数），业务身份参数（爬虫名、类型等）写在装饰器里。

---

Job 模式

适用场景：定时/触发式爬虫，有明确的开始和结束。

最简写法

from spider_sdk import report_job


@report_job(
    reporter_kwargs={
        "spider_name": "news_parser_job",
        "spider_type": "parser",
        "env": "prod",
    },
    run_name="daily_news_parse",
    timeout_sec=1800,   # 平台任务卡死阈值（秒），超过后平台自动标记 timeout
)
def run_parser():
    # 正常返回 → 自动上报 run_end=success
    # 抛出异常 → 自动上报 run_end=failed
    pass

带过程心跳和错误上报

开启 inject_ctx=True，函数签名接收 reporter_ctx，可在过程中调心跳和错误事件：

from spider_sdk import report_job


@report_job(
    reporter_kwargs={
        "spider_name": "news_parser_job",
        "spider_type": "parser",
        "env": "prod",
        "tags": ["news", "parser"],
        "version": "v1.0.0",
        "data_source": "news_site",
        "source_category": "news",
    },
    run_name="daily_news_parse",
    timeout_sec=1800,
    inject_ctx=True,
)
def run_parser(reporter_ctx=None):
    success_count = 0
    fail_count = 0
    total = 100

    for i, url in enumerate(get_url_list()):
        try:
            crawl(url)
            success_count += 1
        except Exception as e:
            fail_count += 1
            reporter_ctx.report_error(
                error_type="CrawlError",
                error_message=str(e),
                stage="fetch",
                item_url=url,
            )

        # 每处理一批上报一次心跳（SDK 内置最小间隔保护，频繁调用安全）
        reporter_ctx.heartbeat(
            progress_current=i + 1,
            progress_total=total,
            metrics={"success_count": success_count, "fail_count": fail_count},
            message="crawling",
        )

    # 返回 dict 且含 "metrics" key，自动作为 run_end 的汇总指标
    return {"metrics": {"items": total, "success_count": success_count, "fail_count": fail_count}}

reporter_ctx 可用方法

方法	说明
reporter_ctx.heartbeat(message, progress_current, progress_total, metrics, last_error_type, last_error_message)	上报心跳/进度，含最小间隔保护（heartbeat_min_sec），可频繁调用
reporter_ctx.report_error(error_type, error_message, error_detail, stage, item_url, item_id, count)	上报单条过程错误事件，超过 error_event_max 上限后自动丢弃

---

Service 模式

适用场景：长驻服务型爬虫（FastAPI、Scrapy 常驻服务等），无明确结束时间。

from spider_sdk import report_service


@report_service(
    reporter_kwargs={
        "spider_name": "parser_service",
        "spider_type": "fastapi_service",
        "env": "prod",
        "tags": ["service", "parser"],
        "version": "v1.0.0",
    },
    heartbeat_interval_sec=30,   # SDK 自动每 30 秒发一次心跳
    heartbeat_message="healthy",
    up_message="service started",
    down_reason="shutdown",
)
def start_service():
    # 进入时自动发 service_up（成功才会在退出时发 service_down）
    # 退出时自动发 service_down
    # 中途每 heartbeat_interval_sec 自动发一次 service_heartbeat
    uvicorn.run(app, host="0.0.0.0", port=8000)

带动态心跳指标

通过 heartbeat_metrics_provider 在每次心跳时采集实时指标：

def get_metrics():
    return {"queue_size": queue.qsize(), "workers": active_workers()}


@report_service(
    reporter_kwargs={"spider_name": "worker_service", "spider_type": "worker", "env": "prod"},
    heartbeat_interval_sec=30,
    heartbeat_metrics_provider=get_metrics,   # 每次心跳前调用，返回值作为 metrics
)
def start_worker():
    run_forever()

注意：heartbeat_metrics_provider 的签名必须与被装饰函数的原始参数一致（或无参），不要声明 reporter_ctx 参数。

service_up / service_down 保证

• service_up 上报成功 → 退出时发 service_down
• service_up 上报失败（网络不通） → 不发 service_down，避免平台收到孤儿下线事件

---

with 上下文管理器模式

适用场景：不想用装饰器，或需要在代码块内精细控制 run 生命周期。

from spider_sdk import SpiderReportContext


with SpiderReportContext(
    reporter_kwargs={
        "spider_name": "manual_job",
        "spider_type": "script",
        "env": "prod",
    },
    run_name="manual_run",
    timeout_sec=600,
) as ctx:
    # __enter__ 自动 start，ctx.run_id 可用
    for url in url_list:
        try:
            crawl(url)
        except Exception as e:
            ctx.report_error(error_type="CrawlError", error_message=str(e), item_url=url)

        ctx.heartbeat(message="running")

    # 正常退出 __exit__ → 自动 success
    # 抛出异常 → 自动 fail，异常继续上抛

SpiderReportContext 可用属性和方法

名称	说明
ctx.run_id	当前 run 的 ID（__enter__ 之后可用）
ctx.heartbeat(...)	同 reporter_ctx.heartbeat
ctx.report_error(...)	同 reporter_ctx.report_error

---

错误上报机制

自动失败上报

情况	结果
函数正常返回	自动 run_end=success
函数抛出未捕获异常	自动 run_end=failed，含堆栈
try/except 吞掉异常（没有 raise）	视为 success，不会触发 failed

过程错误上报（report_error）

• 适合"部分失败但整体成功"的场景，如批量抓取中单条 URL 失败
• 单次 run 最多上报 error_event_max（默认 200）条，超出静默丢弃
• 建议在业务层控制上报量（如前 N 条），避免撑满上限后丢失关键错误

# 推荐：只上报前 3 条，后续只累加计数
if fail_count <= 3:
    reporter_ctx.report_error(
        error_type="ParseError",
        error_message=str(e),
        stage="parse",
        item_url=url,
    )

常见误区：异常被吞导致意外 success

# ❌ 这样写：装饰器看到的是"正常返回"，会上报 success
def main():
    try:
        crawl()
    except Exception as e:
        logging.error(e)   # 吞掉了

# ✅ 需要 failed 时：显式 raise
def main():
    try:
        crawl()
    except Exception as e:
        logging.error(e)
        raise RuntimeError("crawl failed") from e

---

动态身份（并发不串号）

多爬虫并发时，通过 runtime_resolver 在运行时动态注入身份字段，避免不同爬虫实例共用同一个 spider_name：

from spider_sdk import report_job


@report_job(
    reporter_kwargs={"spider_type": "scrapy", "env": "prod"},
    runtime_resolver=lambda spider, **kwargs: {
        "spider_name": spider.name,
        "tags": ["scrapy", spider.name],
        "data_source": spider.start_urls[0] if spider.start_urls else None,
    },
)
def run_spider(spider):
    spider.start()

runtime_resolver 接收被装饰函数的全部参数，返回的字段会覆盖 reporter_kwargs 里的同名字段（优先级最高）。

---

参数配置总规则

优先级从低到高：

spider_sdk.ini（固定项） < reporter_kwargs（业务身份） < runtime_resolver 返回值（动态覆盖）

• spider_sdk.ini：只放连接和限速参数，所有爬虫共用
• reporter_kwargs：放爬虫身份参数，每个爬虫独立配置
• runtime_resolver：并发/框架场景下动态生成身份，优先级最高

report_job_from_env(...) 为历史函数名，当前等价于 report_job(reporter_kwargs={"mode": "job"}, ...)，从 spider_sdk.ini 读取固定项。

---

关键参数速查

spider_sdk.ini 固定项（8 个）

字段	必填	默认	说明
platform_url	是	无	平台地址
platform_key	是	无	上报密钥（X-Spider-Key Header）
timeout_sec	否	2	SDK 上报 HTTP 超时（秒）
max_retries	否	2	SDK 上报重试次数
heartbeat_min_sec	否	10	心跳最小发送间隔（SDK 防抖，非平台判定）
error_detail_max	否	8192	错误详情字段截断长度（字符数）
error_event_max	否	200	单次 run 最多上报错误事件条数
enable_ip	否	true	是否上报 runtime.ip

report_job 装饰器参数

参数	默认	说明
reporter_kwargs	{}	爬虫身份参数，见下表
run_name	None	本次运行名（可选，平台展示用）
trigger	"schedule"	触发方式：schedule / manual / api
timeout_sec	None	平台任务卡死阈值（秒），上报给平台用于超时判定
inject_ctx	False	True 时注入 reporter_ctx 参数，用于过程心跳/错误上报
ctx_arg_name	"reporter_ctx"	注入参数名（可自定义）
success_metrics	None	固定汇总指标，函数返回 dict 含 metrics key 时自动提取
result_to_metrics	None	自定义函数，从返回值提取指标
runtime_resolver	None	动态身份覆盖函数
fail_error_type	"UnknownError"	异常时上报的错误类型
fail_stage	"other"	异常时上报的阶段

reporter_kwargs 爬虫身份参数

字段	必填	说明
spider_name	是	爬虫名称
spider_type	是	爬虫类型（如 parser / scrapy / fastapi_service）
env	否	环境：dev / staging / prod（默认 prod）
mode	否	运行模式：job / service（装饰器自动设置，一般不需要手填）
tags	否	标签列表或逗号字符串
version	否	爬虫版本
data_source	否	数据来源（站点 URL 或名称）
source_category	否	来源类别（如 news / social_media）

report_service 装饰器参数

参数	默认	说明
reporter_kwargs	{}	爬虫身份参数（同上）
heartbeat_interval_sec	30.0	心跳发送频率（秒）
heartbeat_message	"service heartbeat"	心跳消息
heartbeat_metrics	None	固定心跳指标
heartbeat_metrics_provider	None	动态心跳指标函数，每次心跳前调用
up_message	"service started"	上线消息
down_reason	"shutdown"	下线原因
inject_ctx	False	注入 reporter_ctx（service 模式下 run_id 为 None）
runtime_resolver	None	动态身份覆盖函数
error_type	"ServiceError"	服务异常时上报的错误类型
error_stage	"other"	服务异常时上报的阶段

---

平台判定说明

行为	控制方	参数
任务卡死判定（job）	平台	report_job(timeout_sec=...) 上报给平台
服务失联判定（service）	平台	平台策略，通常为若干心跳周期未收到
心跳发送频率（service）	SDK	heartbeat_interval_sec
心跳最小间隔防抖（job）	SDK	heartbeat_min_sec（ini 配置）
service_down 是否发送	SDK	service_up 成功才发，失败时跳过避免孤儿事件

---

FAQ

1. 缺字段会怎样？

SDK 只记录告警并静默跳过，不抛异常，不影响业务主流程。

2. 平台不可用时会怎样？

start() 上报失败时本地生成 run_id 继续执行；所有后续上报静默丢弃；业务函数正常跑。

3. 为什么不用环境变量？

固定项统一到 spider_sdk.ini，减少部署差异；业务项在装饰器里更直观，方便 code review。

4. heartbeat_min_sec 和 heartbeat_interval_sec 有什么区别？

• heartbeat_min_sec：SDK 防抖阈值，job 模式下手动调 reporter_ctx.heartbeat() 过于频繁时自动丢弃
• heartbeat_interval_sec：service 模式下 SDK 自动心跳线程的发送间隔

5. 多爬虫并发怎么避免串号？

用 runtime_resolver 动态生成 spider_name，每次调用时从函数参数里取爬虫实例名。

6. inject_ctx=True 时函数签名怎么写？

函数必须声明 reporter_ctx=None 参数（或 **kwargs），否则 SDK 会跳过注入并打警告日志。

7. service_up 失败了，service_down 还会发吗？

不会。SDK 检测到 service_up 上报失败（平台网络不通），退出时跳过 service_down，避免平台收到没有对应上线记录的孤儿下线事件。

---

推荐公开接口

接口	场景
report_job(...)	定时/触发式爬虫（装饰器）
report_job_from_env(...)	同上，历史兼容名
report_service(...)	长驻服务型爬虫（装饰器）
SpiderReportContext(...)	需要精细控制的场景（with 语句）

---

测试命令

python -m unittest ai_test/test_readme.py
python -m unittest pip_package/ai_test/test_readme_env.py
python -m unittest ai_test/test_spider_sdk.py
python -m unittest pip_package/ai_test/test_spider_sdk.py

---

说明

sdk_docx 为镜像说明文档；若冲突，以本 README 为准（SSOT）。