1. 从手工整理到自动出榜:行业排行榜的真实痛点
行业排行榜这件事,看起来只是"把数字排个序",真正动手才会发现它是一条完整的生产流水线。数据从哪来、字段怎么统一、不同量纲的指标怎么比较、榜单多久更新一次、异常数据怎么排除,每个环节都需要明确规则。我做过一段时间的行业研究,最头疼的不是算法,而是数据口径——同一个销量指标,A 来源叫 sales,B 来源叫 volume,C 来源用"万"做单位,直接合并必然出错。
手工整理排行榜有四个典型问题。第一是口径不一致,不同来源的字段命名、单位、统计周期都可能不同。第二是更新不可控,行业数据每天都在变,手工只能做到周更甚至月更,发布时榜单已经过时。第三是重复劳动严重,每次整理都要反复打开多个页面、复制粘贴、在表格里核对。第四是可扩展性差,新增一个行业或一种榜单类型,手工流程几乎要重做一遍。
把排行榜生成拆成"采集、清洗、计算、输出"四个环节,每个环节做成可配置、可复用的模块,就能实现自动出榜。这篇内容聚焦用 OpenClaw 完成行业销量、热度、融资数据采集,再经排序计算自动产出排行榜。我会给出可复制的config.toml与settings.json骨架,演示通过 TaoToken 统一 Key 打通 AI 工具配置链路,并附上采集到排序到榜单生成的验证动作与报错排查步骤。适合有基本 Python 能力、想搭建自己榜单系统的读者。
2. TaoToken 前置:统一 Key 打通配置链路
在讲采集和排序之前,先把工具链的配置问题解决掉。做这类数据项目时,经常需要在多个 AI 工具之间切换——写解析器时用代码助手,调排序逻辑时用对话模型,跑 Agent 任务时又要换一套配置。每个工具单独配 Key、单独管额度,时间长了很容易乱。
TaoToken 在这里的作用是提供一个统一的 API 通道,把不同 AI 工具的接入配置收敛到一处。你只需要在 TaoToken 控制台创建一个 Key,然后在各个工具里引用同一个 Key 和 API 地址即可。官网地址是 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,API 端点是 https://taotoken.net/api 。
具体操作上,先到控制台的 API Keys 页面生成一个 Key:https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 。生成后复制保存,后面在settings.json里会用到。如果你需要先验证模型是否可用,可以到模型对话页面测试:https://taotoken.net/model-chat?utm_source=taotoken_aicg_blog_end&utm_content=model-chat&utm_campaign=rewrite 。长期做编码和 Agent 任务的话,Coding Plan 会更合适:https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite 。
注意:Key 只保存在本地配置文件或环境变量里,不要提交到代码仓库。建议用
.gitignore排除settings.json,或者改用环境变量注入。
接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ,里面列出了兼容的接口格式和参数说明。ClaudeCode 相关的配置参考 https://taotoken.net/claude-code-anthropic?utm_source=taotoken_aicg_blog_end&utm_content=claude-code-anthropic&utm_campaign=rewrite 。把 Key 和端点准备好之后,下面进入采集配置。
3. 可复制配置:config.toml 与 settings.json 骨架
整个项目用两个配置文件分离关注点:config.toml管采集任务和排序参数,settings.json管 AI 工具接入和运行环境。这样调整榜单规则时不用碰代码,换工具时也不用改采集逻辑。
先看config.toml。它定义了采集任务列表、行业范围、热度权重和更新周期:
# config.toml [global] industry = "新能源汽车" period = "month" top_n = 50 db_path = "rank.db" timezone = "UTC" [collector] min_delay = 0.5 max_delay = 2.0 timeout = 15 retry_times = 3 respect_robots = true [hot_weights] search_index = 0.4 social_mention = 0.35 news_count = 0.25 [decay] half_life_days = 7.0 enabled = true [[tasks]] task_id = "sales_source_01" name = "example-sales-page" url = "https://example.com/industry/sales" method = "GET" source = "example_site" rank_type = "sales" parse_rule = "parse_sales_page" interval_seconds = 3600 [[tasks]] task_id = "funding_source_01" name = "example-funding-page" url = "https://example.com/industry/funding" method = "GET" source = "example_site" rank_type = "funding" parse_rule = "parse_funding_page" interval_seconds = 7200再看settings.json,它负责 AI 工具接入和运行参数:
{ "api_base": "https://taotoken.net/api", "api_key": "sk-your-taotoken-key", "model": "claude-sonnet", "timeout": 60, "max_retries": 2, "log_level": "INFO", "output_dir": "./output", "snapshot_retention_days": 90 }api_base和api_key就是前面在 TaoToken 控制台拿到的信息。model字段指定默认调用的模型,写解析器或生成榜单说明时可以复用。snapshot_retention_days控制榜单快照保留天数,避免历史数据无限增长。
读取这两个配置的代码可以这样写:
import json import tomllib from pathlib import Path def load_config(path: str = "config.toml") -> dict: with open(path, "rb") as f: return tomllib.load(f) def load_settings(path: str = "settings.json") -> dict: with open(path, "r", encoding="utf-8") as f: return json.load(f) CONFIG = load_config() SETTINGS = load_settings()tomllib是 Python 3.11 起内置的,低版本可以用tomli替代。配置加载完成后,采集器就可以按CONFIG["tasks"]逐条执行。
4. 采集与排序核心实现
配置就绪后,进入采集和排序的核心代码。采集层封装请求发送、超时控制、异常重试和限流;排序层按榜单类型分别计算。
采集器实现如下:
import time import random import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry class OpenClawCollector: def __init__(self, cfg: dict): self.session = requests.Session() retry = Retry( total=cfg.get("retry_times", 3), backoff_factor=0.6, status_forcelist=[429, 500, 502, 503, 504], allowed_methods=["GET", "POST"], ) adapter = HTTPAdapter(max_retries=retry) self.session.mount("http://", adapter) self.session.mount("https://", adapter) self.session.headers.update({ "User-Agent": "OpenClaw-Collector/1.0 (+business-analysis)" }) self.min_delay = cfg.get("min_delay", 0.5) self.max_delay = cfg.get("max_delay", 2.0) def fetch(self, task: dict) -> dict: last_error = None for attempt in range(task.get("retry_times", 3) + 1): try: time.sleep(random.uniform(self.min_delay, self.max_delay)) resp = self.session.request( method=task["method"], url=task["url"], timeout=task.get("timeout", 15), ) resp.raise_for_status() return { "task_id": task["task_id"], "status_code": resp.status_code, "body": resp.text, "collected_at": time.strftime("%Y-%m-%dT%H:%M:%S"), } except Exception as exc: last_error = exc time.sleep((2 ** attempt) + random.random()) raise RuntimeError(f"task {task['task_id']} failed: {last_error}")关键点在于把重试、指数退避、随机延迟封装在同一个入口。随机延迟能降低对目标服务器的压力,也避免被简单的反爬策略拦截。respect_robots提醒我们采集公开数据时要遵守目标站点的 robots 协议和服务条款,只采集允许抓取的内容。
排序层按榜单类型分开实现。销量榜直接降序,热度榜需要归一化后加权,融资榜先按主体聚合再排序:
import math from collections import defaultdict def min_max_normalize(values): if not values: return [] lo, hi = min(values), max(values) if hi == lo: return [0.5 for _ in values] return [(v - lo) / (hi - lo) for v in values] def log_normalize(values): return [math.log1p(v) for v in values] def build_sales_rank(records, industry, period="month", top_n=50): candidates = [ r for r in records if r.get("category") == industry and r.get("metric_name") == "sales" and r.get("period") == period ] candidates.sort(key=lambda x: x["metric_value"], reverse=True) return [ {"rank": i, "item_name": r["item_name"], "sales": r["metric_value"]} for i, r in enumerate(candidates[:top_n], start=1) ] def build_hot_rank(records, industry, weights, top_n=50): candidates = [ r for r in records if r.get("category") == industry and r.get("metric_name") in weights ] item_scores = defaultdict(dict) for r in candidates: item_scores[r["item_name"]][r["metric_name"]] = float(r["metric_value"]) score_map = defaultdict(float) for metric, weight in weights.items(): values = [item_scores[name].get(metric, 0.0) for name in item_scores] normalized = min_max_normalize(log_normalize(values)) for name, nv in zip(item_scores.keys(), normalized): score_map[name] += nv * weight sorted_items = sorted(score_map.items(), key=lambda kv: kv[1], reverse=True) return [ {"rank": i, "item_name": name, "hot_score": round(score, 4)} for i, (name, score) in enumerate(sorted_items[:top_n], start=1) ] def build_funding_rank(records, industry, top_n=50): candidates = [ r for r in records if r.get("category") == industry and r.get("metric_name") == "funding" ] total_by_name = defaultdict(float) rounds_by_name = defaultdict(int) for r in candidates: total_by_name[r["item_name"]] += r["metric_value"] rounds_by_name[r["item_name"]] += 1 sorted_items = sorted(total_by_name.items(), key=lambda kv: kv[1], reverse=True) return [ { "rank": i, "item_name": name, "funding_total": total, "rounds": rounds_by_name[name], } for i, (name, total) in enumerate(sorted_items[:top_n], start=1) ]热度榜对每个子指标先做对数归一化再加权,消除量纲差异。融资榜按主体聚合总额,避免同一公司多轮融资被拆成多条记录。这些函数接收清洗后的标准记录,输出带名次的榜单列表。
5. 验证请求与成功结果
代码写完后需要验证整条链路是否跑通。验证分三步:先确认采集能拿到数据,再确认清洗后字段完整,最后确认榜单输出符合预期。
第一步,单独测试采集器:
from collector import OpenClawCollector from config_loader import CONFIG collector = OpenClawCollector(CONFIG["collector"]) task = CONFIG["tasks"][0] payload = collector.fetch(task) print("status:", payload["status_code"]) print("body length:", len(payload["body"]))成功时输出类似:
status: 200 body length: 48213如果状态码不是 200,或者 body 长度为 0,说明目标地址或解析规则有问题,先排查采集层。
第二步,验证清洗后的记录:
from cleaner import clean_records raw = [ {"item_name": "示例公司A", "category": "新能源汽车", "metric_name": "sales", "metric_value": 12000, "period": "month"}, {"item_name": "示例公司B", "category": "新能源汽车", "metric_name": "sales", "metric_value": 8500, "period": "month"}, ] cleaned = clean_records(raw) print("cleaned count:", len(cleaned)) for r in cleaned: print(r["item_name"], r["metric_name"], r["metric_value"])预期输出:
cleaned count: 2 示例公司A sales 12000 示例公司B sales 8500第三步,跑完整流水线生成榜单:
from ranker import build_sales_rank, build_hot_rank, build_funding_rank sales = build_sales_rank(cleaned, "新能源汽车") print("销量榜:") for item in sales: print(f" {item['rank']} - {item['item_name']} ({item['sales']})")成功时输出:
销量榜: 1 - 示例公司A (12000) 2 - 示例公司B (8500)三步都通过,说明采集、清洗、排序链路已经打通。接下来把结果写入快照表,前端只读快照,不再实时排序。
6. 本篇常见错排查
实际跑的时候会遇到几类典型报错,这里按现象、原因、处理方式列出来。
采集超时或连接被拒。现象是requests.exceptions.ConnectTimeout或ConnectionError。原因通常是目标地址不可达、网络策略限制或超时设置过短。处理方式是先确认地址在浏览器能打开,再适当调大timeout,并检查retry_times是否生效。如果目标站点有频率限制,把min_delay和max_delay调大。
解析后记录数为 0。现象是采集成功但清洗后没有数据。原因多半是解析规则里的 CSS 选择器与页面结构不匹配。处理方式是先把payload["body"]保存到本地文件,用浏览器开发者工具核对选择器,再更新parse_rule对应的解析函数。
单位换算错误导致量级异常。现象是某条记录的metric_value比其他记录大几个数量级。原因是"万""亿"等单位没有正确换算。处理方式是在_parse_number里补充单位判断,并在清洗阶段加一条校验规则,数值超过合理区间就标记待复核。
热度榜权重配置不生效。现象是调整hot_weights后榜单排名没变化。原因是配置加载后没有重新传入build_hot_rank,或者权重字段名与记录里的metric_name不一致。处理方式是打印weights和item_scores的键,确认两边对得上。
TaoToken 请求返回 401 或 403。现象是调用 AI 工具时报鉴权失败。原因是settings.json里的api_key填错或已失效。处理方式是到控制台重新生成 Key,确认api_base是https://taotoken.net/api,然后重启程序。如果还是失败,到接入文档核对请求头格式。
快照表查询变慢。现象是榜单接口响应时间随数据量增长。原因是快照表没有建索引。处理方式是在rank_type和industry上建联合索引,并确保前端读的是快照而不是实时排序。
提示:所有报错都建议先看日志里的
task_id和status_code,大部分问题能定位到具体任务或具体字段。
7. 接入与排障入口
配置链路和排障都走通之后,日常维护主要就是调整采集任务和排序参数。如果你在接入过程中遇到鉴权或端点问题,到 API Keys 页面重新生成 Key 并核对配置:https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 。接口参数和请求格式的细节在接入文档里:https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 。
需要先验证模型是否可用时,用模型对话页面快速测试:https://taotoken.net/model-chat?utm_source=taotoken_aicg_blog_end&utm_content=model-chat&utm_campaign=rewrite 。长期跑编码和 Agent 任务的话,Coding Plan 的额度管理更省心:https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite 。ClaudeCode 相关配置参考:https://taotoken.net/claude-code-anthropic?utm_source=taotoken_aicg_blog_end&utm_content=claude-code-anthropic&utm_campaign=rewrite 。
最后提醒一点:采集公开数据时控制好频率,遵守目标站点的服务条款,只采集允许抓取的内容。榜单的公正性建立在数据质量和规则透明之上,把指标口径、权重和更新时间写清楚,榜单才有长期价值。