omni-parse/tools/panel_hunt.py

"""Охота за панелями new-api: /api/status -> /api/pricing -> бесплатные модели.

Панель new-api (одна из самых частых раздач в абуз-каналах) отдаёт каталог
моделей и цены публично. Нулевая цена = бесплатный доступ = находка для абузов.

Запуск:
    python tools/panel_hunt.py                  # домены из data/radar_sources.json
    python tools/panel_hunt.py --hosts a.com b.com
    python tools/panel_hunt.py --limit 400 --workers 24
    python tools/panel_hunt.py --apply          # записать data/panels.json (источники omni)

Пишет только пути/счётчики. Ключи и токены не трогает.
"""
from __future__ import annotations

import argparse
import concurrent.futures as cf
import json
import ssl
import sys
import urllib.error
import urllib.request
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
DATA = ROOT / "data"
UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0 Safari/537.36"
CTX = ssl.create_default_context()
CTX.check_hostname = False
CTX.verify_mode = ssl.CERT_NONE
MARKERS = ("HeaderNavModules", "system_name", "chats", "quota_per_unit", "display_in_currency", "announcements_enabled")


def _get(url, timeout=8, cap=400000):
    req = urllib.request.Request(url, headers={"User-Agent": UA, "Accept": "application/json"})
    with urllib.request.urlopen(req, timeout=timeout, context=CTX) as r:
        return r.status, r.read(cap).decode("utf-8", "ignore")


def _is_newapi(body: str) -> bool:
    if '"data"' not in body:
        return False
    return any(m in body for m in MARKERS)


def probe(host: str):
    """-> dict | None. Проверяет сам хост и api.<host>."""
    for h in (host, "api." + host if not host.startswith("api.") else host):
        try:
            st, body = _get("https://" + h + "/api/status", timeout=7)
            if st != 200 or not _is_newapi(body):
                continue
        except Exception:
            continue
        rec = {"host": h, "status": "newapi", "models": 0, "free": [], "free_count": 0}
        try:
            st, body = _get("https://" + h + "/api/pricing", timeout=12)
            if st == 200:
                js = json.loads(body)
                data = js.get("data") if isinstance(js, dict) else None
                if isinstance(data, list):
                    rec["models"] = len(data)
                    free = []
                    for m in data:
                        if not isinstance(m, dict):
                            continue
                        try:
                            zero = float(m.get("model_ratio") or 0) == 0 and float(m.get("completion_ratio") or 0) == 0
                        except Exception:
                            zero = False
                        if zero and m.get("model_name"):
                            free.append(m["model_name"])
                    rec["free"] = free[:80]
                    rec["free_count"] = len(free)
        except Exception:
            rec["pricing"] = "err"
        return rec
    return None


def load_hosts(args) -> list:
    if args.hosts:
        return list(dict.fromkeys(args.hosts))
    p = DATA / "radar_sources.json"
    if not p.exists():
        return []
    try:
        d = json.loads(p.read_text(encoding="utf-8"))
    except Exception:
        return []
    out = []
    for item in d.get("domains") or []:
        dom = item[0] if isinstance(item, list) else (item.get("domain") or "")
        if dom:
            out.append(str(dom))
    return list(dict.fromkeys(out))


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--hosts", nargs="*", default=None)
    ap.add_argument("--limit", type=int, default=0, help="0 = все домены")
    ap.add_argument("--workers", type=int, default=24)
    ap.add_argument("--apply", action="store_true", help="записать data/panels.json")
    ap.add_argument("--min-free", type=int, default=0, help="только панели с >= N бесплатных моделей")
    a = ap.parse_args()

    hosts = load_hosts(a)
    if a.limit:
        hosts = hosts[:a.limit]
    if not hosts:
        print("нет доменов: сначала omni radar --apply")
        return 1

    res = []
    with cf.ThreadPoolExecutor(max(4, a.workers)) as ex:
        futs = {ex.submit(probe, h): h for h in hosts}
        for i, f in enumerate(cf.as_completed(futs), 1):
            try:
                r = f.result()
            except Exception:
                r = None
            if r:
                res.append(r)
                print("  %s | new-api | моделей %d | free %d" % (r["host"], r["models"], r["free_count"]))
            if i % 200 == 0:
                print("... проверено %d/%d, найдено %d" % (i, len(hosts), len(res)), flush=True)

    res.sort(key=lambda x: x["free_count"], reverse=True)
    if a.min_free:
        res = [r for r in res if r["free_count"] >= a.min_free]
    print("итог: проверено %d | панелей new-api %d | с бесплатными моделями %d | free-моделей всего %d"
          % (len(hosts), len(res), sum(1 for r in res if r["free_count"]), sum(r["free_count"] for r in res)))

    if a.apply:
        out = DATA / "panels.json"
        prev = {}
        if out.exists():
            try:
                for r in json.loads(out.read_text(encoding="utf-8")).get("panels") or []:
                    prev[r.get("host")] = r
            except Exception:
                prev = {}
        for r in res:
            prev[r["host"]] = r  # свежие данные перекрывают старые, мёртвые не затираются
        payload = {"panels": sorted(prev.values(), key=lambda x: x.get("free_count") or 0, reverse=True)}
        out.write_text(json.dumps(payload, ensure_ascii=False, indent=1), encoding="utf-8")
        print("записано: %s (панелей %d)" % (out, len(payload["panels"])))
    return 0


if __name__ == "__main__":
    sys.exit(main())