From f263d3ffffcae0c213179f8f5d61a092874b6bef Mon Sep 17 00:00:00 2001 From: Vlad Gothbreach Date: Tue, 27 Oct 2026 19:16:46 +0300 Subject: [PATCH] =?UTF-8?q?post=5Fdigest:=20=D0=B7=D0=B0=D0=B3=D0=BE=D0=BB?= =?UTF-8?q?=D0=BE=D0=B2=D0=BE=D0=BA=20=3D=20=D1=81=D0=B0=D0=BC=D0=B0=D1=8F?= =?UTF-8?q?=20=D0=B8=D0=BD=D1=84=D0=BE=D1=80=D0=BC=D0=B0=D1=82=D0=B8=D0=B2?= =?UTF-8?q?=D0=BD=D0=B0=D1=8F=20=D1=84=D1=80=D0=B0=D0=B7=D0=B0,=20=D1=87?= =?UTF-8?q?=D0=B8=D1=81=D1=82=D0=BA=D0=B0=20CAPS/=D0=B3=D0=BE=D0=BC=D0=BE?= =?UTF-8?q?=D0=B3=D0=BB=D0=B8=D1=84=D0=BE=D0=B2/markdown,=20=D0=BA=D0=BB?= =?UTF-8?q?=D0=B8=D0=BA=D0=B0=D0=B1=D0=B5=D0=BB=D1=8C=D0=BD=D1=8B=D0=B9=20?= =?UTF-8?q?=D1=81=D0=B5=D1=80=D0=B2=D0=B8=D1=81,=20=D1=82=D0=B5=D1=81?= =?UTF-8?q?=D1=82=D1=8B=20=D1=84=D0=BE=D1=80=D0=BC=D0=B0=D1=82=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/test_post_digest.py | 83 ++++++++++ tools/post_digest.py | 319 +++++++++++++++++++++++++++++--------- 2 files changed, 330 insertions(+), 72 deletions(-) create mode 100644 tests/test_post_digest.py diff --git a/tests/test_post_digest.py b/tests/test_post_digest.py new file mode 100644 index 0000000..8dfd80e --- /dev/null +++ b/tests/test_post_digest.py @@ -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 "1." in html + assert "1 млн" in html and "токенов" in html + assert "на 90 дней" in html + assert "без карты" in html + assert "1. Суть находки - что дают: объём, условия · сервис - источник: канал → (скрытая ссылка) +Структура поста: + 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 8 # отправить в бота - python tools/post_digest.py --chat + 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 @@ -21,6 +28,7 @@ import os import re import sqlite3 import sys +import time import urllib.parse import urllib.request from pathlib import Path @@ -38,60 +46,178 @@ TAG_RU = {"promo": "промо", "trial": "триал", "keys": "ключи", "l "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]") +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) -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 = {"", "-", "—", "нет"} +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() -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(" ,.;:-") + "…" +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 gain(raw: str, limit: int = 2) -> str: - """Что дают: объёмы и суммы из текста.""" +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: 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) + 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: - txt = flat(raw) - for m in DOMAIN.finditer(txt): - host = m.group(1).lower() - if TME.match(host): - continue - return host + for m in DOMAIN.finditer(flat(raw)): + host = re.sub(r"^www\.", "", m.group(1).lower()) + if not TME.match(host): + return host return "" @@ -99,24 +225,53 @@ 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()] + 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 "") - 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_", "") + 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 "" - 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 + 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]: @@ -130,7 +285,7 @@ def panels_block(limit: int = 5) -> list[str]: 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 "" + 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("") @@ -146,17 +301,27 @@ def counts(): def pick(limit: int) -> list[dict]: - res = abuse.scan(limit=max(limit * 6, 90)) - out, seen = [], set() + res = abuse.scan(limit=max(limit * 8, 120)) + out, seen_sig, seen_host = [], set(), 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): + 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 - key = it.get("sig") or it["id"] - if key in seen: + if re.match(r"^\s*(совет|гайд|лайфхак|инструкция|подборка)\b", body, re.I): 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) if len(out) >= limit: break @@ -167,13 +332,23 @@ 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 = ["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()[:3900] + return NL.join(out).strip()[:4000] def token() -> str: