From 0e3090edd26f0450f85b3d8f0e3bde68f8342580 Mon Sep 17 00:00:00 2001 From: Vlad Gothbreach Date: Tue, 27 Oct 2026 15:22:28 +0300 Subject: [PATCH] =?UTF-8?q?post=5Fdigest:=20=D0=BA=D0=B0=D1=80=D1=82=D0=BE?= =?UTF-8?q?=D1=87=D0=BA=D0=B8=20=D0=B2=D0=BC=D0=B5=D1=81=D1=82=D0=BE=20?= =?UTF-8?q?=D1=81=D1=82=D1=80=D0=BE=D0=BA=D0=B8-=D1=81=D0=B2=D0=B0=D0=BB?= =?UTF-8?q?=D0=BA=D0=B8=20=E2=80=94=20=D0=B7=D0=B0=D0=B3=D0=BE=D0=BB=D0=BE?= =?UTF-8?q?=D0=B2=D0=BE=D0=BA,=20=D0=BE=D0=B1=D1=8A=D1=91=D0=BC/=D1=81?= =?UTF-8?q?=D0=B5=D1=80=D0=B2=D0=B8=D1=81/=D1=82=D0=B5=D0=B3=D0=B8,=20?= =?UTF-8?q?=D1=81=D0=BA=D1=80=D1=8B=D1=82=D0=B0=D1=8F=20=D1=81=D1=81=D1=8B?= =?UTF-8?q?=D0=BB=D0=BA=D0=B0;=20=D0=BF=D0=B0=D0=BD=D0=B5=D0=BB=D0=B8=20?= =?UTF-8?q?=D0=B1=D0=BB=D0=BE=D0=BA=D0=BE=D0=BC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tools/post_digest.py | 213 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 213 insertions(+) create mode 100644 tools/post_digest.py diff --git a/tools/post_digest.py b/tools/post_digest.py new file mode 100644 index 0000000..a83e2b6 --- /dev/null +++ b/tools/post_digest.py @@ -0,0 +1,213 @@ +"""Пост-дайджест абузов для Telegram: карточки, а не свалка строк. + +Формат карточки: + 1. Суть находки + что дают: объём, условия · сервис + источник: канал → (скрытая ссылка) + +Запуск: + python tools/post_digest.py --dry # посмотреть текст + python tools/post_digest.py --limit 8 # отправить в бота + python tools/post_digest.py --chat +В каналы не постит: только личка получателю (--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 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": "реф"} + +# Мусор и не-абузы: новости, навигация, правила канала. +SKIP_TAGS = {"news", "обзор", "review"} +SKIP_WORDS = ("навигация", "закреп", "правила канала", "прайс-лист", "реклама в канале", + "меню канала", "список каналов", "подписывайтесь", "реклама") + +EMOJI = re.compile("[\U0001F000-\U0001FAFF\u2600-\u27BF\u2B00-\u2BFF\uFE0F\u2060\u200B\u200D\u2022\u25AA\u25CF]") +MONEY = re.compile(r"(?:\$\s?\d[\d.,]*|\d[\d.,]*\s?(?:млрд|млн|тыс|k|m|b)\b(?:\s?токен(?:ов|а)?)?)", re.I) +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))\b", re.I) +TME = re.compile(r"^(t\.me|telegram\.me|telegra\.ph)$", re.I) +EMPTY = {"", "-", "—", "нет"} + + +def flat(txt: str) -> str: + txt = EMOJI.sub(" ", txt or "") + return re.sub(r"\s+", " ", txt).strip() + + +def headline(raw: str, n: int = 78) -> str: + """Короткий заголовок: до двоеточия/тире, иначе по границе слова.""" + txt = flat(raw) + for sep in (":", " — ", " – ", ". ", "! ", "? "): + head = txt.split(sep)[0].strip() + if 14 <= len(head) <= n: + return head + if len(txt) <= n: + return txt + cut = txt[:n] + cut = cut[:cut.rfind(" ")] if " " in cut[40:] else cut + return cut.rstrip(" ,.;:-") + "…" + + +def gain(raw: str, limit: int = 2) -> str: + """Что дают: объёмы и суммы из текста.""" + txt = flat(raw) + hits: list[str] = [] + for m in MONEY.finditer(txt): + s = m.group(0).strip(" ,.;:-") + if len(s) < 3 or s.lower() in (x.lower() for x in hits): + continue + if not re.search(r"\d", s): + continue + hits.append(s) + if len(hits) >= limit: + break + return " / ".join(hits) + + +def site(raw: str) -> str: + txt = flat(raw) + for m in DOMAIN.finditer(txt): + host = m.group(1).lower() + if TME.match(host): + continue + 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 [TAG_RU.get(t, t) for t in raw if t and str(t).strip()] + + +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 "") + g = gain(body) + s = site(body) + kinds = "/".join(tags_of(it)[:3]) + meta = " · ".join(b for b in (g, s, kinds) if b) or "подробности в посте" + src = (it.get("source") or "источник").replace("rad_", "").replace("tgp_", "") + url = it.get("url") or "" + out = ["%d. %s" % (i, html.escape(title)), + "%s" % html.escape(meta)] + if url: + out.append('%s' % (html.escape(url), html.escape(src[:34]))) + out.append("") + return out + + +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 = (" · " + 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 * 6, 90)) + out, seen = [], set() + for it in res["items"]: + tags = {t.strip() for t in str(it.get("tags") or "").split(",") if t.strip()} + low = flat("%s %s" % (it.get("title") or "", it.get("text") or ""))[:300].lower() + if tags & SKIP_TAGS or any(w in low for w in SKIP_WORDS): + continue + key = it.get("sig") or it["id"] + if key in seen: + continue + seen.add(key) + 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") + head = "OMNI · дайджест абузов · %s" % today + stat = "находок {:,} · уникальных {:,} · записей в базе {:,}".format(ab, uniq, items).replace(",", " ") + out = [head, stat, ""] + for i, it in enumerate(picks, 1): + out.extend(card(i, it)) + out.extend(panels_block()) + return NL.join(out).strip()[:3900] + + +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()) \ No newline at end of file