Blame for app/handlers/basic.py

"""/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

    @client.on(cmd("admins"))
    @guard(ctx, owner=True)
    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"))

    @client.on(cmd("addadmin"))
    @guard(ctx, owner=True)
    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.")

    @client.on(cmd("removeadmin"))
    @guard(ctx, owner=True)
    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.")

    @client.on(cmd("start"))
    @client.on(cmd("help"))
    @guard(ctx)
    async def h_help(event):
        await say(event, HELP)

    @client.on(cmd("formatting"))
    @guard(ctx)
    async def h_formatting(event):
        await say(event, CHEAT_SHEET)

    @client.on(cmd("rich"))
    @guard(ctx)
    async def h_rich(event):
        await say(event, RICH_HELP)
        await say(event, RICH_HELP_HTML)

    @client.on(cmd("cancel"))
    @guard(ctx)
    async def h_cancel(event):
        await discard_temp(ctx, event.sender_id)
        clear_state(ctx, event.sender_id)
        await say(event, "Cancelled.")

    @client.on(cmd("export"))
    @guard(ctx, owner=True)
    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,
        )

    @client.on(cmd("userbot"))
    @guard(ctx, owner=True)
    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.",
        )

    @client.on(
        events.NewMessage(incoming=True, func=lambda e: e.is_private and not (e.raw_text or "").startswith("/"))
    )
    @guard(ctx)
    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)

    @client.on(events.CallbackQuery())
    @guard(ctx)
    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