2c6281d041
Expand manhuang collect/UI (tabs, markets, guishi/quote probes), add configs and launchd helpers, ops handbooks, and scripts for channel validation and price/threat enrichment. Keys stay in keys.env.example only.
628 lines
24 KiB
Python
Executable file
628 lines
24 KiB
Python
Executable file
#!/usr/bin/env python3
|
||
"""社区/GitHub/公开 API 价源 → 本地缓存（自己渠道）。
|
||
|
||
源（见 configs/manhuang/channels.json）：
|
||
1) howardpen9/awesome-ai-api-proxy prices.latest.json — 中转 token 价
|
||
2) OpenRouter /api/v1/models — 官方聚合
|
||
3) 5sim /v1/guest/prices — 接码
|
||
4) mnfst free data.json + cheahjs README — free 档目录
|
||
5) LiteLLM model_prices JSON — 官方 MSRP 锚
|
||
|
||
写出：~/.cache/manhuang/quotes/community-prices.json
|
||
不把密钥写盘；失败单项记 error，不整崩。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
import time
|
||
import urllib.error
|
||
import urllib.request
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
UA = "joy-manhuang-community-prices/1.0 (+local-cache; no-secrets)"
|
||
OUT_DIR = Path.home() / ".cache" / "manhuang" / "quotes"
|
||
OUT = OUT_DIR / "community-prices.json"
|
||
MSRP_OUT = OUT_DIR / "msrp-litellm.json"
|
||
FREE_OUT = OUT_DIR / "free-llm-lists.json"
|
||
CFG_USER = Path.home() / ".config" / "manhuang" / "channels.json"
|
||
CFG_REPO = Path(__file__).resolve().parents[2] / "configs" / "manhuang" / "channels.json"
|
||
|
||
PROXY_PRICES = (
|
||
"https://raw.githubusercontent.com/howardpen9/awesome-ai-api-proxy"
|
||
"/main/data/prices.latest.json"
|
||
)
|
||
OPENROUTER_MODELS = "https://openrouter.ai/api/v1/models"
|
||
FIVESIM_PRICES = "https://5sim.net/v1/guest/prices"
|
||
MNFST_FREE_JSON = (
|
||
"https://raw.githubusercontent.com/mnfst/awesome-free-llm-apis/main/data.json"
|
||
)
|
||
CHEAHJS_README = (
|
||
"https://raw.githubusercontent.com/cheahjs/free-llm-api-resources/main/README.md"
|
||
)
|
||
LITELLM_PRICES = (
|
||
"https://raw.githubusercontent.com/BerriAI/litellm"
|
||
"/main/model_prices_and_context_window.json"
|
||
)
|
||
|
||
# sku → LiteLLM 模型 key 候选（优先官方 provider，非 azure 前缀）
|
||
LITELLM_SKU_MODELS: dict[str, list[str]] = {
|
||
"openai": ["gpt-4o", "gpt-4.1", "gpt-4o-mini"],
|
||
"chatgpt_pro": ["gpt-4o", "gpt-4o-mini"],
|
||
"claude": ["claude-sonnet-4-20250514", "claude-3-7-sonnet-20250219", "claude-3-5-sonnet-20241022"],
|
||
"claude_max": ["claude-sonnet-4-20250514", "claude-3-7-sonnet-20250219"],
|
||
"deepseek": ["deepseek-chat", "deepseek/deepseek-chat"],
|
||
"openrouter": ["openrouter/openai/gpt-4o-mini", "openrouter/deepseek/deepseek-chat"],
|
||
}
|
||
|
||
# 用于「中转可比价」的 canonical / 名称片段
|
||
RELAY_MODEL_HINTS = (
|
||
"gpt-4o-mini",
|
||
"gpt-4o",
|
||
"claude-3.5-sonnet",
|
||
"claude-sonnet-4",
|
||
"claude-3-5-sonnet",
|
||
"deepseek-v3",
|
||
"deepseek-chat",
|
||
)
|
||
|
||
|
||
def _now() -> str:
|
||
return datetime.now(timezone.utc).isoformat()
|
||
|
||
|
||
def _fetch(url: str, timeout: int = 25) -> tuple[int, bytes]:
|
||
req = urllib.request.Request(
|
||
url,
|
||
headers={"User-Agent": UA, "Accept": "application/json,*/*"},
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=timeout) as r:
|
||
return int(getattr(r, "status", 200) or 200), r.read()
|
||
except urllib.error.HTTPError as e:
|
||
try:
|
||
body = e.read()
|
||
except Exception:
|
||
body = b""
|
||
return int(e.code), body
|
||
except Exception as e:
|
||
return 0, str(e).encode()[:200]
|
||
|
||
|
||
def _load_channels_cfg() -> dict[str, Any]:
|
||
for p in (CFG_USER, CFG_REPO):
|
||
if p.is_file():
|
||
try:
|
||
d = json.loads(p.read_text(encoding="utf-8"))
|
||
if isinstance(d, dict):
|
||
d["_cfg_path"] = str(p)
|
||
return d
|
||
except (OSError, json.JSONDecodeError):
|
||
continue
|
||
return {"_cfg_path": "builtin"}
|
||
|
||
|
||
def _aggregate_proxy_prices(data: dict[str, Any]) -> dict[str, Any]:
|
||
"""把 ~3k token 价压成 provider 摘要 + 代表性模型最低 input 价。"""
|
||
recs = data.get("records") or []
|
||
by_provider: dict[str, dict[str, Any]] = {}
|
||
ladder: list[dict[str, Any]] = []
|
||
|
||
for r in recs:
|
||
if not isinstance(r, dict):
|
||
continue
|
||
unit = str(r.get("unit") or "")
|
||
if "input" not in unit or "cache" in unit:
|
||
continue
|
||
price = r.get("price_usd")
|
||
if price is None:
|
||
continue
|
||
try:
|
||
price_f = float(price)
|
||
except (TypeError, ValueError):
|
||
continue
|
||
if price_f <= 0 or price_f > 500:
|
||
continue
|
||
pid = str(r.get("provider_id") or "unknown")
|
||
pname = str(r.get("provider_name") or pid)
|
||
raw = str(r.get("raw_model_name") or r.get("canonical_model") or "")
|
||
low = raw.lower()
|
||
slot = by_provider.setdefault(
|
||
pid,
|
||
{
|
||
"id": pid,
|
||
"name": pname,
|
||
"n_input": 0,
|
||
"min_input_usd_per_1m": None,
|
||
"sample_models": [],
|
||
"source_url": r.get("source_url"),
|
||
},
|
||
)
|
||
slot["n_input"] += 1
|
||
cur = slot["min_input_usd_per_1m"]
|
||
if cur is None or price_f < float(cur):
|
||
slot["min_input_usd_per_1m"] = price_f
|
||
if len(slot["sample_models"]) < 5 and raw:
|
||
slot["sample_models"].append(raw[:80])
|
||
|
||
for hint in RELAY_MODEL_HINTS:
|
||
if hint in low or hint.replace(".", "-") in low or hint.replace("-", ".") in low:
|
||
ladder.append({
|
||
"provider_id": pid,
|
||
"provider_name": pname,
|
||
"model": raw[:100],
|
||
"hint": hint,
|
||
"price_usd_per_1m_input": price_f,
|
||
"source_url": r.get("source_url"),
|
||
"captured_at": r.get("captured_at"),
|
||
})
|
||
break
|
||
|
||
# 每个 hint 取最低价
|
||
best_by_hint: dict[str, dict[str, Any]] = {}
|
||
for row in ladder:
|
||
h = row["hint"]
|
||
prev = best_by_hint.get(h)
|
||
if not prev or row["price_usd_per_1m_input"] < prev["price_usd_per_1m_input"]:
|
||
best_by_hint[h] = row
|
||
|
||
providers = sorted(
|
||
by_provider.values(),
|
||
key=lambda x: (x["min_input_usd_per_1m"] is None, x["min_input_usd_per_1m"] or 9e9),
|
||
)
|
||
return {
|
||
"ok": True,
|
||
"snapshot_date": data.get("snapshot_date"),
|
||
"generated_at": data.get("generated_at"),
|
||
"record_count": data.get("record_count") or len(recs),
|
||
"providers_n": len(providers),
|
||
"providers": providers[:40],
|
||
"best_by_model_hint": best_by_hint,
|
||
"sku_hints": {
|
||
# 用 gpt-4o-mini 最低 input 作「中转可比指数」（$/1M）
|
||
"relay_index_model": "gpt-4o-mini",
|
||
"relay_index_usd_per_1m": (best_by_hint.get("gpt-4o-mini") or {}).get(
|
||
"price_usd_per_1m_input"
|
||
),
|
||
"openrouter_present": any(p["id"] == "openrouter" for p in providers),
|
||
},
|
||
}
|
||
|
||
|
||
def _openrouter_summary(data: dict[str, Any]) -> dict[str, Any]:
|
||
models = data.get("data") or []
|
||
n = len(models) if isinstance(models, list) else 0
|
||
sample = []
|
||
min_prompt = None
|
||
if isinstance(models, list):
|
||
for m in models[:80]:
|
||
if not isinstance(m, dict):
|
||
continue
|
||
mid = m.get("id") or ""
|
||
pricing = m.get("pricing") or {}
|
||
try:
|
||
pr = float(pricing.get("prompt") or 0)
|
||
except (TypeError, ValueError):
|
||
pr = 0.0
|
||
# openrouter 常是 per-token 美元
|
||
per_1m = pr * 1_000_000 if pr and pr < 1 else pr
|
||
if per_1m and (min_prompt is None or per_1m < min_prompt):
|
||
min_prompt = per_1m
|
||
if any(h in str(mid).lower() for h in ("gpt-4o-mini", "claude", "deepseek")):
|
||
sample.append({"id": mid, "prompt_per_1m_est": round(per_1m, 6) if per_1m else None})
|
||
if len(sample) >= 12:
|
||
break
|
||
return {
|
||
"ok": n > 0,
|
||
"models_n": n,
|
||
"min_prompt_per_1m_est": min_prompt,
|
||
"sample": sample,
|
||
}
|
||
|
||
|
||
def _fivesim_summary(data: dict[str, Any]) -> dict[str, Any]:
|
||
"""国别树 → 我们关心的产品最低价。"""
|
||
# 产品 key 因 5sim 而异；扫所有 product 名含 openai/claude/google/any
|
||
want_products = (
|
||
"openai", "chatgpt", "claude", "anthropic", "google", "gemini",
|
||
"microsoft", "any", "other", "telegram",
|
||
)
|
||
want_countries = {
|
||
"usa": "sms_us",
|
||
"unitedkingdom": "sms_otp",
|
||
"england": "sms_otp",
|
||
"netherlands": "sms_nl",
|
||
"indonesia": "sms_id",
|
||
"russia": "sms_otp",
|
||
"china": "sms_otp",
|
||
"hongkong": "sms_otp",
|
||
"philippines": "sms_otp",
|
||
"india": "sms_otp",
|
||
}
|
||
by_sku: dict[str, list[float]] = {}
|
||
samples: list[dict[str, Any]] = []
|
||
|
||
if not isinstance(data, dict):
|
||
return {"ok": False, "error": "not_object"}
|
||
|
||
for country, products in data.items():
|
||
if not isinstance(products, dict):
|
||
continue
|
||
ckey = str(country).lower().replace(" ", "")
|
||
sku = want_countries.get(ckey)
|
||
if not sku and ckey not in ("usa", "netherlands", "indonesia"):
|
||
# 仍扫 any 国别里的 openai 类
|
||
sku = None
|
||
for prod, operators in products.items():
|
||
pl = str(prod).lower()
|
||
if not any(w in pl for w in want_products):
|
||
continue
|
||
if not isinstance(operators, dict):
|
||
continue
|
||
for op, info in operators.items():
|
||
if not isinstance(info, dict):
|
||
continue
|
||
try:
|
||
cost = float(info.get("cost") or 0)
|
||
except (TypeError, ValueError):
|
||
continue
|
||
if cost <= 0 or cost > 50:
|
||
continue
|
||
target = sku or "sms_otp"
|
||
by_sku.setdefault(target, []).append(cost)
|
||
if len(samples) < 30:
|
||
samples.append({
|
||
"country": country,
|
||
"product": prod,
|
||
"operator": op,
|
||
"cost": cost,
|
||
"count": info.get("count"),
|
||
"sku_hint": target,
|
||
})
|
||
|
||
mins = {k: round(min(v), 4) for k, v in by_sku.items() if v}
|
||
return {
|
||
"ok": bool(mins),
|
||
"min_by_sku": mins,
|
||
"samples": samples[:20],
|
||
"countries_scanned": len(data) if isinstance(data, dict) else 0,
|
||
}
|
||
|
||
|
||
def _mnfst_free_summary(data: dict[str, Any]) -> dict[str, Any]:
|
||
providers = data.get("providers") if isinstance(data, dict) else None
|
||
if not isinstance(providers, list):
|
||
return {"ok": False, "error": "no_providers"}
|
||
rows: list[dict[str, Any]] = []
|
||
models_n = 0
|
||
for p in providers:
|
||
if not isinstance(p, dict):
|
||
continue
|
||
models = p.get("models") if isinstance(p.get("models"), list) else []
|
||
models_n += len(models)
|
||
rows.append({
|
||
"name": p.get("name"),
|
||
"category": p.get("category"),
|
||
"country": p.get("country"),
|
||
"url": p.get("url"),
|
||
"baseUrl": p.get("baseUrl"),
|
||
"models_n": len(models),
|
||
"sample_models": [str(m.get("id") or m.get("name") or "")[:60] for m in models[:4] if isinstance(m, dict)],
|
||
"rate_hint": (models[0].get("rateLimit") if models and isinstance(models[0], dict) else None),
|
||
})
|
||
return {
|
||
"ok": bool(rows),
|
||
"lastUpdated": data.get("lastUpdated"),
|
||
"providers_n": len(rows),
|
||
"models_n": models_n,
|
||
"providers": rows[:40],
|
||
}
|
||
|
||
|
||
def _cheahjs_readme_summary(text: str) -> dict[str, Any]:
|
||
"""README 粗统计：标题段落 + 外链数（非完整解析）。"""
|
||
if not text:
|
||
return {"ok": False, "error": "empty"}
|
||
headings = re.findall(r"(?m)^#{2,3}\s+(.+)$", text)
|
||
links = re.findall(r"https?://[^\s\)\]\>\"']+", text)
|
||
# 常见 provider 关键词命中
|
||
keywords = (
|
||
"OpenRouter", "Groq", "Gemini", "Mistral", "NVIDIA", "Together",
|
||
"Cerebras", "Cohere", "Hugging", "GitHub Models", "Cloudflare",
|
||
)
|
||
hits = [k for k in keywords if k.lower() in text.lower()]
|
||
return {
|
||
"ok": True,
|
||
"bytes": len(text.encode("utf-8", "replace")),
|
||
"headings_n": len(headings),
|
||
"headings_sample": [h.strip()[:80] for h in headings[:16]],
|
||
"links_n": len(set(links)),
|
||
"provider_keywords": hits,
|
||
}
|
||
|
||
|
||
def _litellm_msrp(data: dict[str, Any]) -> dict[str, Any]:
|
||
"""从 LiteLLM 价表抽 demand 相关 MSRP 锚（$/1M tokens）。"""
|
||
if not isinstance(data, dict):
|
||
return {"ok": False, "error": "not_object"}
|
||
anchors: dict[str, Any] = {}
|
||
for sku, keys in LITELLM_SKU_MODELS.items():
|
||
picked = None
|
||
for key in keys:
|
||
row = data.get(key)
|
||
if not isinstance(row, dict):
|
||
continue
|
||
try:
|
||
inp = float(row.get("input_cost_per_token") or 0)
|
||
outp = float(row.get("output_cost_per_token") or 0)
|
||
except (TypeError, ValueError):
|
||
continue
|
||
if inp <= 0 and outp <= 0:
|
||
continue
|
||
picked = {
|
||
"model": key,
|
||
"provider": row.get("litellm_provider"),
|
||
"mode": row.get("mode"),
|
||
"input_usd_per_1m": round(inp * 1_000_000, 6) if inp else None,
|
||
"output_usd_per_1m": round(outp * 1_000_000, 6) if outp else None,
|
||
"max_input_tokens": row.get("max_input_tokens") or row.get("max_tokens"),
|
||
}
|
||
break
|
||
if picked:
|
||
anchors[sku] = picked
|
||
zero_chat = 0
|
||
for _k, v in data.items():
|
||
if not isinstance(v, dict):
|
||
continue
|
||
if v.get("mode") != "chat":
|
||
continue
|
||
try:
|
||
if float(v.get("input_cost_per_token") or -1) == 0:
|
||
zero_chat += 1
|
||
except (TypeError, ValueError):
|
||
pass
|
||
return {
|
||
"ok": bool(anchors),
|
||
"models_in_db": len(data),
|
||
"zero_cost_chat_n": zero_chat,
|
||
"anchors": anchors,
|
||
"source": LITELLM_PRICES,
|
||
}
|
||
|
||
|
||
def probe() -> dict[str, Any]:
|
||
cfg = _load_channels_cfg()
|
||
out: dict[str, Any] = {
|
||
"ts": _now(),
|
||
"ok": True,
|
||
"cfg": cfg.get("_cfg_path"),
|
||
"sources": {},
|
||
"sku_overlay": {}, # sku_id → {price, currency, channel, kind, url, note}
|
||
"free_llm": {},
|
||
"msrp_anchor": {},
|
||
}
|
||
|
||
# 1) awesome-ai-api-proxy
|
||
code, body = _fetch(PROXY_PRICES)
|
||
src: dict[str, Any] = {"http": code, "bytes": len(body)}
|
||
if code == 200 and body:
|
||
try:
|
||
data = json.loads(body.decode("utf-8", "replace"))
|
||
agg = _aggregate_proxy_prices(data if isinstance(data, dict) else {})
|
||
src.update(agg)
|
||
idx = (agg.get("sku_hints") or {}).get("relay_index_usd_per_1m")
|
||
if idx is not None:
|
||
note = f"community token index gpt-4o-mini $/1M input ≈ {idx}"
|
||
for sid in ("relay_premium", "newapi_relay"):
|
||
out["sku_overlay"][sid] = {
|
||
"price": float(idx),
|
||
"currency": "USD",
|
||
"unit": "1m_input",
|
||
"channel": "awesome-ai-api-proxy",
|
||
"channel_id": "gh-proxy-prices",
|
||
"kind": "relay_index",
|
||
"url": "https://howardpen9.github.io/awesome-ai-api-proxy/",
|
||
"title": "relay_token_index",
|
||
"note": note,
|
||
"confidence": 0.7,
|
||
}
|
||
# openrouter 在表里的最低 input
|
||
for p in agg.get("providers") or []:
|
||
if p.get("id") == "openrouter" and p.get("min_input_usd_per_1m") is not None:
|
||
out["sku_overlay"]["openrouter"] = {
|
||
"price": float(p["min_input_usd_per_1m"]),
|
||
"currency": "USD",
|
||
"unit": "1m_input",
|
||
"channel": "awesome-ai-api-proxy/openrouter",
|
||
"channel_id": "gh-proxy-prices",
|
||
"kind": "relay_index",
|
||
"url": "https://openrouter.ai/models",
|
||
"title": "openrouter_min_input",
|
||
"confidence": 0.8,
|
||
}
|
||
break
|
||
except (json.JSONDecodeError, UnicodeError) as e:
|
||
src["error"] = str(e)[:120]
|
||
src["ok"] = False
|
||
else:
|
||
src["ok"] = False
|
||
src["error"] = body[:120].decode("utf-8", "replace") if body else "empty"
|
||
out["sources"]["awesome_ai_api_proxy"] = src
|
||
time.sleep(0.3)
|
||
|
||
# 2) OpenRouter live models
|
||
code, body = _fetch(OPENROUTER_MODELS)
|
||
src2: dict[str, Any] = {"http": code, "bytes": len(body)}
|
||
if code == 200 and body:
|
||
try:
|
||
data = json.loads(body.decode("utf-8", "replace"))
|
||
sm = _openrouter_summary(data if isinstance(data, dict) else {})
|
||
src2.update(sm)
|
||
if sm.get("min_prompt_per_1m_est") and "openrouter" not in out["sku_overlay"]:
|
||
out["sku_overlay"]["openrouter"] = {
|
||
"price": float(sm["min_prompt_per_1m_est"]),
|
||
"currency": "USD",
|
||
"unit": "1m_input",
|
||
"channel": "OpenRouter /models",
|
||
"channel_id": "openrouter-api",
|
||
"kind": "relay_index",
|
||
"url": OPENROUTER_MODELS,
|
||
"title": "openrouter_live_min",
|
||
"confidence": 0.85,
|
||
}
|
||
except (json.JSONDecodeError, UnicodeError) as e:
|
||
src2["ok"] = False
|
||
src2["error"] = str(e)[:120]
|
||
else:
|
||
src2["ok"] = False
|
||
src2["error"] = "fetch_fail"
|
||
out["sources"]["openrouter_models"] = src2
|
||
time.sleep(0.3)
|
||
|
||
# 3) 5sim
|
||
code, body = _fetch(FIVESIM_PRICES, timeout=40)
|
||
src3: dict[str, Any] = {"http": code, "bytes": len(body)}
|
||
if code == 200 and body:
|
||
try:
|
||
data = json.loads(body.decode("utf-8", "replace"))
|
||
sm = _fivesim_summary(data if isinstance(data, dict) else {})
|
||
src3.update(sm)
|
||
for sid, price in (sm.get("min_by_sku") or {}).items():
|
||
out["sku_overlay"][sid] = {
|
||
"price": float(price),
|
||
"currency": "USD",
|
||
"unit": "sms",
|
||
"channel": "5sim guest",
|
||
"channel_id": "5sim-guest",
|
||
"kind": "marketplace",
|
||
"url": "https://5sim.net/",
|
||
"title": f"5sim_min_{sid}",
|
||
"confidence": 0.9,
|
||
}
|
||
except (json.JSONDecodeError, UnicodeError) as e:
|
||
src3["ok"] = False
|
||
src3["error"] = str(e)[:120]
|
||
else:
|
||
src3["ok"] = False
|
||
src3["error"] = "fetch_fail"
|
||
out["sources"]["fivesim"] = src3
|
||
time.sleep(0.3)
|
||
|
||
# 4) free LLM lists — mnfst JSON + cheahjs README
|
||
free_doc: dict[str, Any] = {"ts": _now(), "sources": {}}
|
||
code, body = _fetch(MNFST_FREE_JSON)
|
||
src4: dict[str, Any] = {"http": code, "bytes": len(body)}
|
||
if code == 200 and body:
|
||
try:
|
||
data = json.loads(body.decode("utf-8", "replace"))
|
||
sm = _mnfst_free_summary(data if isinstance(data, dict) else {})
|
||
src4.update(sm)
|
||
free_doc["mnfst"] = sm
|
||
free_doc["sources"]["mnfst"] = {"ok": sm.get("ok"), "http": code}
|
||
except (json.JSONDecodeError, UnicodeError) as e:
|
||
src4["ok"] = False
|
||
src4["error"] = str(e)[:120]
|
||
else:
|
||
src4["ok"] = False
|
||
src4["error"] = "fetch_fail"
|
||
out["sources"]["mnfst_free"] = src4
|
||
time.sleep(0.3)
|
||
|
||
code, body = _fetch(CHEAHJS_README)
|
||
src5: dict[str, Any] = {"http": code, "bytes": len(body)}
|
||
if code == 200 and body:
|
||
text = body.decode("utf-8", "replace")
|
||
sm = _cheahjs_readme_summary(text)
|
||
src5.update(sm)
|
||
free_doc["cheahjs"] = sm
|
||
free_doc["sources"]["cheahjs"] = {"ok": sm.get("ok"), "http": code}
|
||
else:
|
||
src5["ok"] = False
|
||
src5["error"] = "fetch_fail"
|
||
out["sources"]["cheahjs_free"] = src5
|
||
free_doc["ok"] = bool((free_doc.get("mnfst") or {}).get("ok") or (free_doc.get("cheahjs") or {}).get("ok"))
|
||
free_doc["providers_n"] = (free_doc.get("mnfst") or {}).get("providers_n") or 0
|
||
free_doc["models_n"] = (free_doc.get("mnfst") or {}).get("models_n") or 0
|
||
out["free_llm"] = {
|
||
"ok": free_doc["ok"],
|
||
"providers_n": free_doc["providers_n"],
|
||
"models_n": free_doc["models_n"],
|
||
"mnfst_updated": (free_doc.get("mnfst") or {}).get("lastUpdated"),
|
||
"cheahjs_keywords": (free_doc.get("cheahjs") or {}).get("provider_keywords") or [],
|
||
"path": str(FREE_OUT),
|
||
}
|
||
|
||
# 5) LiteLLM MSRP anchors
|
||
code, body = _fetch(LITELLM_PRICES, timeout=60)
|
||
src6: dict[str, Any] = {"http": code, "bytes": len(body)}
|
||
msrp_doc: dict[str, Any] = {"ts": _now(), "ok": False}
|
||
if code == 200 and body:
|
||
try:
|
||
data = json.loads(body.decode("utf-8", "replace"))
|
||
sm = _litellm_msrp(data if isinstance(data, dict) else {})
|
||
src6.update({k: v for k, v in sm.items() if k != "anchors"})
|
||
src6["ok"] = sm.get("ok")
|
||
msrp_doc = {"ts": _now(), **sm}
|
||
out["msrp_anchor"] = sm.get("anchors") or {}
|
||
# 可选：把 msrp 写进 overlay（kind=msrp，不盖 marketplace）
|
||
for sid, a in (sm.get("anchors") or {}).items():
|
||
if not isinstance(a, dict) or a.get("input_usd_per_1m") is None:
|
||
continue
|
||
# 不覆盖已有 marketplace/relay_index；仅补 msrp 字段位
|
||
if sid not in out["sku_overlay"]:
|
||
out["sku_overlay"][sid] = {
|
||
"price": float(a["input_usd_per_1m"]),
|
||
"currency": "USD",
|
||
"unit": "1m_input",
|
||
"channel": f"LiteLLM/{a.get('provider') or 'msrp'}",
|
||
"channel_id": "litellm-msrp",
|
||
"kind": "msrp",
|
||
"url": "https://github.com/BerriAI/litellm",
|
||
"title": a.get("model"),
|
||
"confidence": 0.75,
|
||
"output_usd_per_1m": a.get("output_usd_per_1m"),
|
||
}
|
||
else:
|
||
out["sku_overlay"][sid]["msrp_1m_input"] = a.get("input_usd_per_1m")
|
||
out["sku_overlay"][sid]["msrp_model"] = a.get("model")
|
||
except (json.JSONDecodeError, UnicodeError) as e:
|
||
src6["ok"] = False
|
||
src6["error"] = str(e)[:120]
|
||
else:
|
||
src6["ok"] = False
|
||
src6["error"] = "fetch_fail"
|
||
out["sources"]["litellm_msrp"] = src6
|
||
|
||
out["overlay_n"] = len(out["sku_overlay"])
|
||
out["ok"] = out["overlay_n"] > 0 or any(
|
||
(s or {}).get("ok") for s in out["sources"].values()
|
||
)
|
||
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
||
OUT.write_text(json.dumps(out, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
FREE_OUT.write_text(json.dumps(free_doc, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
MSRP_OUT.write_text(json.dumps(msrp_doc, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
return out
|
||
|
||
|
||
def main() -> int:
|
||
r = probe()
|
||
print(
|
||
"wrote", OUT,
|
||
"overlay", r.get("overlay_n"),
|
||
"proxy_ok", (r.get("sources") or {}).get("awesome_ai_api_proxy", {}).get("ok"),
|
||
"5sim_ok", (r.get("sources") or {}).get("fivesim", {}).get("ok"),
|
||
"free_n", (r.get("free_llm") or {}).get("providers_n"),
|
||
"msrp_n", len(r.get("msrp_anchor") or {}),
|
||
)
|
||
for sid, o in (r.get("sku_overlay") or {}).items():
|
||
print(f" {sid}: {o.get('price')} {o.get('unit')} @ {o.get('channel')}")
|
||
fl = r.get("free_llm") or {}
|
||
print(f"free: providers={fl.get('providers_n')} models={fl.get('models_n')} → {FREE_OUT}")
|
||
print(f"msrp: {list((r.get('msrp_anchor') or {}).keys())} → {MSRP_OUT}")
|
||
return 0 if r.get("ok") else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|
