ui: локальный Tabler dashboard + 150 проверенных TG-источников и probe 595 каналов

tools/tg_pool_probe.py

@@ -0,0 +1,122 @@
+"""Probe public Telegram previews without writing posts to the store.
+
+Run from the repository root: python tools/tg_pool_probe.py
+Each of up to 12 coordinator threads runs one isolated adapter subprocess.
+The 20-second subprocess deadline includes imports, network fallbacks and parsing;
+Future.result(timeout=...) alone would leave hung network threads running.
+Results preserve pool order and username spelling; ok means items > 0.
+"""
+from __future__ import annotations
+
+import argparse
+from concurrent.futures import ThreadPoolExecutor, as_completed
+import json
+import os
+from pathlib import Path
+import re
+import subprocess
+import sys
+from types import SimpleNamespace
+
+ROOT = Path(__file__).resolve().parents[1]
+WORKERS = 12
+TIMEOUT = 20
+LIMIT = 20
+
+
+def load_channels(path: Path) -> list[str]:
+    channels = json.loads(path.read_text(encoding="utf-8"))["channels"]
+    if not isinstance(channels, list):
+        raise ValueError("channels must be a list")
+    names = []
+    for entry in channels:
+        username = entry.get("username") if isinstance(entry, dict) else entry
+        if not isinstance(username, str) or not re.fullmatch(r"[A-Za-z0-9_]+", username):
+            raise ValueError(f"Invalid username in channels: {username!r}")
+        names.append(username)
+    return names
+
+
+def adapter_count(username: str) -> int:
+    sys.path.insert(0, str(ROOT))
+    from omni import feeds
+
+    src = SimpleNamespace(name=f"ab_tg_auto_{username}", category="abuse",
+                          kind="site", url=f"https://t.me/s/{username}",
+                          adapter="tg_web", tags=("abuse", "abuse", "tg"),
+                          limit=LIMIT, detail=False, extra={})
+    # Call the actual adapter and validation directly so unexpected exceptions
+    # reach the subprocess exit status instead of fetch() swallowing them.
+    return len(feeds._valid_items(feeds.ADAPTERS["tg_web"](src, LIMIT) or []))
+
+
+def probe(username: str, timeout: float = TIMEOUT) -> dict:
+    count = 0
+    try:
+        env = dict(os.environ, PYTHONPATH="", PYTHONIOENCODING="utf-8",
+                   PYTHONUTF8="1", PYTHONDONTWRITEBYTECODE="1")
+        result = subprocess.run([sys.executable, str(Path(__file__).resolve()),
+                                 "--probe", username],
+                                cwd=str(ROOT), env=env, capture_output=True,
+                                encoding="utf-8", timeout=timeout, check=False)
+        if result.returncode == 0:
+            value = json.loads(result.stdout)
+            if type(value) is int and 0 <= value <= LIMIT:
+                count = value
+        else:
+            print(f"probe error @{username}: exit={result.returncode} "
+                  f"{result.stderr[-300:].strip()}", file=sys.stderr, flush=True)
+    except subprocess.TimeoutExpired:
+        print(f"probe timeout @{username}: {timeout:g}s", file=sys.stderr, flush=True)
+    except (OSError, ValueError) as exc:
+        print(f"probe error @{username}: {type(exc).__name__}: {exc}",
+              file=sys.stderr, flush=True)
+    return {"username": username, "items": count, "ok": count > 0}
+
+
+def save_results(path: Path, rows: list[dict | None]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps([r for r in rows if r is not None],
+                               ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--input", type=Path, default=ROOT / "data/radar_sources.json")
+    parser.add_argument("--output", type=Path, default=ROOT / "data/tg_pool_probe.json")
+    parser.add_argument("--workers", type=int, choices=range(1, WORKERS + 1), default=WORKERS)
+    parser.add_argument("--timeout", type=float, default=TIMEOUT)
+    parser.add_argument("--probe", help=argparse.SUPPRESS)
+    args = parser.parse_args()
+    if not 0 < args.timeout <= TIMEOUT:
+        parser.error("timeout must be greater than zero and at most 20 seconds")
+    if args.probe is not None:
+        if not re.fullmatch(r"[A-Za-z0-9_]+", args.probe):
+            parser.error("invalid username")
+        print(json.dumps(adapter_count(args.probe)))
+        return 0
+    if args.input.resolve() == args.output.resolve():
+        parser.error("output must not overwrite the input pool")
+    names = load_channels(args.input)
+    rows = [None] * len(names)
+    print(f"channels={len(names)} workers={args.workers} timeout={args.timeout:g}s limit={LIMIT}", flush=True)
+    with ThreadPoolExecutor(max_workers=args.workers) as executor:
+        pending = {executor.submit(probe, name, args.timeout): i
+                   for i, name in enumerate(names)}
+        for done, future in enumerate(as_completed(pending), 1):
+            rows[pending[future]] = future.result()
+            if done % 25 == 0 or done == len(names):
+                save_results(args.output, rows)
+                positive = sum(r is not None and r["ok"] for r in rows)
+                print(f"checked={done}/{len(names)} positive={positive}", flush=True)
+    save_results(args.output, rows)
+    print(json.dumps({"checked": len(rows),
+                      "positive": sum(r["ok"] for r in rows),
+                      "zero": sum(r["items"] == 0 for r in rows),
+                      "items_ge_10": sum(r["items"] >= 10 for r in rows),
+                      "output": str(args.output)}, ensure_ascii=False), flush=True)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())