Spaces:
Paused
Paused
| """/start /help /cancel /export /userbot, plus the routers for free text and inline-button presses (/import is in backup.py).""" | |
| from __future__ import annotations | |
| import io | |
| import json | |
| import logging | |
| from telethon import events | |
| from ..common import Ctx, UserError, cmd, clear_state, discard_temp, guard, say, show | |
| from ..picker import PAGE_CB | |
| from ..richmsg import RICH_HELP, RICH_HELP_HTML | |
| from ..richtext import CHEAT_SHEET | |
| from ..tgutil import esc | |
| log = logging.getLogger(__name__) | |
| HELP = ( | |
| "<b>Channel Post Bot</b>\n\n" | |
| "<b>Posts</b>\n" | |
| "/new - create a post (text or media, with buttons)\n" | |
| "/posts - your posts and drafts: edit text, caption, media and buttons, or delete them - a published post is deleted in the channel and in My posts (a button post also has “Mass replace links”)\n" | |
| "/autopost 114 20 - post “Episodes 01 to 20”, “21 to 40” ... up to 114, each with a download button, then fill in the links\n" | |
| "/formatting - bold, italic, links, quotes ...: Telegram's own formatting is kept, or type HTML / Markdown instead\n" | |
| "/rich - rich messages: headings, lists, tables, expandable sections, math, maps, pictures, buttons ... typed as HTML / Markdown, or sent as a Telegram rich message\n" | |
| "/cancel - stop what you're doing\n\n" | |
| "<b>Channels</b>\n" | |
| "/addchannel - register a channel (the bot must be an admin there)\n" | |
| "/channels - list your channels, or take one out (the bot asks whether its saved posts in My posts are deleted too)\n\n" | |
| "<b>Owner tools</b>\n" | |
| "/broadcast [time] - send one message to all channels; with a time (50m, 1h, 5d) it is deleted again from all channels and My posts afterwards\n" | |
| "/replace <post link> - reply to a new media message to replace that post's media and caption (where Telegram refuses the bot, the userbot does it)\n" | |
| "/swap @old @new - swap the username inside t.me links (button links and hyperlinks) in all channels\n" | |
| "/handleswap @old @new - change the username @old written in posts (text, captions, button names) to @new in all channels; links stay\n" | |
| "/undo - undo the last /swap or /handleswap\n" | |
| "/repost @old @new - copy every post of a channel in order (links swapped, replies, pins and links between posts kept), then delete the old posts\n" | |
| "/shift @source @destination [from_id] [to_id] [-c] [-all] [--no-media] - copy posts from one channel into another (buttons kept, listed in My posts); " | |
| "a channel Telegram restricts (copyright strike) is copied from My posts; " | |
| "-c: links to the source's posts in all your channels are changed to the copies; -all: the destination also gets the source's name, description and photo; " | |
| "--no-media: no files, only the text and buttons\n" | |
| "/sync - compare every channel with My posts right now (posts deleted, edited or added by others, the older posts from before the bot was added; service messages are deleted)\n" | |
| "/userbot - status of the helper account that, by itself and without asking, deletes and edits the posts the bot is not allowed to\n" | |
| "/export - download a JSON backup of your channels and posts\n" | |
| "/import - put a backup made by /export back (only adds: nothing that exists is changed or deleted)" | |
| "\n/admins - list admins (owner)\n/addadmin 123456789 - add an admin (owner)\n/removeadmin 123456789 - remove an admin (owner)" | |
| ) | |
| def register(ctx: Ctx) -> None: | |
| client, db = ctx.client, ctx.db | |
| async def h_admins(event): | |
| ids = sorted(ctx.admins | set(ctx.cfg.admins)) | |
| await say(event, "Admins: " + (", ".join(map(str, ids)) if ids else "none")) | |
| async def h_addadmin(event): | |
| parts = (event.raw_text or "").split() | |
| if len(parts) != 2 or not parts[1].isdigit() or int(parts[1]) <= 0: | |
| raise UserError("Use /addadmin followed by a numeric Telegram user ID.") | |
| uid = int(parts[1]) | |
| if ctx.cfg.is_owner(uid): | |
| raise UserError("That user is already an owner.") | |
| if not await db.add_admin(uid): | |
| await say(event, f"{uid} is already an admin.") | |
| return | |
| ctx.admins.add(uid) | |
| await say(event, f"Added {uid} as an admin.") | |
| async def h_removeadmin(event): | |
| parts = (event.raw_text or "").split() | |
| if len(parts) != 2 or not parts[1].isdigit() or int(parts[1]) <= 0: | |
| raise UserError("Use /removeadmin followed by a numeric Telegram user ID.") | |
| uid = int(parts[1]) | |
| if not await db.remove_admin(uid): | |
| await say(event, f"{uid} is not an admin.") | |
| return | |
| ctx.admins.discard(uid) | |
| await say(event, f"Removed {uid} as an admin.") | |
| async def h_help(event): | |
| await say(event, HELP) | |
| async def h_formatting(event): | |
| await say(event, CHEAT_SHEET) | |
| async def h_rich(event): | |
| await say(event, RICH_HELP) | |
| await say(event, RICH_HELP_HTML) | |
| async def h_cancel(event): | |
| await discard_temp(ctx, event.sender_id) | |
| clear_state(ctx, event.sender_id) | |
| await say(event, "Cancelled.") | |
| async def h_export(event): | |
| data = await db.export_all() | |
| buf = io.BytesIO(json.dumps(data, ensure_ascii=False, indent=1).encode("utf-8")) | |
| buf.name = "channel-posts-export.json" | |
| await event.respond( | |
| f"📦 Backup: {len(data['channels'])} channel(s), {len(data['posts'])} post(s). " | |
| "To put it back (on this bot or a new one), send /import and then this file.", | |
| file=buf, | |
| ) | |
| async def h_userbot(event): | |
| ub = ctx.userbot | |
| if ub is None or not ub.enabled: | |
| await say( | |
| event, | |
| "🤖 <b>Userbot: not set up.</b>\n" | |
| "The bot does its work itself first. If Telegram refuses - deleting posts older than 48 hours, or " | |
| "editing a post that belongs to another bot (/replace, text and caption edits) - a helper account, " | |
| "the userbot, takes over by itself, without asking you each time: it joins the channel through an " | |
| "invite link the bot makes, the bot gives it the “Delete messages” or “Edit messages of others” right " | |
| "if it has the “Add new admins” right, and gives the rights back afterwards.\n\n" | |
| "To set it up: on your own computer run <code>python make_userbot_session.py</code> with a " | |
| "<b>separate</b> Telegram account, and put the printed text into the bot's secrets as " | |
| "<code>USERBOT_SESSION</code>. Then send /start to this bot from that account (new media reaches the " | |
| "userbot through its chat with the bot - that is what makes a replace quick). See the README.", | |
| ) | |
| return | |
| if not ub.ready: | |
| await ub.connect() | |
| kept = "stays an admin" if ub.keep_admin else "is made admin only while it deletes or edits" | |
| await say( | |
| event, | |
| f"🤖 <b>Userbot:</b> {esc(ub.describe())}\nIt {kept} (USERBOT_KEEP_ADMIN).\n" | |
| "It must have started this bot (/start from that account): new media for an edit goes through its chat " | |
| "with the bot, which is what makes a big file as quick as a small one.", | |
| ) | |
| async def h_free_text(event): | |
| st = ctx.state.get(event.sender_id) | |
| if not st: | |
| await say(event, "Use /new to create a post or /posts to manage your posts. /help lists everything.") | |
| return | |
| fn = ctx.text_handlers.get(st["mode"]) | |
| if fn is None: | |
| clear_state(ctx, event.sender_id) | |
| await say(event, "That step is no longer active. Start again with /new.") | |
| return | |
| await fn(event, st) | |
| async def h_callback(event): | |
| try: | |
| await event.answer() # stop the loading spinner right away | |
| except Exception: | |
| pass | |
| data = event.data.decode("utf-8", "replace") | |
| prefix, *parts = data.split(":") | |
| fn = ctx.callbacks.get(prefix) | |
| if fn is None: | |
| await event.respond("This button is no longer active. Use /new or /posts.") | |
| return | |
| await fn(event, parts) | |
| async def cb_noop(event, parts): | |
| await event.answer() | |
| async def cb_cancel(event, parts): | |
| uid = event.sender_id | |
| await discard_temp(ctx, uid) | |
| clear_state(ctx, uid) | |
| await show(event, "Cancelled.") | |
| async def cb_channel_page(event, parts): | |
| """◀ ▶ under a channel chooser: draw the same chooser again on another page.""" | |
| kind, arg, page = parts[0], parts[1], int(parts[2]) | |
| draw = ctx.pickers.get(kind) | |
| if draw is None: | |
| raise UserError("This list is no longer active. Start the command again.") | |
| await draw(event, arg, page) | |
| ctx.callbacks["noop"] = cb_noop | |
| ctx.callbacks["cx"] = cb_cancel | |
| ctx.callbacks[PAGE_CB] = cb_channel_page | |