105 lines
3.8 KiB
Python
105 lines
3.8 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 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": "тулза"}
|
|
|
|
|
|
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:
|
|
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((it.get("title") or it.get("text") or "")[:110].strip())
|
|
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("")
|
|
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())
|