abuse: теги accs + продажа подписок, строже порог; panels: реальные free-имена и 130/175 с бесплатными
This commit is contained in:
120
tools/refresh_panels_free.py
Normal file
120
tools/refresh_panels_free.py
Normal file
@@ -0,0 +1,120 @@
|
||||
"""Перепроверка реестра 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())
|
||||
Reference in New Issue
Block a user