115 lines
4.0 KiB
Python
115 lines
4.0 KiB
Python
"""
|
||
Конфигурация доступа к Telegram Bot API через internal_relay (retransbot).
|
||
|
||
Все исходящие вызовы Bot API должны идти на RELAY_URL, а не на api.telegram.org.
|
||
"""
|
||
|
||
import os
|
||
from typing import Tuple
|
||
|
||
from dotenv import load_dotenv
|
||
|
||
|
||
def get_relay_url() -> str:
|
||
"""Базовый URL internal_relay без завершающего слэша."""
|
||
load_dotenv()
|
||
relay = (os.getenv("RELAY_URL") or "").strip().rstrip("/")
|
||
if not relay:
|
||
raise RuntimeError(
|
||
"Требуется переменная окружения RELAY_URL "
|
||
"(базовый URL internal_relay, например http://127.0.0.1:8081)"
|
||
)
|
||
# Допускаем host:port без схемы — httpx требует http:// или https://
|
||
if "://" not in relay:
|
||
relay = f"http://{relay}"
|
||
return relay
|
||
|
||
|
||
def configure_telebot_api() -> str:
|
||
"""
|
||
Настраивает pyTelegramBotAPI (telebot.apihelper) на работу через relay.
|
||
Вызывать до создания TeleBot.
|
||
"""
|
||
from telebot import apihelper
|
||
|
||
relay = get_relay_url()
|
||
apihelper.API_URL = f"{relay}/bot{{0}}/{{1}}"
|
||
apihelper.FILE_URL = f"{relay}/file/bot{{0}}/{{1}}"
|
||
return relay
|
||
|
||
|
||
def get_ptb_base_urls() -> Tuple[str, str]:
|
||
"""
|
||
base_url и base_file_url для python-telegram-bot.Bot.
|
||
Формат как в Bot API: .../bot и .../file/bot (библиотека сама дописывает token/method).
|
||
"""
|
||
relay = get_relay_url()
|
||
return f"{relay}/bot", f"{relay}/file/bot"
|
||
|
||
|
||
def bot_api_method_url(token: str, method: str) -> str:
|
||
"""Полный URL метода Bot API через relay (для сырых httpx/requests вызовов)."""
|
||
return f"{get_relay_url()}/bot{token}/{method}"
|
||
|
||
|
||
# Публикация в два шага (send → edit ссылок): пауза и повторы при гонке с Telegram
|
||
EDIT_LINKS_INITIAL_DELAY_SEC = 3
|
||
EDIT_LINKS_MAX_ATTEMPTS = 5
|
||
EDIT_LINKS_RETRY_INTERVAL_SEC = 2
|
||
|
||
|
||
def extract_text_link_urls(entities) -> list:
|
||
"""URL из entities типа text_link (ответ Telegram Bot API или python-telegram-bot)."""
|
||
urls = []
|
||
for ent in entities or []:
|
||
if isinstance(ent, dict):
|
||
if ent.get("type") == "text_link" and ent.get("url"):
|
||
urls.append(ent["url"])
|
||
elif getattr(ent, "type", None) == "text_link":
|
||
url = getattr(ent, "url", None)
|
||
if url:
|
||
urls.append(url)
|
||
return urls
|
||
|
||
|
||
def published_message_has_expected_links(
|
||
entities,
|
||
expected_urls,
|
||
*,
|
||
result_text: str | None = None,
|
||
min_length: int | None = None,
|
||
) -> bool:
|
||
"""
|
||
Проверка, что в сообщении есть именно footer-ссылки публикации.
|
||
|
||
Нельзя проверять «любой text_link» или слова «Инфо»/«Подписка» в тексте:
|
||
в теле события могут быть свои <a href> из replace_vk_links.
|
||
"""
|
||
expected_urls = [u for u in (expected_urls or []) if u]
|
||
if not expected_urls:
|
||
return True
|
||
|
||
if min_length is not None and result_text is not None:
|
||
if len(result_text) < min_length:
|
||
return False
|
||
|
||
found = set(extract_text_link_urls(entities))
|
||
return all(url in found for url in expected_urls)
|
||
|
||
|
||
def published_message_has_links(text: str, entities=None, expected_urls=None) -> bool:
|
||
"""Проверка footer-ссылок; expected_urls обязателен при вызове из publisher."""
|
||
if expected_urls is not None:
|
||
return published_message_has_expected_links(
|
||
entities,
|
||
expected_urls,
|
||
result_text=text,
|
||
min_length=None,
|
||
)
|
||
return bool(extract_text_link_urls(entities))
|
||
|
||
|
||
def published_text_has_links(text: str) -> bool:
|
||
"""Устаревший вызов без expected_urls — только для совместимости."""
|
||
return published_message_has_links(text)
|