213 lines
7.8 KiB
Python
213 lines
7.8 KiB
Python
"""Пост-дайджест абузов для Telegram: карточки, а не свалка строк.
|
||
|
||
Формат карточки:
|
||
<b>1. Суть находки</b>
|
||
что дают: объём, условия · сервис
|
||
источник: канал → (скрытая ссылка)
|
||
|
||
Запуск:
|
||
python tools/post_digest.py --dry # посмотреть текст
|
||
python tools/post_digest.py --limit 8 # отправить в бота
|
||
python tools/post_digest.py --chat <id>
|
||
В каналы не постит: только личка получателю (--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 = ["<b>%d. %s</b>" % (i, html.escape(title)),
|
||
"<i>%s</i>" % html.escape(meta)]
|
||
if url:
|
||
out.append('<a href="%s">%s</a>' % (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 = ["<b>ПАНЕЛИ С БЕСПЛАТНЫМИ МОДЕЛЯМИ</b>"]
|
||
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 — <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 * 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 = "<b>OMNI · дайджест абузов</b> · %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()) |