"""Дайджест абузов для Telegram: карточки с группировкой, без сырых заголовков. Структура поста: OMNI · дайджест абузов · 27.10.2026 новых за сутки 412 · всего 6 746 находок (5 182 уникальных) КЛЮЧИ И ТОКЕНЫ 1. GLM-5.3 без цензуры 1 млн токенов · 1 мес · без карты platform.experientiallabs.ai · пост (HarnessLLM) ПАНЕЛИ С БЕСПЛАТНЫМИ МОДЕЛЯМИ newapi.prorisehub.com — 229 free из 454 (qwen-flash-c, dall-e-3) Запуск: python tools/post_digest.py --dry # текст в консоль python tools/post_digest.py --limit 10 # отправить в бота python tools/post_digest.py --chat --limit 6 В каналы не постит: только личка получателю (--chat, дефолт OMNI_CHAT). """ from __future__ import annotations import argparse import datetime as dt import html import json import os import re import sqlite3 import sys import time import urllib.parse import urllib.request from pathlib import Path ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) from omni import abuse # noqa: E402 NL = chr(10) TAG_RU = {"promo": "промо", "trial": "триал", "keys": "ключи", "leak": "слив", "resale": "перепродажа", "accs": "аккаунты", "credits": "кредиты", "invite": "инвайты", "ltd": "lifetime", "freeapi": "free-API", "combo": "комбо", "tool": "тулза", "coupon": "купон", "nokyc": "без KYC", "api": "API", "sub": "подписка", "unlimited": "безлимит", "referral": "реф"} KEYS = {"keys", "credits", "promo", "freeapi", "api", "unlimited", "leak", "combo"} ACCS = {"accs", "trial", "invite", "coupon", "resale", "nokyc", "sub"} SKIP_TAGS = {"news", "обзор", "review"} SKIP_WORDS = ("навигация", "закреп", "правила канала", "прайс-лист", "реклама в канале", "меню канала", "список каналов", "подписывайтесь", "дайджест", "за сутки", "подборка", "итоги", "абуза недели", "жирных абуз", "weekly", "теряют", "выкатил", "представил", "анонсировал", "обзор рынка", "исследован", "скоростью", "тренд") EMOJI = re.compile("[\U0001F000-\U0001FAFF\u2600-\u27BF\u2B00-\u2BFF\uFE0F\u2060\u200B\u200D\u2022\u25AA\u25CF\u25AB\u25A0\u25A1\u25E6\u2B1B]") MONEY = re.compile(r"(?:\$\s?\d[\d.,]*|\d[\d.,]*\s?(?:млрд|млн|тыс|k|m|b)\b(?:\s?токен(?:ов|а)?)?)", re.I) PERIOD = re.compile(r"(на\s?\d+\s?(?:дн(?:ей|я|и)?|мес(?:яц(?:а|ев)?)?|год(?:а)?)" r"|\d+\s?(?:дн(?:ей|я)|мес(?:яцев|яца)?)\b" r"|навсегда|lifetime|ежемесячно|пожизненно|бессрочно)", re.I) COND = [(r"без карты|карта не нужна|без привязки карты", "без карты"), (r"без kyc|no kyc|без верификации", "без KYC"), (r"только для новых|для новых пользователей|new users only", "только новым"), (r"нужен vpn|требуется vpn|через vpn", "нужен VPN"), (r"без капчи", "без капчи"), (r"без email|без почты", "без почты")] DOMAIN = re.compile(r"\b((?:[a-z0-9][a-z0-9-]{1,30}\.){1,3}(?:ai|com|io|dev|app|net|org|me|pro|cc|sh|xyz|top|site|so|ru|cloud|tech))\b", re.I) TME = re.compile(r"^(t\.me|telegram\.me|telegra\.ph|youtu\.be|youtube\.com)$", re.I) LEAD = re.compile(r"^(совет|гайд|инструкция|лайфхак|подборка|новость|обзор|релиз)\s*[:.\-—]\s*", re.I) NUM = re.compile(r"^\s*\d+[.)]\s*") UNIT = {"млрд": 1e9, "b": 1e9, "млн": 1e6, "m": 1e6, "тыс": 1e3, "k": 1e3} HOMO = str.maketrans({"C": "С", "c": "с", "A": "А", "a": "а", "B": "В", "E": "Е", "H": "Н", "K": "К", "M": "М", "O": "О", "P": "Р", "T": "Т", "X": "Х", "y": "у", "o": "о", "p": "р", "x": "х"}) def dehomoglyph(word: str) -> str: """Латинские гомоглифы внутри кириллических слов: Cовет → Совет.""" core = word.strip('.,:;!?()[]«»"\'') if not core or not re.search(r"[А-Яа-яЁё]", core): return word if re.search(r"[A-Za-z]", core) and len(re.findall(r"[А-Яа-яЁё]", core)) >= 3: return word.replace(core, core.translate(HOMO), 1) return word def norm_text(txt: str) -> str: return " ".join(dehomoglyph(w) for w in txt.split()) MARK = re.compile(r"[*_`~]{1,3}") def flat(txt: str) -> str: txt = EMOJI.sub(" ", txt or "") txt = MARK.sub(" ", txt) return re.sub(r"\s+", " ", txt).strip() ABBR = {"API", "LLM", "GPT", "AI", "GLM", "GPU", "CPU", "KYC", "SSO", "VPN", "IDE", "CLI", "SDK", "JSON", "HTML", "TTS", "STT", "TLS", "SSL", "URL", "JWT", "REST", "CSV", "PDF", "SQL", "AWS", "GCP", "VPS", "AGI", "RAG", "MCP", "SMS", "OTP", "PRO", "MAX", "ULTRA", "TURBO", "FREE", "API-KEY"} def smart_case(txt: str) -> str: """Крик из каналов → нормальный регистр; аббревиатуры и тарифы не трогаем.""" words = txt.split() shout = [w for w in words if len(re.sub(r"[^\wА-Яа-яЁё]", "", w)) >= 6 and w.isupper()] if len(shout) >= 2 or (shout and len(shout) / max(len(words), 1) > 0.25): out = [] for w in words: core = re.sub(r"[^\wА-Яа-яЁё-]", "", w) if w.isupper() and len(core) >= 2 and core not in ABBR: out.append(w.lower()) elif w.isupper() and len(core) == 1 and re.search(r"[А-ЯЁ]", core): out.append(w.lower()) else: out.append(w) txt = " ".join(out) letters = [c for c in txt if c.isalpha()] if letters: upper = sum(1 for c in letters if c.isupper()) / len(letters) if upper > 0.5 and len(txt) > 12: txt = " ".join(w if re.sub(r"[^\wА-Яа-яЁё-]", "", w) in ABBR else w.lower() for w in txt.split()) return txt[:1].upper() + txt[1:] if txt else txt TAIL_STOP = ("на", "в", "с", "со", "к", "ко", "по", "и", "или", "для", "от", "до", "за", "из", "о", "об", "при", "под", "над", "что", "как") def tail_clean(txt: str) -> str: """Убрать висячие предлоги/союзы на конце обрезанной фразы.""" words = txt.rstrip(" ,.;:-").split() while words and words[-1].lower() in TAIL_STOP: words.pop() return " ".join(words).rstrip(" ,.;:-") def headline(raw: str, n: int = 72) -> str: """Заголовок = самая информативная фраза: цифры, объём, сервис, без CAPS.""" txt = smart_case(NUM.sub("", LEAD.sub("", norm_text(flat(raw))))) parts = [x for x in re.split(r"(?:\s[—–]\s|:|(?<=[.!?])\s|(?<=[A-Za-zА-Яа-яЁё0-9])\s(?=[А-ЯЁA-Z][а-яёa-z]))", txt) if x.strip()] best, best_sc = "", -99 for x in parts: x = smart_case(x.strip(" ,.;:-")) if len(x) < 12: continue if len(x) > n: cut = x[:n] cut = cut[:cut.rfind(" ")] if " " in cut[40:] else cut x = tail_clean(cut) + "…" sc = 0 if re.search(r"\d", x): sc += 3 if re.search(r"токен|кредит|бесплат|free|млрд|млн|дн(ей|я)|мес", x, re.I): sc += 2 if re.search(r"[A-Za-z]{3,}", x): sc += 1 if 18 <= len(x) <= 62: sc += 1 if re.search(r"[А-ЯЁ]{5,}", x): sc -= 3 if x.endswith("…"): sc -= 1 if sc > best_sc: best, best_sc = x, sc return tail_clean(best) or txt[:n] def _value(hit: str) -> float: m = re.search(r"([\d.,]+)\s?(млрд|млн|тыс|k|m|b)?", hit, re.I) if not m: return 0.0 try: num = float(m.group(1).replace(",", ".")) except ValueError: return 0.0 return num * UNIT.get((m.group(2) or "").lower(), 1) def gain(raw: str) -> str: """Объём выгоды: токены/кредиты важнее цены; цена — только как запасной вариант.""" txt = flat(raw) hits = [m.group(0).strip(" ,.;:-") for m in MONEY.finditer(txt)] hits = [h for h in hits if re.search(r"\d", h)] if not hits: return "" vol = [h for h in hits if not h.strip().startswith("$")] money = [h for h in hits if h.strip().startswith("$") and _value(h) >= 50] pool = vol or money or [] if not pool: return "" best = max(pool, key=_value) return re.sub(r"\b(млрд|млн|тыс)\b", lambda m: m.group(1).lower(), best, flags=re.I) def period(raw: str) -> str: m = PERIOD.search(flat(raw)) return m.group(0).lower() if m else "" def conds(raw: str) -> str: txt = flat(raw).lower() found = [label for pat, label in COND if re.search(pat, txt)] return ", ".join(dict.fromkeys(found)) def site(raw: str) -> str: for m in DOMAIN.finditer(flat(raw)): host = re.sub(r"^www\.", "", m.group(1).lower()) if not TME.match(host): return host return "" def tags_of(it: dict) -> list[str]: raw = it.get("tags") or [] if isinstance(raw, str): raw = [t.strip() for t in raw.split(",")] return [str(t).strip() for t in raw if t and str(t).strip()] def kind(it: dict) -> str: tags = set(tags_of(it)) if tags & KEYS: return "keys" if tags & ACCS: return "accs" return "tools" CAT_TITLE = {"keys": "КЛЮЧИ И ТОКЕНЫ", "accs": "АККАУНТЫ И ТРИАЛЫ", "tools": "ТУЛЗЫ И ПРОЧЕЕ"} def card(i: int, it: dict) -> list[str]: body = "%s %s" % (it.get("title") or "", it.get("text") or "") title = headline(it.get("title") or it.get("text") or "") low_title = title.lower() bits = [b for b in (gain(body), period(body), conds(body)) if b and b.lower() not in low_title] meta = " · ".join(bits) if bits else "детали в посте" host = site(body) src = it.get("source") or "источник" for p in ("rad_", "tgp_", "pn_"): src = src.replace(p, "") url = it.get("url") or "" tail = (' · пост' % html.escape(url)) if url else "" if host: line = '%s%s (%s)' % ( html.escape(host), html.escape(host), tail, html.escape(src[:26])) else: line = "%s%s" % (html.escape(src[:26]), tail) return ["%d. %s" % (i, html.escape(title)), "%s" % html.escape(meta), line, ""] def fresh_24h() -> int: c = sqlite3.connect(str(ROOT / "data" / "omni.sqlite3")) try: edge = int(time.time()) - 86400 return c.execute("SELECT COUNT(*) FROM abuse_idx a JOIN items i ON i.id = a.id " "WHERE i.published >= ?", (edge,)).fetchone()[0] except sqlite3.Error: return 0 def panels_block(limit: int = 5) -> list[str]: pj = ROOT / "data" / "panels.json" if not pj.exists(): return [] rows = [x for x in json.load(pj.open(encoding="utf-8"))["panels"] 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 = ["ПАНЕЛИ С БЕСПЛАТНЫМИ МОДЕЛЯМИ"] for x in rows[:limit]: names = [str(n) for n in (x.get("free") or [])[:2]] tail = (" (%s)" % html.escape(", ".join(names))) if names else "" out.append("%s — %d free из %d%s" % ( html.escape(str(x.get("host"))), int(x["free_count"]), int(x.get("models") or 0), tail)) out.append("") return out def counts(): c = sqlite3.connect(str(ROOT / "data" / "omni.sqlite3")) items = c.execute("SELECT COUNT(*) FROM items").fetchone()[0] ab = c.execute("SELECT COUNT(*) FROM abuse_idx").fetchone()[0] uniq = c.execute("SELECT COUNT(DISTINCT sig) FROM abuse_idx").fetchone()[0] return items, ab, uniq def pick(limit: int) -> list[dict]: res = abuse.scan(limit=max(limit * 8, 120)) out, seen_sig, seen_host = [], set(), set() for it in res["items"]: tags = {t.lower() for t in tags_of(it)} body = norm_text(flat("%s %s" % (it.get("title") or "", it.get("text") or ""))) if tags & SKIP_TAGS or any(w in body[:300].lower() for w in SKIP_WORDS): continue if re.match(r"^\s*(совет|гайд|лайфхак|инструкция|подборка)\b", body, re.I): continue src_name = str(it.get("source") or "") if src_name.startswith("pn_") or "/api/pricing" in str(it.get("url") or ""): continue sig = it.get("sig") or it["id"] host = site(body) if sig in seen_sig or (host and host in seen_host): continue if not gain(body) and not host: continue seen_sig.add(sig) if host: seen_host.add(host) out.append(it) if len(out) >= limit: break return out def build(limit: int) -> str: picks = pick(limit) items, ab, uniq = counts() today = dt.date.today().strftime("%d.%m.%Y") out = ["OMNI · дайджест абузов · %s" % today, "новых за сутки {:,} · всего {:,} находок ({:,} уникальных)".format( fresh_24h(), ab, uniq).replace(",", " "), ""] groups: dict[str, list[dict]] = {"keys": [], "accs": [], "tools": []} for it in picks: groups[kind(it)].append(it) n = 0 for key in ("keys", "accs", "tools"): if not groups[key]: continue out.append("%s" % CAT_TITLE[key]) for it in groups[key]: n += 1 out.extend(card(n, it)) out.extend(panels_block()) return NL.join(out).strip()[:4000] def token() -> str: p = ROOT / ".bot_token" t = p.read_text(encoding="utf-8").strip() if p.exists() else os.environ.get("OMNI_BOT_TOKEN", "") if not t: raise SystemExit("нет .bot_token") return t def send(text: str, chat: str) -> dict: data = urllib.parse.urlencode({"chat_id": chat, "text": text, "parse_mode": "HTML", "disable_web_page_preview": "true"}).encode() url = "https://api.telegram.org/bot%s/sendMessage" % token() with urllib.request.urlopen(urllib.request.Request(url, data=data), timeout=40) as r: return json.load(r) def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--limit", type=int, default=8) ap.add_argument("--chat", default=os.environ.get("OMNI_CHAT", "7448683285")) ap.add_argument("--dry", action="store_true") a = ap.parse_args() text = build(a.limit) if a.dry: print(text) return 0 r = send(text, a.chat) res = r.get("result") or {} print("ok", r.get("ok"), "| message_id", res.get("message_id"), "| chat", (res.get("chat") or {}).get("id"), "| длина", len(text)) return 0 if r.get("ok") else 1 if __name__ == "__main__": sys.exit(main())