150 lines
5.5 KiB
Python
150 lines
5.5 KiB
Python
"""Отправить топ находок в Telegram через бота проекта.
|
||
|
||
Запуск: python tools/push_top.py [--limit 12] [--chat <id>] [--dry]
|
||
Токен берётся из .bot_token, получатель — --chat или OMNI_CHAT (дефолт 7448683285).
|
||
Ничего не постит в каналы: только личное сообщение получателю.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
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
|
||
|
||
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": "реф"}
|
||
|
||
|
||
NOISE = ("навигация по каналу", "навигация", "закреп", "правила канала", "прайс-лист",
|
||
"реклама в канале", "меню канала", "список каналов")
|
||
EMOJI = re.compile("[🀀-☀-➿⬀-⯿️
]")
|
||
|
||
|
||
def clean(txt: str, n: int = 110) -> str:
|
||
txt = EMOJI.sub(" ", txt or "")
|
||
txt = re.sub(r"\s+", " ", txt).strip()
|
||
if len(txt) <= n:
|
||
return txt
|
||
cut = txt[:n]
|
||
if " " in cut[40:]:
|
||
cut = cut[:cut.rfind(" ")]
|
||
return cut.rstrip(" ,.;:-") + "…"
|
||
|
||
|
||
def is_noise(title: str) -> bool:
|
||
low = (title or "").lower()
|
||
return any(k in low for k in NOISE)
|
||
|
||
|
||
def panel_block(limit: int = 4) -> list[str]:
|
||
import json
|
||
from pathlib import Path
|
||
pj = ROOT / "data" / "panels.json"
|
||
if not pj.exists():
|
||
return []
|
||
rows = json.load(pj.open(encoding="utf-8"))["panels"]
|
||
rows = [x for x in rows 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 = ", ".join(str(n) for n in (x.get("free") or [])[:3])
|
||
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),
|
||
(" · " + html.escape(names)) if names else ""))
|
||
out.append("")
|
||
return out
|
||
|
||
|
||
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 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 build(limit: int) -> str:
|
||
res = abuse.scan(limit=max(limit * 4, 60))
|
||
picks, seen = [], set()
|
||
for it in res["items"]:
|
||
key = it.get("sig") or it["id"]
|
||
if key in seen or is_noise(it.get("title") or ""):
|
||
continue
|
||
seen.add(key)
|
||
picks.append(it)
|
||
if len(picks) >= limit:
|
||
break
|
||
items, ab, uniq = counts()
|
||
out = ["<b>OMNI · ТОП НАХОДОК</b>",
|
||
f"в базе {items:,} записей · находок {ab:,} · уникальных {uniq:,}".replace(",", " "),
|
||
""]
|
||
for i, it in enumerate(picks, 1):
|
||
raw = it.get("tags") or []
|
||
raw = raw.split(",") if isinstance(raw, str) else list(raw)
|
||
tags = ", ".join(TAG_RU.get(t, t) for t in raw if t)
|
||
title = html.escape(clean(it.get("title") or it.get("text") or ""))
|
||
tags_s = f" · <i>{html.escape(tags)}</i>" if tags else ""
|
||
dup = f" · копий {it['dups']}" if it.get("dups") else ""
|
||
link = it.get("url") or ""
|
||
head = f"{i}. <b>{it.get('abuse_score', 0)}</b>{tags_s}{dup}"
|
||
out.append(head)
|
||
out.append(title)
|
||
if link:
|
||
out.append(f'<a href="{html.escape(link)}">{html.escape(link[:78])}</a>')
|
||
out.append("")
|
||
out.extend(panel_block())
|
||
return "\n".join(out)[:3900]
|
||
|
||
|
||
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 = f"https://api.telegram.org/bot{token()}/sendMessage"
|
||
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=12)
|
||
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())
|