手动跑几次就乱，问题通常不在抓取脚本本身，而在缺少三样东西：固定节奏、增量位置、写盘规则。last30days-skill 这类按近 30 天窗口抓跨平台评论的任务，建议把它收敛成一个“可被计划任务调用的入口命令”，每次运行只做固定几步：读游标 → 抓评论 → 按话题分桶 → 安全写回 → 记日志 → 更新游标。下面按这五步给出可直接替换的配置与骨架，平台字段名、目录路径需要结合自己的环境确认。

适用场景：需要持续收集近 30 天跨平台评论、又不希望每次全量重抓。操作方向：先固定入口命令与输出目录结构，再补游标文件和运行日志，最后把落盘改成“临时文件 + 原子替换”。验证方式：连续跑两次，对比新增条数、游标时间戳是否前移、旧分桶文件是否被改动。边界：平台返回字段缺失、话题为空、请求中断时，只能保守跳过并记录，不追求一次补齐。

用计划任务封装一次运行命令

把抓取逻辑包成一个可以带参数执行的入口，计划任务只负责按时间触发它，不要在 crontab 里写业务逻辑。工作目录、配置文件、输出目录都用绝对路径，避免 cron 环境变量差异导致相对路径解析到别处。

# crontab -e
# 每天 08:10 跑一次；标准输出与错误输出分开留存
10 8 * * * cd /opt/last30days && mkdir -p logs data && \
  /usr/bin/python3 -m skill.run `--config` conf/sources.yaml `--window` 30d \
  >> logs/run.out 2>> logs/run.err

重定向写法有两种取舍：>> logs/run.out 2>> logs/run.err 把正常日志和异常堆栈分开，排查时先看 err 文件更快；>> logs/run.log 2>&1 合并成一条时间线，适合按顺序还原现场。两种都是追加写，不要用单个 > 覆盖，否则上一次的报错会被冲掉。如果系统用 systemd，把同样的命令放进 ExecStart=，输出交给 journald，再用 StandardOutput=append: 指定文件也可以。

cron 不保证上一次跑完再跑下一次，抓取耗时不稳定时建议加一个锁文件（如 flock -n /tmp/last30days.lock），让重叠的那次直接退出，而不是两个进程同时写同一个分桶文件。

记录上次抓取位置作为增量游标

增量靠一个游标文件实现：记录每个平台最后一次成功抓到的位置（时间戳或游标 ID），下次运行从这里往后取。游标按平台分开存，某个平台失败时不影响其他平台的进度。

# data/cursor.json
{
  "schema": 1,
  "updated_at": "<ISO8601 时间>",
  "sources": {
    "platform_a": {"last_ts": "<ISO8601>", "last_id": "<评论ID>"},
    "platform_b": {"last_ts": "<ISO8601>", "last_id": null}
  }
}

写入游标的时机放在“数据已经落盘成功之后”，顺序不能反：先写数据文件，再更新游标。如果先更新游标后写盘失败，这批数据就永久丢了。

无游标时的兜底逻辑要写死，不要依赖默认值：读不到文件或解析失败，就按 now - 30d 作为起点，并把这轮标记为 bootstrap，在日志里写明是首次运行还是游标损坏。起点窗口别顺手放大，窗口越大重复越多，去重成本越高。同一平台返回的时间戳时区不一致时，统一换算到 UTC 再比较，避免出现“游标比实际数据新、后面几天全被跳过”的情况。

按话题字段做分桶并命名文件

分桶键优先选平台返回的结构化字段，例如 topic、topic_id，其次是 tags 的第一个元素；只有这些都缺失时才退回关键词匹配，因为关键词匹配结果不稳定，同一批评论可能被分到不同桶。分桶前统一做归一化：去首尾空白、转小写、把路径分隔符和空白替换成下划线，限制长度。

# 文件名规则：data/by_topic/<topic>/<source>-<YYYYMMDD>.jsonl
# 例：data/by_topic/退款流程/platform_a-20250101.jsonl

def bucket_name(item):
    raw = item.get("topic") or item.get("topic_id") or (item.get("tags") or [None])[0]
    if not raw or not str(raw).strip():
        return "_untagged"
    name = re.sub(r"[\\s/\\\\:*?\"<>|]+", "_", str(raw).strip().lower())
    return name[:64]

空话题不要丢，统一落到 _untagged 目录，方便事后回看是平台没给字段还是解析出错。按天分文件比按话题单文件更好维护：单文件会随天增长变大，一旦写坏整桶都要重来。文件名里带来源和日期，话题名只做目录名，便于 ls data/by_topic/ 直接按话题取用。

写回本地时保证不覆盖旧数据

评论是追加型数据，行为选 append 模式，用行分隔 JSON（jsonl），每条一行，写入前用 last_id 或 (source, comment_id) 做一次去重。追加失败或进程被杀，最坏只是最后一行残缺，读取端跳过解析失败的行即可。

但汇总类文件（话题索引、统计快照）不能直接覆盖写，否则一次失败就把历史结果冲掉。做法是写临时文件再原子替换：

tmp = path + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
    json.dump(payload, f, ensure_ascii=False)
    f.flush()
    os.fsync(f.fileno())
os.replace(tmp, path)   # 同分区内的原子替换

回滚方式分两种：替换前把旧文件 os.replace(path, path + ".bak") 留一份，替换后校验新文件能正常解析，解析失败就把 .bak 换回去；追加写则不做回滚，只在日志里记录写入的行区间，出问题时按行号截断到上一次成功的位置。临时文件一定要和目标文件在同一分区，跨分区 os.replace 会退化成复制加删除，中途失败仍可能留下半个文件。

留一份可查的运行日志

日志要能回答“这轮是抓取失败还是写入失败”，所以按阶段记录，而不是只记一句成功或异常。建议每次运行生成一个 run_id，所有输出带上它。

{
  "run_id": "<时间戳>",
  "stage": "fetch | bucket | write | cursor",
  "source": "platform_a",
  "cursor_before": "<ISO8601>",
  "cursor_after": "<ISO8601>",
  "fetched": 0,
  "skipped_dup": 0,
  "written": 0,
  "files": ["data/by_topic/.../platform_a-20250101.jsonl"],
  "error": null,
  "exit_code": 0
}

成功判据建议写成可判断的条件：进程退出码为 0，且每个来源的状态是 ok 或 skipped（无新数据），且本轮所有 files 路径在磁盘上存在、大小不为 0，且游标时间已前移。任一来源报错就返回非 0 退出码，但不要因此阻止其他来源继续写盘；游标只对成功的来源推进，失败的来源保留旧游标，下次自然重试。这样在 logs/run.err 里能看到异常堆栈，在结构化日志里能看到是哪一步断的。