vault ключей + мост abuse-radar + китайские источники: всё в одном реестре

- omni/vault.py: пул ключей парсеров (env-first, data/vault.json, ротация, cooldown 401/402/403=1ч, 429=10м), только маски в выводе
- tools/build_vault.py: сборка пула из MASTER_KEYS.csv и локальных дампов (значения не в git)
- tools/vault_check.py: живая проверка с маскированием
- searchers: firecrawl/tavily/exa/serper/jina/apify берут ключ из vault, если нет env
- omni radar: домены+каналы из radar_server.db -> реестр, --import-posts заливает посты каналов
- radar-источники помечены opt_in: общий обход не тормозят (869 против 3337 всего)
- omni/sources_cn.py + tools/wave_cn.py: 9 живых китайских источников (52pojie, nodeseek, kanxue, anquanke, 4hou, xz.aliyun, ithome, cnbeta, v2ex)
- net.get(proxy=...): принудительный прокси для CF-площадок
- /api/keys + строка ключей в панели
- tests/test_vault_radar.py: 7 тестов, всего 29 passed

omni/cli.py

@@ -298,6 +298,122 @@ def cmd_pump(a):
     return 0
 
 
+def cmd_keys(a):
+    """Сводка vault: сколько ключей на сервис, сколько готово, что остывает."""
+    from . import vault
+    st = vault.stats()
+    if a.service:
+        st = [x for x in st if x["service"] == a.service]
+    if not st:
+        print("vault пуст — собери: python tools/build_vault.py")
+        return 1
+    w = max(len(x["service"]) for x in st)
+    print("%-*s %8s %8s %9s  %s" % (w, "сервис", "ключей", "готово", "остывают", "env"))
+    for x in st:
+        print("%-*s %8d %8d %9d  %s" % (w, x["service"], x["keys"], x["ready"],
+                                        x["cooling"], "*" if x["env"] else ""))
+    if a.verbose:
+        print()
+        for x in st:
+            for r in (vault._load()[0].get(x["service"]) or []):
+                print("  %-*s %s %-7s %s" % (w, x["service"], vault.mask(r["key"]),
+                                             r.get("status") or "?", (r.get("note") or "")[:30]))
+    return 0
+
+
+def cmd_radar(a):
+    """Мост из abuse-radar: домены и Telegram-каналы радара -> реестр omni."""
+    import json as _json
+    import sqlite3 as _sq
+    from pathlib import Path as _P
+    from . import sources as _src
+
+    cand = [a.db] if a.db else [
+        str(_P("C:/Users/User/tmp/abuse_radar/radar_server.db")),
+        str(_P("C:/Users/User/tmp/abuse_radar/radar.db")),
+        "data/radar.db",
+    ]
+    db = next((c for c in cand if _P(c).exists()), "")
+    if not db:
+        print("radar.db не найден, укажи --db")
+        return 2
+
+    c = _sq.connect(db)
+    doms = c.execute("SELECT domain, score, info FROM domains ORDER BY score DESC").fetchall()
+    chans = c.execute("SELECT name, ok, fails FROM channels WHERE ok=1").fetchall()
+    posts = c.execute("SELECT chan, mid, date, text FROM messages").fetchall() if a.import_posts else []
+    c.close()
+
+    if a.import_posts:
+        def head_line(txt):
+            for ln in (txt or "").splitlines():
+                ln = ln.strip()
+                if len(ln) >= 20 and sum(ch.isalpha() for ch in ln) >= 10:
+                    return ln[:150]
+            return ((txt or "").strip().splitlines() or [""])[0][:150]
+
+        items, seen_u = [], set()
+        for chan, mid, date, text in posts:
+            text = (text or "").strip()
+            if len(text) < 30:
+                continue
+            url_u = "https://t.me/%s/%s" % (chan, mid)
+            if url_u in seen_u:
+                continue
+            seen_u.add(url_u)
+            items.append(Item("rad_" + chan, "post", url=url_u, title=head_line(text), text=text,
+                              author="@" + chan, published=(date or "")[:19],
+                              source="rad_" + chan, extra={"via": "radar"}))
+        if a.limit:
+            items = items[:a.limit]
+        n_new = n_dup = 0
+        for i in range(0, len(items), 500):
+            n1, n2 = store.add(items[i:i + 500])
+            n_new += n1
+            n_dup += n2
+        print("постов радара: %d | новых %d, дублей %d" % (len(items), n_new, n_dup))
+
+    have = {s.url.rstrip('/').lower() for s in _src.all_sources()}
+    have_names = {s.name for s in _src.all_sources()}
+
+    new_dom, new_chan = [], []
+    for dom, score, info in doms:
+        if a.min_score and (score or 0) < a.min_score:
+            continue
+        url = "https://" + dom
+        if url.lower() in have or ("rad_" + dom.replace('.', '_').replace('-', '_')) in have_names:
+            continue
+        title = ""
+        try:
+            title = (_json.loads(info or "{}") or {}).get("title") or ""
+        except Exception:
+            pass
+        new_dom.append((dom, score, title))
+    for name, ok, fails in chans:
+        if ("tg_" + name) in have_names:
+            continue
+        new_chan.append(name)
+
+    if a.limit:
+        new_dom, new_chan = new_dom[:a.limit], new_chan[:a.limit]
+    print("радар: %s" % db)
+    print("доменов всего %d, новых для omni %d | каналов живых %d, новых %d"
+          % (len(doms), len(new_dom), len(chans), len(new_chan)))
+    for dom, score, title in new_dom[:5]:
+        print("  %-28s score %-4s %s" % (dom, score, (title or "")[:44]))
+    if new_chan[:5]:
+        print("  каналы:", ", ".join(new_chan[:8]))
+    if not a.apply:
+        print("dry-run: добавь --apply, чтобы записать в data/radar_sources.json")
+        return 0
+
+    out = _P("data") / "radar_sources.json"
+    out.write_text(_json.dumps({"domains": new_dom, "channels": new_chan},
+                               ensure_ascii=False, indent=1), encoding="utf-8")
+    print("записано: %s (доменов %d, каналов %d)" % (out, len(new_dom), len(new_chan)))
+    return 0
+
+
 def main(argv=None):
     try:
         sys.stdout.reconfigure(encoding="utf-8", errors="replace")
@@ -405,6 +521,19 @@ def main(argv=None):
     sp.add_argument("--apply", action="store_true")
     sp.set_defaults(fn=cmd_pump)
 
+    sp = sub.add_parser("keys", help="ключи парсеров из vault: сервисы, счётчики, остывание")
+    sp.add_argument("--service", default="", help="только один сервис")
+    sp.add_argument("-v", "--verbose", action="store_true", help="по ключам, масками")
+    sp.set_defaults(fn=cmd_keys)
+
+    sp = sub.add_parser("radar", help="забрать источники из abuse-radar (домены + каналы)")
+    sp.add_argument("--db", default="", help="путь к radar.db (по умолчанию ищет рядом)")
+    sp.add_argument("--apply", action="store_true", help="записать в реестр omni")
+    sp.add_argument("--min-score", type=int, default=20, help="минимальный score домена")
+    sp.add_argument("--import-posts", action="store_true", help="залить посты каналов радара в базу")
+    sp.add_argument("--limit", type=int, default=0)
+    sp.set_defaults(fn=cmd_radar)
+
     args = p.parse_args(argv)
     return args.fn(args)
 

omni/net.py

@@ -117,8 +117,11 @@ def post_json(url, payload, headers=None, timeout=None):
         return Resp(url, 0, "", "curl-post", f"{type(e).__name__}: {e}"[:160])
 
 
-def get(url, headers=None, timeout=None, json_mode=False):
-    """Лестница: httpx -> curl_cffi -> curl_cffi с прокси. Возвращает первый осмысленный ответ."""
+def get(url, headers=None, timeout=None, json_mode=False, proxy=None):
+    """Лестница: httpx -> curl_cffi -> curl_cffi с прокси. Возвращает первый осмысленный ответ.
+
+    proxy=... — принудительный прокси для конкретного источника (напр. CF-форумы через DE).
+    """
     timeout = timeout or _cfg.get("timeout", 25)
     host = urllib.parse.urlsplit(url).netloc
     _throttle(host)
@@ -132,6 +135,13 @@ def get(url, headers=None, timeout=None, json_mode=False):
     if ck:
         h["Cookie"] = ck
 
+    if proxy:
+        forced = _try_curl(url, h, timeout, proxy)
+        if forced.ok and not forced.blocked:
+            return forced
+        alt = _try_httpx(url, h, timeout, proxy)
+        return alt if alt.status == 200 else forced
+
     first = _try_httpx(url, h, timeout, None)
     if first.ok and not first.blocked:
         return first

omni/providers/searchers.py

@@ -14,8 +14,18 @@ def _mk(provider, title, url, text="", score=0.0, published="", extra=None):
                 published=published, extra=extra or {})
 
 
+ENV2SVC = {"FIRECRAWL_API_KEY": "firecrawl", "TAVILY_API_KEY": "tavily", "EXA_API_KEY": "exa",
+           "SERPER_API_KEY": "serper", "JINA_API_KEY": "jina", "APIFY_TOKEN": "apify"}
+
+
 def _key(name):
-    return config.KEYS.get(name)
+    """env первым, иначе ключ из vault (data/vault.json), иначе пусто."""
+    v = config.KEYS.get(name)
+    if v:
+        return v
+    from .. import vault
+    svc = ENV2SVC.get(name)
+    return vault.get(svc) if svc else ""
 
 
 def available(name):

omni/serve.py

@@ -236,6 +236,15 @@ class Handler(BaseHTTPRequestHandler):
             items = [{"key": k, "label": taxonomy.LABELS.get(k, k), "n": v} for k, v in cnt.items()]
             return self._json({"ok": True, "types": items, "total": sum(cnt.values())})
 
+        if path == "/api/keys":
+            from . import vault as _v
+            rows = _v.stats()
+            if q.get("service"):
+                rows = [x for x in rows if x["service"] == q["service"][0]]
+            return self._json({"ok": True, "services": rows,
+                               "total_keys": sum(x["keys"] for x in rows),
+                               "total_ready": sum(x["ready"] for x in rows)})
+
         if path == "/api/runs":
             try:
                 limit = _intv(q, "limit", 80, 1, 500)

omni/sources.py

@@ -207,6 +207,58 @@ SOURCES = _dedup(SOURCES_BASE + [Source(**d) for d in EXTRA] + [Source(**d) for
 BY_NAME = {s.name: s for s in SOURCES}
 CATEGORIES = sorted({s.category for s in SOURCES})
 
+_RADAR = {"at": 0.0, "rows": []}
+
+
+def radar_sources():
+    """Источники из abuse-radar: data/radar_sources.json (см. omni radar --apply).
+
+    Домены идут html_listing (root страница с маркерами), каналы — tg_web.
+    Файл перечитывается при изменении mtime, реестр не разрастается дублями.
+    """
+    import json
+    from pathlib import Path
+    p = Path(__file__).resolve().parents[1] / "data" / "radar_sources.json"
+    try:
+        mtime = p.stat().st_mtime
+    except OSError:
+        return []
+    if _RADAR["rows"] and _RADAR["at"] == mtime:
+        return _RADAR["rows"]
+    rows = []
+    try:
+        d = json.loads(p.read_text(encoding="utf-8"))
+    except Exception:
+        return []
+    for item in d.get("domains") or []:
+        dom, score, title = (list(item) + ["", 0, ""])[:3] if isinstance(item, list) else (item.get("domain"), item.get("score"), item.get("title"))
+        name = "rad_" + str(dom).replace(".", "_").replace("-", "_")
+        rows.append(Source(name=name, category="abuse", kind="site", url="https://" + str(dom),
+                           adapter="html_listing", tags=("radar", "abuse"),
+                           limit=12, detail=False, extra={"score": score, "title": (title or "")[:120], "opt_in": True}))
+    for ch in d.get("channels") or []:
+        name = "tg_" + str(ch).lstrip("@")
+        rows.append(Source(name=name, category="abuse", kind="tg", url="https://t.me/" + str(ch).lstrip("@"),
+                           adapter="tg_web", tags=("tg", "abuse", "radar"), limit=25,
+                           detail=False, extra={"opt_in": True}))
+    _RADAR.update(at=mtime, rows=rows)
+    return rows
+
+
+def _cn_sources():
+    """Китайские/азиатские источники из сгенерированного omni/sources_cn.py."""
+    try:
+        from .sources_cn import CN
+    except Exception:
+        return []
+    out = []
+    for d in CN:
+        try:
+            out.append(Source(**d))
+        except Exception:
+            continue
+    return out
+
 
 def all_sources():
     """Реестр + пользовательские источники из панели, без дублей по url."""
@@ -219,7 +271,7 @@ def all_sources():
              ("name", "category", "kind", "url", "adapter", "tags", "limit", "detail", "extra")}
         extra.append(Source(**d))
     seen, out = set(), []
-    for s in SOURCES + extra:
+    for s in SOURCES + _cn_sources() + extra + radar_sources():
         key = (s.url or "").rstrip("/").lower()
         if key and key in seen:
             continue
@@ -227,15 +279,27 @@ def all_sources():
         out.append(s)
     return out
 
-def enabled_sources():
-    """Только включённые: выключенные в панели источники не дёргаем."""
+def enabled_sources(include_opt_in=False):
+    """Только включённые: выключенные в панели источники не дёргаем.
+
+    Источники из abuse-radar помечены extra["opt_in"] — их 2000+, в общий обход
+    они не лезут; берутся только когда явно просят (--tags radar / --only / category abuse).
+    """
     from . import source_state
     fl = source_state.flags()
-    return [s for s in all_sources() if fl.get(s.name, {}).get("enabled", True)]
+    out = []
+    for s in all_sources():
+        if not fl.get(s.name, {}).get("enabled", True):
+            continue
+        if not include_opt_in and (s.extra or {}).get("opt_in"):
+            continue
+        out.append(s)
+    return out
 
 
 def pick(category=None, only=None, tags=None, respect_enabled=True):
-    out = enabled_sources() if respect_enabled else all_sources()
+    explicit = bool(only or tags or category)
+    out = enabled_sources(include_opt_in=explicit) if respect_enabled else all_sources()
     if category:
         cats = {c.strip() for c in category.split(",") if c.strip()}
         out = [s for s in out if s.category in cats]

omni/sources_cn.py

@@ -0,0 +1,17 @@
+"""Китайские и азиатские источники (проверены живыми запросами tools/wave_cn.py).
+
+Форумы реверса/хостинга, security-новости, tech-ленты. Все — публичные списки,
+без ключей и кук. Подключаются в реестр omni как категории forum/security/news.
+"""
+
+CN = [
+    {"name": "cn_52pojie", "category": "forum", "kind": "site", "url": "https://www.52pojie.cn/forum.php?mod=guide&view=new", "adapter": "html_listing", "tags": ("cn", "reverse", "abuse"), "limit": 20},
+    {"name": "cn_nodeseek", "category": "forum", "kind": "site", "url": "https://www.nodeseek.com/", "adapter": "html_listing", "tags": ("cn", "hosting"), "limit": 20},
+    {"name": "cn_anquanke", "category": "security", "kind": "site", "url": "https://www.anquanke.com/", "adapter": "html_listing", "tags": ("cn", "security"), "limit": 20},
+    {"name": "cn_4hou", "category": "security", "kind": "site", "url": "https://www.4hou.com/", "adapter": "html_listing", "tags": ("cn", "security"), "limit": 15},
+    {"name": "cn_kanxue", "category": "forum", "kind": "site", "url": "https://bbs.kanxue.com/", "adapter": "html_listing", "tags": ("cn", "reverse"), "limit": 20},
+    {"name": "cn_ithome", "category": "news", "kind": "site", "url": "https://www.ithome.com/rss/", "adapter": "rss", "tags": ("cn", "tech"), "limit": 20},
+    {"name": "cn_cnbeta", "category": "news", "kind": "site", "url": "https://www.cnbeta.com.tw/backend.php", "adapter": "rss", "tags": ("cn", "tech"), "limit": 20},
+    {"name": "cn_v2ex_rss", "category": "forum", "kind": "site", "url": "https://www.v2ex.com/index.xml", "adapter": "rss", "tags": ("cn", "community"), "limit": 20},
+    {"name": "cn_baidu_sec", "category": "security", "kind": "site", "url": "https://xz.aliyun.com/", "adapter": "html_listing", "tags": ("cn", "security"), "limit": 15},
+]

omni/vault.py

@@ -0,0 +1,128 @@
+"""Vault: ключи сервисов с ротацией и остыванием.
+
+Источник — data/vault.json (собирается tools/build_vault.py из дампов ключей).
+Значения живут только на диске и в памяти: в логах и панели печатается маска.
+
+Правила:
+  * env важнее файла (FIRECRAWL_API_KEY и т.п.) — ручное управление не ломается;
+  * ключ, получивший 401/402/403, уходит в остывание на час; 429 — на 10 минут;
+  * round-robin, чтобы не долбить один ключ и не выжигать квоту.
+
+Использование:
+  from . import vault
+  key = vault.get("firecrawl")           # или "" если ничего нет
+  vault.mark("firecrawl", key, "dead")   # 401/402/403
+  vault.mark("firecrawl", key, "rate")   # 429
+"""
+import json
+import os
+import random
+import time
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+VAULT = ROOT / "data" / "vault.json"
+STATE = ROOT / "data" / "vault_state.json"
+
+ENV = {
+    "firecrawl": "FIRECRAWL_API_KEY",
+    "tavily": "TAVILY_API_KEY",
+    "exa": "EXA_API_KEY",
+    "serper": "SERPER_API_KEY",
+    "jina": "JINA_API_KEY",
+    "apify": "APIFY_TOKEN",
+    "2captcha": "TWOCAPTCHA_KEY",
+    "capmonster": "CAPMONSTER_KEY",
+    "telegram": "TG_BOT_TOKEN",
+}
+
+COOLDOWN = {"dead": 3600, "rate": 600, "empty": 1800}
+
+_cache = {"data": None, "state": None, "at": 0.0}
+
+
+def _load():
+    now = time.time()
+    if _cache["data"] is not None and now - _cache["at"] < 30:
+        return _cache["data"], _cache["state"]
+    data, state = {}, {}
+    try:
+        data = json.loads(VAULT.read_text(encoding="utf-8"))
+    except Exception:
+        data = {}
+    try:
+        state = json.loads(STATE.read_text(encoding="utf-8"))
+    except Exception:
+        state = {}
+    _cache.update(data=data, state=state, at=now)
+    return data, state
+
+
+def _save_state(state):
+    try:
+        STATE.parent.mkdir(parents=True, exist_ok=True)
+        STATE.write_text(json.dumps(state, ensure_ascii=False), encoding="utf-8")
+    except Exception:
+        pass
+    _cache["state"] = state
+
+
+def services():
+    data, _ = _load()
+    return sorted(data)
+
+
+def keys(service):
+    data, _ = _load()
+    return [r["key"] for r in (data.get(service) or []) if r.get("key")]
+
+
+def mask(k):
+    k = k or ""
+    return (k[:4] + "…" + k[-2:]) if len(k) > 8 else "…"
+
+
+def get(service, offset=0):
+    """Живой ключ сервиса: env, иначе ротация по пулу с учётом остывания."""
+    env_name = ENV.get(service)
+    if env_name and os.environ.get(env_name):
+        return os.environ[env_name].strip()
+    pool = keys(service)
+    if not pool:
+        return ""
+    _, state = _load()
+    st = state.get(service, {})
+    now = time.time()
+    avail = []
+    for k in pool:
+        until = st.get(k, 0)
+        if until <= now:
+            avail.append(k)
+    if not avail:
+        # всё остывает — берём тот, у кого срок истекает раньше
+        k = min(pool, key=lambda x: st.get(x, 0))
+        return k
+    return avail[(offset + int(now)) % len(avail)] if len(avail) > 1 else avail[0]
+
+
+def mark(service, key, kind="dead"):
+    if not key:
+        return
+    _, state = _load()
+    st = state.setdefault(service, {})
+    st[key] = time.time() + COOLDOWN.get(kind, 600)
+    _save_state(state)
+
+
+def stats():
+    data, state = _load()
+    out = []
+    now = time.time()
+    for svc in sorted(data):
+        rows = data[svc]
+        st = state.get(svc, {})
+        cooling = sum(1 for r in rows if st.get(r["key"], 0) > now)
+        out.append({"service": svc, "keys": len(rows), "cooling": cooling,
+                    "ready": len(rows) - cooling,
+                    "env": bool(ENV.get(svc) and os.environ.get(ENV[svc]))})
+    return out