120 lines
4.7 KiB
Python
120 lines
4.7 KiB
Python
"""Перепроверка реестра new-api панелей: жив ли /api/pricing и сколько моделей бесплатных.
|
||
|
||
Запуск: python tools/refresh_panels_free.py [--workers 12] [--timeout 15] [--write]
|
||
|
||
Пишет data/panels.json (+ .bak) только с --write. В выводе — сводка и топ панелей.
|
||
Секретов не касается: только публичные /api/pricing и /api/models.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import shutil
|
||
import urllib.request
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
PANELS = ROOT / "data" / "panels.json"
|
||
UA = "curl/8.5.0"
|
||
|
||
|
||
def _get(url: str, timeout: int):
|
||
req = urllib.request.Request(url, headers={"User-Agent": UA, "Accept": "application/json"})
|
||
with urllib.request.urlopen(req, timeout=timeout) as r:
|
||
return r.getcode(), r.read()
|
||
|
||
|
||
def probe(entry: dict, timeout: int) -> dict:
|
||
host = str(entry.get("host") or entry.get("url") or "").strip()
|
||
host = host.replace("https://", "").replace("http://", "").strip("/")
|
||
out = dict(entry)
|
||
out["host"] = host
|
||
out["checked"] = True
|
||
if not host:
|
||
out.update(alive=False, free_count=0, models=0, why="no-host")
|
||
return out
|
||
base = "https://" + host
|
||
for path in ("/api/pricing", "/api/models"):
|
||
try:
|
||
code, body = _get(base + path, timeout)
|
||
except Exception as e:
|
||
out.update(alive=False, why=type(e).__name__)
|
||
continue
|
||
if code != 200:
|
||
out.update(alive=False, why="http %s" % code)
|
||
continue
|
||
try:
|
||
js = json.loads(body.decode("utf-8", "replace"))
|
||
except Exception:
|
||
out.update(alive=False, why="bad-json")
|
||
continue
|
||
if path == "/api/models":
|
||
rows = js.get("data") if isinstance(js, dict) else js
|
||
if isinstance(rows, list):
|
||
out.update(alive=True, why="models-only", models=len(rows), free_count=int(out.get("free_count") or 0))
|
||
return out
|
||
rows = js.get("data") if isinstance(js, dict) else None
|
||
if not isinstance(rows, list):
|
||
out.update(alive=False, why="no-data")
|
||
continue
|
||
free = 0
|
||
names: list[str] = []
|
||
for m in rows:
|
||
if not isinstance(m, dict):
|
||
continue
|
||
ratio = m.get("model_ratio")
|
||
quota = m.get("quota_type")
|
||
try:
|
||
freeish = (ratio is not None and float(ratio) == 0.0) or (quota == 0)
|
||
except Exception:
|
||
freeish = False
|
||
if freeish:
|
||
free += 1
|
||
nm = m.get("model_name") or m.get("model")
|
||
if isinstance(nm, str) and len(names) < 12:
|
||
names.append(nm)
|
||
out.update(alive=True, why="pricing", models=len(rows), free_count=free, free=names)
|
||
return out
|
||
return out
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--workers", type=int, default=12)
|
||
ap.add_argument("--timeout", type=int, default=15)
|
||
ap.add_argument("--write", action="store_true")
|
||
a = ap.parse_args()
|
||
|
||
raw = json.loads(PANELS.read_text(encoding="utf-8"))
|
||
rows = raw["panels"] if isinstance(raw, dict) else raw
|
||
print("в реестре:", len(rows))
|
||
|
||
with ThreadPoolExecutor(max_workers=a.workers) as ex:
|
||
res = list(ex.map(lambda e: probe(e, a.timeout), rows))
|
||
|
||
alive = [r for r in res if r.get("alive")]
|
||
with_free = [r for r in alive if int(r.get("free_count") or 0) > 0]
|
||
total_free = sum(int(r.get("free_count") or 0) for r in with_free)
|
||
print("живых /api/pricing:", len(alive), "| с бесплатными:", len(with_free), "| free-моделей:", total_free)
|
||
print("было: с бесплатными %d" % sum(1 for r in rows if int(r.get("free_count") or 0) > 0))
|
||
|
||
with_free.sort(key=lambda r: -int(r.get("free_count") or 0))
|
||
for r in with_free[:10]:
|
||
print(" %-28s free %-4s всего %s" % (r["host"], r.get("free_count"), r.get("models")))
|
||
|
||
from collections import Counter
|
||
print("причины отказов:", Counter(r.get("why") for r in res if not r.get("alive")).most_common(6))
|
||
|
||
if a.write:
|
||
if PANELS.exists():
|
||
shutil.copy2(PANELS, PANELS.with_suffix(".json.bak"))
|
||
payload = dict(raw) if isinstance(raw, dict) else {"panels": res}
|
||
payload["panels"] = res
|
||
PANELS.write_text(json.dumps(payload, ensure_ascii=False, indent=1), encoding="utf-8")
|
||
print("записано:", PANELS)
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main()) |