abuse: строгий free_count (prompt и completion = 0) — 90 панелей вместо 131; sync_panel_items держит тексты панелей в актуальном виде; push_top отправляет топ находок в бота

tools/push_top.py

@@ -10,6 +10,7 @@ import argparse
 import html
 import json
 import os
+import re
 import sqlite3
 import sys
 import urllib.parse
@@ -24,7 +25,50 @@ from omni import abuse  # noqa: E402
 TAG_RU = {"promo": "промо", "trial": "триал", "keys": "ключи", "leak": "слив",
           "resale": "перепродажа", "accs": "аккаунты", "credits": "кредиты",
           "invite": "инвайты", "ltd": "lifetime", "freeapi": "free-API",
-          "combo": "комбо", "tool": "тулза"}
+          "combo": "комбо", "tool": "тулза", "coupon": "купон", "nokyc": "без KYC",
+          "api": "API", "sub": "подписка", "unlimited": "безлимит", "referral": "реф"}
+
+
+NOISE = ("навигация по каналу", "навигация", "закреп", "правила канала", "прайс-лист",
+         "реклама в канале", "меню канала", "список каналов")
+EMOJI = re.compile("[🀀-🫿☀-➿⬀-⯿️⁠​‍

]")
+
+
+def clean(txt: str, n: int = 110) -> str:
+    txt = EMOJI.sub(" ", txt or "")
+    txt = re.sub(r"\s+", " ", txt).strip()
+    if len(txt) <= n:
+        return txt
+    cut = txt[:n]
+    if " " in cut[40:]:
+        cut = cut[:cut.rfind(" ")]
+    return cut.rstrip(" ,.;:-") + "…"
+
+
+def is_noise(title: str) -> bool:
+    low = (title or "").lower()
+    return any(k in low for k in NOISE)
+
+
+def panel_block(limit: int = 4) -> list[str]:
+    import json
+    from pathlib import Path
+    pj = ROOT / "data" / "panels.json"
+    if not pj.exists():
+        return []
+    rows = json.load(pj.open(encoding="utf-8"))["panels"]
+    rows = [x for x in rows if int(x.get("free_count") or 0) > 0]
+    rows.sort(key=lambda x: -int(x.get("free_count") or 0))
+    if not rows:
+        return []
+    out = ["<b>ПАНЕЛИ С БЕСПЛАТНЫМИ МОДЕЛЯМИ</b>"]
+    for x in rows[:limit]:
+        names = ", ".join(str(n) for n in (x.get("free") or [])[:3])
+        out.append("%s — <b>%d</b> free из %d%s" % (
+            html.escape(str(x.get("host"))), int(x["free_count"]), int(x.get("models") or 0),
+            (" · " + html.escape(names)) if names else ""))
+    out.append("")
+    return out
 
 
 def token() -> str:
@@ -48,7 +92,7 @@ def build(limit: int) -> str:
     picks, seen = [], set()
     for it in res["items"]:
         key = it.get("sig") or it["id"]
-        if key in seen:
+        if key in seen or is_noise(it.get("title") or ""):
             continue
         seen.add(key)
         picks.append(it)
@@ -62,7 +106,7 @@ def build(limit: int) -> str:
         raw = it.get("tags") or []
         raw = raw.split(",") if isinstance(raw, str) else list(raw)
         tags = ", ".join(TAG_RU.get(t, t) for t in raw if t)
-        title = html.escape((it.get("title") or it.get("text") or "")[:110].strip())
+        title = html.escape(clean(it.get("title") or it.get("text") or ""))
         tags_s = f" · <i>{html.escape(tags)}</i>" if tags else ""
         dup = f" · копий {it['dups']}" if it.get("dups") else ""
         link = it.get("url") or ""
@@ -72,6 +116,7 @@ def build(limit: int) -> str:
         if link:
             out.append(f'<a href="{html.escape(link)}">{html.escape(link[:78])}</a>')
         out.append("")
+    out.extend(panel_block())
     return "\n".join(out)[:3900]
 
 

tools/refresh_panels_free.py

@@ -64,9 +64,16 @@ def probe(entry: dict, timeout: int) -> dict:
             if not isinstance(m, dict):
                 continue
             ratio = m.get("model_ratio")
+            cratio = m.get("completion_ratio")
             quota = m.get("quota_type")
             try:
-                freeish = (ratio is not None and float(ratio) == 0.0) or (quota == 0)
+                # Строго как адаптер newapi_pricing: модель бесплатна только если
+                # нулевые И prompt, И completion. Мягкое «ratio==0 или quota==0»
+                # завышало счётчик в разы (760 моделей на панели с 1 бесплатной).
+                if ratio is None and cratio is None:
+                    freeish = (quota == 0)
+                else:
+                    freeish = float(ratio or 0) == 0.0 and float(cratio or 0) == 0.0
             except Exception:
                 freeish = False
             if freeish:

tools/shots.py

@@ -26,8 +26,11 @@ SECRET = re.compile(r"\b(sk-|pk-|ek_live_|ghp_|xoxb-)[A-Za-z0-9_\-]{12,}")
 
 def main():
     bad = []
+    only = set(sys.argv[1].split(",")) if len(sys.argv) > 1 else None
     with Camoufox(headless=True, geoip=False, humanize=True) as b:
         for route, name in SHOTS:
+            if only and name not in only:
+                continue
             p = b.new_page(viewport={"width": 1500, "height": 1350})
             p.goto(BASE + route, wait_until="domcontentloaded", timeout=90000)
             n = 0

tools/sync_panel_items.py

@@ -0,0 +1,63 @@
+"""Синхронизировать тексты pn_*-записей с data/panels.json.
+
+Панель переотдаёт тот же url, поэтому store писал её один раз и число
+бесплатных моделей в выдаче застывало. Скрипт переписывает title/text/score
+по актуальному free_count (строгое правило: prompt и completion = 0).
+
+Запуск: python tools/sync_panel_items.py [--dry]
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import sqlite3
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def plural(n: int) -> str:
+    if n % 10 == 1 and n % 100 != 11:
+        return "модель"
+    if n % 10 in (2, 3, 4) and n % 100 not in (12, 13, 14):
+        return "модели"
+    return "моделей"
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--dry", action="store_true")
+    a = ap.parse_args()
+    rows = json.load((ROOT / "data" / "panels.json").open(encoding="utf-8"))["panels"]
+    by = {}
+    for x in rows:
+        host = str(x.get("host") or "").strip()
+        if not host:
+            continue
+        by["pn_" + host.replace(".", "_").replace("-", "_")] = (
+            host, int(x.get("free_count") or 0), int(x.get("models") or 0), x.get("free") or [])
+    c = sqlite3.connect(str(ROOT / "data" / "omni.sqlite3"))
+    c.execute("PRAGMA busy_timeout=60000")
+    n = 0
+    for src, (host, free, total, names) in by.items():
+        item = c.execute("SELECT id FROM items WHERE source=? LIMIT 1", (src,)).fetchone()
+        if not item:
+            continue
+        title = "%s: %d бесплатных %s из %d" % (host, free, plural(free), total)
+        text = title + ("\n" + ", ".join(str(x) for x in names[:12]) if names else "")
+        if a.dry:
+            n += 1
+            continue
+        c.execute("UPDATE items SET title=?, text=? WHERE id=?", (title, text, item[0]))
+        c.execute("UPDATE items_fts SET title=?, text=? WHERE rowid=?", (title, text, item[0]))
+        c.execute("UPDATE abuse_idx SET score=? WHERE id=?", (70 + min(free // 3, 110), item[0]))
+        n += 1
+    if not a.dry:
+        c.commit()
+    print("синхронизировано панелей:", n, "| с free:", sum(1 for v in by.values() if v[1] > 0))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())