mirror of
https://gitverse.ru/kpa39l/vesti.git
synced 2026-09-29 01:50:03 +00:00
kpa39l
104 lines
4.2 KiB
Python
104 lines
4.2 KiB
Python
# Публикация карточек в Telegram через Bot API.
|
|
# Бот: @dedinit_vesti_<direction>_<lang>_bot (шаблон из конфига .env).
|
|
# Режим подтверждения: веб вызывает publish() только после явного клика.
|
|
# Метрики: getChatMemberCount / getMessage (views) для опубликованных.
|
|
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent.parent
|
|
BOT_TOKEN = os.getenv("VESTI_BOT_TOKEN", "")
|
|
API = "https://api.telegram.org"
|
|
TIMEOUT = 20.0
|
|
|
|
|
|
def _bot_url(method: str) -> str:
|
|
if not BOT_TOKEN:
|
|
raise RuntimeError("VESTI_BOT_TOKEN не задан (dry-run недоступен для сети)")
|
|
return f"{API}/bot{BOT_TOKEN}/{method}"
|
|
|
|
|
|
def channel_username(direction: str, lang: str) -> str:
|
|
"""@dedinit_vesti_<direction>_<lang>_bot — по спеке; канал — тот же без _bot."""
|
|
return f"@dedinit_vesti_{direction}_{lang}_bot"
|
|
|
|
|
|
def _upload_photo(chat_id: str, photo_path: str, caption: str = "") -> int:
|
|
"""Отправляет фото (медиа первым). Возвращает message_id."""
|
|
with open(photo_path, "rb") as f:
|
|
r = httpx.post(
|
|
_bot_url("sendPhoto"),
|
|
data={"chat_id": chat_id, "caption": caption[:1024]},
|
|
files={"photo": f},
|
|
timeout=TIMEOUT,
|
|
)
|
|
r.raise_for_status()
|
|
return r.json()["result"]["message_id"]
|
|
|
|
|
|
def _send_message(chat_id: str, text: str, parse_mode: str = "HTML") -> int:
|
|
r = httpx.post(
|
|
_bot_url("sendMessage"),
|
|
json={"chat_id": chat_id, "text": text, "parse_mode": parse_mode},
|
|
timeout=TIMEOUT,
|
|
)
|
|
r.raise_for_status()
|
|
return r.json()["result"]["message_id"]
|
|
|
|
|
|
def publish_card(card: dict, direction: str = "linux", lang: str = "ru", chat_id: str | None = None) -> dict:
|
|
"""Публикует карточку {media, text} в канал. Возвращает {message_id, media_message_id}.
|
|
|
|
dry-run: если VESTI_BOT_TOKEN не задан — возвращает плейсхолдер (не отправляет).
|
|
"""
|
|
if not BOT_TOKEN:
|
|
return {"message_id": 0, "media_message_id": 0, "dry_run": True}
|
|
|
|
chat = chat_id or channel_username(direction, lang)
|
|
media_msg = None
|
|
if card.get("media"):
|
|
media_msg = _upload_photo(chat, card["media"], caption=card["text"][:1024])
|
|
msg = _send_message(chat, card["text"])
|
|
return {"message_id": msg, "media_message_id": media_msg, "dry_run": False}
|
|
|
|
|
|
def publish_multi(card: dict, directions: list[str], lang: str = "ru") -> dict:
|
|
"""Публикует карточку во ВСЕ каналы направлений. Возвращает {direction: {message_id, media_message_id, dry_run}}.
|
|
|
|
directions: список направлений (напр. ['linux', 'ai']).
|
|
При пустом списке — публикация в направление по умолчанию (card['direction']).
|
|
"""
|
|
if not directions:
|
|
directions = [card.get("direction") or "linux"]
|
|
out = {}
|
|
for d in directions:
|
|
out[d] = publish_card(card, direction=d, lang=lang)
|
|
return out
|
|
|
|
|
|
def get_views(chat_id: str, message_id: int) -> int:
|
|
"""Возвращает просмотры опубликованного сообщения через Bot API (getMessage → views)."""
|
|
if not BOT_TOKEN:
|
|
return 0
|
|
try:
|
|
r = httpx.get(_bot_url("getMessage"), params={"chat_id": chat_id, "message_id": message_id}, timeout=TIMEOUT)
|
|
r.raise_for_status()
|
|
m = r.json()["result"]
|
|
return int(m.get("views") or m.get("forwards") or 0)
|
|
except Exception as e:
|
|
print(f" ! get_views failed {chat_id}/{message_id}: {e}")
|
|
return 0
|
|
|
|
|
|
def get_views_multi(directions: list[str], message_ids: dict, lang: str = "ru") -> dict:
|
|
"""Собирает просмотры по каждому направлению. Возвращает {direction: views}."""
|
|
out = {}
|
|
for d in directions:
|
|
mid = (message_ids.get(d) or {}).get("message_id") or 0
|
|
if mid:
|
|
out[d] = get_views(channel_username(d, lang), mid)
|
|
else:
|
|
out[d] = 0
|
|
return out |