抓取 OpenRouter 每天的免费模型 Top10 榜单 Python代码！

================== 代码开始

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
OpenRouter 免费模型 Top10 每日抓取

数据源（均为公开接口，无需 API Key）：
1. https://openrouter.ai/api/v1/models -> 免费模型清单（pricing.prompt == "0"）
2. https://openrouter.ai/api/frontend/v1/rankings/* -> 官方榜单（仅覆盖全球 Top10，用于补充真实 token 量）

产出：
reports/YYYY-MM-DD.md 当天 Markdown 榜单
data/YYYY-MM-DD.json 当天结构化数据
history.jsonl 追加式历史记录（用于对比排名变化）

用法：
python fetch_free_top10.py # 抓今天
python fetch_free_top10.py --top 20 # 改成 Top20
"""

import argparse
import csv
import gzip
import json
import math
import os
import ssl
import sys
import time
import urllib.error
import urllib.request
from datetime import datetime, timezone, timedelta

BASE = os.path.dirname(os.path.abspath(__file__))
REPORTS = os.path.join(BASE, "reports")
DATA = os.path.join(BASE, "data")
HISTORY = os.path.join(BASE, "history.jsonl")

MODELS_URL = "https://openrouter.ai/api/v1/models"
RANK_ENDPOINTS = [
"natural-language",
"programming-language",
"context-length",
"image-output",
"rerank-documents",
]
RANK_URL = "https://openrouter.ai/api/frontend/v1/rankings/{}"

UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0 Safari/537.36"
CST = timezone(timedelta(hours=8))

def http_get(url, timeout=60, retries=4):
"""带重试的 GET，返回 bytes。"""
ctx = ssl.create_default_context()
last = None
for attempt in range(retries):
try:
req = urllib.request.Request(
url,
headers={"User-Agent": UA, "Accept-Encoding": "gzip", "Accept": "application/json"},
)
with urllib.request.urlopen(req, timeout=timeout, context=ctx) as r:
data = r.read()
if r.headers.get("Content-Encoding") == "gzip":
data = gzip.decompress(data)
return data
except Exception as e: # noqa: BLE001
last = e
wait = 2 ** attempt
print(f" [retry {attempt + 1}/{retries}] {type(e).__name__}: {e} -> sleep {wait}s", file=sys.stderr)
time.sleep(wait)
raise RuntimeError(f"GET failed after {retries} tries: {url} ({last})")

def get_json(url, **kw):
return json.loads(http_get(url, **kw).decode("utf-8"))

# ---------------------------------------------------------------- 抓取

def fetch_free_models():
"""返回免费模型列表。判定：prompt 与 completion 均为 0。"""
data = get_json(MODELS_URL)
models = data.get("data", [])
free = []
for m in models:
pr = m.get("pricing") or {}
try:
p = float(pr.get("prompt", "1"))
c = float(pr.get("completion", "1"))
except (TypeError, ValueError):
continue
if p == 0 and c == 0:
free.append(m)
return models, free

def fetch_rankings():
"""抓官方榜单，返回 {endpoint: {'bucket': date, 'tokens': {model: n}}}。"""
out = {}
for ep in RANK_ENDPOINTS:
try:
d = get_json(RANK_URL.format(ep))
series = d.get("data") or []
if not series:
continue
last = series[-1]
out[ep] = {"bucket": last.get("x"), "tokens": last.get("ys") or {}}
print(f" rankings/{ep}: bucket={last.get('x')} entries={len(out[ep]['tokens'])}")
except Exception as e: # noqa: BLE001
print(f" rankings/{ep}: FAILED ({e})", file=sys.stderr)
out[ep] = {"bucket": None, "tokens": {}, "error": str(e)}
return out

# ---------------------------------------------------------------- 可用性探测

DEFAULT_KEY_FILE = os.path.join(os.path.expanduser("~"), ".workbuddy-ai", "models.json")

def load_api_key(explicit=None):
"""取 OpenRouter key：优先命令行，其次 WorkBuddy 的 models.json。"""
if explicit:
return explicit
env = os.environ.get("OPENROUTER_API_KEY")
if env:
return env
if os.path.exists(DEFAULT_KEY_FILE):
try:
cfg = json.load(open(DEFAULT_KEY_FILE, encoding="utf-8"))
for m in cfg if isinstance(cfg, list) else [cfg]:
if str(m.get("vendor", "")).lower() == "openrouter" and m.get("apiKey"):
return m["apiKey"]
except Exception: # noqa: BLE001
pass
return None

def probe_model(model_id, key, timeout=60):
"""对单个模型发一次最小请求，返回 (status, note)。"""
body = json.dumps({
"model": model_id,
"messages": [{"role": "user", "content": "hi"}],
"max_tokens": 4,
}).encode()
req = urllib.request.Request(
"https://openrouter.ai/api/v1/chat/completions",
data=body, method="POST",
headers={"Authorization": "Bearer " + key, "Content-Type": "application/json",
"Accept-Encoding": "gzip"},
)
try:
with urllib.request.urlopen(req, timeout=timeout) as r:
return r.status, "OK"
except urllib.error.HTTPError as e:
raw = e.read()
if e.headers.get("Content-Encoding") == "gzip":
raw = gzip.decompress(raw)
try:
msg = json.loads(raw.decode("utf-8", "replace"))["error"]["message"]
except Exception: # noqa: BLE001
msg = raw.decode("utf-8", "replace")[:120]
return e.code, msg
except Exception as e: # noqa: BLE001
return None, f"{type(e).__name__}: {e}"

def classify(status, note):
"""把探测结果翻译成人话。"""
if status == 200:
return "✅ 可用"
if status == 402:
return "❌ 需账户充值"
if status == 404:
return "❌ 模型不存在"
if status == 403:
low = (note or "").lower()
if "region" in low:
return "❌ 地区限制"
if "agentic" in low or "harness" in low or "coding agent" in low:
return "❌ 仅限 agent 环境"
return "❌ 无权限"
if status == 429:
return "❌ 触发限流"
return f"❌ {status}" if status else "❌ 网络失败"

def probe_all(rows, key, top_n):
"""对前 top_n 个模型做真实可用性探测。"""
out = {}
for r in rows[:top_n]:
s, note = probe_model(r["id"], key)
out[r["id"]] = {"status": s, "note": note, "label": classify(s, note)}
print(f" probe {r['id']:<46} {s} {out[r['id']]['label']}") return out # ---------------------------------------------------------------- 排名打分 def score_models(free, rankings): """ 综合打分。官方榜单只覆盖全球 Top10，多数免费模型拿不到 token 量， 因此以「上线时间新」为主序，「上下文长度」为辅， 官方榜出现情况作为加权加成与真实热度标记。 """ # 各维度取最大 token 量（不跨维度累加：维度间口径不同，累加会失真） token_max = {} bucket_of = {} dims_of = {} for ep, info in rankings.items(): for mid, n in (info.get("tokens") or {}).items(): if n > token_max.get(mid, 0):
token_max[mid] = n
bucket_of.setdefault(mid, []).append(info.get("bucket"))
dims_of.setdefault(mid, []).append(ep)

now = int(time.time())
max_ctx = max([m.get("context_length") or 0 for m in free] or [1]) or 1
all_tokens = [v for v in token_max.values() if v > 0]
log_max = (max((math.log10(v) for v in all_tokens), default=0.0)) or 1.0

rows = []
for m in free:
mid = m["id"]
created = int(m.get("created") or 0)
ctx = int(m.get("context_length") or 0)
tokens = token_max.get(mid, 0)
age_days = max(0, (now - created) // 86400) if created else 9999

# 新鲜度：越新越高（0~100），90 天以上视为 0
freshness = max(0.0, 100.0 * (1 - age_days / 90.0)) if created else 0.0
ctx_score = 100.0 * ctx / max_ctx

# 官方榜加成：进过榜加权，未进榜不罚分；token 量做对数归一化
official_bonus = 35.0 if tokens > 0 else 0.0
token_score = 15.0 * (math.log10(tokens) / log_max) if tokens > 0 else 0.0

total = 0.50 * freshness + 0.28 * ctx_score + official_bonus + token_score

rows.append({
"id": mid,
"name": m.get("name") or mid,
"context_length": ctx,
"created": created,
"created_date": datetime.fromtimestamp(created, CST).strftime("%Y-%m-%d") if created else "",
"age_days": age_days,
"is_free_suffix": mid.endswith(":free"),
"tier": "free" if mid.endswith(":free") else "promo",
"official_tokens": int(tokens),
"ranked_dims": sorted(set(dims_of.get(mid, []))),
"ranked_buckets": sorted(set(b for b in bucket_of.get(mid, []) if b)),
"supports_tools": "tools" in (m.get("supported_parameters") or []),
"supports_vision": "image" in (m.get("supported_parameters") or []),
"reasoning": bool(m.get("reasoning")),
"score": round(total, 2),
"modalities": m.get("architecture", {}).get("modality") or "",
"description": (m.get("description") or "").strip().replace("\n", " ")[:160],
})

rows.sort(key=lambda r: (-r["score"], -r["context_length"], r["id"]))
for i, r in enumerate(rows, 1):
r["rank"] = i
return rows

# ---------------------------------------------------------------- 历史对比

def load_history():
if not os.path.exists(HISTORY):
return {}
hist = {}
with open(HISTORY, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
rec = json.loads(line)
except json.JSONDecodeError:
continue
hist[rec.get("date")] = {m["id"]: m.get("rank") for m in rec.get("models", [])}
return hist

def dedupe_history():
"""同一天重跑会让 history.jsonl 出现重复日期行，只保留最后一条。"""
if not os.path.exists(HISTORY):
return
latest = {}
order = []
total = 0
with open(HISTORY, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
rec = json.loads(line)
except json.JSONDecodeError:
continue
total += 1
d = rec.get("date")
if d not in latest:
order.append(d)
latest[d] = line
if total == len(latest):
return # 无重复日期，无需重写
with open(HISTORY, "w", encoding="utf-8") as f:
for d in order:
f.write(latest[d] + "\n")

# ---------------------------------------------------------------- 输出

def fmt_int(n):
return f"{n:,}" if isinstance(n, (int, float)) else "-"

def write_markdown(date_str, rows, all_rows, rankings, changes, total_models, total_free, prev_date=None, probes=None):
probes = probes or {}
has_probe = bool(probes)
L = []
L.append(f"# OpenRouter 免费模型 Top{len(rows)} — {date_str}\n")
L.append(f"> 抓取时间：{datetime.now(CST).strftime('%Y-%m-%d %H:%M:%S')} (UTC+8)　"
f"平台模型总数：{total_models}　免费模型总数：{total_free}\n")

moved = [c for c in changes if c["delta"] not in (None, 0)]
baseline = f"（基准：{prev_date}）" if prev_date else ""
if moved or any(c["delta"] is None for c in changes):
up = [c for c in changes if c["delta"] is not None and c["delta"] < 0] down = [c for c in changes if c["delta"] is not None and c["delta"] > 0]
new = [c for c in changes if c["delta"] is None]
bits = []
if new:
bits.append("新进榜 " + ", ".join(f"`{c['id']}`" for c in new))
if up:
bits.append("上升 " + ", ".join(f"`{c['id']}`(+{-c['delta']})" for c in up[:5]))
if down:
bits.append("下降 " + ", ".join(f"`{c['id']}`(-{c['delta']})" for c in down[:5]))
if not moved and not new:
bits.append("无变化，全部模型排名与上次一致")
L.append(f"**较昨日变化**{baseline}：" + "；".join(bits) + "\n")
else:
L.append(f"**较昨日变化**{baseline}：无变化，全部模型排名与上次一致。\n")

head = "| 排名 | 模型名称 | 模型 ID | 上下文 | 上线日期 | 工具调用 |"
sep = "|---:|---|---|---:|:---:|:---:|"
if has_probe:
head += " 实测可用性 |"
sep += ":---:|"
head += " 较昨日 |"
sep += ":---:|"
L.append(head)
L.append(sep)
chg_map = {c["id"]: c for c in changes}
for r in rows:
c = chg_map.get(r["id"])
if c is None or c["delta"] is None:
arrow = "新进榜"
elif c["delta"] < 0: arrow = f"↑ {-c['delta']}" elif c["delta"] > 0:
arrow = f"↓ {c['delta']}"
else:
arrow = "—"
ctx = fmt_int(r["context_length"])
line = (f"| {r['rank']} | {r['name']} | `{r['id']}` | {ctx} | {r['created_date']} | "
f"{'✅' if r['supports_tools'] else '—'} |")
if has_probe:
p = probes.get(r["id"])
line += f" {p['label'] if p else '—'} |"
line += f" {arrow} |"
L.append(line)

if has_probe:
usable = [r for r in rows if (probes.get(r["id"]) or {}).get("status") == 200]
L.append(f"\n## 立即可用（免费额度内，实测通过）— {len(usable)}/{len(rows)} 个\n")
if usable:
L.append("| 排名 | 模型 ID | 上下文 | 工具调用 | 模型名称 |")
L.append("|---:|---|---:|:---:|---|")
for r in usable:
L.append(f"| {r['rank']} | `{r['id']}` | {fmt_int(r['context_length'])} | "
f"{'✅' if r['supports_tools'] else '—'} | {r['name']} |")
else:
L.append("本次未探测到可用模型。")

blocked = [r for r in rows if probes.get(r["id"]) and probes[r["id"]]["status"] != 200]
if blocked:
L.append(f"\n## 不可用（{len(blocked)} 个）及原因\n")
L.append("| 模型 ID | 状态码 | 原因 | 官方说明 |")
L.append("|---|---:|---|---|")
for r in blocked:
p = probes[r["id"]]
L.append(f"| `{r['id']}` | {p['status']} | {p['label']} | {(p['note'] or '')[:90]} |")

L.append(f"\n## 全部免费模型（按综合分，共 {total_free} 个）\n")
L.append("| # | 模型名称 | 模型 ID | 上下文 | 上线日期 | 工具调用 | 视觉 | 官方榜 token | 分数 | 免费类型 | 较昨日 |")
L.append("|---:|---|---|---:|:---:|:---:|:---:|---:|---:|---|---:|")
chg_all = {c["id"]: c for c in changes}
for r in all_rows:
c = chg_all.get(r["id"])
if c is None or c["delta"] is None:
arrow = "新进榜"
elif c["delta"] < 0: arrow = f"↑ {-c['delta']}" elif c["delta"] > 0:
arrow = f"↓ {c['delta']}"
else:
arrow = "—"
tok = fmt_int(r["official_tokens"]) if r["official_tokens"] else "未上榜"
tier = "`:free` 免费额度" if r.get("tier") == "free" else "零价推广（可能需充值）"
L.append(f"| {r['rank']} | {r['name']} | `{r['id']}` | {fmt_int(r['context_length'])} | "
f"{r['created_date']} | {'✅' if r['supports_tools'] else '—'} | "
f"{'✅' if r['supports_vision'] else '—'} | {tok} | {r['score']} | {tier} | {arrow} |")

L.append("\n## 数据来源说明\n")
L.append(f"- 免费模型清单：`{MODELS_URL}`（判定 pricing.prompt == 0 且 completion == 0）")
L.append("- 官方榜单接口（返回全球 Top10 时间序列，用于补充真实 token 量）：")
for ep, info in rankings.items():
L.append(f" - `{RANK_URL.format(ep)}` → 数据周期 {info.get('bucket')}"
+ (f"，⚠️ {info['error']}" if info.get("error") else ""))
L.append("\n> 注：OpenRouter 官方榜单仅公布全球 Top10，多数免费模型不在榜内。"
"本榜单的「排名」由模型上线时间与上下文长度综合生成，"
"「官方榜 token」列显示的是该模型真实进入官方榜时的 token 量（未上榜则为 0）。\n")

path = os.path.join(REPORTS, f"{date_str}.md")
with open(path, "w", encoding="utf-8") as f:
f.write("\n".join(L))
return path

def write_csv(date_str, rows):
path = os.path.join(DATA, f"{date_str}.csv")
cols = ["rank", "id", "name", "context_length", "created_date", "score", "tier",
"official_tokens", "supports_tools", "supports_vision", "reasoning", "is_free_suffix"]
with open(path, "w", encoding="utf-8-sig", newline="") as f:
w = csv.DictWriter(f, fieldnames=cols, extrasaction="ignore")
w.writeheader()
for r in rows:
w.writerow(r)
return path

def main():
ap = argparse.ArgumentParser()
ap.add_argument("--top", type=int, default=10, help="榜单长度，默认 10")
ap.add_argument("--probe", action="store_true",
help="用 OpenRouter key 实测每个模型是否真的可用（消耗免费额度）")
ap.add_argument("--key", help="OpenRouter API Key（默认自动从 ~/.workbuddy-ai/models.json 读取）")
args = ap.parse_args()

os.makedirs(REPORTS, exist_ok=True)
os.makedirs(DATA, exist_ok=True)

date_str = datetime.now(CST).strftime("%Y-%m-%d")
print(f"[1/4] 抓取模型清单 {MODELS_URL}")
all_models, free = fetch_free_models()
total_models = len(all_models)
print(f" 模型总数 {total_models}，免费模型 {len(free)}")

print(f"[2/4] 抓取官方榜单 {len(RANK_ENDPOINTS)} 个维度")
rankings = fetch_rankings()

print("[3/4] 计算排名")
ranked = score_models(free, rankings)
top = ranked[: args.top]

probes = {}
if args.probe:
key = load_api_key(args.key)
if not key:
print(" ⚠️ 未找到 OpenRouter key，跳过可用性探测", file=sys.stderr)
else:
print(f" 实测可用性（key {key[:12]}...{key[-4:]}）")
probes = probe_all(ranked, key, args.top)

hist = load_history()
# 只与「今天之前」最近一次快照对比，避免重跑当天时拿自己当基准
prior = [d for d in sorted(hist) if d < date_str] prev = prior[-1] if prior else None prev_map = hist.get(prev, {}) if prev else {} changes = [] for r in ranked: old = prev_map.get(r["id"]) # delta = 新名次 - 旧名次：负数 = 名次数字变小 = 排名上升（箭头 ↑） changes.append({ "id": r["id"], "prev_rank": old, "delta": (r["rank"] - old) if old else None, }) print("[4/4] 输出报告") md = write_markdown(date_str, top, ranked, rankings, changes, total_models, len(free), prev, probes) csv_path = write_csv(date_str, ranked) with open(os.path.join(DATA, f"{date_str}.json"), "w", encoding="utf-8") as f: json.dump({ "date": date_str, "fetched_at": datetime.now(CST).isoformat(), "total_models": total_models, "free_count": len(free), "probes": probes, "models": ranked, "rankings_meta": {k: {"bucket": v.get("bucket"), "error": v.get("error")} for k, v in rankings.items()}, }, f, ensure_ascii=False, indent=2) with open(HISTORY, "a", encoding="utf-8") as f: f.write(json.dumps({ "date": date_str, "total_free": len(free), "models": [{"id": r["id"], "rank": r["rank"], "score": r["score"]} for r in ranked], }, ensure_ascii=False) + "\n") dedupe_history() print("\n=== OpenRouter 免费模型 Top%d (%s) ===" % (len(top), date_str)) for r in top: tok = fmt_int(r["official_tokens"]) if r["official_tokens"] else "—" print(f"{r['rank']:>2}. {r['id']:<52} ctx={fmt_int(r['context_length']):>10} tokens={tok:>14}")
print(f"\n报告: {md}\nCSV : {csv_path}")

if __name__ == "__main__":
main()

===================== 代码结束

学习资料见知识星球。

以上就是今天要分享的技巧，你学会了吗？若有什么问题，欢迎在下方留言。

快来试试吧，小琥 my21ke007。获取 1000个免费 Excel模板福利​​​​！

更多技巧， www.excelbook.cn

欢迎 加入 零售创新 知识星球，知识星球主要以数据分析、报告分享、数据工具讨论为主；