388 lines
16 KiB
Python
388 lines
16 KiB
Python
"""Дайджест абузов для 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 <id> --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 = (' · <a href="%s">пост</a>' % html.escape(url)) if url else ""
|
||
if host:
|
||
line = '<a href="https://%s">%s</a>%s <i>(%s)</i>' % (
|
||
html.escape(host), html.escape(host), tail, html.escape(src[:26]))
|
||
else:
|
||
line = "%s%s" % (html.escape(src[:26]), tail)
|
||
return ["<b>%d. %s</b>" % (i, html.escape(title)),
|
||
"<i>%s</i>" % 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 = ["<b>ПАНЕЛИ С БЕСПЛАТНЫМИ МОДЕЛЯМИ</b>"]
|
||
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 — <b>%d</b> 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 = ["<b>OMNI · дайджест абузов</b> · %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("<b>%s</b>" % 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()) |