post_digest: заголовок = самая информативная фраза, чистка CAPS/гомоглифов/markdown, кликабельный сервис, тесты формата
Some checks failed
ci / test (3.10) (push) Has been cancelled
ci / test (3.11) (push) Has been cancelled
ci / test (3.12) (push) Has been cancelled

This commit is contained in:
Vlad Gothbreach
2026-10-27 19:16:46 +03:00
parent 2612aeb206
commit f263d3ffff
2 changed files with 330 additions and 72 deletions

83
tests/test_post_digest.py Normal file
View File

@@ -0,0 +1,83 @@
"""Формат дайджеста: чистка заголовков, объёмов, условий и группировка."""
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "tools"))
import post_digest as d # noqa: E402
def test_homoglyph_fix():
assert d.dehomoglyph("Cовет") == "Совет"
assert d.dehomoglyph("GitHub") == "GitHub"
def test_shout_to_sentence():
out = d.smart_case("DEEPSEEK ПРОДОЛЖАЕТ РАЗДАВАТЬ ТОКЕНЫ")
assert out.startswith("Deepseek")
assert not any(w.isupper() and len(w) > 4 for w in out.split())
def test_abbr_kept():
out = d.smart_case("FREE ДОСТУП К ТОПОВЫМ LLM ЧЕРЕЗ GITHUB-OAUTH")
assert "LLM" in out
def test_headline_drops_lead_and_caps():
out = d.headline("Cовет: не обязательно тратить большие деньги на инструменты ИИ")
assert not out.lower().startswith("cовет")
assert len(out) <= 80
def test_headline_no_dangling_preposition():
out = d.headline("RU 200 000 000 бесплатных токенов и тариф PRO на", n=40)
assert not out.rstrip("…").endswith(" на")
def test_gain_prefers_volume_over_price():
assert d.gain("подписка $10, дают 5 млн токенов") == "5 млн токенов"
def test_gain_skips_small_money():
assert d.gain("подписка за $10 на неделю") == ""
def test_period_and_conds():
txt = "90 дней PRO бесплатно, без карты и без KYC"
assert d.period(txt) == "90 дней"
assert d.conds(txt) == "без карты, без KYC"
def test_kind_grouping():
assert d.kind({"tags": "keys,trial"}) == "keys"
assert d.kind({"tags": ["accs"]}) == "accs"
assert d.kind({"tags": ""}) == "tools"
def test_card_render_offline():
it = {
"title": "**1 млн бесплатных токенов** на QWEN3.8-OMNI-FLASH",
"text": "дают 1 млн токенов на 90 дней, без карты и без KYC, ключ по ссылке",
"url": "https://example.com/post/1",
"source": "demo_channel",
"tags": "keys,trial",
}
html = chr(10).join(d.card(1, it))
assert "<b>1." in html
assert "1 млн" in html and "токенов" in html
assert "на 90 дней" in html
assert "без карты" in html
assert "<a href=" in html and "пост" in html
assert "**" not in html
def test_build_renders():
try:
text = d.build(3)
except Exception as exc: # sandbox без наполненной базы
import pytest
pytest.skip(str(exc))
assert "OMNI" in text and "КЛЮЧИ" in text.upper()
assert len(text) <= 4000

View File

@@ -1,14 +1,21 @@
"""Пост-дайджест абузов для Telegram: карточки, а не свалка строк. """Дайджест абузов для Telegram: карточки с группировкой, без сырых заголовков.
Формат карточки: Структура поста:
<b>1. Суть находки</b> 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 --dry # текст в консоль
python tools/post_digest.py --limit 8 # отправить в бота python tools/post_digest.py --limit 10 # отправить в бота
python tools/post_digest.py --chat <id> python tools/post_digest.py --chat <id> --limit 6
В каналы не постит: только личка получателю (--chat, дефолт OMNI_CHAT). В каналы не постит: только личка получателю (--chat, дефолт OMNI_CHAT).
""" """
from __future__ import annotations from __future__ import annotations
@@ -21,6 +28,7 @@ import os
import re import re
import sqlite3 import sqlite3
import sys import sys
import time
import urllib.parse import urllib.parse
import urllib.request import urllib.request
from pathlib import Path from pathlib import Path
@@ -38,59 +46,177 @@ TAG_RU = {"promo": "промо", "trial": "триал", "keys": "ключи", "l
"combo": "комбо", "tool": "тулза", "coupon": "купон", "nokyc": "без KYC", "combo": "комбо", "tool": "тулза", "coupon": "купон", "nokyc": "без KYC",
"api": "API", "sub": "подписка", "unlimited": "безлимит", "referral": "реф"} "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_TAGS = {"news", "обзор", "review"}
SKIP_WORDS = ("навигация", "закреп", "правила канала", "прайс-лист", "реклама в канале", SKIP_WORDS = ("навигация", "закреп", "правила канала", "прайс-лист", "реклама в канале",
"меню канала", "список каналов", "подписывайтесь", "реклама") "меню канала", "список каналов", "подписывайтесь", "дайджест", "за сутки",
"подборка", "итоги", "абуза недели", "жирных абуз", "weekly",
"теряют", "выкатил", "представил", "анонсировал", "обзор рынка",
"исследован", "скоростью", "тренд")
EMOJI = re.compile("[\U0001F000-\U0001FAFF\u2600-\u27BF\u2B00-\u2BFF\uFE0F\u2060\u200B\u200D\u2022\u25AA\u25CF]") 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) 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) PERIOD = re.compile(r"(на\s?\d+\s?(?:дн(?:ей|я|и)?|мес(?:яц(?:а|ев)?)?|год(?:а)?)"
TME = re.compile(r"^(t\.me|telegram\.me|telegra\.ph)$", re.I) r"|\d+\s?(?:дн(?:ей|я)|мес(?:яцев|яца)?)\b"
EMPTY = {"", "-", "—", "нет"} 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: def flat(txt: str) -> str:
txt = EMOJI.sub(" ", txt or "") txt = EMOJI.sub(" ", txt or "")
txt = MARK.sub(" ", txt)
return re.sub(r"\s+", " ", txt).strip() return re.sub(r"\s+", " ", txt).strip()
def headline(raw: str, n: int = 78) -> str: ABBR = {"API", "LLM", "GPT", "AI", "GLM", "GPU", "CPU", "KYC", "SSO", "VPN", "IDE",
"""Короткий заголовок: до двоеточия/тире, иначе по границе слова.""" "CLI", "SDK", "JSON", "HTML", "TTS", "STT", "TLS", "SSL", "URL", "JWT", "REST",
txt = flat(raw) "CSV", "PDF", "SQL", "AWS", "GCP", "VPS", "AGI", "RAG", "MCP", "SMS", "OTP",
for sep in (":", " — ", " – ", ". ", "! ", "? "): "PRO", "MAX", "ULTRA", "TURBO", "FREE", "API-KEY"}
head = txt.split(sep)[0].strip()
if 14 <= len(head) <= n:
return head def smart_case(txt: str) -> str:
if len(txt) <= n: """Крик из каналов → нормальный регистр; аббревиатуры и тарифы не трогаем."""
return txt words = txt.split()
cut = txt[:n] 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 cut = cut[:cut.rfind(" ")] if " " in cut[40:] else cut
return cut.rstrip(" ,.;:-") + "…" 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 gain(raw: str, limit: int = 2) -> str: 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) txt = flat(raw)
hits: list[str] = [] hits = [m.group(0).strip(" ,.;:-") for m in MONEY.finditer(txt)]
for m in MONEY.finditer(txt): hits = [h for h in hits if re.search(r"\d", h)]
s = m.group(0).strip(" ,.;:-") if not hits:
if len(s) < 3 or s.lower() in (x.lower() for x in hits): return ""
continue vol = [h for h in hits if not h.strip().startswith("$")]
if not re.search(r"\d", s): money = [h for h in hits if h.strip().startswith("$") and _value(h) >= 50]
continue pool = vol or money or []
hits.append(s) if not pool:
if len(hits) >= limit: return ""
break best = max(pool, key=_value)
return " / ".join(hits) 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: def site(raw: str) -> str:
txt = flat(raw) for m in DOMAIN.finditer(flat(raw)):
for m in DOMAIN.finditer(txt): host = re.sub(r"^www\.", "", m.group(1).lower())
host = m.group(1).lower() if not TME.match(host):
if TME.match(host):
continue
return host return host
return "" return ""
@@ -99,24 +225,53 @@ def tags_of(it: dict) -> list[str]:
raw = it.get("tags") or [] raw = it.get("tags") or []
if isinstance(raw, str): if isinstance(raw, str):
raw = [t.strip() for t in raw.split(",")] raw = [t.strip() for t in raw.split(",")]
return [TAG_RU.get(t, t) for t in raw if t and str(t).strip()] 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]: def card(i: int, it: dict) -> list[str]:
body = "%s %s" % (it.get("title") or "", it.get("text") or "") body = "%s %s" % (it.get("title") or "", it.get("text") or "")
title = headline(it.get("title") or it.get("text") or "") title = headline(it.get("title") or it.get("text") or "")
g = gain(body) low_title = title.lower()
s = site(body) bits = [b for b in (gain(body), period(body), conds(body))
kinds = "/".join(tags_of(it)[:3]) if b and b.lower() not in low_title]
meta = " · ".join(b for b in (g, s, kinds) if b) or "подробности в посте" meta = " · ".join(bits) if bits else "детали в посте"
src = (it.get("source") or "источник").replace("rad_", "").replace("tgp_", "") host = site(body)
src = it.get("source") or "источник"
for p in ("rad_", "tgp_", "pn_"):
src = src.replace(p, "")
url = it.get("url") or "" url = it.get("url") or ""
out = ["<b>%d. %s</b>" % (i, html.escape(title)), tail = (' · <a href="%s">пост</a>' % html.escape(url)) if url else ""
"<i>%s</i>" % html.escape(meta)] if host:
if url: line = '<a href="https://%s">%s</a>%s <i>(%s)</i>' % (
out.append('<a href="%s">%s</a>' % (html.escape(url), html.escape(src[:34]))) html.escape(host), html.escape(host), tail, html.escape(src[:26]))
out.append("") else:
return out 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]: def panels_block(limit: int = 5) -> list[str]:
@@ -130,7 +285,7 @@ def panels_block(limit: int = 5) -> list[str]:
out = ["<b>ПАНЕЛИ С БЕСПЛАТНЫМИ МОДЕЛЯМИ</b>"] out = ["<b>ПАНЕЛИ С БЕСПЛАТНЫМИ МОДЕЛЯМИ</b>"]
for x in rows[:limit]: for x in rows[:limit]:
names = [str(n) for n in (x.get("free") or [])[:2]] names = [str(n) for n in (x.get("free") or [])[:2]]
tail = (" · " + html.escape(", ".join(names))) if names else "" tail = (" (%s)" % html.escape(", ".join(names))) if names else ""
out.append("%s — <b>%d</b> free из %d%s" % ( 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)) html.escape(str(x.get("host"))), int(x["free_count"]), int(x.get("models") or 0), tail))
out.append("") out.append("")
@@ -146,17 +301,27 @@ def counts():
def pick(limit: int) -> list[dict]: def pick(limit: int) -> list[dict]:
res = abuse.scan(limit=max(limit * 6, 90)) res = abuse.scan(limit=max(limit * 8, 120))
out, seen = [], set() out, seen_sig, seen_host = [], set(), set()
for it in res["items"]: for it in res["items"]:
tags = {t.strip() for t in str(it.get("tags") or "").split(",") if t.strip()} tags = {t.lower() for t in tags_of(it)}
low = flat("%s %s" % (it.get("title") or "", it.get("text") or ""))[:300].lower() body = norm_text(flat("%s %s" % (it.get("title") or "", it.get("text") or "")))
if tags & SKIP_TAGS or any(w in low for w in SKIP_WORDS): if tags & SKIP_TAGS or any(w in body[:300].lower() for w in SKIP_WORDS):
continue continue
key = it.get("sig") or it["id"] if re.match(r"^\s*(совет|гайд|лайфхак|инструкция|подборка)\b", body, re.I):
if key in seen:
continue continue
seen.add(key) 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) out.append(it)
if len(out) >= limit: if len(out) >= limit:
break break
@@ -167,13 +332,23 @@ def build(limit: int) -> str:
picks = pick(limit) picks = pick(limit)
items, ab, uniq = counts() items, ab, uniq = counts()
today = dt.date.today().strftime("%d.%m.%Y") today = dt.date.today().strftime("%d.%m.%Y")
head = "<b>OMNI · дайджест абузов</b> · %s" % today out = ["<b>OMNI · дайджест абузов</b> · %s" % today,
stat = "находок {:,} · уникальных {:,} · записей в базе {:,}".format(ab, uniq, items).replace(",", " ") "новых за сутки {:,} · всего {:,} находок ({:,} уникальных)".format(
out = [head, stat, ""] fresh_24h(), ab, uniq).replace(",", " "),
for i, it in enumerate(picks, 1): ""]
out.extend(card(i, it)) 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()) out.extend(panels_block())
return NL.join(out).strip()[:3900] return NL.join(out).strip()[:4000]
def token() -> str: def token() -> str: