реестр: дедуп источников строго по url (2 падавших теста зелёные); инкрементальный abuse-индекс + tgsearch из предыдущей сессии

tools/panel_hunt.py

@@ -0,0 +1,153 @@
+"""Охота за панелями 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())