84 lines
3.6 KiB
Python
84 lines
3.6 KiB
Python
"""Заливка истории Telegram-каналов через свою сессию (Telethon) в базу omni.
|
||
|
||
Веб-превью t.me/s даёт ~20 постов, сессия — 200+. Скрипт берёт живую сессию из
|
||
пула аудита, тянет каналы и складывает посты как обычные записи (source = tg_<канал>),
|
||
чтобы они получили типы, скор и попали в панель/бот.
|
||
|
||
Запуск:
|
||
python tools/tgpump.py --pool /c/Users/User/tmp/tg_chat_grow/alive_orca.json \
|
||
--channels thehackernews,cvefeed,true_secator --limit 200 [--apply]
|
||
|
||
Без --apply только считает (dry-run). Секретов в коде нет: api_id/hash из env
|
||
TG_API_ID/TG_API_HASH (по умолчанию официальные desktop-значения), путь к сессии —
|
||
из запуска.
|
||
"""
|
||
import argparse
|
||
import json
|
||
import os
|
||
import sys
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||
|
||
from omni import store, tgsess # noqa: E402
|
||
from omni.feeds import _item # noqa: E402
|
||
|
||
|
||
def pick_session(pool_path, skip=()):
|
||
rows = [r for r in json.loads(Path(pool_path).read_text(encoding="utf-8"))
|
||
if r.get("state") == "alive" and r["session"] not in skip]
|
||
if not rows:
|
||
raise SystemExit("в пуле нет живых сессий")
|
||
return rows[0]["session"]
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--pool", default="/c/Users/User/tmp/tg_chat_grow/alive_orca.json")
|
||
ap.add_argument("--sessions-root", default=str(Path.home() / ".orca" / "sessions"))
|
||
ap.add_argument("--channels", required=True, help="через запятую или @файл")
|
||
ap.add_argument("--limit", type=int, default=200)
|
||
ap.add_argument("--apply", action="store_true")
|
||
a = ap.parse_args()
|
||
|
||
chans = a.channels
|
||
if chans.startswith("@"):
|
||
chans = Path(chans[1:]).read_text(encoding="utf-8").replace("\n", ",")
|
||
handles = [c.strip() for c in chans.split(",") if c.strip()]
|
||
|
||
sess = pick_session(a.pool)
|
||
os.environ["OMNI_TG_SESSION"] = str(Path(a.sessions_root) / sess)[:-8]
|
||
os.environ.setdefault("OMNI_TG_API_ID", os.environ.get("TG_API_ID", "2040"))
|
||
os.environ.setdefault("OMNI_TG_API_HASH",
|
||
os.environ.get("TG_API_HASH", "b18441a1ff607e10a989891a5462e627"))
|
||
os.environ["OMNI_TG_LIMIT"] = str(a.limit)
|
||
|
||
print(f"сессия {sess} | каналов {len(handles)} | лимит {a.limit}")
|
||
res = tgsess.fetch(handles, a.limit)
|
||
total_posts, items, per = 0, [], []
|
||
for rec in res:
|
||
h = rec["handle"]
|
||
posts = rec.get("posts") or []
|
||
total_posts += len(posts)
|
||
per.append((h, len(posts), rec.get("error", "")))
|
||
src = SimpleNamespace(name="tg_" + h.replace("-", "_"), tags=["abuse", "tg"])
|
||
for p in posts:
|
||
if len(p["text"]) < 12:
|
||
continue
|
||
items.append(_item(src, "post", p["url"], p["title"], p["text"],
|
||
author="@" + h, published=p["date"],
|
||
extra={"views": p["views"], "handle": h, "via": "session"}))
|
||
print("постов получено:", total_posts, "| к записи:", len(items))
|
||
for h, n, err in per:
|
||
print(f" {h:24} {n:4} {err}")
|
||
|
||
if not a.apply:
|
||
print("dry-run. добавь --apply чтобы записать в базу")
|
||
return
|
||
new, dup = store.add(items)
|
||
print(f"записано новых: {new}, дублей: {dup}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main() |