如何用Skills做Listing每日巡检（硬核完整技术方案）

0. 需求背景与业务目标

🎯 一句话目标： 基于领星已有运营日志，把“运营主动查异常”升级为“系统识别异常并主动告警”。

0.1 业务背景与能力缺口

评分、Review、父子体关系和购物车归属会直接影响 Listing 的转化、广告承接和销售稳定性。目前运营仍需每天逐个检查链接，监控规模扩大后容易出现耗时、漏检和无法追溯的问题。

• 行业现状： 领星、赛狐已经能够记录 Listing 的关键运营变化。
• 缺少规则： 现有监控模块无法针对四类业务场景配置监控对象、判定条件和接收人。
• 缺少触达： 日志主要用于被动查询和事后追溯，异常仍要依赖运营主动发现。

有日志不等于有监控，能追溯不等于能预警。真正缺少的是从“变化数据”到“运营行动”的自动化链路。

建设思路： 整合日志接入、规则判定、状态管理、事件去重和飞书通知，形成统一的轻量级告警工具。首期由本地 Python 通过领星 Open API 与官方 Python SDK 采集，不重复建设网页抓取链路。

0.2 需要监控的四类变化

• 新增差评： 识别新增 1～3 星 Review，不重复告警历史存量。
• 评分变化： 记录评分旧值、新值和变化时间，升分与降分均留痕。
• 父体成员变化： 运营希望判断同一父 ASIN 下的子 ASIN 集合和数量是否发生变化，包括成员减少、增加，以及数量不变但成员被替换。
• 购物车丢失： 识别 Buy Box 从自有卖家切换给其他卖家或无人持有；仅价格变化不判定为掉购物车。

0.3 业务目标

1. 降低人工巡检成本： 自动覆盖全部启用 ASIN，运营只处理异常。
2. 实现主动触达： 在每日巡检窗口内识别变化并通过飞书告警。
3. 控制误报与重复： 区分数据源失败和业务正常，同一事件最多发送一次。
4. 形成追溯闭环： 告警可反查 ASIN、店铺、原始日志、状态变化和任务批次。
5. 保持轻量可演进： 先以 Skill、Python 和 SQLite 落地，后续可平滑迁移 MySQL。

🔔 业务闭环： 运营日志 → 规则判定 → 状态比较与事件去重 → 飞书主动告警 → 运营处理

0.4 建设范围

本期建设

• 配置化 ASIN 清单与领星自动运营日志增量读取。
• 四类事件判定、SQLite 留痕、飞书告警及失败审计。

本期不做

• 不建设分钟级或实时监控。
• 不自动调价、修改 Listing、调整广告或执行其他运营动作。
• 不接入 Sorftime 等第三方数据源，不推断数据源未提供的评论明细。

0.5 验收成功标准

• ✅ 每日任务覆盖全部启用目标，未完成目标必须有明确失败原因。
• ✅ 四类标准测试样例均能按本文口径正确识别。
• ✅ 重复拉取或重复执行不会产生重复事件和重复业务告警。
• ✅ 接口超时、返回异常或解析失败只记录技术失败，不发送运营告警，也不误判为链接正常。
• ✅ 任一告警均可反查事件时间、变化前后值、原始日志、event_key 和任务批次。

✅ 方案结论： 采用“Codex 每日计划任务 + Listing Monitor Skill + Python 采集程序 + 领星 Open API/官方 SDK + SQLite + 飞书签名 Webhook”的轻量实现。没有常驻服务、Java/Spring、Redis 或消息队列；Python 负责采集、分页、重试、规则、落库和通知，模型只负责启动任务并读取紧凑汇总。

数据边界： 主链路只读取领星 Open API：运营日志负责差评、评分和 Buybox 事件，Listing 接口每日完整快照负责父体成员变化；不接入 Sorftime 或网页抓取。Open API 原始响应直接在本地解析并写入 SQLite，不进入模型上下文；SQLite 保存目标、状态、父体快照、事件、Outbox 和运行记录。

0.6 飞书告警效果

后续可以迭代的功能：将Listing和人员绑定，告警时可以直接 @到人；有能力的可以跟自己的ERP系统建立ACK机制联动，考核运营的响应率。

1. 需求与监控口径

📌 关键口径： 父体规则不能只依赖运营日志中的单条“父 ASIN 变动”，也不能只比较子体数量。必须每天完整拉取 erp_listing，按父体构建去重后的子 ASIN 集合，再与上一份完整快照做集合差异比较。

2. 数据源组合与职责边界

• 运营日志： 提供 Review 变动、评分变动和 Buybox 变动等已识别事件，适合低成本增量判断。
• Listing 快照： erp_listing 提供 asin、parent_asin、store_id、marketplace_id 等当前关系，用于重建父体成员集合。
• 双源互补： 日志回答“发生过什么”，快照回答“当前父体下有哪些子体”；两路在统一事件模型和告警状态机处汇合。
• 失败语义： 任一数据源调用失败、分页不完整或关键字段缺失，都记为 UNKNOWN 并写入技术失败记录；不得发送运营告警，也不得把未知状态解释成业务异常。

因此，本方案不是“只读运营日志”，而是“日志增量 + Listing 每日完整快照”的轻量组合。两者均由 Python 通过领星 Open API/官方 SDK 读取，不增加网页抓取或第三方数据源；原始响应不进入模型上下文。

领星官方说明：运营日志

3. 总体数据流

每日任务并行读取两路数据：运营日志经过合并、标准化和事件去重；Listing 经过完整分页校验、父体分组和成员集合比较。两路结果统一进入规则引擎、SQLite 事件账本、Outbox 与飞书通知。

4. 领星 Open API 与 Python SDK 采集设计

4.1 SDK 适配与采集层次

适配层不得把官方 SDK 的具体方法名、请求字段和返回包装泄漏到规则层。SDK 包装器是唯一接口边界；接口路径、鉴权方式、批量上限、分页字段、限流和历史时间范围须在收到领星 Open API 文档与 SDK 后逐项确认，再冻结为可测试契约。

4.2 采集契约与查询窗口

operation_logs = collector.fetch_operation_logs(
    target_asins=target_asins,
    window_start=scheduled_date - timedelta(days=2),
    window_end=scheduled_date,
)

listing_pages = collector.fetch_listing_pages(
    sid=sid,
    marketplace_id=marketplace_id,
    require_complete=True,
)

4.3 Listing 快照查询与父体聚合

领星 Listing Open API（对应现有 erp_listing 业务能力，具体 SDK 方法名待官方文档确认）不直接返回子体数量时，可利用每条记录中的 asin、parent_asin、store_id 和 marketplace_id 在本地构建父体成员集合。应按店铺和站点完整分页拉取目标范围，禁止只取第一页或只筛选在售状态。

def build_parent_sets(rows):
    groups = defaultdict(set)
    for row in rows:
        child = normalize_asin(row.get("asin"))
        parent = normalize_asin(row.get("parent_asin"))
        if not child or not parent or child == parent:
            continue
        if str(row.get("is_delete")) == "1":
            continue
        key = (row["store_id"], row["marketplace_id"], parent)
        groups[key].add(child)
    return groups

📌 数据口径： 该规则监控的是“领星当前可见的父体成员集合”。任一分页失败、总数不一致或字段缺失时，状态必须记为 UNKNOWN，不得把不完整结果解释为子体减少。

4.4 运营日志 Open API 返回结构与兼容策略

现有链路的观测样本显示：返回结果按日期分组，每个日期项可能包含多个日志数组，B096S5YCC6 的系统自动日志出现在 operation_log_data，而不是只出现在 auto_log_data。切换官方 SDK 后必须用 Open API 原始响应重新验证字段路径，不能直接照搬旧链路的包装层结构，也不能只读某一个数组。

{
  "type_name": "表现",
  "content": "[Rating变动] 新增5个Rating",
  "gmt_modified": "2026-08-31 23:59:59",
  "asin": "B096S5YCC6",
  "sid": 13345,
  "is_detail": false
}

解析顺序：

1. 遍历日期列表，兼容日期字段 r_date / rdate。
2. 合并 operation_log_data 与 auto_log_data；null 一律按空数组处理。
3. log_data 只作版本兼容：必须确认是系统自动事件后才纳入，避免把人工备注当成告警。
4. 只处理 Review变动、评分变动、Buybox变动三类日志；“父 ASIN 变动”仅保存为辅助证据，不作为父体成员告警的唯一依据。
5. 按事件时间、ASIN、店铺和内容构造幂等键，重复轮询不得重复通知。

5. 统一事件模型

{
  "event_type": "PARENT_MEMBERS_CHANGED",
  "subject_type": "PARENT_ASIN",
  "subject_key": "B0DM7H5QZ8",
  "old_count": 6,
  "new_count": 5,
  "added_asins": [],
  "removed_asins": ["B0XXXXXXX"],
  "source_type": "LISTING_SNAPSHOT"
}

6. 四类规则的判定细节

6.1 新增差评

• 匹配表现类 Review变动 自动日志，并解析新增 1–3 星 Review 数量。
• 数量大于 0 产生 NEGATIVE_REVIEW_ADDED，告警级别为 P2；只增加总 Review、没有 1～3 星信息时仅记账，不发差评告警。
• 领星该类日志同一天最多一条，告警内容应说明“领星本日汇总新增数量”，不把它解释成单条评价详情。

6.2 评分变化

• 匹配 评分变动，解析旧评分和新评分。
• 新评分低于旧评分：核心链接发送 P1，普通链接发送 P2；高于旧评分默认只记账，开启正向通知时发送 P3；相等则仅记审计日志。
• Rating变动 是评分数量变化，不等于星级分数变化，二者必须使用不同事件类型。

6.3 父体成员集合与数量变化

💡 能力边界： 领星运营日志只记录单个 ASIN 的“父 ASIN 变动”，不直接提供父体下子体数量。因此本规则必须以 erp_listing 每日完整快照为主，运营日志仅用于解释成员为何加入或移出。

• 首次完整快照只建立基线，不发送业务告警。
• 按 sid + marketplace_id + parent_asin 分组，对子 ASIN 去重后生成有序集合及 SHA-256 哈希。
• 比较集合而不是只比较数量：即使数量保持 5 → 5，只要新增或移除成员非空，仍产生事件。
• removed_asins 非空时，核心父体发送 P0、普通父体发送 P1；只有 added_asins 时发送 P3。
• 告警展示旧数量、新数量、新增 ASIN、移除 ASIN和本次快照时间。
• 分页不完整、总数不一致或关键字段缺失时只记录技术失败，不发送运营告警，也不更新基线。

added = current.children - previous.children
removed = previous.children - current.children

if not added and not removed:
    return Decision.silent()

if removed:
    alert_level = "P0" if target.business_level == "CORE" else "P1"
else:
    alert_level = "P3"

return Decision.alert(
    event_type="PARENT_MEMBERS_CHANGED",
    alert_level=alert_level,
    old_count=len(previous.children),
    new_count=len(current.children),
    added_asins=sorted(added),
    removed_asins=sorted(removed),
)

6.4 掉购物车

• 匹配 Buybox变动，区分“价格变化”和“持有者变化”。
• 只有最新持有者不命中配置的自营卖家名称/别名时，才产生 BUYBOX_LOST：核心链接发送 P0，普通链接发送 P1；价格变化不作为掉购物车告警。
• 最新持有者重新命中自营卖家时产生 BUYBOX_REGAINED，发送 P3 恢复通知。
• erp_listing 不返回 Buybox 持有者，因此不能用它代替 Buybox 判定。首次接入需由运营在领星页面确认当前持有者并配置自营卖家别名；未完成时状态标为 UNKNOWN，不做“已掉购物车”的推断。

7. 告警状态机

业务异常和采集异常必须分开。运营日志或 Listing 快照失败时，状态只能是“数据未知”；不得据此推断掉购物车、评分下降或父体成员减少。只有完整快照比较产生的成员差异，才能进入父体业务告警状态。

8. 技术选型与运行形态

本方案不建设常驻 Web 服务，也不引入 Java、Spring、Redis、消息队列或独立部署环境。一次每日巡检就是一次短任务：Codex 定时任务唤起 Skill，Skill 启动本地 Python 入口；Python 通过领星 Open API/官方 SDK 获取运营日志和 Listing 完整快照，并完成批量调度、分页、重试、解析、父体聚合、规则判定、SQLite 落库与飞书投递。模型不读取原始响应，只接收紧凑运行汇总。

Skill 与 Plugin 的选择： 当前只有一个内部工作流，直接开发 Skill 最轻。Plugin 不是调度器，也不会替代 SQLite；它只是把 Skill、可选 SDK/通知依赖和展示资源打包，适合后续需要跨团队安装、版本发布时再做。OpenAI 官方也建议先用 Skill 迭代单个工作流，稳定并需要分发时再封装 Plugin。

OpenAI：Scheduled tasks

OpenAI：Build skills

OpenAI：Build plugins

9. Skill 与目录设计

Skill 负责定义固定启动方式、配置边界和摘要输出；本地 Python 负责通过领星官方 SDK 采集，以及所有可重复验证的业务逻辑和飞书投递。原始响应在进程内直接标准化、落库，不经模型中转，从根源上避免大批量数据占用上下文。

lingxing-listing-alert-monitor/
├── SKILL.md                         # 启动方式、边界、失败语义、摘要契约
├── agents/openai.yaml              # 展示信息；不声明领星数据工具依赖
├── scripts/
│   ├── run_daily.py                # 每日短任务入口与紧凑汇总
│   ├── lingxing_client.py          # 官方 SDK 包装、鉴权与响应解包
│   ├── collector.py                # 批量、分页、限流、重试和完整性校验
│   ├── monitor.py                  # 目标配置与本地管理 CLI
│   ├── storage.py                  # SQLite Store 实现与短事务
│   ├── parsers.py                  # 领星运营日志解析器注册表
│   ├── snapshots.py                # Listing 父体聚合与集合哈希
│   ├── rules.py                    # 日志事件、父体集合与 P0～P3 分级规则
│   ├── render_feishu.py            # 告警卡片数据生成
│   └── send_feishu.py              # Webhook 主通道与 OpenAPI 后备
├── migrations/001_init_sqlite.sql
├── references/
│   ├── event-patterns.yaml         # 已验证的脱敏日志样本与匹配规则
│   └── severity-policy.yaml        # P0～P3 默认等级与正向通知开关
└── tests/fixtures/                 # 脱敏日志、Listing 快照与 SDK 响应样本

运行产物：~/Documents/AI-Ecommerce-Runs/listing-monitor/runs/<run_slug>/
数据库：~/Documents/AI-Ecommerce-Runs/listing-monitor/listing-monitor.sqlite

脚本契约：

10. SQLite 库表设计

数据库只保存运行状态和审计证据，不保存领星或飞书凭据。时间统一保存 ISO-8601 UTC 文本；展示时再转 Asia/Shanghai。JSON 字段在 SQLite 中保存为 TEXT，迁移 MySQL 时改为 JSON 类型。

PRAGMA foreign_keys = ON;
PRAGMA journal_mode = WAL;
PRAGMA busy_timeout = 5000;

-- 监控目标：定义需要巡检的 ASIN 或父 ASIN
CREATE TABLE IF NOT EXISTS monitor_target (
    id                      INTEGER PRIMARY KEY AUTOINCREMENT, -- 本地主键
    target_type             TEXT NOT NULL CHECK (target_type IN ('ASIN', 'PARENT_ASIN')), -- 目标类型
    target_value            TEXT NOT NULL,                    -- ASIN 或父 ASIN
    business_level          TEXT NOT NULL DEFAULT 'NORMAL' CHECK (business_level IN ('CORE', 'NORMAL')), -- 核心/普通链接
    sid                     INTEGER NOT NULL,                 -- 领星店铺 ID
    marketplace_id          TEXT NOT NULL,                    -- Amazon Marketplace ID
    own_seller_aliases_json TEXT NOT NULL DEFAULT '[]',       -- 自营卖家名称及别名 JSON 数组
    notify_chat_id          TEXT NOT NULL,                    -- 飞书告警群 chat_id
    enabled                 INTEGER NOT NULL DEFAULT 1 CHECK (enabled IN (0, 1)), -- 是否启用
    created_at              TEXT NOT NULL,                    -- 创建时间，ISO-8601 UTC
    updated_at              TEXT NOT NULL,                    -- 更新时间，ISO-8601 UTC
    UNIQUE (target_type, sid, marketplace_id, target_value)
);

-- ASIN 当前状态：保存日志规则的最新可比较状态
CREATE TABLE IF NOT EXISTS listing_state (
    target_id               INTEGER PRIMARY KEY,              -- monitor_target.id，仅用于 ASIN 目标
    rating_score            REAL,                             -- 最新星级分数
    buybox_holder           TEXT,                             -- 归一化后的最新 Buybox 持有者
    buybox_owned            INTEGER CHECK (buybox_owned IN (0, 1) OR buybox_owned IS NULL), -- 1 自营，0 非自营，NULL 未知
    last_source_event_time  TEXT,                             -- 已推进状态的最新源事件时间
    updated_at              TEXT NOT NULL,                    -- 状态更新时间，ISO-8601 UTC
    FOREIGN KEY (target_id) REFERENCES monitor_target(id) ON DELETE CASCADE
);

-- 每日任务运行记录：用于幂等、统计和故障审计
CREATE TABLE IF NOT EXISTS job_run (
    id                      INTEGER PRIMARY KEY AUTOINCREMENT, -- 运行批次主键
    run_key                 TEXT NOT NULL UNIQUE,             -- 逻辑运行唯一键
    scheduled_date          TEXT NOT NULL,                    -- 计划执行日期，Asia/Shanghai，YYYY-MM-DD
    window_start            TEXT NOT NULL,                    -- 查询窗口起点，ISO-8601 UTC
    window_end              TEXT NOT NULL,                    -- 查询窗口终点，ISO-8601 UTC，左闭右开
    status                  TEXT NOT NULL CHECK (status IN ('RUNNING', 'SUCCESS', 'PARTIAL', 'FAILED')), -- 运行状态
    attempt_count           INTEGER NOT NULL DEFAULT 1,       -- 当前逻辑任务尝试次数
    target_total            INTEGER NOT NULL DEFAULT 0,       -- 本次启用目标总数
    target_success          INTEGER NOT NULL DEFAULT 0,       -- 成功目标数
    target_failed           INTEGER NOT NULL DEFAULT 0,       -- 失败目标数
    new_event_count         INTEGER NOT NULL DEFAULT 0,       -- 新增事件数
    sent_alert_count        INTEGER NOT NULL DEFAULT 0,       -- 成功发送告警数
    error_summary_json      TEXT,                             -- 失败摘要 JSON
    started_at              TEXT NOT NULL,                    -- 首次开始时间，ISO-8601 UTC
    finished_at             TEXT                              -- 最终结束时间，ISO-8601 UTC
);

-- 父体成员快照：保存每个父 ASIN 当日完整子体集合
CREATE TABLE IF NOT EXISTS parent_snapshot (
    id                      INTEGER PRIMARY KEY AUTOINCREMENT, -- 快照主键
    run_id                  INTEGER NOT NULL,                 -- 所属运行批次
    target_id               INTEGER NOT NULL,                 -- monitor_target.id，仅用于 PARENT_ASIN 目标
    snapshot_date           TEXT NOT NULL,                    -- 快照业务日期，Asia/Shanghai，YYYY-MM-DD
    child_count             INTEGER NOT NULL CHECK (child_count >= 0), -- 去重后子 ASIN 数量
    child_asins_json        TEXT NOT NULL,                    -- 排序后的子 ASIN JSON 数组
    child_set_hash          TEXT NOT NULL,                    -- 子体集合规范化 JSON 的 SHA-256
    source_total_rows       INTEGER NOT NULL,                 -- 本次店铺/站点 Listing 原始总行数
    is_complete             INTEGER NOT NULL CHECK (is_complete IN (0, 1)), -- 分页是否完整
    observed_at             TEXT NOT NULL,                    -- 观测完成时间，ISO-8601 UTC
    FOREIGN KEY (run_id) REFERENCES job_run(id) ON DELETE CASCADE,
    FOREIGN KEY (target_id) REFERENCES monitor_target(id) ON DELETE CASCADE,
    UNIQUE (run_id, target_id)
);

CREATE INDEX IF NOT EXISTS idx_parent_snapshot_lookup
    ON parent_snapshot(target_id, snapshot_date DESC);

-- 统一事件账本：保存日志事件、快照差异事件及原始证据
CREATE TABLE IF NOT EXISTS operation_event (
    id                      INTEGER PRIMARY KEY AUTOINCREMENT, -- 事件主键
    event_key               TEXT NOT NULL UNIQUE,             -- 跨重跑稳定的幂等键
    target_id               INTEGER NOT NULL,                 -- 关联监控目标
    event_type              TEXT NOT NULL,                    -- 标准业务事件类型
    alert_level             TEXT CHECK (alert_level IN ('P0', 'P1', 'P2', 'P3') OR alert_level IS NULL), -- 告警级别；NULL 表示只记事件
    parse_status            TEXT NOT NULL CHECK (parse_status IN ('PARSED', 'UNPARSED', 'IGNORED')), -- 解析状态
    parser_version          TEXT NOT NULL,                    -- 解析器规则版本
    source_type             TEXT NOT NULL CHECK (source_type IN ('OPERATION_LOG', 'LISTING_SNAPSHOT', 'SYSTEM')), -- 事件来源
    source_bucket           TEXT NOT NULL,                    -- 来源数组或工具分桶
    source_event_time       TEXT NOT NULL,                    -- 业务事件发生时间
    source_log_id           TEXT,                             -- 领星原始日志 ID，无则为空
    old_value               TEXT,                             -- 可标量表达的旧值
    new_value               TEXT,                             -- 可标量表达的新值
    change_detail_json      TEXT,                             -- 集合差异等结构化变化明细
    raw_content             TEXT NOT NULL,                    -- 原始日志正文或快照差异摘要
    raw_payload_json        TEXT NOT NULL,                    -- 原始数据或最小证据 JSON
    observed_at             TEXT NOT NULL,                    -- 系统观测时间，ISO-8601 UTC
    FOREIGN KEY (target_id) REFERENCES monitor_target(id) ON DELETE CASCADE
);

CREATE INDEX IF NOT EXISTS idx_event_target_time
    ON operation_event(target_id, source_event_time DESC);
CREATE INDEX IF NOT EXISTS idx_event_type_time
    ON operation_event(event_type, source_event_time DESC);

-- 通知 Outbox：保证事件入库与飞书发送解耦且可重试
CREATE TABLE IF NOT EXISTS alert_outbox (
    id                      INTEGER PRIMARY KEY AUTOINCREMENT, -- Outbox 主键
    event_id                INTEGER NOT NULL,                 -- 关联事件 ID
    channel                 TEXT NOT NULL DEFAULT 'FEISHU',   -- 通知渠道
    status                  TEXT NOT NULL CHECK (status IN ('PENDING', 'SENDING', 'SENT', 'FAILED', 'DEAD')), -- 投递状态
    payload_json            TEXT NOT NULL,                    -- 待发送飞书卡片 JSON
    attempt_count           INTEGER NOT NULL DEFAULT 0,       -- 已尝试发送次数
    next_attempt_at         TEXT,                             -- 下次允许重试时间
    message_id              TEXT,                             -- 飞书发送成功后的消息 ID
    last_error              TEXT,                             -- 最近一次失败摘要
    created_at              TEXT NOT NULL,                    -- 创建时间，ISO-8601 UTC
    updated_at              TEXT NOT NULL,                    -- 状态更新时间，兼作 SENDING 租约判断
    sent_at                 TEXT,                             -- 最终发送成功时间
    FOREIGN KEY (event_id) REFERENCES operation_event(id) ON DELETE CASCADE,
    UNIQUE (event_id, channel)
);

CREATE INDEX IF NOT EXISTS idx_outbox_due
    ON alert_outbox(status, next_attempt_at);

10.1 表职责与关键约束

10.2 字段数据字典

SQLite 不支持 MySQL 的列级 COMMENT 语法，因此字段释义同时维护在本节和 SQL 行注释中。表内“迁移类型”是未来迁移到 MySQL 8 时的建议类型，不代表当前需要执行迁移。

10.2.1 monitor_target

10.2.2 listing_state

10.2.3 job_run

10.2.4 parent_snapshot

10.2.5 operation_event

10.2.6 alert_outbox

一致性约束： child_count 必须等于子体数组长度；is_complete = 0 的快照不得参与业务比较；buybox_owned = NULL 表示未知而不是掉购物车；alert_outbox.status = SENT 时必须同时写入 message_id 和 sent_at。

11. 每日执行主流程

每日任务查询最近 3 个自然日的运营日志，同时完整分页拉取当日 Listing。日志依靠 event_key 去重；父体依靠完整快照、成员集合哈希和差集判断。外部 Open API/SDK 请求和飞书发送不得放在数据库事务内，事务只包围本地事件、状态、快照与 Outbox 写入。

def run_daily(scheduled_date):
    run = store.start_or_resume_run(
        run_key=f"listing-monitor:{scheduled_date}",
        window=last_three_days(scheduled_date),
    )
    collector = LingxingCollector.from_environment()

    # 外部请求全部在事务外完成，原始响应不进入模型上下文
    try:
        log_rows = collector.fetch_operation_logs(
            target_asins=store.list_asins(), window=run.window
        )
        ingest_operation_logs(run, log_rows)
    except ProviderError as error:
        store.record_source_failure(run.id, "operation_logs", error)

    scopes = group_parent_targets_by_scope(store.list_parent_targets())
    for scope, targets in scopes.items():
        try:
            page_result = collector.fetch_listing_pages(scope)
            require_complete_pages(page_result)
            parent_sets = build_parent_sets(page_result.rows)
        except ProviderError as error:
            store.record_scope_failure(run.id, scope, error)
            continue

        persist_parent_decisions(run, targets, parent_sets)

    dispatch_pending_outbox()
    store.finish_run(run.id)
    return build_compact_run_summary(run.id)

11.1 顺序与乱序处理

运营日志使用 3 天重叠窗口，旧事件可能晚于新事件返回；只有不早于 listing_state.last_source_event_time 的事件才能推进当前状态。父体快照按 snapshot_date 比较最近两份完整快照，不完整快照不更新基线，也不参与业务差异判断。

11.2 event_key 生成

def build_log_event_key(item, target):
    if item.get("log_id"):
        canonical = f"lingxing:{item['log_id']}"
    else:
        canonical = "|".join([
            str(target.sid), target.target_value.upper(),
            normalize_space(item.get("type_name")),
            normalize_space(item.get("content")),
            normalize_time(item.get("gmt_modified")),
        ])
    return sha256(canonical.encode()).hexdigest()

def build_snapshot_event_key(target, previous, current):
    canonical = "|".join([
        "lingxing-snapshot", current.snapshot_date,
        str(target.sid), target.marketplace_id,
        target.target_value.upper(),
        previous.child_set_hash, current.child_set_hash,
    ])
    return sha256(canonical.encode()).hexdigest()

日志事件优先使用领星 log_id；无 log_id 时使用稳定字段哈希。父体快照差异使用“目标 + 快照日期 + 旧集合哈希 + 新集合哈希”生成 event_key。技术幂等键不能简化为“同一天同类型”，否则会吞掉同日多次 Buybox 变化或不同成员替换。

12. 关键逻辑判断

12.1 自动日志筛选

1. 按日期遍历结果，兼容 r_date 和 rdate。
2. 将 operation_log_data、auto_log_data 的 null 转为空数组后合并。
3. log_data 默认不进入业务判定；只有明确存在系统标识且事件类型在白名单时才兼容纳入。
4. 业务规则白名单仅包含 Review变动、评分变动、Buybox变动；父体成员变化由 Listing 快照路径产生。
5. 未知格式落 UNPARSED 并记录技术失败，不发送运营告警；禁止让大模型根据自然语言猜测业务状态。

12.2 业务状态转换表

12.3 告警等级与响应时限

飞书业务告警统一使用 P0、P1、P2、P3 四级，不再使用 INFO、TECH 或 NONE 作为告警级别。未达到业务告警条件的事件仍可写入 operation_event，但 alert_level 为空且不生成 Outbox。

def resolve_alert_level(event, target, policy):
    core = target.business_level == "CORE"

    if event.type in {"BUYBOX_LOST", "PARENT_MEMBER_REMOVED"}:
        return "P0" if core else "P1"
    if event.type == "RATING_DECREASED":
        return "P1" if core else "P2"
    if event.type == "NEGATIVE_REVIEW_ADDED":
        return "P2"
    if event.type in {"PARENT_MEMBER_ADDED", "BUYBOX_REGAINED"}:
        return "P3"
    if event.type == "RATING_INCREASED" and policy.notify_positive:
        return "P3"
    return None

P0 启用约束： 目标必须由运营显式配置 business_level = CORE。未标记的目标一律按 NORMAL 处理，程序不得根据销量、名称或模型推断核心程度。

💡 发现时效边界： 当前方案每日运行一次，因此 P0 约束的是“系统发现后的处置时限”，不等于异常发生后 15 分钟内发现。若业务要求真正的 P0 实时发现，CORE 目标的巡检频率必须提升到 5～15 分钟。

12.4 父体成员集合比较

def compare_parent_members(target, previous, current):
    if not current.is_complete:
        store.record_technical_failure("Listing 快照不完整")
        return Decision.silent()
    if previous is None:
        return Decision.baseline_only()

    added = current.children - previous.children
    removed = previous.children - current.children
    if not added and not removed:
        return Decision.silent()

    if removed:
        level = "P0" if target.business_level == "CORE" else "P1"
    else:
        level = "P3"

    return Decision.parent_members_changed(
        alert_level=level,
        old_count=len(previous.children),
        new_count=len(current.children),
        added_asins=sorted(added),
        removed_asins=sorted(removed),
    )

比较对象必须是集合，不是单独的数量字段。数量用于告警展示；集合差集才是业务判定依据。运营日志中的“父 ASIN 变动”可以附加到告警作为证据，但不得单独替代完整快照比较。

12.5 Buybox 持有者归一化

def normalize_seller(value):
    return collapse_spaces(value).strip().casefold()

def is_owned(holder, aliases):
    if holder is None or not holder.strip():
        return None
    normalized = normalize_seller(holder)
    return normalized in {normalize_seller(x) for x in aliases}

owned = is_owned(parsed.new_holder, target.own_seller_aliases)
if owned is None:
    store.record_technical_failure("Buybox 持有者为空")
    return Decision.silent()
if owned is False and previous.buybox_owned != 0:
    level = "P0" if target.business_level == "CORE" else "P1"
    return Decision.alert(level, "BUYBOX_LOST")
if owned is True and previous.buybox_owned == 0:
    return Decision.alert("P3", "BUYBOX_REGAINED")
return Decision.silent()

别名必须做精确归一化匹配，不使用模糊包含。例如卖家名 “ABC” 不能匹配 “ABC Outlet”。首次接入时 buybox_owned 为 null；如果第一条有效持有者事件显示为非自营卖家，仍按异常告警，但卡片需标记“首次观测”。

12.6 解析器版本

每类 content 使用独立解析器，并在 raw_payload_json 中保留原始数据。operation_event.parser_version 初始值为 v1；规则升级时通过离线命令重放 UNPARSED 事件，不直接修改已经发送过的业务结论。

13. 飞书告警设计与通知可靠性

13.1 通知幂等与投递一致性

Outbox 状态转换： PENDING → SENDING → SENT；发送失败转 FAILED 并设置 next_attempt_at，达到 3 次后转 DEAD。每次每日任务开始时先处理上次遗留的 PENDING/FAILED，再拉取新数据。领星 Open API/SDK 调用失败只把本轮标记为 PARTIAL，不生成任何“掉购物车/评分下降”等业务事件。

13.2 告警卡片规范

飞书投递保留现有轻量方案：send_feishu.py --transport auto 对固定告警群优先使用签名 Webhook；lark-cli 只用于首次配置和人工诊断；需要应用身份能力时再走已有 OpenAPI 客户端。当前卡片仅包含 Amazon OpenLink，不需要回调，因此不强制切换官方 SDK。只有出现多群动态路由、回执查询、交互回调或统一应用治理需求时，才升级为飞书官方 SDK 主通道。飞书只主动发送业务告警，技术失败仅记录在 job_run.error_summary_json 和任务汇总中。

13.2.1 卡片通用数据协议

{
  "event_key": "sha256...",
  "event_type": "PARENT_MEMBERS_CHANGED",
  "alert_level": "P1",
  "subject": {"type": "PARENT_ASIN", "value": "B0DM7H5QZ8"},
  "context": {
    "business_level": "NORMAL",
    "sid": 13345,
    "marketplace_id": "ATVPDKIKX0DER"
  },
  "occurred_at": "2026-09-02T01:10:00Z",
  "observed_at": "2026-09-02T01:12:08Z",
  "change": {
    "old_count": 6,
    "new_count": 5,
    "added_asins": [],
    "removed_asins": ["B0D5CLSMFB"]
  },
  "links": {"amazon": "configured-url"}
}

13.2.2 P0～P2 业务告警

🔴 P0｜B096S5YCC6 购物车丢失

业务标识： 核心链接

店铺 / 站点： 示例店铺 / 美国站

持有者： OWN STORE → OTHER SELLER

发生 / 发现： 08:55 / 09:10

处置要求： 立即检查价格、库存和跟卖情况

按钮： 查看亚马逊链接

event_key: 90c2…4aa7

💡 P1｜B0DM7H5QZ8 父体成员移除

业务标识： 普通链接

店铺 / 站点： 示例店铺 / 美国站

子体数量： 6 → 5

移除： B0D5CLSMFB

新增： 无

快照时间： 09:12；分页完整

按钮： 查看亚马逊父体

event_key: 5ad0…1e3b

💡 P2｜B096S5YCC6 新增差评

店铺 / 站点： 示例店铺 / 美国站

变化： 新增 1～3 星 Review 2 条

发生 / 发现： 08:41 / 09:10

数据类型： Review 变动

按钮： 查看亚马逊链接

event_key: 8f31…c912

💡 P2｜B096S5YCC6 评分下降

业务标识： 普通链接

店铺 / 站点： 示例店铺 / 美国站

星级变化： 4.5 → 4.4（-0.1）

发生 / 发现： 08:46 / 09:10

按钮： 查看亚马逊链接

event_key: b1a8…72ef

13.2.3 P3 低优先级与恢复通知

🔵 P3｜B0DM7H5QZ8 父体新增成员

店铺 / 站点： 示例店铺 / 美国站

子体数量： 5 → 6

新增： B0D5CM8VZ9

移除： 无

按钮： 查看亚马逊父体

event_key: 3bb1…10d9

✅ P3｜B096S5YCC6 购物车已恢复

店铺 / 站点： 示例店铺 / 美国站

持有者： OTHER SELLER → OWN STORE

恢复时间： 2026-09-02 10:15

关联异常： BUYBOX_LOST / 90c2…4aa7

按钮： 查看亚马逊链接

13.2.4 生成与发送规则

14. 每日调度与配置

建议每日 09:10（Asia/Shanghai）运行一次，查询当天及前两天。该频率满足“每日看链接”需求，但不承诺 Buybox 实时发现：当天发生的多次 Buybox 变化会在下一次日任务中统一读取并逐条去重处理。

计划任务提示词应保持短且确定：

执行 Listing Monitor Skill 的每日巡检：
1. 使用计划日期启动本地 run_daily.py；
2. 由 Python 通过领星官方 SDK 读取近 3 天运营日志与当日完整 Listing 快照；
3. 由 Collector 完成批量/分片、分页、限流、重试和完整性校验；
4. 由本地规则完成去重、状态比较、SQLite 与 Outbox 写入；
5. 由 send_feishu.py 投递待发送业务告警并回写结果；
6. 只向任务返回成功、失败、新事件和告警数量等紧凑摘要；
7. 禁止把原始响应放入模型上下文，禁止把无数据或不完整分页推断为业务正常。

本地 Scheduled Task 需要电脑保持开机且 ChatGPT/Codex 桌面应用运行。监控目标通过轻量 CLI 维护：target add、target list、target enable、target disable。领星 Open API/SDK 凭据和飞书凭据均由本机环境变量或安全配置管理，不写入 Skill、SQLite、日志、模型上下文或告警正文。

15. SQLite 迁移 MySQL 设计

业务层只依赖 Store 接口，不直接散落 sqlite3 SQL。当前实现为 SqliteStore；未来新增 MySqlStore，解析器、规则、Skill 和飞书通知流程无需修改。

迁移步骤： 冻结一次调度 → SQLite 导出 CSV/JSONL → 在 MySQL 建表和唯一索引 → 按 monitor_target、listing_state、job_run、parent_snapshot、operation_event、alert_outbox 顺序导入 → 校验行数、集合哈希及唯一键 → 切换 Store 配置 → 影子执行一次不发消息的巡检 → 恢复计划任务。

16. 测试与评审门槛

📌 MVP 交付边界： 一个 Skill、一组小型 Python 模块、1 个 SQLite 文件、1 个每日计划任务。不建设服务端，不维护常驻进程，不引入应用框架；Python 负责领星采集、确定性规则、落库和飞书投递，模型只负责启动任务并读取紧凑汇总。