jsonscraper
市场

如何通过 TikTok 公开视频评估潜在客户线索质量:Python 分步流水线

按关键词收集视频样本,提取潜在需求信号,并将可核查的数据交给分析师或 AI 分类器

编辑政策 报告错误

销售团队总会从不同渠道收到同一个问题:潜在客户正在尝试解决哪些任务?手动搜索 TikTok 会得到一堆视频,但很难在一周后重复搜索、比较结果,也难以解释为什么某条视频会进入报告。对于小规模研究,可以将几个精确查询的公开视频结果转化为可复现的数据流:保存原始响应,提取视频和作者字段,去除重复项,并单独评估购买或选择产品意向的迹象。

文章:如何通过 TikTok 公开视频评估潜在客户线索质量:Python 分步流水线
摘要:面向营销人员和开发者的实用方案:通过 TikTok Scraper API 按受众搜索词查找公开视频,保存原始响应、去重视频,并依据明确规则排序。文中提供 Python 脚本,可处理两种响应结构,并安全传递 API 密钥。
图片用途:封面——核心概念
章节:
视觉主题:特写编辑插画
© jsonscraper · AI 生成插画

下面介绍一个使用 jsonscraper 的 TikTok Scraper API 编写的 Python 实用方案。例如,一家视频剪辑服务商希望了解创作者如何描述字幕、模板和导出方面的难题。脚本会收集搜索返回的结果,之后由分析师检查分类,并决定哪些表述适合用于受众研究。这是一种组织公开视频观察的方法,并非自动统计所有有购买意向的用户。

问题:受众搜索词已经存在,但研究难以复现

手动研究很容易丢失上下文。一位员工搜索“如何添加字幕”,另一位搜索“Reels 自动字幕”,第三位只保存自己喜欢的视频。最终表格混杂了不同的查询、日期、指标和筛选标准。当负责人要求重复分析或比较两周的数据时,团队不仅要还原结果,还得重建整个搜索过程。

更好的做法是预先定义观察单位:一条找到的视频,以及它的说明文字、标识符、链接、作者和可获取的公开指标。每条视频还要保存原始搜索词和采集时间。这样便能分别回答以下问题:出现了哪些表述、提到了哪些问题、哪些视频获得了更多公开互动,以及已保存的快照之间发生了什么变化。

以一家开发字幕和短视频制作工具的公司为例。它希望找到创作者直接讨论手动转录、语音识别错误、字幕时间轴,或为多个平台准备内容的公开视频。不能仅凭一条视频就把作者认定为“潜在客户”或“买家”:研究人员可以按照可观察到的迹象为内容排序,再人工核实结论。

解决方案:搜索、记录、清理、评估

这条流水线分为四个阶段。首先,团队建立一份小型搜索词表,每个查询对应受众的一项具体任务。然后,脚本调用视频搜索并保存完整的 JSON 响应。第三阶段将不同的响应结构规范化为每条视频一行的统一记录,并按标识符去重。最后,由规则或单独的 AI 分类器为内容设定人工审核优先级,同时保留评分原因。

将这些阶段分开有助于复现。原始响应可以帮助排查字段变化或规范化错误。规范化表格便于筛选和可视化。单独保存评分表,则可以在无需重新调用 API 的情况下调整规则,前提是保存的数据适合再次处理,且符合团队的保留期限。

作为基础查询,可以使用“自动字幕错误”“如何快速添加字幕”和“商业短视频剪辑”。这些只是词表示例,实际措辞应根据受众语言和产品进行调整。搜索词需要与结果一同保存,因为不同措辞会形成不同的样本集。

TikTok Scraper API 提供什么

在 jsonscraper 发布的文档中,视频搜索使用基础地址 https://tiktok.evelode.com 下的 GET 路由 /searchVideo。必填参数为 keyword;文档列出的可选参数包括 count、cursor、region、date_posted、sort_by 和 cache_timeout。例如,sort_by 接受文档中列出的 most-liked、relevance 和 date-posted;date_posted 接受 this-week 或 all-time 等值。参数清单和路由用途详见TikTok API 端点参考文档。

密钥通过查询参数 license_key 传递。下面的示例从环境变量 JSONSCRAPER_TIKTOK_LICENSE_KEY 读取密钥,以免将其存入代码或写入日志。请根据当前的访问条件设置认证方式和参数;TikTok Scraper API 页面介绍了产品,并提供演示和文档入口。

解析搜索响应时,应根据实际结构处理,而不是假设只有一种数组。在已检查的示例中,API 返回顶层状态和一个 tiktok 对象;视频可能位于 tiktok.aweme_list 中,搜索结果也可能出现在 tiktok.search_item_list[].aweme_info 中。两种结构都应检查。如果字段存在,找到的视频对象中会包含视频标识符、说明文字、链接、发布时间、作者信息和统计数据,例如播放量、点赞数、评论数和分享数。应将指标和字段视为可选项:缺少值时,代码不应伪造零值。

搜索分页使用 cursor。如果脚本继续获取结果,应将返回值作为不透明的下一页令牌传入,而不是自行计算。对于初始原型,每个查询只收集一页是合理的:这样更易于检查,也能让负载可预测。之后可以加入分页,同时限制页数并防止重复使用游标。

准备运行环境

示例使用 Python 3.8 或更高版本,以及标准库。在 macOS 或 Linux 的 shell 中设置环境变量,例如:

export JSONSCRAPER_TIKTOK_LICENSE_KEY="你的密钥"
python tiktok_demand.py

在 Windows PowerShell 中可这样设置:

$env:JSONSCRAPER_TIKTOK_LICENSE_KEY="你的密钥"
python .\tiktok_demand.py

脚本会为每个关键词分别发起请求。它会检查 HTTP 错误、JSON 响应和外层状态,然后从已确认的两种数组结构中提取对象。原始响应逐行写入 JSONL 文件:每条记录包含搜索词、观察时间和原始响应。第二部分将结果整理成字段统一的 CSV。代码中不包含密钥文件,也不会打印密钥内容。

示例:收集并规范化搜索结果

import csv
import json
import os
import time
from datetime import datetime, timezone
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
from urllib.request import Request, urlopen

BASE_URL = "https://tiktok.evelode.com/searchVideo"
QUERIES = [
    "自动字幕错误",
    "如何快速添加字幕",
    "商业短视频剪辑",
]


def extract_videos(payload):
    """支持已观察到的两种搜索响应结构。"""
    tiktok = payload.get("tiktok") or {}
    videos = []

    aweme_list = tiktok.get("aweme_list") or []
    if isinstance(aweme_list, list):
        videos.extend(item for item in aweme_list if isinstance(item, dict))

    search_items = tiktok.get("search_item_list") or []
    if isinstance(search_items, list):
        for item in search_items:
            if not isinstance(item, dict):
                continue
            video = item.get("aweme_info")
            if isinstance(video, dict):
                videos.append(video)

    return videos


def fetch(query, license_key):
    params = {
        "keyword": query,
        "count": 20,
        "sort_by": "relevance",
        "date_posted": "this-month",
        "region": "US",
        "license_key": license_key,
    }
    url = BASE_URL + "?" + urlencode(params)
    request = Request(url, headers={"Accept": "application/json"})

    try:
        with urlopen(request, timeout=45) as response:
            payload = json.loads(response.read().decode("utf-8"))
    except HTTPError as exc:
        raise RuntimeError("API 返回 HTTP 状态码 {}".format(exc.code)) from exc
    except (URLError, TimeoutError, json.JSONDecodeError) as exc:
        raise RuntimeError("无法完成 API 请求或解析响应") from exc

    if payload.get("status") != "ok":
        raise RuntimeError("API 响应状态不是 ok")
    return payload


def metric(statistics, key):
    value = statistics.get(key)
    return value if isinstance(value, (int, float)) else ""


def main():
    license_key = os.environ.get("JSONSCRAPER_TIKTOK_LICENSE_KEY")
    if not license_key:
        raise SystemExit("请在环境变量中设置 JSONSCRAPER_TIKTOK_LICENSE_KEY")

    raw_path = Path("tiktok_raw.jsonl")
    csv_path = Path("tiktok_videos.csv")
    seen_ids = set()
    rows = []

    with raw_path.open("a", encoding="utf-8") as raw_file:
        for query in QUERIES:
            observed_at = datetime.now(timezone.utc).isoformat()
            payload = fetch(query, license_key)
            raw_file.write(json.dumps({
                "query": query,
                "observed_at": observed_at,
                "response": payload,
            }, ensure_ascii=False) + "\n")

            for video in extract_videos(payload):
                video_id = video.get("aweme_id")
                if not video_id or video_id in seen_ids:
                    continue
                seen_ids.add(video_id)

                author = video.get("author") or {}
                stats = video.get("statistics") or {}
                rows.append({
                    "query": query,
                    "observed_at": observed_at,
                    "video_id": video_id,
                    "created_at": video.get("create_time", ""),
                    "description": video.get("desc", ""),
                    "share_url": video.get("share_url", ""),
                    "author_username": author.get("unique_id", ""),
                    "author_nickname": author.get("nickname", ""),
                    "views": metric(stats, "play_count"),
                    "likes": metric(stats, "digg_count"),
                    "comments": metric(stats, "comment_count"),
                    "shares": metric(stats, "share_count"),
                })
            time.sleep(1)

    columns = [
        "query", "observed_at", "video_id", "created_at", "description",
        "share_url", "author_username", "author_nickname", "views",
        "likes", "comments", "shares",
    ]
    with csv_path.open("w", newline="", encoding="utf-8") as csv_file:
        writer = csv.DictWriter(csv_file, fieldnames=columns)
        writer.writeheader()
        writer.writerows(rows)

    print("已将 {} 条不重复的返回视频保存至 {}".format(len(rows), csv_path))


if __name__ == "__main__":
    main()

此处将 count 设为每个查询请求的样本数量。脚本不会假设 API 必须返回恰好这么多条结果:实际行数取决于响应。请求之间的暂停只是控制运行节奏的简单措施,并不表示服务存在某个限额或要求。用于生产环境的定时任务时,应加入设有上限的重试机制、错误监控和可接受的请求频率规则,并依据自身套餐条件进行设置。

如何阅读生成的表格

CSV 中每一行代表一条通过某个搜索词找到的、唯一的返回视频。query 字段标明记录来源,observed_at 表示快照时间,video_id 用作去重键。说明文字和链接便于分析师核查上下文,公开计数器则反映采集时的互动情况。若要进行长期比较,应分别保存多次观察结果;需要查看指标变化时,不要覆盖历史记录。

播放量、点赞数、评论数和分享数是 API 在请求时返回的公开统计快照。个别对象可能缺少这些数据,因此 CSV 中的空值表示响应里没有该数值字段。它与零值不同,计算汇总时应单独处理。此外,搜索结果只是针对特定词语、地区、时间筛选条件和请求参数返回的样本。它们描述的是实际获取到的视频,而非平台上所有内容或受众的完整规模。

初步评估潜在相关性时,可以建立一份透明的规则表:与问题表述匹配的情况、提及所用工具、询问价格或进行比较、是否包含具体工作场景。最好分别保存各项特征和评分理由,而不是只记录一个不透明的分数。例如,可以把询问自动转录准确度的视频标记为“待核查:讨论了质量”,把泛泛的内容建议视频标记为“主题相关性较低”。

接入 AI,但不要让它充当裁判

积累一定样本后,可以让模型按照预先设定的方案标注说明文字:问题、提及的工具、提问语气,以及是否存在产品选择信号。只向模型传递所需字段,例如说明文字和记录标识符,并要求其返回含标签和简短理由的固定 JSON。将指令版本、模型和结果与原始记录一同保存,以便重复标注或由人工抽样核查。

有效的分类提示应将结论限制在可观察到的迹象上。可以要求模型判断文本是否明确提到字幕问题、是否请求推荐工具,或是否讨论工作流程。如果信息不足,应设置“上下文不足”类别。模型不应仅凭一条公开视频推断作者身份、购买意向或人口统计特征。最终产物是供分析的主题队列,而非用于自动联系的人员名单。

要评估标注质量,可以随机抽取一部分样本,由人工标注相同类别。逐项比较一致性并检查错误,例如模型可能将讽刺误判为抱怨,或把教育性建议当成个人经历。完善搜索词表和指令,通常比立刻增加模型复杂度更有用。报告中保存原始文本和链接时,应遵循既定的数据保留与访问政策。

定时自动化与质量控制

脚本成功手动运行后,可以通过 cron、云平台的任务调度器,或 n8n 之类的编排工具定时运行。实际的节点顺序不变:读取查询列表、调用 API、检查状态、保存原始响应、规范化视频、写入新快照并生成报告。密钥应通过运行环境的秘密管理机制传递,而不是写进表格、工作流文本或错误消息。

自动任务最好包含四项检查。第一,请求是否成功完成且响应是否为 JSON。第二,外层状态是否符合预期值。第三,处理器是否能识别两种搜索结构,并妥善处理缺失字段。第四,新出现的不重复视频数量和错误数量是否写入日志。如果结果突然为空,报告应将零样本显示为一次运行的状态,而不应在没有说明的情况下将其与普通成功报告混为一谈。

在验证完第一页之后,再添加分页。搜索路由的文档列出了 cursor 参数;保存返回的令牌,并在下一次请求中原样传递。限制每个搜索词的最大页数;如果没有新条目、没有下一个游标,或游标已经处理过,则停止循环。这样可以控制运行时间和采集规模。游标的存在意味着可以请求继续获取结果,但并不保证结果涵盖所有发布内容。

同一天重复运行时,返回结果可能重叠。因此,去重键应使用稳定的视频标识符,快照历史则应作为包含观察日期的独立记录保存。如果需要汇总视图,可基于快照构建:这样既能区分首次发现的视频,也能识别出现在多个搜索词中或之后再次获取的视频。

指标:如何进行合理比较

运营监控可关注每个查询找到的不重复视频数、说明文字非空的比例、可用互动数据的分布,以及特定主题出现的频率。比较不同查询时,应保持参数和运行周期一致。如果一次运行使用 this-week,另一次使用 this-month,样本量差异可能只是时间窗口造成的,因此没有进一步分析时,不能将其解释为兴趣变化。

要比较同一条视频随时间变化的指标,需要保存多个快照,并采用统一的存储方案。每个日期都保留原始值,再计算各次观察之间的差异。如果实际运行间隔不同,就不要将差异称作“单日增长”。同样,按点赞数排序有助于找出样本中较受关注的视频,却不能衡量作者或产品的商业成效。

面向营销团队的最终报告可以包含三个层面:附有可点击链接的受众表述示例、问题主题分组,以及采集样本的定量描述。每项结论都应注明观察数量、搜索词和采集周期。这既能为负责人提供背景,也便于分析师基于同一份原始数据重复筛选。

常见错误及规避方法

将空字段当作零值

如果统计数据或作者信息缺失,上面的代码会将单元格留空。这保留了未知值与测得的零值之间的区别。进行分析时,应明确规定缺失值处理方式,并注明各项计算排除的行数。

假设搜索始终只返回一个数组

搜索响应中可能包含 aweme_list,也可能包含嵌套的 search_item_list[].aweme_info。处理器会检查两种结构,并忽略类型不符的元素。如果结构发生变化,保存的原始响应有助于快速找到数据的新位置;测试用的 JSON 固定样本则可用于在下次运行前验证修复。

把同一条视频重复算作多个信号

一条视频可能与多个搜索词匹配。示例会在最终 CSV 中只保留一次,并将当前顺序中首次匹配的查询写入字段。如果需要保留所有“视频—查询”对应关系,应建立单独的关联表,或为每个标识符保存查询列表,而不是丢失记录来源。

比较不可比的样本

搜索词、地区、时间筛选条件、排序方式和运行日期都会影响样本。请将参数与响应一同保存;进行变量比较时,每次只改变一个因素。这样有助于判断结果变化是由搜索结果行为导致,还是实验设置发生了改变。

把相关性说成购买意向

视频说明文字和公开计数器能提供内容特征和受众反应的迹象,但它们本身无法说明谁准备购买、某人是否看过广告,或转化率如何。请将这些信息用于提出假设、验证主题和准备内容。销售成效应通过自身的营销活动数据评估,并确保它与这项研究在时间和受众上具有可比性。

如何扩展这条流水线

基础版本稳定运行后,可以添加配置文件来保存查询列表和运行参数,从而无需修改 Python 代码即可调整词表。下一步可以建立查询与主题类别的对应关系,例如“识别质量”“剪辑速度”和“导出”。这些类别应由团队定义,并通过实例验证;API 返回搜索数据,语义分类体系则由研究人员设定。

若要替代 CSV,可使用关系型数据库,建立运行记录表、视频表和指标快照表。对视频标识符建立唯一索引,可以防止目录重复;快照则能跟踪可用计数器的变化。观察日期应使用 UTC 保存,并记录规范化器版本。提取逻辑变化后,可以重新处理归档的 JSONL,并比较结果。

在 n8n 中,可以按计划启动采集,并通过独立的 Python 服务或节点执行规范化代码,前提是所选环境支持所需操作。第一阶段不一定要接入模型:先确保采集稳定,再衡量人工分类所需时间以及重复出现的类别。只有当 AI 能减少具体的重复工作,且其标签可以核查时,才值得使用。

如果实际问题不是按表述进行搜索,而是定期观察某个账号,官方参考文档还提供了单独的 /getUserFeed 路由,支持 query、max_cursor 和 count 参数。针对话题标签,则有使用 cursor 参数的 /getHashtagFeed。这属于不同的数据采集场景:选择 endpoint 应从研究问题出发,而不是用一种样本替代另一种。

结果:可复现的样本,而不再是一堆收藏

运行后,团队会得到两项有用成果:原始响应存档,以及包含搜索词、观察日期、链接、说明文字和可用指标的不重复视频表。团队可以据此人工归纳受众问题,将文本交给受控分类流程,制作每周摘要,并比较多次快照。代码会明确处理请求错误、检查外层状态、区分两种搜索结构,并将缺失指标保留为空值。

对团队来说,实际价值在于更透明地判断哪些问题值得在访谈、落地页或广告文案中验证。如果研究需要其他类型的数据,可以调整搜索词或采用相应的文档路由。改造流水线前,可先查看TikTok API 文档核对参数;产品信息和演示入口则位于TikTok Scraper API 页面。

这种方法解决了最初的问题:将 TikTok 搜索变成可复现的研究流程。脚本收集实际返回的视频,保存其来源和快照;随后由人工或可核查的分类器将内容转化为需求假设。团队由此获得下一步决策的依据,同时也能明确区分公开视频内容的迹象与经过验证的销售结果。

人们

尚未列出本文的人员。

继续阅读如何用 Python 归档 TikTok 搜索结果
阅读下一篇文章

将阅读内容转化为可用的集成

探索 jsonscraper 社交数据 API、测试请求并构建下一个工作流。

查看 API