diff --git a/TGBRIDGE.md b/TGBRIDGE.md new file mode 100644 index 0000000..855e43f --- /dev/null +++ b/TGBRIDGE.md @@ -0,0 +1,208 @@ +# Telegram bridge — rebuild notes + +Everything below is in `tgbridge/`. Routes, behaviour and env-var names stay +backwards compatible: **your existing `credentials.env` works unchanged.** + +--- + +## 1. What you need to do + +One optional step, for the avatars you asked for: + +1. Make a channel in your server (e.g. `#tg-avatar-cache`) and deny + `@everyone` → *View Channel*. +2. Right-click it → **Copy Channel ID**. +3. Paste it into the `AVATAR_CACHE_CHANNEL_ID=` line already added at the + bottom of `credentials.env`. + +Leave it empty and everything else still works — forwarded messages just keep +the default webhook avatar. + +Optional: `pip install lottie` to make animated `.tgs` stickers actually +animate. Without it they fall back to a static thumbnail, which is still +better than the broken attachment they produce today. + +`main.py` now loads the bridge as one extension: + +```python +await bot.load_extension('tgbridge') +``` + +**Behavioural change worth knowing:** the Telegram poller now starts when the +extension loads, not from `on_ready`. That's what fixes the duplicate-poller +bug (A3), but it means startup happens before the Discord gateway connects. A +Telegram failure at that moment is caught and retried in the background (5 +attempts, backing off 30 s → 10 min) — it will **not** stop the rest of the bot +from booting. Verified with a deliberately invalid token: the bot loaded fine, +both cogs registered, and `!bridge status` showed the error. + +--- + +## 2. Bugs fixed + +These were all live defects, not hypotheticals. + +| # | Bug | Effect before | +|---|---|---| +| A1 | `parse_mode='Markdown'` with an unescaped body | **Every** Discord message containing a stray `*`, `_`, `` ` `` or `[` was rejected by Telegram with a 400 and silently dropped. Now HTML parse mode with proper escaping, plus a plain-text retry if Telegram still complains. | +| A2 | No `allowed_mentions` on webhook posts | Anyone in the Telegram group could type `@everyone` and ping your whole server. Now `{"parse": []}` on every post. | +| A3 | `forwarder.start()` called from `on_ready` | `on_ready` fires again after every gateway reconnect, starting a second poller on the same token → Telegram 409 Conflict and duplicated messages. Now starts once at extension load, with a `_running` guard. | +| A4 | `document.mime_type.startswith(...)` | `mime_type` is nullable; any document without one raised into the bare `except` and the message was dropped. | +| A5 | `temp_in_path` used in `finally`, assigned in `try` | If `NamedTemporaryFile` raised, cleanup threw `UnboundLocalError` and masked the real error. | +| A6 | ffmpeg cleanup after `wait_for`, no kill on timeout | A conversion timeout leaked the temp file **and** left an orphaned ffmpeg process forever. Now killed and reaped in every path. | + +--- + +## 3. Stickers — measured, not guessed + +Your old code ran all three Telegram sticker formats through one GIF +conversion. I tested each: + +| Format | Before | Now | +|---|---|---| +| **Static `.webp`** | Re-encoded to GIF. Measured: the antialiased edge went from **228 alpha levels to 2**, i.e. a visible jagged halo on every sticker — and the file got **4× bigger** (942 B → 3771 B) and was downscaled 512→320. | Passed through untouched. Discord renders `.webp` natively with full alpha. Zero CPU. | +| **Animated `.tgs`** | ffmpeg exits **183 "Invalid data found"** (it's gzipped Lottie JSON, ffmpeg cannot read it). The `returncode == 0` check failed, so the raw `.tgs` was uploaded as an **unopenable junk attachment**. | Rendered via the optional `lottie` library; falls back to Telegram's own static thumbnail + emoji. | +| **Video `.webm`** | Converted to GIF, same alpha crushing. | Converted to **animated WebP** — loops in Discord, keeps 8-bit alpha, and measured **7.4× smaller** than the GIF (16 KB vs 122 KB). | + +**One subtle bug worth calling out.** ffmpeg's *default* VP9 decoder silently +discards the alpha channel of VP9-in-WebM — every transparent pixel decodes to +opaque black. Only `libvpx-vp9`, selected **before** `-i`, reads it. Telegram +video stickers are exactly this format. Verified: without the flag, 1 alpha +level; with it, 217. This is now handled in `media.py:_decoder_args`. + +Converted stickers are cached in sqlite by `file_unique_id`, so each one is +only ever converted once. + +--- + +## 4. Profile-photo avatars + +Telegram's only file URL embeds your bot token +(`api.telegram.org/file/bot/...`), so it can never go in a webhook +payload. Instead the photo is re-uploaded once to your private cache channel +and that link is used as `avatar_url`. + +I verified against the live API that **Discord re-hosts the image at post +time**: `author.avatar` comes back as a fresh hash distinct from the webhook's +own. So the source link only needs to be alive at the moment of the POST, and +already-forwarded messages keep their avatar forever. + +Caching behaviour, all verified by test: + +- repeat lookups cost **zero** API calls +- users with no photo (or privacy-hidden) are cached negatively, so it doesn't + re-ask on every message +- when the hosted link goes stale it re-uploads **from the stored blob**, + without re-downloading from Telegram +- when someone changes their photo it's detected and refreshed +- channel posts use the chat photo instead of a user photo + +--- + +## 5. Features added + +**Reply threading (both directions).** A Telegram reply renders as a Discord +quote block with a clickable jump link to the original forwarded message; a +Discord reply becomes a native Telegram reply. Backed by a message-ID map in +`data/tgbridge.db`. + +**Edit sync (both directions).** Telegram `edited_message` updates patch the +Discord message; Discord edits patch the Telegram one. The Discord side uses +`on_raw_message_edit`, **not** `on_message_edit` — the cached variant never +fires for messages sent before the last restart, so edit sync would have +silently worked in testing and failed in production. + +**Delete sync (Discord → Telegram only).** Including bulk deletes. The reverse +isn't possible: Telegram's Bot API gives bots no delete events at all. + +**Albums.** Telegram sends an album as N separate updates, so a 5-photo post +used to become 5 separate Discord messages. Now debounced (1.5 s) into one +message with up to 10 attachments. All N Telegram IDs map to the single +Discord message, so a reply to *any* photo in the album still resolves. + +**Message types that used to say `*(unsupported message type)*`:** voice notes +(with duration), video notes, audio (with artist/title), polls (question + +options), locations and venues (with a maps link), contacts, dice rolls, +join/leave notices, pins, and "forwarded from X" attribution. + +**Rich text.** Telegram entities → Discord markdown and back: bold, italic, +underline, strikethrough, spoilers, inline code, code blocks with language, +links, and blockquotes. + +> Telegram entity offsets are counted in **UTF-16 code units**, not Python +> characters, so any message containing an emoji shifts every following offset +> and naive slicing corrupts the formatting. The converter works on a UTF-16 +> buffer and decodes once at the end. Verified with emoji before, inside and +> after formatted spans. + +**Discord references resolved.** `<@123>`, `<@&123>`, `<#123>` and +`<:emoji:123>` used to forward to Telegram as raw snowflakes; they now become +readable names. + +**Long messages split** instead of being truncated at 2000 chars (`content[:2000]`), +on the nicest available boundary. Same on the Telegram side at 4096. + +**Rate limits and retries.** One shared `aiohttp` session instead of a new one +per message; 429s now wait the `retry_after` Discord asks for; 5xx retries with +exponential backoff. + +**Oversized uploads** are detected and posted as a notice naming the file, +instead of vanishing. + +**Markdown injection blocked.** Telegram text is escaped before it reaches +Discord, and Discord text is HTML-escaped before it reaches Telegram, so +neither side can inject formatting or HTML into the other. + +**Route auto-discovery.** Any `TELEGRAM_GROUP_ID_` + +`DISCORD_WEBHOOK_URL_` + `DISCORD_CHANNEL_ID_` trio becomes a +route automatically — a third bridged channel is now an `.env` edit, not a code +change. `data/bridge_routes.json` can add or override routes. Your existing +`main` and `shitpost` routes were confirmed to load unchanged. + +**`!bridge status`** — routes, uptime, forwarded counts, webhook success/failure, +cached avatars and stickers, whether Lottie is installed, and the last error. + +**Explicit echo guard.** The bridge now checks the sender against its own bot +ID. Previously loops were avoided only incidentally, which would have bitten as +soon as edit sync or a second token was added. + +--- + +## 6. Files + +| File | Purpose | +|---|---| +| `__init__.py` | extension entry point; wires and starts everything | +| `config.py` | route discovery from env + optional JSON | +| `store.py` | sqlite: message map, avatar cache, sticker cache | +| `entities.py` | Telegram entities ↔ Discord markdown ↔ Telegram HTML | +| `media.py` | ffmpeg / Lottie conversion, with process kill + cleanup | +| `avatars.py` | profile photos → Discord CDN | +| `webhook.py` | Discord webhook client: session, retries, splitting | +| `forwarder.py` | Telegram → Discord | +| `discord_to_telegram.py` | Discord → Telegram | +| `status_cog.py` | `!bridge status` | + +New database at `data/tgbridge.db` (created automatically, `data/` is already +gitignored). Message mappings are pruned after 30 days. + +--- + +## 7. What I could not test + +I have no access to your Telegram group or a live Discord channel, so the +end-to-end path is **unverified against the real APIs**. What I did verify: + +- every module imports, and `main.py` parses +- both directions exercised with mocked Telegram/Discord objects, covering + ~25 message shapes (text, entities, photos, albums, replies, edits, echo + guard, null MIME types, polls, locations, dice, voice, splitting) +- ffmpeg conversions run on real generated media, with alpha fidelity measured +- avatar caching logic, including staleness and photo-change paths +- route discovery against your actual `credentials.env` +- the Discord avatar re-hosting question, against the live API (one test + message, deleted immediately afterwards) + +Worth watching on first run: whether Telegram accepts the generated HTML for +unusual messages (there's a plain-text fallback if it doesn't), and whether the +bot has permission to delete messages in the Telegram group for delete sync. diff --git a/main.py b/main.py index 98ecf8d..6f8cbbd 100644 --- a/main.py +++ b/main.py @@ -5,9 +5,6 @@ import os import logging import asyncio -from tgbridge.forwarder import TelegramForwarder -from tgbridge.discord_to_telegram import DiscordToTelegramForwarder - # from twitterbridge.rss import TwitterRSSForwarder load_dotenv('credentials.env') @@ -18,19 +15,8 @@ intents.members = True bot = commands.Bot(command_prefix='!', intents=intents, help_command=None) -# Telegram → Discord -forwarder = TelegramForwarder(token=os.getenv('TELEGRAM_BOT_TOKEN')) -if os.getenv('TELEGRAM_GROUP_ID') and os.getenv('DISCORD_WEBHOOK_URL'): - forwarder.add_route(int(os.getenv('TELEGRAM_GROUP_ID')), os.getenv('DISCORD_WEBHOOK_URL')) -if os.getenv('TELEGRAM_GROUP_ID_SHITPOST') and os.getenv('DISCORD_WEBHOOK_URL_SHITPOST'): - forwarder.add_route(int(os.getenv('TELEGRAM_GROUP_ID_SHITPOST')), os.getenv('DISCORD_WEBHOOK_URL_SHITPOST')) - -# Discord → Telegram (reverse bridge) -d_to_tg = DiscordToTelegramForwarder(bot=bot, tg_token=os.getenv('TELEGRAM_BOT_TOKEN')) -if os.getenv('DISCORD_CHANNEL_ID') and os.getenv('TELEGRAM_GROUP_ID'): - d_to_tg.add_route(int(os.getenv('DISCORD_CHANNEL_ID')), int(os.getenv('TELEGRAM_GROUP_ID'))) -if os.getenv('DISCORD_CHANNEL_ID_SHITPOST') and os.getenv('TELEGRAM_GROUP_ID_SHITPOST'): - d_to_tg.add_route(int(os.getenv('DISCORD_CHANNEL_ID_SHITPOST')), int(os.getenv('TELEGRAM_GROUP_ID_SHITPOST'))) +# The Telegram bridge discovers its own routes from credentials.env and +# starts polling when the extension loads — see tgbridge/config.py. # twitter = TwitterRSSForwarder( # webhook_url=os.getenv('TWITTER_DISCORD_WEBHOOK'), @@ -44,7 +30,6 @@ if os.getenv('DISCORD_CHANNEL_ID_SHITPOST') and os.getenv('TELEGRAM_GROUP_ID_SHI @bot.event async def on_ready(): print(f'[+] {bot.user} is online!') - asyncio.create_task(forwarder.start()) # asyncio.create_task(twitter.start()) # asyncio.create_task(twitter_mao.start()) # Clear stale global slash commands @@ -55,7 +40,7 @@ async def on_ready(): async def main(): async with bot: - await bot.add_cog(d_to_tg) + await bot.load_extension('tgbridge') await bot.load_extension('cogs.general') await bot.load_extension('cogs.chance') await bot.load_extension('cogs.gifs') @@ -80,7 +65,9 @@ async def main(): try: await bot.start(os.getenv('DISCORD_TOKEN')) finally: - await forwarder.stop() + # bot.close() unloads extensions, so tgbridge.teardown stops the + # Telegram poller and closes its aiohttp session for us + pass # twitter.stop() # twitter_mao.stop() # await twitter.close() diff --git a/tgbridge/__init__.py b/tgbridge/__init__.py index 8b13789..8998f53 100644 --- a/tgbridge/__init__.py +++ b/tgbridge/__init__.py @@ -1 +1,89 @@ +#two-way telegram <-> discord bridge, loaded as an extension## +"""tgbridge — two-way Telegram <-> Discord bridge. +Polling starts at extension load, not on_ready: on_ready fires again after +every gateway reconnect and a second poller on one token means telegram 409s. +""" +###################################################################### +import asyncio +import logging +import os + +from discord.ext import commands + +from . import config +from .discord_to_telegram import DiscordToTelegramForwarder +from .forwarder import TelegramForwarder +from .status_cog import BridgeStatus + +log = logging.getLogger(__name__) + +_forwarder: TelegramForwarder | None = None + + +async def setup(bot: commands.Bot) -> None: + global _forwarder + + token = os.getenv("TELEGRAM_BOT_TOKEN") + if not token: + log.warning("TELEGRAM_BOT_TOKEN is not set — bridge not loaded") + return + + routes = config.load_routes() + if not routes: + log.warning("no bridge routes configured — bridge not loaded") + return + + replay = os.getenv("TELEGRAM_REPLAY_BACKLOG", "").lower() in ("1", "true", "yes") + + _forwarder = TelegramForwarder( + token=token, + discord_bot=bot, + avatar_channel_id=config.avatar_channel_id(), + routes=routes, + drop_pending=not replay, + ) + reverse = DiscordToTelegramForwarder(bot=bot, tg_token=token, routes=routes) + + await bot.add_cog(reverse) + await bot.add_cog(BridgeStatus(bot, _forwarder, reverse)) + bot.tg_bridge = _forwarder + + # setup() runs before the gateway connects, so raising here stops the whole + # bot booting. retry in the background instead; !bridge status shows it. + try: + await _forwarder.start() + except Exception as exc: + log.error("telegram bridge failed to start (%s) — retrying in the " + "background; the rest of the bot is unaffected", exc) + _forwarder.last_error = f"start failed: {exc}" + # keep the reference; a bare task can be gc'd mid-flight + bot.tg_bridge_retry = asyncio.create_task(_retry_start(_forwarder)) + + +async def _retry_start(forwarder) -> None: + delay = 30 + for attempt in range(5): + await asyncio.sleep(delay) + try: + await forwarder.start() + log.info("telegram bridge recovered on attempt %d", attempt + 1) + return + except Exception as exc: + log.warning("telegram bridge retry %d failed: %s", attempt + 1, exc) + forwarder.last_error = f"start failed: {exc}" + delay = min(delay * 2, 600) + log.error("telegram bridge gave up after 5 retries — restart the bot") + + +async def teardown(bot: commands.Bot) -> None: + global _forwarder + task = getattr(bot, "tg_bridge_retry", None) + if task and not task.done(): + task.cancel() + if _forwarder is not None: + await _forwarder.stop() + _forwarder = None + + +__all__ = ["TelegramForwarder", "DiscordToTelegramForwarder", "setup", "teardown"] diff --git a/tgbridge/avatars.py b/tgbridge/avatars.py new file mode 100644 index 0000000..fa74f8a --- /dev/null +++ b/tgbridge/avatars.py @@ -0,0 +1,165 @@ +#telegram profile photos re-hosted as discord webhook avatars## +"""Profile photos for forwarded messages. + +Telegram's file url embeds the bot token, so it can never go in a webhook +payload. The photo is uploaded once to a private discord channel and that link +is used as avatar_url — discord re-hosts it at post time, so the link only has +to be alive during the POST. + +Best-effort: every failure returns None and the message keeps the default avatar. +""" +###################################################################### +import asyncio +import io +import logging +import time + +import discord + +from . import store + +log = logging.getLogger(__name__) + + +class AvatarCache: + def __init__(self, discord_bot, channel_id: int | None): + self.bot = discord_bot + self.channel_id = channel_id + self._tg_bot = None + self._locks: dict[int, "object"] = {} + self.hits = 0 + self.uploads = 0 + + def bind_telegram(self, tg_bot): + self._tg_bot = tg_bot + + @property + def enabled(self) -> bool: + return bool(self.channel_id and self._tg_bot) + + def _lock(self, key: int): + # stop two messages from one user racing to upload the same photo + lock = self._locks.get(key) + if lock is None: + lock = self._locks[key] = asyncio.Lock() + return lock + + # ── telegram side ────────────────────────────────────── + + async def _current_photo(self, user_id: int): + """-> (file_id, file_unique_id) for a user's current photo, or None.""" + photos = await self._tg_bot.get_user_profile_photos(user_id, limit=1) + # empty is normal: no photo set, or privacy hides it + if not photos or not photos.photos or not photos.photos[0]: + return None + sizes = photos.photos[0] + pick = next((s for s in sizes if s.width >= 160), sizes[-1]) + return pick.file_id, pick.file_unique_id + + async def _chat_photo(self, chat): + photo = getattr(chat, "photo", None) + if not photo: + return None + return photo.small_file_id, photo.small_file_unique_id + + async def _download(self, file_id: str) -> bytes | None: + try: + tg_file = await self._tg_bot.get_file(file_id) + buf = io.BytesIO() + await tg_file.download_to_memory(buf) + return buf.getvalue() + except Exception as exc: + log.debug("avatar download failed: %s", exc) + return None + + # ── discord side ─────────────────────────────────────── + + async def _upload(self, key: int, image: bytes) -> str | None: + """Re-host the image in the cache channel and return its url.""" + channel = self.bot.get_channel(self.channel_id) + if channel is None: + try: + channel = await self.bot.fetch_channel(self.channel_id) + except Exception as exc: + log.warning("avatar cache channel %s unreachable: %s", + self.channel_id, exc) + return None + try: + file = discord.File(io.BytesIO(image), filename=f"{key}.jpg") + message = await channel.send(file=file) + self.uploads += 1 + return message.attachments[0].url + except Exception as exc: + log.warning("avatar upload failed: %s", exc) + return None + + # ── public ──────────────────────────────────────────── + + async def url_for_user(self, user) -> str | None: + if not self.enabled or user is None: + return None + return await self._resolve(user.id, lambda: self._current_photo(user.id)) + + async def url_for_chat(self, chat) -> str | None: + if not self.enabled or chat is None: + return None + return await self._resolve(chat.id, lambda: self._chat_photo(chat)) + + async def _resolve(self, key: int, fetch_photo) -> str | None: + try: + async with self._lock(key): + return await self._resolve_inner(key, fetch_photo) + except Exception as exc: + log.debug("avatar resolution failed for %s: %s", key, exc) + return None + + async def _resolve_inner(self, key: int, fetch_photo) -> str | None: + now = int(time.time()) + cached = await store.get_avatar(key) + + # fresh link, checked recently — nothing to do + if cached and cached["cdn_url"] and \ + now - cached["uploaded_at"] < store.AVATAR_URL_TTL and \ + now - cached["checked_at"] < store.AVATAR_RECHECK: + self.hits += 1 + return cached["cdn_url"] + + # known to have no photo, not time to re-check yet + if cached and not cached["file_unique_id"] and \ + now - cached["checked_at"] < store.AVATAR_RECHECK: + return None + + photo = None + if not cached or now - cached["checked_at"] >= store.AVATAR_RECHECK: + try: + photo = await fetch_photo() + except Exception as exc: + log.debug("profile photo lookup failed for %s: %s", key, exc) + photo = None + + if photo is None: + await store.put_avatar(key, None, None, None, 0, now) + return None + + # same photo: refresh the link from stored bytes, no re-download + if cached and photo and photo[1] == cached["file_unique_id"] \ + and cached["image"]: + if cached["cdn_url"] and now - cached["uploaded_at"] < store.AVATAR_URL_TTL: + await store.touch_avatar(key, now) + self.hits += 1 + return cached["cdn_url"] + url = await self._upload(key, cached["image"]) + if url: + await store.put_avatar(key, photo[1], cached["image"], url, now, now) + return url + + # new or changed photo: download, host, remember + if photo is None: + return cached["cdn_url"] if cached else None + image = await self._download(photo[0]) + if not image: + return cached["cdn_url"] if cached else None + url = await self._upload(key, image) + if url: + await store.put_avatar(key, photo[1], image, url, now, now) + return url diff --git a/tgbridge/config.py b/tgbridge/config.py new file mode 100644 index 0000000..c140b1b --- /dev/null +++ b/tgbridge/config.py @@ -0,0 +1,119 @@ +#bridge routes, discovered from env instead of hardcoded in main.py## +"""Route configuration for the bridge. + +Any TELEGRAM_GROUP_ID_ + DISCORD_WEBHOOK_URL_ + +DISCORD_CHANNEL_ID_ trio becomes a route, so a new bridged channel is +an .env edit. The unsuffixed trio still works, as "main". +data/bridge_routes.json adds to or overrides them. +""" +###################################################################### +import json +import logging +import os +from dataclasses import dataclass +from pathlib import Path + +log = logging.getLogger(__name__) + +ROUTES_FILE = Path(__file__).parent.parent / "data" / "bridge_routes.json" + +_TG_PREFIX = "TELEGRAM_GROUP_ID" +_WH_PREFIX = "DISCORD_WEBHOOK_URL" +_CH_PREFIX = "DISCORD_CHANNEL_ID" + + +@dataclass +class Route: + name: str + tg_chat_id: int + dc_channel_id: int | None = None + webhook_url: str | None = None + + @property + def tg_to_dc(self) -> bool: + return bool(self.webhook_url) + + @property + def dc_to_tg(self) -> bool: + return bool(self.dc_channel_id) + + +def _int_or_none(value): + try: + return int(str(value).strip()) + except (TypeError, ValueError): + return None + + +def _env_routes() -> dict[str, Route]: + routes: dict[str, Route] = {} + for key, value in os.environ.items(): + if not key.startswith(_TG_PREFIX): + continue + suffix = key[len(_TG_PREFIX):] # "" or "_SHITPOST" + chat_id = _int_or_none(value) + if chat_id is None: + log.warning("%s is not a valid chat id, skipping", key) + continue + name = suffix.lstrip("_").lower() or "main" + routes[name] = Route( + name=name, + tg_chat_id=chat_id, + dc_channel_id=_int_or_none(os.getenv(_CH_PREFIX + suffix)), + webhook_url=os.getenv(_WH_PREFIX + suffix) or None, + ) + return routes + + +def _file_routes() -> dict[str, Route]: + if not ROUTES_FILE.exists(): + return {} + try: + raw = json.loads(ROUTES_FILE.read_text(encoding="utf-8")) + except Exception as exc: + log.error("could not read %s: %s", ROUTES_FILE, exc) + return {} + + routes = {} + for index, entry in enumerate(raw or []): + chat_id = _int_or_none(entry.get("telegram_chat_id")) + if chat_id is None: + log.warning("route %d in %s has no telegram_chat_id", index, ROUTES_FILE) + continue + name = entry.get("name") or f"route{index}" + routes[name] = Route( + name=name, + tg_chat_id=chat_id, + dc_channel_id=_int_or_none(entry.get("discord_channel_id")), + webhook_url=entry.get("webhook_url") or None, + ) + return routes + + +def load_routes() -> list[Route]: + """Env routes first; the json file wins on a name clash.""" + routes = _env_routes() + routes.update(_file_routes()) + + usable = [] + for route in routes.values(): + if not (route.tg_to_dc or route.dc_to_tg): + log.warning("route %r has neither a webhook nor a channel — ignored", + route.name) + continue + usable.append(route) + + for route in usable: + directions = [] + if route.tg_to_dc: + directions.append("tg->dc") + if route.dc_to_tg: + directions.append("dc->tg") + log.info("bridge route %r: chat %s <-> channel %s (%s)", + route.name, route.tg_chat_id, route.dc_channel_id, + ", ".join(directions)) + return usable + + +def avatar_channel_id() -> int | None: + return _int_or_none(os.getenv("AVATAR_CACHE_CHANNEL_ID")) diff --git a/tgbridge/discord_to_telegram.py b/tgbridge/discord_to_telegram.py index e426fd4..08654bd 100644 --- a/tgbridge/discord_to_telegram.py +++ b/tgbridge/discord_to_telegram.py @@ -1,96 +1,346 @@ -# Discord → Telegram forwarding +#discord-to-telegram forwarding## +"""Discord -> Telegram forwarding. + +Sends as HTML, not Markdown: telegram 400s on any unpaired * _ ` or [, which +silently dropped most real messages. Only html can escape arbitrary user text. +""" +###################################################################### +import html import io +import logging + import discord from discord.ext import commands -from telegram import Bot +from telegram import (Bot, InputMediaAudio, InputMediaDocument, + InputMediaPhoto, InputMediaVideo, ReplyParameters) from telegram.error import TelegramError +from . import entities, store + +log = logging.getLogger(__name__) + +TEXT_LIMIT = 4096 +CAPTION_LIMIT = 1024 +MEDIA_GROUP_MAX = 10 + class DiscordToTelegramForwarder(commands.Cog): - """ - Listens for Discord messages in configured channels and forwards them to - the paired Telegram group. + """Mirrors configured Discord channels into their paired Telegram chats.""" - Routes are added via add_route(discord_channel_id, telegram_chat_id). - Bot messages and webhook messages are silently ignored to prevent loops. - """ - - def __init__(self, bot: commands.Bot, tg_token: str): + def __init__(self, bot: commands.Bot, tg_token: str, routes=None): self.bot = bot self._tg_bot = Bot(token=tg_token) - # discord_channel_id (int) -> telegram_chat_id (int) + # discord_channel_id -> telegram_chat_id self.routes: dict[int, int] = {} + for route in routes or []: + if route.dc_to_tg and route.tg_chat_id: + self.routes[route.dc_channel_id] = route.tg_chat_id + self.forwarded = 0 + self.last_error: str | None = None def add_route(self, discord_channel_id: int, telegram_chat_id: int): - self.routes[discord_channel_id] = telegram_chat_id + self.routes[int(discord_channel_id)] = int(telegram_chat_id) - # -- helpers -------------------------------------------------------------- - - def _format_header(self, message: discord.Message) -> str: - """Return a bold sender label: *Username* (Discord)""" - name = message.author.display_name - return f"{discord.utils.escape_markdown(name)}" - - async def _send_text(self, chat_id: int, text: str): + async def cog_unload(self): try: - await self._tg_bot.send_message( - chat_id=chat_id, - text=text, - parse_mode='Markdown', - ) - except TelegramError as e: - print(f'[D->TG] send_message failed: {e}') + await self._tg_bot.shutdown() + except Exception: + pass - async def _send_attachment(self, chat_id: int, attachment: discord.Attachment, caption: str): - """Download a Discord attachment and upload it to Telegram.""" + # ── helpers ───────────────────────────────────────────────────────────── + + def _header(self, message: discord.Message) -> str: + return f"{html.escape(message.author.display_name)}" + + def _body(self, message: discord.Message) -> str: + content = entities.resolve_discord_references(message, message.content or "") + return entities.to_telegram_html(content) + + async def _reply_params(self, message: discord.Message): + """Point the telegram message at whatever this replies to.""" + ref = getattr(message, "reference", None) + if not ref or not ref.message_id: + return None try: - file_bytes = await attachment.read() - buf = io.BytesIO(file_bytes) - buf.name = attachment.filename + mapped = await store.tg_for_dc(message.channel.id, ref.message_id) + if not mapped: + return None + return ReplyParameters(message_id=mapped[1], + allow_sending_without_reply=True) + except Exception as exc: + log.debug("reply lookup failed: %s", exc) + return None - mime = attachment.content_type or '' + @staticmethod + def _split(text: str, limit: int) -> list[str]: + if not text: + return [] + if len(text) <= limit: + return [text] + chunks, rest = [], text + while len(rest) > limit: + window = rest[:limit] + cut = window.rfind("\n") + if cut < limit // 2: + cut = window.rfind(" ") + if cut < limit // 2: + cut = limit + chunks.append(rest[:cut]) + rest = rest[cut:].lstrip() + if rest: + chunks.append(rest) + return chunks - if mime.startswith('image/') and not mime.startswith('image/gif'): - await self._tg_bot.send_photo(chat_id=chat_id, photo=buf, caption=caption, parse_mode='Markdown') - elif mime.startswith('video/') or mime == 'image/gif': - await self._tg_bot.send_video(chat_id=chat_id, video=buf, caption=caption, parse_mode='Markdown') - elif mime.startswith('audio/'): - await self._tg_bot.send_audio(chat_id=chat_id, audio=buf, caption=caption, parse_mode='Markdown') - else: - await self._tg_bot.send_document(chat_id=chat_id, document=buf, caption=caption, parse_mode='Markdown') + async def _send_text(self, chat_id: int, text: str, reply=None): + sent = [] + for chunk in self._split(text, TEXT_LIMIT): + try: + msg = await self._tg_bot.send_message( + chat_id=chat_id, text=chunk, parse_mode="HTML", + reply_parameters=reply) + except TelegramError as exc: + # last resort: drop formatting rather than lose the message + log.warning("HTML send failed (%s) — retrying as plain text", exc) + self.last_error = str(exc) + try: + msg = await self._tg_bot.send_message( + chat_id=chat_id, text=_strip_tags(chunk), + reply_parameters=reply) + except TelegramError as exc2: + log.error("plain-text send also failed: %s", exc2) + self.last_error = str(exc2) + continue + sent.append(msg) + reply = None + return sent - except TelegramError as e: - print(f'[D->TG] send_attachment failed ({attachment.filename}): {e}') - except Exception as e: - print(f'[D->TG] unexpected error for attachment {attachment.filename}: {e}') + # ── attachments ───────────────────────────────────────────────────────── - # -- event handler -------------------------------------------------------- + @staticmethod + def _kind(attachment: discord.Attachment) -> str: + mime = (attachment.content_type or "").lower() + name = (attachment.filename or "").lower() + if mime.startswith("image/gif") or name.endswith(".gif"): + return "animation" + if mime.startswith("image/"): + return "photo" + if mime.startswith("video/"): + return "video" + if mime.startswith("audio/"): + return "audio" + return "document" + + async def _buffer(self, attachment: discord.Attachment): + buf = io.BytesIO(await attachment.read()) + buf.name = attachment.filename + return buf + + async def _send_attachments(self, chat_id, attachments, caption, reply): + """Send attachments, batching into media groups where allowed.""" + # photos and videos can share a group; documents and audio cannot + visual, documents, audio, animations = [], [], [], [] + for attachment in attachments: + kind = self._kind(attachment) + (visual if kind in ("photo", "video") + else audio if kind == "audio" + else animations if kind == "animation" + else documents).append((kind, attachment)) + + sent = [] + caption_left = caption + + async def flush_group(items, factory): + nonlocal caption_left, reply + for batch in _chunks(items, MEDIA_GROUP_MAX): + # read each attachment once and reuse the buffer + prepared = [] + for index, (kind, attachment) in enumerate(batch): + buf = await self._buffer(attachment) + caption = None + if index == 0 and caption_left: + caption = caption_left[:CAPTION_LIMIT] + caption_left = None + prepared.append((kind, buf, caption)) + try: + if len(prepared) == 1: + kind, buf, caption = prepared[0] + sent.extend(await self._send_single( + chat_id, kind, buf, caption, reply)) + else: + media = [] + for kind, buf, caption in prepared: + kwargs = {"media": buf} + if caption: + kwargs["caption"] = caption + kwargs["parse_mode"] = "HTML" + media.append(factory(kind, kwargs)) + sent.extend(await self._tg_bot.send_media_group( + chat_id=chat_id, media=media, reply_parameters=reply)) + reply = None + except TelegramError as exc: + log.error("attachment send failed: %s", exc) + self.last_error = str(exc) + + await flush_group(visual, lambda kind, kw: + InputMediaPhoto(**kw) if kind == "photo" + else InputMediaVideo(**kw)) + await flush_group(animations, lambda kind, kw: InputMediaDocument(**kw)) + await flush_group(documents, lambda kind, kw: InputMediaDocument(**kw)) + await flush_group(audio, lambda kind, kw: InputMediaAudio(**kw)) + + if caption_left: + sent.extend(await self._send_text(chat_id, caption_left, reply)) + return sent + + async def _send_single(self, chat_id, kind, buf, caption, reply): + """One attachment on its own — use the richest endpoint for its type.""" + common = {"chat_id": chat_id, "caption": caption, + "parse_mode": "HTML" if caption else None, + "reply_parameters": reply} + if kind == "photo": + return [await self._tg_bot.send_photo(photo=buf, **common)] + if kind == "video": + return [await self._tg_bot.send_video(video=buf, **common)] + if kind == "audio": + return [await self._tg_bot.send_audio(audio=buf, **common)] + if kind == "animation": + return [await self._tg_bot.send_animation(animation=buf, **common)] + return [await self._tg_bot.send_document(document=buf, **common)] + + async def _send_stickers(self, chat_id, stickers, reply): + """Discord stickers have no telegram equivalent, send what we can.""" + sent = [] + for sticker in stickers: + try: + fmt = getattr(sticker, "format", None) + name = getattr(fmt, "name", "") or "" + if name.lower() == "lottie": + sent.extend(await self._send_text( + chat_id, f"(sticker: {html.escape(sticker.name)})", + reply)) + else: + data = await sticker.read() + buf = io.BytesIO(data) + buf.name = f"{sticker.name}.{'gif' if name.lower() == 'gif' else 'png'}" + sent.append(await self._tg_bot.send_document( + chat_id=chat_id, document=buf, reply_parameters=reply)) + reply = None + except Exception as exc: + log.warning("sticker forward failed: %s", exc) + return sent + + # ── events ────────────────────────────────────────────────────────────── @commands.Cog.listener() async def on_message(self, message: discord.Message): - # ignore bots (including ourselves) and webhook posts to prevent echo loops + # ignoring bots and webhooks is what breaks the echo loop if message.author.bot or message.webhook_id: return - chat_id = self.routes.get(message.channel.id) if chat_id is None: return - header = self._format_header(message) - body = message.content or '' + try: + reply = await self._reply_params(message) + header = self._header(message) + body = self._body(message) + caption = f"{header}\n{body}".strip() if body else header - # text portion - if body or not message.attachments: - full_text = f"{header}\n{body}".strip() - await self._send_text(chat_id, full_text) - - # attachments - for i, attachment in enumerate(message.attachments): - # first attachment gets the header as caption; rest are bare - if i == 0: - caption = header if body else f"{header}\n{body}".strip() + sent = [] + if message.attachments: + sent += await self._send_attachments( + chat_id, message.attachments, caption, reply) + elif getattr(message, "stickers", None): + sent += await self._send_text(chat_id, caption, reply) + sent += await self._send_stickers(chat_id, message.stickers, None) else: - caption = '' - await self._send_attachment(chat_id, attachment, caption) + sent += await self._send_text(chat_id, caption, reply) - print(f'[D->TG] {message.author} (channel {message.channel.id}) -> tg chat {chat_id}') + if sent: + await store.link(chat_id, [m.message_id for m in sent], + message.channel.id, message.id, "dc") + self.forwarded += 1 + log.info("[D->TG] %s (channel %s) -> chat %s", + message.author, message.channel.id, chat_id) + except Exception as exc: + self.last_error = f"{type(exc).__name__}: {exc}" + log.exception("[D->TG] forward failed") + + @commands.Cog.listener() + async def on_raw_message_edit(self, payload: discord.RawMessageUpdateEvent): + # raw, not on_message_edit: the cached one never fires for messages + # sent before the last restart + chat_id = self.routes.get(payload.channel_id) + if chat_id is None: + return + data = payload.data or {} + if data.get("author", {}).get("bot") or data.get("webhook_id"): + return + content = data.get("content") + if content is None: + return # embed-only update, not an edit + + try: + mapped = await store.tg_for_dc(payload.channel_id, payload.message_id) + if not mapped: + return + _, tg_msg_id = mapped + + message = payload.cached_message + author = (message.author.display_name if message + else data.get("author", {}).get("global_name") + or data.get("author", {}).get("username") or "unknown") + resolved = entities.resolve_discord_references(message, content) \ + if message else content + text = (f"{html.escape(str(author))}\n" + f"{entities.to_telegram_html(resolved)}\n" + f"(edited)")[:TEXT_LIMIT] + try: + await self._tg_bot.edit_message_text( + chat_id=chat_id, message_id=tg_msg_id, text=text, + parse_mode="HTML") + except TelegramError: + # the original was media, so the text is in the caption + await self._tg_bot.edit_message_caption( + chat_id=chat_id, message_id=tg_msg_id, + caption=text[:CAPTION_LIMIT], parse_mode="HTML") + except TelegramError as exc: + log.debug("[D->TG] edit sync failed: %s", exc) + except Exception: + log.exception("[D->TG] edit sync error") + + @commands.Cog.listener() + async def on_raw_message_delete(self, payload: discord.RawMessageDeleteEvent): + await self._delete(payload.channel_id, [payload.message_id]) + + @commands.Cog.listener() + async def on_raw_bulk_message_delete( + self, payload: discord.RawBulkMessageDeleteEvent): + await self._delete(payload.channel_id, list(payload.message_ids)) + + async def _delete(self, channel_id: int, message_ids: list[int]): + chat_id = self.routes.get(channel_id) + if chat_id is None: + return + for message_id in message_ids: + try: + mapped = await store.tg_for_dc(channel_id, message_id) + if not mapped: + continue + await self._tg_bot.delete_message(chat_id=chat_id, + message_id=mapped[1]) + except TelegramError as exc: + # telegram refuses deletes older than 48h, which is normal + log.debug("[D->TG] delete sync skipped: %s", exc) + except Exception: + log.exception("[D->TG] delete sync error") + + +def _chunks(items, size): + for index in range(0, len(items), size): + yield items[index:index + size] + + +def _strip_tags(text: str) -> str: + import re + return html.unescape(re.sub(r"<[^>]+>", "", text)) diff --git a/tgbridge/entities.py b/tgbridge/entities.py new file mode 100644 index 0000000..1486de6 --- /dev/null +++ b/tgbridge/entities.py @@ -0,0 +1,261 @@ +#telegram entities <-> discord markdown <-> telegram html## +"""Rich-text conversion between Telegram and Discord. + +Telegram offsets count UTF-16 code units, not Python chars — one emoji shifts +every later offset, so we slice a utf-16-le buffer and decode once at the end. +""" +###################################################################### +import html +import logging +import re + +log = logging.getLogger(__name__) + +# characters that would otherwise be interpreted as Discord markdown +_DISCORD_SPECIAL = set("\\*_~`|") + +# Telegram entity type -> (opening marker, closing marker) +_SIMPLE_MARKERS = { + "bold": ("**", "**"), + "italic": ("*", "*"), + "underline": ("__", "__"), + "strikethrough": ("~~", "~~"), + "spoiler": ("||", "||"), + "code": ("`", "`"), +} + + +def _utf16_units(text: str) -> list[bytes]: + """Split text into UTF-16 code units — the unit Telegram offsets count.""" + raw = text.encode("utf-16-le") + return [raw[i:i + 2] for i in range(0, len(raw), 2)] + + +def _ascii_char(unit: bytes): + """Return the ASCII character for a code unit, or None if it isn't ASCII.""" + if unit[1] == 0 and unit[0] < 128: + return chr(unit[0]) + return None + + +def to_discord_markdown(text: str, entities) -> str: + """Render Telegram text + entities as Discord markdown.""" + if not text: + return "" + try: + return _convert(text, entities or []) + except Exception as exc: # pragma: no cover - safety net + log.warning("entity conversion failed, sending plain text: %s", exc) + return escape_discord(text) + + +def escape_discord(text: str) -> str: + """Neutralise Discord markdown in an untrusted string.""" + out = [] + at_line_start = True + for ch in text: + if ch in _DISCORD_SPECIAL: + out.append("\\" + ch) + elif ch == ">" and at_line_start: + out.append("\\>") + else: + out.append(ch) + at_line_start = ch == "\n" + return "".join(out) + + +def _markers_for(entity): + """-> (open, close, is_literal, is_quote) for one Telegram entity.""" + etype = entity.type + if etype in _SIMPLE_MARKERS: + opener, closer = _SIMPLE_MARKERS[etype] + return opener, closer, etype == "code", False + if etype == "pre": + lang = getattr(entity, "language", None) or "" + return f"```{lang}\n", "\n```", True, False + if etype == "text_link": + return "[", f"]({entity.url})", False, False + if etype == "text_mention": + # a mention of a Telegram user with no public @username + return "**", "**", False, False + if etype in ("blockquote", "expandable_blockquote"): + return "> ", "", False, True + # url/mention/hashtag/email/custom_emoji need no markup of their own + return None, None, False, False + + +def _convert(text: str, entities) -> str: + units = _utf16_units(text) + total = len(units) + + # events[i] = markers to emit immediately before code unit i + opens: dict[int, list] = {} + closes: dict[int, list] = {} + + for index, ent in enumerate(entities): + opener, closer, literal, quote = _markers_for(ent) + if opener is None: + continue + start = ent.offset + end = ent.offset + ent.length + if start < 0 or end > total or start >= end: + continue + # longer spans open first; the index makes closing LIFO so two + # entities on the same span don't emit crossed tags + opens.setdefault(start, []).append((-ent.length, index, opener, literal, quote)) + closes.setdefault(end, []).append((ent.length, -index, closer, literal, quote)) + + out = bytearray() + literal_depth = 0 + quote_depth = 0 + at_line_start = True + + def emit(s: str): + out.extend(s.encode("utf-16-le")) + + for i in range(total + 1): + for _, _, closer, literal, quote in sorted(closes.get(i, [])): + if closer: + emit(closer) + if literal: + literal_depth -= 1 + if quote: + quote_depth -= 1 + for _, _, opener, literal, quote in sorted(opens.get(i, [])): + if literal: + literal_depth += 1 + if quote: + quote_depth += 1 + if opener: + emit(opener) + at_line_start = False + + if i == total: + break + + unit = units[i] + ch = _ascii_char(unit) + if ch == "\n" and quote_depth: + # keep every line of a quote inside the quote + emit("\n" + "> " * quote_depth) + at_line_start = False + elif literal_depth or ch is None: + out.extend(unit) + at_line_start = False + elif ch in _DISCORD_SPECIAL: + emit("\\" + ch) + at_line_start = False + elif ch == ">" and at_line_start: + emit("\\>") + at_line_start = False + else: + out.extend(unit) + at_line_start = ch == "\n" + + return bytes(out).decode("utf-16-le") + + +# ── discord → telegram ───────────────────────────────────────────────────── +# telegram's Markdown mode 400s on any unpaired * _ ` or [ — html is the only +# parse mode where arbitrary user text can be escaped safely + +_CODE_BLOCK = re.compile(r"```(?:([a-zA-Z0-9+#-]*)\n)?(.*?)```", re.DOTALL) +_INLINE_CODE = re.compile(r"`([^`\n]+)`") +_MD_LINK = re.compile(r"\[([^\]]+)\]\((https?://[^\s)]+)\)") +_BOLD = re.compile(r"\*\*(.+?)\*\*", re.DOTALL) +_UNDERLINE = re.compile(r"__(.+?)__", re.DOTALL) +_STRIKE = re.compile(r"~~(.+?)~~", re.DOTALL) +_SPOILER = re.compile(r"\|\|(.+?)\|\|", re.DOTALL) +# the leading \S keeps "* * *" from being read as an empty italic span +_ITALIC = re.compile(r"(?") +_USER_MENTION = re.compile(r"<@!?(\d+)>") +_ROLE_MENTION = re.compile(r"<@&(\d+)>") +_CHANNEL_MENTION = re.compile(r"<#(\d+)>") +_TIMESTAMP = re.compile(r"") + + +def resolve_discord_references(message, content: str) -> str: + """Replace raw <@id> / <#id> / <:emoji:id> markup with readable names.""" + if not content: + return "" + guild = getattr(message, "guild", None) + + def user_repl(match): + uid = int(match.group(1)) + member = guild.get_member(uid) if guild else None + if member is None: + for user in getattr(message, "mentions", []): + if user.id == uid: + member = user + break + return f"@{member.display_name}" if member else "@unknown" + + def role_repl(match): + role = guild.get_role(int(match.group(1))) if guild else None + return f"@{role.name}" if role else "@role" + + def channel_repl(match): + chan = guild.get_channel(int(match.group(1))) if guild else None + return f"#{chan.name}" if chan else "#channel" + + content = _USER_MENTION.sub(user_repl, content) + content = _ROLE_MENTION.sub(role_repl, content) + content = _CHANNEL_MENTION.sub(channel_repl, content) + content = _CUSTOM_EMOJI.sub(lambda m: f":{m.group(1)}:", content) + content = _TIMESTAMP.sub(lambda m: f"", content) + return content + + +def to_telegram_html(content: str) -> str: + """Convert Discord markdown to the small HTML subset Telegram accepts.""" + if not content: + return "" + try: + return _to_html(content) + except Exception as exc: # pragma: no cover - safety net + log.warning("markdown conversion failed, sending escaped text: %s", exc) + return html.escape(content) + + +def _to_html(content: str) -> str: + # park code behind placeholders so its contents aren't read as markdown + stash: list[str] = [] + + def park(rendered: str) -> str: + stash.append(rendered) + return f"\x00{len(stash) - 1}\x00" + + def code_block(match): + lang, body = match.group(1), match.group(2) + body = html.escape(body.strip("\n")) + if lang: + return park(f'
{body}
') + return park(f"
{body}
") + + content = _CODE_BLOCK.sub(code_block, content) + content = _INLINE_CODE.sub( + lambda m: park(f"{html.escape(m.group(1))}"), content) + + # a link's url must not be escaped as text + content = _MD_LINK.sub( + lambda m: park(f'' + f"{html.escape(m.group(1))}"), content) + + content = html.escape(content) + + content = _BOLD.sub(r"\1", content) + content = _UNDERLINE.sub(r"\1", content) + content = _STRIKE.sub(r"\1", content) + content = _SPOILER.sub(r'\1', content) + content = _ITALIC.sub(r"\1", content) + content = _ITALIC_US.sub(r"\1", content) + + # telegram has no inline quote tag, so keep a visual marker + content = re.sub(r"(?m)^>\s?", "| ", content) + + for index, rendered in enumerate(stash): + content = content.replace(f"\x00{index}\x00", rendered) + return content diff --git a/tgbridge/forwarder.py b/tgbridge/forwarder.py index 282440c..71e266b 100644 --- a/tgbridge/forwarder.py +++ b/tgbridge/forwarder.py @@ -1,261 +1,484 @@ -# this is where telegram-to-discord forwarding lives -import io -import json -import aiohttp -from telegram.ext import Application, MessageHandler, filters +#this is where telegram-to-discord forwarding lives## +"""Telegram -> Discord forwarding.""" +###################################################################### import asyncio -import tempfile -import os -import shutil +import io +import logging +import time + +from telegram.ext import Application, MessageHandler, filters + +from . import config, entities, media, store +from .avatars import AvatarCache +from .webhook import WebhookClient + +log = logging.getLogger(__name__) + +# an album arrives as several updates; wait this long after the last one +ALBUM_DEBOUNCE = 1.5 +ALBUM_MAX_FILES = 10 +DOWNLOAD_TIMEOUT = 90.0 +REPLY_SNIPPET = 120 + class TelegramForwarder: - def __init__(self, token): + def __init__(self, token, discord_bot=None, avatar_channel_id=None, + routes=None, drop_pending=True): self.token = token - self.routes = {} + self.bot = discord_bot + self.drop_pending = drop_pending + # chat_id -> Route + self.routes: dict[int, config.Route] = {} + for route in routes or []: + if route.tg_to_dc: + self.routes[route.tg_chat_id] = route + + self.webhook = WebhookClient() + self.avatars = AvatarCache(discord_bot, avatar_channel_id) self._app = None + self._running = False + self._albums: dict[str, dict] = {} + self.started_at = None + self.forwarded = 0 + self.last_error = None + + # ── lifecycle ─────────────────────────────────────────────────────────── def add_route(self, group_id, webhook_url): - self.routes[group_id] = webhook_url + """Back-compatible single-route registration.""" + existing = self.routes.get(int(group_id)) + if existing: + existing.webhook_url = webhook_url + else: + self.routes[int(group_id)] = config.Route( + name=str(group_id), tg_chat_id=int(group_id), + webhook_url=webhook_url) async def start(self): - self._app = Application.builder().token(self.token).build() - self._app.add_handler(MessageHandler(filters.ALL, self._on_message)) + # on_ready fires again on every reconnect; a second poller on the same + # token means telegram 409s and duplicated posts + if self._running: + log.debug("telegram forwarder already running — start() ignored") + return + self._running = True + try: + await store.init_db() + removed = await store.prune() + if removed: + log.info("pruned %d expired message mappings", removed) - await self._app.initialize() - await self._app.start() - await self._app.updater.start_polling(drop_pending_updates=True) - print(f'telegram forwarder started — watching {len(self.routes)} groups') + self._app = Application.builder().token(self.token).build() + self._app.add_handler(MessageHandler(filters.ALL, self._on_update)) + self._app.add_error_handler(self._on_error) + + await self._app.initialize() + await self._app.start() + await self._app.updater.start_polling( + drop_pending_updates=self.drop_pending) + + self.avatars.bind_telegram(self._app.bot) + self.started_at = time.time() + log.info("telegram forwarder started — watching %d chat(s), " + "avatars %s", len(self.routes), + "on" if self.avatars.enabled else "off") + except Exception: + self._running = False + raise async def stop(self): + self._running = False + for pending in list(self._albums.values()): + task = pending.get("task") + if task: + task.cancel() + self._albums.clear() if self._app: - await self._app.updater.stop() - await self._app.stop() - await self._app.shutdown() - print('telegram forwarder stopped') + try: + if self._app.updater.running: + await self._app.updater.stop() + await self._app.stop() + await self._app.shutdown() + except Exception as exc: + log.warning("error during telegram shutdown: %s", exc) + self._app = None + await self.webhook.close() + log.info("telegram forwarder stopped") - # internal helpers - def _sender_name(self, update): - user = update.effective_user + async def _on_error(self, update, context): + self.last_error = str(context.error) + log.error("telegram update failed: %s", context.error) + + # ── naming ────────────────────────────────────────────────────────────── + + def _sender_name(self, message) -> str: + user = getattr(message, "from_user", None) if user: - name = user.first_name or '' + name = (user.first_name or "").strip() if user.last_name: - name += f' {user.last_name}' - return name.strip() or user.username or 'unknown' - - # channel posts have no user — use chat title - chat = update.effective_chat - if chat and chat.title: + name = f"{name} {user.last_name}".strip() + return name or user.username or "unknown" + chat = getattr(message, "sender_chat", None) or getattr(message, "chat", None) + if chat and getattr(chat, "title", None): return chat.title - return 'unknown' + return "unknown" + # ── downloads ─────────────────────────────────────────────────────────── - - async def _download_file(self, file_id): + async def _download(self, file_id: str): + """-> (bytes, filename). Raises on failure.""" tg_file = await asyncio.wait_for( - self._app.bot.get_file(file_id), timeout=30.0 - ) + self._app.bot.get_file(file_id), timeout=30.0) buf = io.BytesIO() await asyncio.wait_for( - tg_file.download_to_memory(buf), timeout=60.0 - ) - buf.seek(0) - filename = tg_file.file_path.split('/')[-1] if tg_file.file_path else 'file' - return buf.read(), filename + tg_file.download_to_memory(buf), timeout=DOWNLOAD_TIMEOUT) + path = tg_file.file_path or "" + return buf.getvalue(), (path.split("/")[-1] or "file") - async def _send_to_webhook(self, webhook_url, username, content='', file_data=None, filename=None): - async with aiohttp.ClientSession() as session: - data = aiohttp.FormData() - payload = {'username': username} - if content: - payload['content'] = content[:2000] + # ── context lines ─────────────────────────────────────────────────────── - if file_data and filename: - data.add_field('payload_json', json.dumps(payload), content_type='application/json') - data.add_field('file', file_data, filename=filename) + async def _reply_line(self, message, route) -> str: + """Render a quote of whatever this message is replying to.""" + replied = getattr(message, "reply_to_message", None) + if not replied: + return "" + try: + who = self._sender_name(replied) + snippet = (replied.text or replied.caption or "").strip() + if not snippet: + snippet = _describe_media(replied) + snippet = " ".join(snippet.split()) + if len(snippet) > REPLY_SNIPPET: + snippet = snippet[:REPLY_SNIPPET].rstrip() + "…" + snippet = entities.escape_discord(snippet) + who = entities.escape_discord(who) + + link = await self._jump_link(route, replied.message_id) + arrow = f"[↩ reply to **{who}**]({link})" if link else f"↩ reply to **{who}**" + return f"> {arrow}: {snippet}\n" if snippet else f"> {arrow}\n" + except Exception as exc: + log.debug("could not render reply context: %s", exc) + return "" + + async def _jump_link(self, route, tg_msg_id: int): + """Link to the Discord copy of a Telegram message, if we forwarded it.""" + if not self.bot: + return None + try: + mapped = await store.dc_for_tg(route.tg_chat_id, tg_msg_id) + if not mapped: + return None + channel_id, message_id = mapped + channel = self.bot.get_channel(channel_id) + guild_id = channel.guild.id if channel and channel.guild else None + if not guild_id: + return None + return (f"https://discord.com/channels/" + f"{guild_id}/{channel_id}/{message_id}") + except Exception: + return None + + def _forward_line(self, message) -> str: + """Note where a forwarded message originally came from.""" + try: + origin = getattr(message, "forward_origin", None) + source = None + if origin is not None: + otype = getattr(origin, "type", "") + if otype == "user" and getattr(origin, "sender_user", None): + user = origin.sender_user + source = (user.first_name or "") + ( + f" {user.last_name}" if user.last_name else "") + elif otype == "hidden_user": + source = getattr(origin, "sender_user_name", None) + elif otype == "chat" and getattr(origin, "sender_chat", None): + source = origin.sender_chat.title + elif otype == "channel" and getattr(origin, "chat", None): + source = origin.chat.title + if not source: + return "" + return f"-# ↪ forwarded from {entities.escape_discord(source.strip())}\n" + except Exception: + return "" + + # ── rendering ─────────────────────────────────────────────────────────── + + async def _render_text(self, message) -> str: + body = message.text or message.caption or "" + ents = message.entities if message.text else message.caption_entities + return entities.to_discord_markdown(body, ents) + + async def _render_sticker(self, message): + """-> (files, extra_text), per sticker format.""" + sticker = message.sticker + emoji = sticker.emoji or "" + cached = await store.get_media(sticker.file_unique_id) + if cached: + return [cached], emoji + + data, filename = await self._download(sticker.file_id) + stem = filename.rsplit(".", 1)[0] or "sticker" + + if sticker.is_video: + # .webm vp9 with alpha -> animated webp keeps the transparency + converted = await media.to_animated_webp(data, ".webm") + if converted: + out = (f"{stem}.webp", converted) + await store.put_media(sticker.file_unique_id, *out) + return [out], emoji + return [(filename, data)], emoji + + if sticker.is_animated: + # .tgs is gzipped lottie, ffmpeg cannot read it + rendered = await media.render_tgs(data) + if rendered: + out = (f"{stem}.gif", rendered) + await store.put_media(sticker.file_unique_id, *out) + return [out], emoji + thumb = getattr(sticker, "thumbnail", None) + if thumb: + try: + tdata, tname = await self._download(thumb.file_id) + return [(tname, tdata)], emoji + except Exception as exc: + log.debug("sticker thumbnail failed: %s", exc) + return [], f"{emoji} *(animated sticker)*".strip() + + # static .webp renders natively with full alpha; converting to gif + # only destroyed the antialiased edges + return [(filename, data)], emoji + + async def _render_media(self, message): + """-> (files, extra_text) for whatever media this message carries.""" + if message.sticker: + return await self._render_sticker(message) + + if message.photo: + data, filename = await self._download(message.photo[-1].file_id) + return [(filename, data)], "" + + animation = getattr(message, "animation", None) + document = getattr(message, "document", None) + # mime_type is nullable + doc_mime = (getattr(document, "mime_type", None) or "") if document else "" + + if animation or message.video or doc_mime.startswith("video"): + if animation: + file_id, is_animation = animation.file_id, True + elif message.video: + file_id, is_animation = message.video.file_id, False else: - async with session.post(webhook_url, json=payload) as resp: - if resp.status not in (200, 204): - print(f'webhook post failed: {resp.status} {await resp.text()}') - return + file_id, is_animation = document.file_id, False - async with session.post(webhook_url, data=data) as resp: - if resp.status not in (200, 204): - print(f'webhook post failed: {resp.status} {await resp.text()}') + data, filename = await self._download(file_id) + suffix = "." + filename.rsplit(".", 1)[-1] if "." in filename else ".mp4" - # main message handler - async def _on_message(self, update, context): + # animations never have audio; elsewhere a silent clip is a gif in + # spirit and should loop, one with sound is a real video + convert = is_animation or not await media.has_audio_track(data, suffix) + if convert: + gif = await media.to_gif(data, suffix) + if gif: + return [(filename.rsplit(".", 1)[0] + ".gif", gif)], "" + return [(filename, data)], "" + + if document: + data, filename = await self._download(document.file_id) + return [(filename, data)], "" + + if message.voice: + data, filename = await self._download(message.voice.file_id) + return [(filename, data)], f"-# 🎤 voice message ({_duration(message.voice)})" + + if message.video_note: + data, filename = await self._download(message.video_note.file_id) + return [(filename, data)], f"-# 📹 video note ({_duration(message.video_note)})" + + if message.audio: + audio = message.audio + data, filename = await self._download(audio.file_id) + label = " — ".join(x for x in (audio.performer, audio.title) if x) + return [(filename, data)], f"-# 🎵 {label}" if label else "" + + return [], _describe_nonmedia(message) + + # ── dispatch ──────────────────────────────────────────────────────────── + + async def _on_update(self, update, context): message = update.effective_message if not message: return - - chat_id = update.effective_chat.id if update.effective_chat else None - if chat_id not in self.routes: + chat = update.effective_chat + route = self.routes.get(chat.id) if chat else None + if route is None: return - webhook_url = self.routes[chat_id] - sender = self._sender_name(update) - caption = message.caption or '' + # echo guard: never re-forward what the bridge itself posted + me = getattr(self._app.bot, "id", None) + sender = getattr(message, "from_user", None) + if me and sender and sender.id == me: + return - # Log incoming message media types for debugging - print(f"[HANDLER] Types - photo:{bool(message.photo)} video:{bool(message.video)} animation:{bool(message.animation)} document:{bool(message.document)}", flush=True) + is_edit = bool(update.edited_message or update.edited_channel_post) try: - if message.photo: - photo = message.photo[-1] - file_data, filename = await self._download_file(photo.file_id) - text = caption.strip() - await self._send_to_webhook(webhook_url, sender, text, file_data, filename) - - # video and document handling are now managed by the enhanced GIF block below - - # --- START OF ENHANCED GIF HANDLING BLOCK --- - # Telegram may send a GIF as an `animation` (the normal case) *or* - # as a muted `video` / `document` with a video MIME type. - # Videos with an audio track are real videos → sent as-is. - # Silent videos are treated as GIFs → converted for Discord auto-loop. - elif getattr(message, "animation", None) or ( - getattr(message, "document", None) and message.document.mime_type.startswith("video") - ) or message.video: - if message.animation: - file_id = message.animation.file_id - is_animation = True - elif message.video: - file_id = message.video.file_id - is_animation = False - else: - file_id = message.document.file_id - is_animation = False - - print(f"[GIF-HANDLER] Detected potential media (file_id={file_id})", flush=True) - file_data, filename = await self._download_file(file_id) - - # Probe for audio track to decide: real video or GIF? - convert_to_gif = is_animation # animations never have audio - - if not is_animation and shutil.which('ffprobe'): - tmp_path = None - try: - with tempfile.NamedTemporaryFile(suffix='.mp4', delete=False) as tmp: - tmp.write(file_data) - tmp_path = tmp.name - - probe = await asyncio.create_subprocess_exec( - 'ffprobe', '-v', 'error', - '-select_streams', 'a', - '-show_entries', 'stream=codec_type', - '-of', 'csv=p=0', - tmp_path, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE - ) - stdout, _ = await asyncio.wait_for(probe.communicate(), timeout=15.0) - - # ffprobe outputs "audio" if an audio stream is present - convert_to_gif = b'audio' not in stdout - if not convert_to_gif: - print("[GIF-HANDLER] Audio track detected — sending as video", flush=True) - except Exception as e: - print(f"[GIF-HANDLER] ffprobe failed, defaulting to convert: {e}", flush=True) - convert_to_gif = True - finally: - if tmp_path and os.path.exists(tmp_path): - os.remove(tmp_path) - - if convert_to_gif: - # >>> FFMPEG GIF CONVERSION START >>> - try: - if not shutil.which('ffmpeg'): - raise RuntimeError("ffmpeg binary not found in PATH") - - with tempfile.NamedTemporaryFile(suffix='.mp4', delete=False) as temp_mp4: - temp_mp4.write(file_data) - temp_mp4_path = temp_mp4.name - - temp_gif_path = temp_mp4_path + '.gif' - - process = await asyncio.create_subprocess_exec( - 'ffmpeg', '-y', '-i', temp_mp4_path, - '-filter_complex', 'fps=15,scale=320:-1:flags=lanczos,split[s0][s1];[s0]palettegen[p];[s1][p]paletteuse', - '-loop', '0', temp_gif_path, - stdout=asyncio.subprocess.DEVNULL, - stderr=asyncio.subprocess.PIPE - ) - - stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=30.0) - - if process.returncode == 0 and os.path.exists(temp_gif_path): - gif_size = os.path.getsize(temp_gif_path) - if gif_size < 10 * 1024 * 1024: - with open(temp_gif_path, 'rb') as f: - file_data = f.read() - filename = filename.rsplit('.', 1)[0] + '.gif' - else: - print(f"[GIF-HANDLER] ffmpeg error: {stderr.decode()}", flush=True) - - if os.path.exists(temp_mp4_path): os.remove(temp_mp4_path) - if os.path.exists(temp_gif_path): os.remove(temp_gif_path) - except Exception as e: - print(f'ffmpeg conversion skipped/failed: {e}', flush=True) - # <<< FFMPEG GIF CONVERSION END <<< - - text = caption.strip() - await self._send_to_webhook(webhook_url, sender, text, file_data, filename) - - elif message.document: - file_data, filename = await self._download_file(message.document.file_id) - text = caption.strip() - await self._send_to_webhook(webhook_url, sender, text, file_data, filename) - - elif message.sticker: - sticker_text = message.sticker.emoji or '(sticker)' - text = sticker_text - # >>> STICKER TO GIF CONVERSION START >>> - file_data, filename = await self._download_file(message.sticker.file_id) - try: - # Determine original file suffix for temporary file - suffix = os.path.splitext(filename)[1] or '.dat' - with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as temp_in: - temp_in.write(file_data) - temp_in_path = temp_in.name - - temp_gif_path = temp_in_path + '.gif' - - process = await asyncio.create_subprocess_exec( - 'ffmpeg', '-y', '-i', temp_in_path, - '-filter_complex', 'fps=15,scale=320:-1:flags=lanczos,split[s0][s1];[s0]palettegen[p];[s1][p]paletteuse', - '-loop', '0', temp_gif_path, - stdout=asyncio.subprocess.DEVNULL, - stderr=asyncio.subprocess.PIPE - ) - - stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=30.0) - - if process.returncode == 0 and os.path.exists(temp_gif_path): - gif_size = os.path.getsize(temp_gif_path) - if gif_size < 10 * 1024 * 1024: - with open(temp_gif_path, 'rb') as f: - file_data = f.read() - filename = filename.rsplit('.', 1)[0] + '.gif' - else: - print('[STICKER-HANDLER] Converted GIF exceeds size limit; sending original file', flush=True) - else: - print(f"[STICKER-HANDLER] ffmpeg error: {stderr.decode()}", flush=True) - except Exception as e: - print(f'Sticker conversion skipped/failed: {e}', flush=True) - finally: - if os.path.exists(temp_in_path): - os.remove(temp_in_path) - if os.path.exists(temp_gif_path): - os.remove(temp_gif_path) - await self._send_to_webhook(webhook_url, sender, text, file_data, filename) - # <<< STICKER TO GIF CONVERSION END <<< - - elif message.text: - text = message.text - await self._send_to_webhook(webhook_url, sender, text) - + if is_edit: + await self._handle_edit(route, message) + elif message.media_group_id: + self._buffer_album(route, message) else: - text = '*(unsupported message type)*' - await self._send_to_webhook(webhook_url, sender, text) + await self._forward(route, [message]) + except Exception as exc: + self.last_error = f"{type(exc).__name__}: {exc}" + log.exception("failed to forward message from %s", + self._sender_name(message)) - except Exception as e: - print(f'failed to forward message from {sender}: {e}') \ No newline at end of file + def _buffer_album(self, route, message): + key = f"{route.tg_chat_id}:{message.media_group_id}" + pending = self._albums.get(key) + if pending is None: + pending = self._albums[key] = {"messages": [], "task": None, + "route": route} + pending["messages"].append(message) + if pending["task"]: + pending["task"].cancel() + pending["task"] = asyncio.create_task(self._flush_album(key)) + + async def _flush_album(self, key): + try: + await asyncio.sleep(ALBUM_DEBOUNCE) + except asyncio.CancelledError: + return + pending = self._albums.pop(key, None) + if not pending or not pending["messages"]: + return + messages = sorted(pending["messages"], key=lambda m: m.message_id) + try: + await self._forward(pending["route"], messages) + except Exception as exc: + self.last_error = f"{type(exc).__name__}: {exc}" + log.exception("failed to forward album") + + async def _forward(self, route, messages): + """Post one or more Telegram messages as a single Discord message.""" + lead = messages[0] + sender = self._sender_name(lead) + + files, notes = [], [] + for message in messages: + try: + got, note = await self._render_media(message) + except Exception as exc: + log.warning("media render failed: %s", exc) + got, note = [], "*(attachment could not be fetched)*" + files.extend(got) + if note: + notes.append(note) + + body_parts = [] + for message in messages: + rendered = await self._render_text(message) + if rendered: + body_parts.append(rendered) + + prefix = self._forward_line(lead) + await self._reply_line(lead, route) + content = prefix + "\n".join(body_parts + notes).strip() + + if not content.strip() and not files: + return + + avatar_url = None + try: + if getattr(lead, "from_user", None): + avatar_url = await self.avatars.url_for_user(lead.from_user) + else: + avatar_url = await self.avatars.url_for_chat( + getattr(lead, "sender_chat", None) or lead.chat) + except Exception as exc: + log.debug("avatar lookup failed: %s", exc) + + result = await self.webhook.send( + route.webhook_url, sender, content, + files=files[:ALBUM_MAX_FILES] or None, avatar_url=avatar_url) + + self.forwarded += 1 + if result and result.get("id"): + # every telegram id in an album maps to the one discord post, so + # a reply to any photo in it resolves + await store.link(route.tg_chat_id, + [m.message_id for m in messages], + int(result["channel_id"]), int(result["id"]), "tg") + + async def _handle_edit(self, route, message): + mapped = await store.dc_for_tg(route.tg_chat_id, message.message_id) + if not mapped: + log.debug("edit for unmapped message %s — ignoring", message.message_id) + return + _, dc_msg_id = mapped + content = await self._render_text(message) + if not content: + return + await self.webhook.edit(route.webhook_url, dc_msg_id, + content + "\n-# *(edited)*") + + +# ── small helpers ─────────────────────────────────────────────────────────── + +def _duration(obj) -> str: + seconds = getattr(obj, "duration", 0) or 0 + return f"{seconds // 60}:{seconds % 60:02d}" + + +def _describe_media(message) -> str: + """Short label for a message quoted in a reply.""" + for attr, label in (("photo", "photo"), ("sticker", "sticker"), + ("animation", "GIF"), ("video", "video"), + ("voice", "voice message"), ("video_note", "video note"), + ("audio", "audio"), ("document", "file")): + if getattr(message, attr, None): + return f"({label})" + return "" + + +def _describe_nonmedia(message) -> str: + """The message types that carry no file at all.""" + poll = getattr(message, "poll", None) + if poll: + options = "\n".join(f"• {entities.escape_discord(o.text)}" + for o in poll.options) + return f"📊 **{entities.escape_discord(poll.question)}**\n{options}" + + location = getattr(message, "location", None) + venue = getattr(message, "venue", None) + if venue: + return (f"📍 **{entities.escape_discord(venue.title)}**\n" + f"{entities.escape_discord(venue.address)}\n" + f"") + if location: + return (f"📍 location: ") + + contact = getattr(message, "contact", None) + if contact: + name = " ".join(x for x in (contact.first_name, contact.last_name) if x) + return (f"👤 **{entities.escape_discord(name)}**\n" + f"{entities.escape_discord(contact.phone_number or '')}") + + dice = getattr(message, "dice", None) + if dice: + return f"{dice.emoji} rolled **{dice.value}**" + + if getattr(message, "new_chat_members", None): + names = ", ".join(entities.escape_discord( + u.first_name or u.username or "someone") + for u in message.new_chat_members) + return f"-# → {names} joined the group" + + left = getattr(message, "left_chat_member", None) + if left: + name = entities.escape_discord(left.first_name or left.username or "someone") + return f"-# ← {name} left the group" + + if getattr(message, "pinned_message", None): + return "-# 📌 pinned a message" + + return "" diff --git a/tgbridge/media.py b/tgbridge/media.py new file mode 100644 index 0000000..679b71d --- /dev/null +++ b/tgbridge/media.py @@ -0,0 +1,196 @@ +#ffmpeg + lottie conversion for stickers, gifs and video## +"""Media conversion for the bridge. + +Two rules here: a timed-out subprocess is always killed and reaped, and temp +paths are bound before the try so the finally can never UnboundLocalError. +""" +###################################################################### +import asyncio +import logging +import os +import shutil +import tempfile + +log = logging.getLogger(__name__) + +FFMPEG = shutil.which("ffmpeg") +FFPROBE = shutil.which("ffprobe") + +MAX_CONVERTED_BYTES = 8 * 1024 * 1024 # bigger than this, discord rejects it + +_GIF_FILTER = ("fps=15,scale=320:-1:flags=lanczos,split[s0][s1];" + "[s0]palettegen[p];[s1][p]paletteuse") + +_lottie_state = None # None = not yet checked + + +def have_ffmpeg() -> bool: + return FFMPEG is not None + + +async def _run(args, timeout: float) -> tuple[int, bytes]: + """Run a subprocess, always killing and reaping it on timeout.""" + proc = await asyncio.create_subprocess_exec( + *args, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + try: + stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout) + return proc.returncode, (stdout or b"") + (stderr or b"") + except asyncio.TimeoutError: + log.warning("%s timed out after %ss — killing", args[0], timeout) + try: + proc.kill() + except ProcessLookupError: + pass + try: + await proc.wait() + except Exception: + pass + raise + + +def _write_temp(data: bytes, suffix: str) -> str: + fd, path = tempfile.mkstemp(suffix=suffix) + with os.fdopen(fd, "wb") as handle: + handle.write(data) + return path + + +def _unlink(path): + if path and os.path.exists(path): + try: + os.remove(path) + except OSError as exc: + log.debug("could not remove temp file %s: %s", path, exc) + + +async def has_audio_track(data: bytes, suffix: str = ".mp4") -> bool: + """True if the clip carries audio — i.e. it is a real video, not a GIF.""" + if not FFPROBE: + return False + src = None + try: + src = await asyncio.to_thread(_write_temp, data, suffix) + code, out = await _run([ + FFPROBE, "-v", "error", "-select_streams", "a", + "-show_entries", "stream=codec_type", "-of", "csv=p=0", src, + ], timeout=15.0) + return b"audio" in out + except Exception as exc: + log.debug("ffprobe failed (%s) — assuming no audio", exc) + return False + finally: + await asyncio.to_thread(_unlink, src) + + +def _decoder_args(in_suffix: str) -> list[str]: + """Input flags that go before -i. + + ffmpeg's native vp9 decoder silently drops the alpha of VP9-in-WebM (every + transparent pixel comes out opaque black). Only libvpx-vp9 reads it, and + only if selected before the input. Telegram video stickers are this format. + """ + return ["-c:v", "libvpx-vp9"] if in_suffix == ".webm" else [] + + +async def _convert(data: bytes, in_suffix: str, out_suffix: str, + args, timeout: float): + """Shared conversion body. Returns bytes, or None if it did not work out.""" + if not FFMPEG: + return None + src = dst = None + try: + src = await asyncio.to_thread(_write_temp, data, in_suffix) + dst = src + out_suffix + code, out = await _run([FFMPEG, "-y", "-v", "error"] + + _decoder_args(in_suffix) + ["-i", src] + + list(args) + [dst], timeout=timeout) + if code != 0 or not os.path.exists(dst): + log.debug("ffmpeg rc=%s: %s", code, out.decode(errors="replace")[:300]) + return None + size = os.path.getsize(dst) + if size > MAX_CONVERTED_BYTES: + log.info("converted media too large (%d bytes) — keeping original", size) + return None + with open(dst, "rb") as handle: + return handle.read() + except Exception as exc: + log.warning("conversion failed: %s", exc) + return None + finally: + await asyncio.to_thread(_unlink, src) + await asyncio.to_thread(_unlink, dst) + + +async def to_animated_webp(data: bytes, in_suffix: str = ".webm"): + """Convert to animated WebP — discord loops it and it keeps 8-bit alpha. + + GIF only has on/off transparency, which is what haloed every sticker edge. + """ + return await _convert( + data, in_suffix, ".webp", + ["-c:v", "libwebp_anim", "-pix_fmt", "yuva420p", "-lossless", "0", + "-q:v", "70", "-loop", "0", "-an", "-fps_mode", "passthrough"], + timeout=45.0) + + +async def to_gif(data: bytes, in_suffix: str = ".mp4"): + """Convert a silent clip to a looping GIF (Discord autoplays these).""" + return await _convert(data, in_suffix, ".gif", + ["-filter_complex", _GIF_FILTER, "-loop", "0"], + timeout=45.0) + + +# ── lottie (.tgs animated stickers) ──────────────────────────────────────── +# .tgs is gzipped lottie json; ffmpeg exits 183 on it. optional — without the +# lib the caller falls back to the sticker's static thumbnail. + +def lottie_available() -> bool: + global _lottie_state + if _lottie_state is None: + try: + import lottie # noqa: F401 + from lottie.parsers.tgs import parse_tgs # noqa: F401 + _lottie_state = True + log.info("lottie renderer available — .tgs stickers will animate") + except Exception: + _lottie_state = False + log.info("lottie not installed — .tgs stickers fall back to thumbnail " + "(pip install lottie to animate them)") + return _lottie_state + + +def _render_tgs_sync(data: bytes): + from lottie.parsers.tgs import parse_tgs + from lottie.exporters.gif import export_gif + + src = dst = None + try: + src = _write_temp(data, ".tgs") + dst = src + ".gif" + animation = parse_tgs(src) + export_gif(animation, dst, skip_frames=2, fps=25) + if not os.path.exists(dst): + return None + if os.path.getsize(dst) > MAX_CONVERTED_BYTES: + log.info("rendered .tgs too large — falling back to thumbnail") + return None + with open(dst, "rb") as handle: + return handle.read() + finally: + _unlink(src) + _unlink(dst) + + +async def render_tgs(data: bytes): + """Render an animated .tgs sticker, or None if that isn't possible here.""" + if not lottie_available(): + return None + try: + return await asyncio.wait_for( + asyncio.to_thread(_render_tgs_sync, data), timeout=60.0) + except Exception as exc: + log.warning("tgs render failed (%s) — falling back to thumbnail", exc) + return None diff --git a/tgbridge/status_cog.py b/tgbridge/status_cog.py new file mode 100644 index 0000000..2a7d4d0 --- /dev/null +++ b/tgbridge/status_cog.py @@ -0,0 +1,99 @@ +"""!bridge — health and routing readout for the Telegram bridge.""" +###################################################################### +import logging +import time + +import discord +from discord.ext import commands + +from . import media, store + +log = logging.getLogger(__name__) + + +def _uptime(started_at) -> str: + if not started_at: + return "not running" + seconds = int(time.time() - started_at) + days, seconds = divmod(seconds, 86400) + hours, seconds = divmod(seconds, 3600) + minutes, seconds = divmod(seconds, 60) + if days: + return f"{days}d {hours}h {minutes}m" + if hours: + return f"{hours}h {minutes}m" + return f"{minutes}m {seconds}s" + + +class BridgeStatus(commands.Cog): + def __init__(self, bot, forwarder, reverse): + self.bot = bot + self.forwarder = forwarder + self.reverse = reverse + + @commands.command(name="bridge") + async def bridge(self, ctx, action: str = "status"): + """Show Telegram bridge routes, counters and health.""" + if action != "status": + await ctx.reply("usage: `!bridge status`") + return + + forwarder = self.forwarder + embed = discord.Embed( + title="Telegram bridge", + colour=discord.Colour.green() if forwarder and forwarder._running + else discord.Colour.red()) + + embed.add_field( + name="Polling", + value=("running — up " + _uptime(forwarder.started_at)) + if forwarder and forwarder._running else "stopped", + inline=True) + + avatars = forwarder.avatars if forwarder else None + embed.add_field( + name="Avatars", + value=(f"on ({avatars.hits} hits, {avatars.uploads} uploads)" + if avatars and avatars.enabled + else "off — set AVATAR_CACHE_CHANNEL_ID"), + inline=True) + + embed.add_field( + name="Lottie (.tgs)", + value="installed" if media.lottie_available() else "not installed", + inline=True) + + lines = [] + for chat_id, route in (forwarder.routes.items() if forwarder else []): + channel = self.bot.get_channel(route.dc_channel_id) \ + if route.dc_channel_id else None + target = channel.mention if channel else f"`{route.dc_channel_id}`" + arrows = "↔" if route.dc_to_tg and route.tg_to_dc else "→" + lines.append(f"**{route.name}** `{chat_id}` {arrows} {target}") + embed.add_field(name=f"Routes ({len(lines)})", + value="\n".join(lines) or "none configured", + inline=False) + + webhook = forwarder.webhook if forwarder else None + counters = [ + f"TG → Discord: **{forwarder.forwarded if forwarder else 0}**", + f"Discord → TG: **{self.reverse.forwarded if self.reverse else 0}**", + ] + if webhook: + counters.append(f"webhook posts: {webhook.sent} ok / {webhook.failed} failed") + try: + counters.append(f"cached avatars: {await store.count_avatars()}") + counters.append(f"cached stickers: {await store.count_media()}") + except Exception: + pass + embed.add_field(name="Counters", value="\n".join(counters), inline=False) + + errors = [e for e in (getattr(forwarder, "last_error", None), + getattr(self.reverse, "last_error", None), + getattr(webhook, "last_error", None)) if e] + if errors: + embed.add_field(name="Last error", + value="```" + "\n".join(errors)[:900] + "```", + inline=False) + + await ctx.reply(embed=embed) diff --git a/tgbridge/store.py b/tgbridge/store.py new file mode 100644 index 0000000..3ad524d --- /dev/null +++ b/tgbridge/store.py @@ -0,0 +1,221 @@ +#msgmap, avatar cache and sticker cache for the telegram bridge## +"""SQLite persistence for the Telegram bridge — async wrapper layer.""" +###################################################################### +import asyncio +import sqlite3 +import logging +import time +from pathlib import Path + +log = logging.getLogger(__name__) + +_PKG_DIR = Path(__file__).parent +DB_PATH = _PKG_DIR.parent / "data" / "tgbridge.db" + +MSGMAP_TTL = 60 * 60 * 24 * 30 # 30 days, then the row is pruned +AVATAR_URL_TTL = 60 * 60 * 12 # 12h — re-upload before the cdn link dies +AVATAR_RECHECK = 60 * 60 * 6 # 6h — re-ask telegram if the photo changed + + +def _connect() -> sqlite3.Connection: + return sqlite3.connect(str(DB_PATH), timeout=10) + + +def _init_db_sync() -> None: + DB_PATH.parent.mkdir(parents=True, exist_ok=True) + with _connect() as conn: + conn.execute("PRAGMA journal_mode=WAL") + conn.executescript(""" + CREATE TABLE IF NOT EXISTS msgmap ( + tg_chat_id INTEGER NOT NULL, + tg_msg_id INTEGER NOT NULL, + dc_channel_id INTEGER NOT NULL, + dc_msg_id INTEGER NOT NULL, + origin TEXT NOT NULL, + created_at INTEGER NOT NULL, + PRIMARY KEY (tg_chat_id, tg_msg_id) + ); + CREATE INDEX IF NOT EXISTS idx_msgmap_dc + ON msgmap (dc_channel_id, dc_msg_id); + CREATE INDEX IF NOT EXISTS idx_msgmap_age + ON msgmap (created_at); + + CREATE TABLE IF NOT EXISTS avatar_cache ( + tg_user_id INTEGER PRIMARY KEY, + file_unique_id TEXT, + image BLOB, + cdn_url TEXT, + uploaded_at INTEGER NOT NULL DEFAULT 0, + checked_at INTEGER NOT NULL DEFAULT 0 + ); + + CREATE TABLE IF NOT EXISTS media_cache ( + file_unique_id TEXT PRIMARY KEY, + filename TEXT NOT NULL, + data BLOB NOT NULL, + created_at INTEGER NOT NULL + ); + """) + log.info("tgbridge database initialised at %s", DB_PATH) + + +async def init_db() -> None: + await asyncio.to_thread(_init_db_sync) + + +# ── message id mapping ───────────────────────────────────────────────────── +# many-to-one: an album's N telegram ids all point at one discord message + +def _link_sync(tg_chat_id, tg_msg_ids, dc_channel_id, dc_msg_id, origin): + now = int(time.time()) + rows = [(tg_chat_id, mid, dc_channel_id, dc_msg_id, origin, now) + for mid in tg_msg_ids] + with _connect() as conn: + conn.executemany( + "INSERT OR REPLACE INTO msgmap " + "(tg_chat_id, tg_msg_id, dc_channel_id, dc_msg_id, origin, created_at) " + "VALUES (?, ?, ?, ?, ?, ?)", rows) + + +async def link(tg_chat_id: int, tg_msg_ids, dc_channel_id: int, + dc_msg_id: int, origin: str) -> None: + """Record that these Telegram message(s) correspond to one Discord message.""" + if isinstance(tg_msg_ids, int): + tg_msg_ids = [tg_msg_ids] + tg_msg_ids = [m for m in tg_msg_ids if m is not None] + if not tg_msg_ids: + return + await asyncio.to_thread(_link_sync, tg_chat_id, tg_msg_ids, + dc_channel_id, dc_msg_id, origin) + + +def _dc_for_tg_sync(tg_chat_id, tg_msg_id): + with _connect() as conn: + row = conn.execute( + "SELECT dc_channel_id, dc_msg_id FROM msgmap " + "WHERE tg_chat_id = ? AND tg_msg_id = ?", + (tg_chat_id, tg_msg_id)).fetchone() + return tuple(row) if row else None + + +async def dc_for_tg(tg_chat_id: int, tg_msg_id: int): + """-> (discord_channel_id, discord_message_id) or None.""" + return await asyncio.to_thread(_dc_for_tg_sync, tg_chat_id, tg_msg_id) + + +def _tg_for_dc_sync(dc_channel_id, dc_msg_id): + with _connect() as conn: + row = conn.execute( + "SELECT tg_chat_id, tg_msg_id FROM msgmap " + "WHERE dc_channel_id = ? AND dc_msg_id = ? " + "ORDER BY tg_msg_id ASC LIMIT 1", + (dc_channel_id, dc_msg_id)).fetchone() + return tuple(row) if row else None + + +async def tg_for_dc(dc_channel_id: int, dc_msg_id: int): + """-> (telegram_chat_id, telegram_message_id) or None.""" + return await asyncio.to_thread(_tg_for_dc_sync, dc_channel_id, dc_msg_id) + + +def _prune_sync(): + cutoff = int(time.time()) - MSGMAP_TTL + with _connect() as conn: + return conn.execute( + "DELETE FROM msgmap WHERE created_at < ?", (cutoff,)).rowcount + + +async def prune() -> int: + return await asyncio.to_thread(_prune_sync) + + +# ── avatar cache ─────────────────────────────────────────────────────────── + +def _get_avatar_sync(tg_user_id): + with _connect() as conn: + row = conn.execute( + "SELECT file_unique_id, image, cdn_url, uploaded_at, checked_at " + "FROM avatar_cache WHERE tg_user_id = ?", (tg_user_id,)).fetchone() + if not row: + return None + return {"file_unique_id": row[0], + "image": bytes(row[1]) if row[1] else None, + "cdn_url": row[2], "uploaded_at": row[3], "checked_at": row[4]} + + +async def get_avatar(tg_user_id: int): + return await asyncio.to_thread(_get_avatar_sync, tg_user_id) + + +def _put_avatar_sync(tg_user_id, file_unique_id, image, cdn_url, + uploaded_at, checked_at): + with _connect() as conn: + conn.execute( + "INSERT OR REPLACE INTO avatar_cache " + "(tg_user_id, file_unique_id, image, cdn_url, uploaded_at, checked_at) " + "VALUES (?, ?, ?, ?, ?, ?)", + (tg_user_id, file_unique_id, + sqlite3.Binary(image) if image else None, + cdn_url, uploaded_at, checked_at)) + + +async def put_avatar(tg_user_id: int, file_unique_id, image, cdn_url, + uploaded_at: int, checked_at: int) -> None: + await asyncio.to_thread(_put_avatar_sync, tg_user_id, file_unique_id, + image, cdn_url, uploaded_at, checked_at) + + +def _touch_avatar_sync(tg_user_id, checked_at): + with _connect() as conn: + conn.execute("UPDATE avatar_cache SET checked_at = ? WHERE tg_user_id = ?", + (checked_at, tg_user_id)) + + +async def touch_avatar(tg_user_id: int, checked_at: int) -> None: + """Record that we asked Telegram and the photo had not changed.""" + await asyncio.to_thread(_touch_avatar_sync, tg_user_id, checked_at) + + +def _count_avatars_sync(): + with _connect() as conn: + return conn.execute("SELECT COUNT(*) FROM avatar_cache").fetchone()[0] + + +async def count_avatars() -> int: + return await asyncio.to_thread(_count_avatars_sync) + + +# ── converted media cache ────────────────────────────────────────────────── + +def _get_media_sync(file_unique_id): + with _connect() as conn: + row = conn.execute( + "SELECT filename, data FROM media_cache WHERE file_unique_id = ?", + (file_unique_id,)).fetchone() + return (row[0], bytes(row[1])) if row else None + + +async def get_media(file_unique_id: str): + """-> (filename, data) for a previously converted sticker, or None.""" + return await asyncio.to_thread(_get_media_sync, file_unique_id) + + +def _put_media_sync(file_unique_id, filename, data): + with _connect() as conn: + conn.execute( + "INSERT OR REPLACE INTO media_cache " + "(file_unique_id, filename, data, created_at) VALUES (?, ?, ?, ?)", + (file_unique_id, filename, sqlite3.Binary(data), int(time.time()))) + + +async def put_media(file_unique_id: str, filename: str, data: bytes) -> None: + await asyncio.to_thread(_put_media_sync, file_unique_id, filename, data) + + +def _count_media_sync(): + with _connect() as conn: + return conn.execute("SELECT COUNT(*) FROM media_cache").fetchone()[0] + + +async def count_media() -> int: + return await asyncio.to_thread(_count_media_sync) diff --git a/tgbridge/webhook.py b/tgbridge/webhook.py new file mode 100644 index 0000000..efe1435 --- /dev/null +++ b/tgbridge/webhook.py @@ -0,0 +1,201 @@ +#discord webhook poster — shared session, retries, splitting## +"""Discord webhook client for the bridge. + +One shared session, allowed_mentions on every payload, 429 backoff, splitting +instead of truncation, and ?wait=true so we learn the message id for edits. +""" +###################################################################### +import asyncio +import json +import logging + +import aiohttp + +log = logging.getLogger(__name__) + +CONTENT_LIMIT = 2000 +MAX_RETRIES = 4 +# discord never tells us the limit (it moves with boost tier), so react to the +# actual rejection instead of guessing a threshold +_TOO_LARGE_STATUSES = {413} + +NO_MENTIONS = {"parse": []} + + +class WebhookClient: + """Shared-session Discord webhook poster.""" + + def __init__(self): + self._session: aiohttp.ClientSession | None = None + self.sent = 0 + self.failed = 0 + self.last_error: str | None = None + + async def session(self) -> aiohttp.ClientSession: + if self._session is None or self._session.closed: + timeout = aiohttp.ClientTimeout(total=120) + self._session = aiohttp.ClientSession(timeout=timeout) + return self._session + + async def close(self): + if self._session and not self._session.closed: + await self._session.close() + self._session = None + + # ── helpers ────────────────────────────────────────────────────────── + + @staticmethod + def split_content(text: str, limit: int = CONTENT_LIMIT) -> list[str]: + """Break text into discord-sized chunks on the nicest boundary.""" + if not text: + return [] + if len(text) <= limit: + return [text] + + chunks, remaining = [], text + while len(remaining) > limit: + window = remaining[:limit] + cut = window.rfind("\n") + if cut < limit // 2: + cut = window.rfind(" ") + if cut < limit // 2: + cut = limit # no sane boundary, hard cut + chunks.append(remaining[:cut].rstrip()) + remaining = remaining[cut:].lstrip() + if remaining: + chunks.append(remaining) + return [c for c in chunks if c] + + async def _request(self, method: str, url: str, *, build_body=None, + json_body=None): + """Request with retries. build_body is a callable because a FormData + cannot be replayed once consumed, so each attempt rebuilds it.""" + session = await self.session() + delay = 1.0 + for attempt in range(MAX_RETRIES): + kwargs = {} + if build_body is not None: + kwargs["data"] = build_body() + elif json_body is not None: + kwargs["json"] = json_body + try: + async with session.request(method, url, **kwargs) as resp: + if resp.status in (200, 204): + self.sent += 1 + if resp.status == 204: + return None + try: + return await resp.json() + except Exception: + return None + + body = (await resp.text())[:400] + + if resp.status == 429: + retry_after = delay + try: + retry_after = float((await resp.json()).get( + "retry_after", delay)) + except Exception: + pass + log.warning("rate limited, waiting %.2fs", retry_after) + await asyncio.sleep(min(retry_after + 0.25, 30)) + continue + + if resp.status in _TOO_LARGE_STATUSES: + raise PayloadTooLarge(body) + + if 500 <= resp.status < 600: + await asyncio.sleep(delay) + delay *= 2 + continue + + # 4xx won't improve on retry + self.failed += 1 + self.last_error = f"{resp.status} {body}" + log.error("webhook %s %s -> %s %s", method, url, resp.status, body) + return None + + except PayloadTooLarge: + raise + except (aiohttp.ClientError, asyncio.TimeoutError) as exc: + log.warning("webhook request failed (attempt %d): %s", + attempt + 1, exc) + self.last_error = str(exc) + await asyncio.sleep(delay) + delay *= 2 + + self.failed += 1 + self.last_error = self.last_error or "exhausted retries" + return None + + # ── public api ─────────────────────────────────────────────────────── + + async def send(self, webhook_url: str, username: str, content: str = "", + files: list | None = None, avatar_url: str | None = None): + """Post to a webhook, splitting long text and attaching files. + + files is a list of (filename, bytes). Returns the first message, which + is what replies and edits anchor to. + """ + chunks = self.split_content(content) or ([""] if files else []) + if not chunks: + return None + + first = None + for index, chunk in enumerate(chunks): + is_last = index == len(chunks) - 1 + attach = files if (is_last and files) else None + payload = { + "username": username[:80] or "unknown", + "allowed_mentions": NO_MENTIONS, + } + if chunk: + payload["content"] = chunk + if avatar_url: + payload["avatar_url"] = avatar_url + + result = await self._post_one(webhook_url, payload, attach) + if index == 0: + first = result + return first + + async def _post_one(self, webhook_url: str, payload: dict, files): + url = webhook_url + "?wait=true" + if not files: + return await self._request("POST", url, json_body=payload) + + def build(): + form = aiohttp.FormData() + form.add_field("payload_json", json.dumps(payload), + content_type="application/json") + for index, (filename, data) in enumerate(files): + form.add_field(f"files[{index}]", data, filename=filename) + return form + + try: + return await self._request("POST", url, build_body=build) + except PayloadTooLarge: + names = ", ".join(name for name, _ in files) + log.info("upload rejected as too large: %s", names) + notice = payload.get("content", "") + payload = dict(payload) + payload["content"] = ( + f"{notice}\n*(attachment too large for this server: " + f"{names})*").strip()[:CONTENT_LIMIT] + return await self._request("POST", url, json_body=payload) + + async def edit(self, webhook_url: str, message_id: int, content: str): + """Mirror an edit onto an already-forwarded message.""" + url = f"{webhook_url}/messages/{message_id}" + body = {"content": content[:CONTENT_LIMIT], + "allowed_mentions": NO_MENTIONS} + return await self._request("PATCH", url, json_body=body) + + async def delete(self, webhook_url: str, message_id: int): + url = f"{webhook_url}/messages/{message_id}" + return await self._request("DELETE", url) + + +class PayloadTooLarge(Exception): + """Discord rejected the upload for size."""