skill-video-lapian/scripts/fetch_comments.py

#!/usr/bin/env python3
"""fetch_comments.py — 采集评论区(可选, VOC 网关翻页).

用法:
  python3 fetch_comments.py --aweme-id 7677548065384877346 --workdir /tmp/lapian [--max 100]
产物: workdir/comments.json = [{nickname,text,likes,...}...] (按点赞排序)
"""
import argparse
import json
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from common import voc_call  # noqa: E402


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--aweme-id", required=True)
    ap.add_argument("--workdir", required=True)
    ap.add_argument("--max", type=int, default=100)
    args = ap.parse_args()

    comments, cursor, seen = [], 0, set()
    while len(comments) < args.max:
        resp = voc_call("douyin/app/v3/fetch_video_comments",
                        {"aweme_id": args.aweme_id, "cursor": cursor, "count": 20})
        if resp.get("error"):
            print(f"[comments] 网关错误: {resp['error']}")
            break
        data = resp.get("data") or {}
        batch = data.get("comments") or (data.get("data") or {}).get("comments") or []
        if not batch:
            break
        new = 0
        for c in batch:
            cid = c.get("cid") or c.get("comment_id") or f"{c.get('user', {}).get('nickname')}|{c.get('create_time')}"
            if cid in seen:
                continue
            seen.add(cid)
            comments.append({
                "nickname": (c.get("user") or {}).get("nickname") or "",
                "text": c.get("text") or "",
                "likes": c.get("digg_count") or 0,
                "reply_count": c.get("reply_comment_total") or 0,
                "time": c.get("create_time") or 0,
                "ip_label": c.get("ip_label") or "",
            })
            new += 1
        print(f"[comments] cursor={cursor} +{new} (累计 {len(comments)})")
        cursor += len(batch)
        if new == 0:
            break

    comments.sort(key=lambda x: -x["likes"])
    out = os.path.join(args.workdir, "comments.json")
    os.makedirs(args.workdir, exist_ok=True)
    json.dump(comments, open(out, "w", encoding="utf-8"), ensure_ascii=False, indent=2)
    print(f"[comments] {len(comments)} 条 → {out}")


if __name__ == "__main__":
    main()