采集器开发指南
SecLens 采用众包采集架构:任何人都可以开发独立的采集器,通过 Ingest API 向平台推送安全情报数据。
快速开始
- 获取凭据:联系管理员获取 Collector Token(
slct_前缀)。 - 开发采集器:使用任何语言编写采集逻辑,将数据格式化为
BulletinCreate结构。 - 推送数据:调用
POST /v1/ingest/bulletins提交采集结果。 - 部署运行:使用 Cron、Airflow 或任何调度系统定期运行采集器。
Ingest API
POST /v1/ingest/bulletins
Authorization: Bearer slct_your_token_here
Content-Type: application/json
[
{
"source": {
"source_slug": "your_source_slug",
"external_id": "unique-item-id",
"origin_url": "https://example.com/advisory/123"
},
"content": {
"title": "安全公告标题",
"summary": "简要描述",
"published_at": "2026-01-15T10:00:00Z"
},
"severity": "high",
"labels": ["vulnerability"],
"topics": ["official_bulletin"]
}
]
Python 采集器示例
import requests
INGEST_URL = "https://your-seclens.example.com/v1/ingest/bulletins"
TOKEN = "slct_your_token_here"
def collect():
# 从目标源获取数据
resp = requests.get("https://example.com/api/advisories", timeout=30)
resp.raise_for_status()
items = resp.json()
# 格式化为 BulletinCreate 结构
bulletins = []
for item in items:
bulletins.append({
"source": {
"source_slug": "example_advisories",
"external_id": str(item["id"]),
"origin_url": item["url"],
},
"content": {
"title": item["title"],
"summary": item.get("description"),
"published_at": item["published"],
},
})
# 推送到 SecLens
headers = {
"Authorization": f"Bearer {TOKEN}",
"Content-Type": "application/json",
}
resp = requests.post(INGEST_URL, json=bulletins, headers=headers, timeout=30)
resp.raise_for_status()
print(f"提交成功: {resp.json()}")
if __name__ == "__main__":
collect()
数据字段说明
source_slug(必填):数据源标识,如aliyun_securityexternal_id(推荐):原始唯一 ID,用于去重origin_url(推荐):原文链接title(必填):标题,1-500 字符published_at(推荐):发布时间,ISO 8601 格式severity:严重程度(critical / high / medium / low / info)labels:标签列表topics:主题分类
注意事项
- 单次提交上限 200 条
- 重复数据(相同
source_slug+external_id)会自动去重 - Token 有速率限制,默认每小时 600 次请求
- 完整 API 文档请参考 /docs(Swagger UI)