如何拉取API所有新增传入数据并通过webhook推送

你对接的这个Roblox公开目录搜索接口没有提供原生的事件推送/Webhook能力，要实现新数据自动推送，最稳妥的落地方案是定时轮询+增量ID比对，不需要复杂架构，少量代码即可稳定运行。

核心实现逻辑 #

• 你当前使用的接口参数SortType=3本身就是按商品最新上架时间排序，ResultsPerPage=1每次只返回1条最新结果，CreatorID=1固定筛选Roblox官方发布的商品，天然适配新内容检测场景
• 本地持久化存储所有已经推送过的商品唯一标识，直接用接口返回的AssetId即可，该字段是Roblox资产的全局唯一ID
• 按固定间隔请求接口拉取最新数据，和本地存储的已推送ID列表做比对：如果返回的资产ID不在本地列表中，判定为新上架商品，立刻将数据按你的Webhook要求组装后发起POST推送，推送成功后把新ID写入本地存储，避免重复推送
• 脚本首次启动时，先将当前接口返回的最新资产ID写入本地存储，不会把历史存量数据误判为新内容推送

最简可运行参考代码（Python） #

import time
import json
import requests

# 基础配置
TARGET_API = "https://search.roblox.com/catalog/json?SortType=3&ResultsPerPage=1&CreatorID=1"
WEBHOOK_URL = "替换为你自己的Webhook接收地址"
POLL_INTERVAL = 60  # 轮询间隔，单位：秒
STORAGE_FILE = "pushed_asset_ids.json"

# 加载已推送过的资产ID集合
def load_pushed_ids():
    try:
        with open(STORAGE_FILE, "r", encoding="utf-8") as f:
            return set(json.load(f))
    except FileNotFoundError:
        return set()

# 保存已推送的资产ID
def save_pushed_ids(id_set):
    with open(STORAGE_FILE, "w", encoding="utf-8") as f:
        json.dump(list(id_set), f, ensure_ascii=False)

# 推送数据到Webhook
def send_webhook(asset_data):
    # 可根据自己使用的Webhook格式要求修改请求体，适配Discord、企业微信、飞书等不同平台
    req_body = {
        "content": f"检测到Roblox官方新上架商品：{asset_data['Name']}，资产ID：{asset_data['AssetId']}"
    }
    requests.post(WEBHOOK_URL, json=req_body, timeout=10)

if __name__ == "__main__":
    pushed_ids = load_pushed_ids()
    # 首次运行初始化存储，避免推送历史数据
    if not pushed_ids:
        init_resp = requests.get(TARGET_API, timeout=10)
        init_resp.raise_for_status()
        init_data = init_resp.json()
        if init_data:
            pushed_ids.add(init_data[0]["AssetId"])
            save_pushed_ids(pushed_ids)
    # 启动轮询检测
    while True:
        try:
            resp = requests.get(TARGET_API, timeout=10)
            resp.raise_for_status()
            latest_assets = resp.json()
            if not latest_assets:
                time.sleep(POLL_INTERVAL)
                continue
            latest_asset = latest_assets[0]
            latest_id = latest_asset["AssetId"]
            if latest_id not in pushed_ids:
                send_webhook(latest_asset)
                pushed_ids.add(latest_id)
                save_pushed_ids(pushed_ids)
        except Exception as e:
            print(f"检测流程出错：{str(e)}")
        time.sleep(POLL_INTERVAL)

落地注意事项 #

• 轮询间隔不建议设置低于30秒，请求频率过高容易触发Roblox接口的限流规则，导致IP被临时拦截，普通场景下60秒的间隔完全可以满足新商品实时检测需求
• 如果需要一次覆盖更多最新商品，可以把URL参数里的ResultsPerPage=1调整为更大的数值（建议不超过20），遍历返回的所有资产做ID比对即可，不会漏过新内容
• 本地存储用JSON文件足够轻量稳定，如果后续要做分布式部署，也可以替换为Redis、SQLite等存储方案，核心比对逻辑不需要改动
• 脚本长期运行直接部署在常驻服务器上，用systemd、PM2等进程守护工具托管即可，不需要额外依赖其他服务

内容的提问来源于stack exchange，提问作者St0na